From 27072e492d8b5c931fa2dbe30da11b4d0ad257ac Mon Sep 17 00:00:00 2001 From: fox0430 Date: Sun, 5 Jul 2026 17:28:29 +0900 Subject: [PATCH] Deduplicate execInTransactionImpl by delegating to queryInTransactionImpl --- .../pg_client/transaction_helpers.nim | 152 +++++------------- 1 file changed, 39 insertions(+), 113 deletions(-) diff --git a/async_postgres/pg_client/transaction_helpers.nim b/async_postgres/pg_client/transaction_helpers.nim index 829a32d..4b630a2 100644 --- a/async_postgres/pg_client/transaction_helpers.nim +++ b/async_postgres/pg_client/transaction_helpers.nim @@ -7,14 +7,15 @@ import std/[options] import ../[async_backend, pg_protocol, pg_connection, pg_types] import ./core -proc execInTransactionImpl( +proc queryInTransactionImpl( conn: PgConnection, beginSql: string, sql: string, params: seq[Option[seq[byte]]], paramOids: seq[int32], paramFormats: seq[int16], -): Future[string] {.async.} = + resultFormats: seq[int16], +): Future[QueryResult] {.async.} = conn.checkReady() conn.state = csBusy @@ -24,7 +25,7 @@ proc execInTransactionImpl( else: newSeq[int16](params.len) - # Pipeline: Parse+Bind+Execute for BEGIN, user SQL, COMMIT + single Sync + # Pipeline: Parse+Bind+Execute for BEGIN, user SQL (with Describe), COMMIT + Sync conn.sendBuf.setLen(0) # BEGIN conn.sendBuf.addParse("", beginSql) @@ -32,7 +33,8 @@ proc execInTransactionImpl( conn.sendBuf.addExecute("", 0) # User SQL conn.sendBuf.addParse("", sql, paramOids) - conn.sendBuf.addBind("", "", formats, params) + conn.sendBuf.addBind("", "", formats, params, resultFormats) + conn.sendBuf.addDescribe(dkPortal, "") conn.sendBuf.addExecute("", 0) # COMMIT conn.sendBuf.addParse("", "COMMIT") @@ -42,19 +44,31 @@ proc execInTransactionImpl( conn.sendBuf.addSync() await conn.sendBufMsg() - # Parse response: 3 phases (BEGIN=0, user=1, COMMIT=2) + var qr = QueryResult() var phase = 0 - var userCommandTag = "" var queryError: ref PgQueryError block recvLoop: while true: - while (let opt = conn.nextMessage(); opt.isSome): + while (let opt = conn.nextMessage(qr.data, addr qr.rowCount); opt.isSome): let msg = opt.get case msg.kind of bmkParseComplete, bmkBindComplete: discard - of bmkRowDescription, bmkDataRow: + of bmkRowDescription: + var fields = msg.fields + var cf: seq[int16] + var co: seq[int32] + if resultFormats.len > 0: + cf = deriveColFmts(resultFormats, fields.len) + co = newSeq[int32](fields.len) + for i in 0 ..< fields.len: + co[i] = fields[i].typeOid + fields[i].formatCode = cf[i] + qr.fields = fields + qr.data = newRowData(int16(qr.fields.len), cf, co) + qr.data.fields = qr.fields + of bmkNoData: discard of bmkEmptyQueryResponse: # Empty/comment-only user SQL yields EmptyQueryResponse instead of @@ -63,7 +77,7 @@ proc execInTransactionImpl( inc phase of bmkCommandComplete: if phase == 1: - userCommandTag = msg.commandTag + qr.commandTag = msg.commandTag inc phase of bmkErrorResponse: if queryError == nil: @@ -73,10 +87,8 @@ proc execInTransactionImpl( if conn.state != csClosed: conn.state = csReady if queryError != nil: - # Error occurred: if we're in a failed transaction, ROLLBACK to - # release it. Swallow any failure so a ROLLBACK error cannot mask - # the original query error (which is what the caller needs to see); - # the public proc's outer wait(timeout) bounds the cleanup. + # ROLLBACK a failed transaction, swallowing any failure so it cannot + # mask the query error; the outer wait(timeout) bounds the cleanup. if msg.txStatus == tsInFailedTransaction: try: discard await conn.simpleExec("ROLLBACK") @@ -92,7 +104,20 @@ proc execInTransactionImpl( discard await conn.fillRecvBuf() - return userCommandTag + return qr + +proc execInTransactionImpl( + conn: PgConnection, + beginSql: string, + sql: string, + params: seq[Option[seq[byte]]], + paramOids: seq[int32], + paramFormats: seq[int16], +): Future[string] {.async.} = + let qr = await queryInTransactionImpl( + conn, beginSql, sql, params, paramOids, paramFormats, @[] + ) + return qr.commandTag proc execInTransaction*( conn: PgConnection, @@ -152,105 +177,6 @@ proc execInTransaction*( tag = await execInTransactionImpl(conn, beginSql, sql, values, oids, formats) return initCommandResult(tag) -proc queryInTransactionImpl( - conn: PgConnection, - beginSql: string, - sql: string, - params: seq[Option[seq[byte]]], - paramOids: seq[int32], - paramFormats: seq[int16], - resultFormats: seq[int16], -): Future[QueryResult] {.async.} = - conn.checkReady() - conn.state = csBusy - - let formats = - if paramFormats.len > 0: - paramFormats - else: - newSeq[int16](params.len) - - # Pipeline: Parse+Bind+Execute for BEGIN, user SQL (with Describe), COMMIT + Sync - conn.sendBuf.setLen(0) - # BEGIN - conn.sendBuf.addParse("", beginSql) - conn.sendBuf.addBind("", "", @[], @[]) - conn.sendBuf.addExecute("", 0) - # User SQL - conn.sendBuf.addParse("", sql, paramOids) - conn.sendBuf.addBind("", "", formats, params, resultFormats) - conn.sendBuf.addDescribe(dkPortal, "") - conn.sendBuf.addExecute("", 0) - # COMMIT - conn.sendBuf.addParse("", "COMMIT") - conn.sendBuf.addBind("", "", @[], @[]) - conn.sendBuf.addExecute("", 0) - # Single Sync - conn.sendBuf.addSync() - await conn.sendBufMsg() - - var qr = QueryResult() - var phase = 0 - var queryError: ref PgQueryError - - block recvLoop: - while true: - while (let opt = conn.nextMessage(qr.data, addr qr.rowCount); opt.isSome): - let msg = opt.get - case msg.kind - of bmkParseComplete, bmkBindComplete: - discard - of bmkRowDescription: - var fields = msg.fields - var cf: seq[int16] - var co: seq[int32] - if resultFormats.len > 0: - cf = deriveColFmts(resultFormats, fields.len) - co = newSeq[int32](fields.len) - for i in 0 ..< fields.len: - co[i] = fields[i].typeOid - fields[i].formatCode = cf[i] - qr.fields = fields - qr.data = newRowData(int16(qr.fields.len), cf, co) - qr.data.fields = qr.fields - of bmkNoData: - discard - of bmkEmptyQueryResponse: - # Empty/comment-only user SQL yields EmptyQueryResponse instead of - # CommandComplete; advance the phase anyway so the trailing COMMIT's - # CommandComplete isn't captured as the user statement's tag. - inc phase - of bmkCommandComplete: - if phase == 1: - qr.commandTag = msg.commandTag - inc phase - of bmkErrorResponse: - if queryError == nil: - queryError = newPgQueryError(msg.errorFields) - of bmkReadyForQuery: - conn.txStatus = msg.txStatus - if conn.state != csClosed: - conn.state = csReady - if queryError != nil: - # ROLLBACK a failed transaction, swallowing any failure so it cannot - # mask the query error; the outer wait(timeout) bounds the cleanup. - if msg.txStatus == tsInFailedTransaction: - try: - discard await conn.simpleExec("ROLLBACK") - except CancelledError as e: - # Don't swallow cancellation (e.g. the outer wait(timeout) - # cancelling this future under chronos) — propagate it. - raise e - except CatchableError: - discard - raise queryError - break recvLoop - else: - discard - await conn.fillRecvBuf() - - return qr - proc queryInTransaction*( conn: PgConnection, sql: string,