Repository navigation
Send UpdateWorkflowExecution to Postgres in one round trip - #12867
Open
ali-khokhar-nvidia wants to merge 45 commits into
Open
ali-khokhar-nvidia wants to merge 45 commits into
ali-khokhar-nvidia wants to merge 45 commits into
Conversation
A pipelined transaction queues its statements, sends BEGIN and all of them in one pgx batch on Flush, then answers the same calls from the saved results. Only pgx in its default cached-statement mode pipelines; other connections report ErrPipelineUnsupported.
UpdateCurrent updates without a new run record their statements against the expected lock results, flush them as one batch, then replay the update against the flushed results so the existing checks build the existing errors before COMMIT or ROLLBACK. Record errors, replay divergence and connections that cannot pipeline fall back to the sequential transaction.
applyTasks followed map iteration order, so a pipelined replay could queue the task inserts in a different order than its record pass and fall back to the sequential transaction.
The record pass ran against a transaction without UpdateExecutionsWithCondition, so it queued the lock and the plain update while the replay sent the conditional update. It also always queued the current execution update, which the replay skips when the locked row is unchanged. Either way every pipelined update diverged and ran again sequentially. A pipelined transaction now implements UpdateExecutionsWithCondition and UpdateCurrentExecutionsIfChanged, which leaves an unchanged current row alone in SQL, so the record pass and the replay queue the same statements.
The pipelined update took three round trips: the autocommit history_node insert, the batch, and a separate COMMIT after the client checked the batch's results. Locks were held across that last round trip. The batch now carries the history nodes and no transaction control, so Postgres runs it as one implicit transaction and commits it at the batch's Sync. The checks the client made on the results are now statements that fail the transaction: the shard range ID, the current run ID and the executions update's row count each divide by a matching row count. When a statement fails nothing was written, so the request reruns the sequential path, which returns today's errors. Any other error leaves the outcome unknown and returns Unavailable. A request that starts a new history branch still runs sequentially. Recording and replaying the transaction is removed, along with the task category ordering and the finishTx extraction that only replay needed.
interfaces_mock.go was not regenerated when PipelinedTx and TxPipeliner were added, so go generate left it changed.
pgx sends a batch with one Sync in every exec mode, and in simple_protocol it sends one multi-statement query, which Postgres also runs as one implicit transaction. The pipeline suite now runs for every pgx test variant.
Only the pgx driver can send the batch, so BeginPipelinedTx returns ErrPipelineUnsupported for lib/pq before the store records a transaction and checks out a connection it cannot use. Commit no longer reports Unsupported.
The caller depends on nothing having been written, not on why, so the sentinel names that guarantee. The wrapped error still carries the cause.
A Commit whose context ends while it waits for a pooled connection sent nothing, yet it reported an unknown outcome, which makes the shard reacquire. It now returns ErrPipelineNotWritten, so the sequential path runs and returns the AppendHistoryTimeoutError it returned before pipelining.
Most of them are condition failures, such as a shard that moved, which the sequential path returns as the typed errors persistence already keeps out of the logs.
A pipelined update that the database rejects or the store cannot record falls back to the sequential path with the same end state, so the suites passed either way. On pgx, TestUpdate_NotZombie now also requires its update, which carries every kind of row, to begin no sequential transaction.
TestUpdateCurrentExecutionsIfChanged passed without the IS DISTINCT FROM condition, since rewriting a row with its own values leaves the same row. A trigger that fails any update of the row now shows the pipeline skips it, as the sequential path does.
It passes whether the rejected batch's connection went back to the pool or was closed, so it only shows that no connection was pooled mid-batch.
Postgres reports the implicit commit's error as the last statement's result in a multi-statement query, so a trailing SELECT 1 keeps it apart from the statements' own errors.
ali-khokhar-nvidia
marked this pull request as ready for review
November 4, 2026 22:37
… pipeline The pipelined transaction's UpdateCurrentExecutions leaves an unchanged row alone, and expectingTx reports its one row, so the separate conditional updater interface and its branch in assertRunIDAndUpdateCurrentExecution go.
One path for every mode, without copying the pgx config on each Commit to read the mode.
An error that pgconn.SafeToRetry reports happened before anything reached Postgres, so the update can run sequentially instead of returning an unknown outcome.
A custom SQL.Connect can return connections that are not pgx's, which Commit cannot batch on.
…ents are prepared
…table Adds the mode and new workflow gates and a BeginPipelinedTx error, which no test covered.
A 100ms timeout could fire before the batch was sent, which Commit reports as not written.
Checks the error type with ErrorAs and discards the PgError that requireRejected returns where the test does not use it.
IgnoreCurrent updates never read the current row, so they now run pipelined. BypassCurrent stays sequential because expectingTx asserts that the run is current. The gate and the recorder cover NewWorkflowNewEvents, as the sequential path does. Commit runs ConvertError once on the raw error and wraps ErrPipelineNotWritten outside it, so a rejection that calls for a new connection pool refreshes the pool. BeginPipelinedTx decides support from the pool's driver, and its error goes through convertSQLError. Tests cover the SafeToRetry branch, CHASM, closed-workflow and buffered-event updates through the pipeline, and the whole updated current row for both archetypes.
|
|
The recorder no longer runs updateWorkflowExecutionTx against a made-up current row. For UpdateCurrent it queues the run ID lock and an update that writes the current row only when it differs, then applies the mutation. Both paths build the current row with buildCurrentExecutionRow. The pipeline interfaces moved to their own file, which mockgen does not read. New tests update a run that is no longer current and check the current row after an update.
…ially Each check statement carries the error it stands for, so a failed check returns the error type the sequential path returns for it. The connection is taken at Begin, so a non-pgx connection picks the sequential path before anything is sent. The current execution update and its change check share one column list.
The shard and current execution checks branch on a pipelined transaction, so the recorder runs updateWorkflowExecutionTx instead of a copy of it, and buildCurrentExecutionRow goes back inline. Simple protocol takes the sequential path, because Postgres logs the whole batch with its values when a check fails there, and the trailing SELECT 1 that only simple protocol needed goes. Each check query names its check in a SQL comment, which the server log shows. A recording error is converted as txExecute converts it, and a context that ends during recording returns AppendHistoryTimeoutError without sending the batch.
A DB in simple protocol took a pooled connection on every update only to return it and take another for the sequential path. Every connection of a DB shares the settings that decide it, so the first rejection answers later calls. send no longer reports whether anything was written. After a failed batch pgx either drains the connection or closes it, and database/sql drops a closed one at the next checkout, so returning ErrBadConn only discarded healthy connections.
Only Commit and Rollback return the connection BeginPipelinedTx takes, and database/sql does not release a Conn when its context ends, so a panic while recording kept it checked out until the process exited. The update now rolls back and panics again. TestPGXExecModes runs the pipeline and mutable state suites in the other pgx exec modes that pipeline, since pgx sends a batch through a different function in each mode.
A deferred function runs while a panic unwinds, so the rollback needs no recover and the panic needs no re-raise past the linter. The defer also replaces the two explicit rollbacks, and is skipped once Commit has taken the connection back.
BeginPipelinedTx copied the pgx config on every update to read the exec mode, and lib/pq ran a driver check on every update. Every connection of a DB shares those settings, so the first connection now decides, and the driver check is gone.
A context that ended before pgx sent the batch reacquired the shard. pgx marks an error it returns before sending anything as safe to retry, so Commit wraps it with ErrPipelineNotSent, and the store returns the error Begin and the sequential path return for a context that ended. The check before Commit goes, as Commit covers that window too.
A custom SQL.Connect can return another driver's connections after the handle rebuilds the pool, and Commit asserted a pgx connection.
A malformed branch returned Unavailable from the pipelined path after it took a connection, while the sequential path returns the parse error. Building the rows at the store's entry returns that error before any transaction, and neither path parses them again.
Nothing reads the statements after Rollback, and describe_exec prepares every batch, so the interface no longer counts round trips. The simple protocol test takes the test's context.
A socket that broke after checkout fails the batch write before anything is sent. Main's history insert retries that on another connection, so the store now runs the sequential path instead of returning Unavailable and rebuilding the pool. The recorder queues a copy of each history row, as the Postgres insert negates TxnID in place.
database/sql checks the context before taking a connection, so the sequential path returns its own error for it without connecting. The check is on the context rather than the error: a pgx connect timeout wraps DeadlineExceeded while the caller's context is alive, and stays a Begin error.
In describe_exec mode, and on a statement's first use in the cache modes, pgx sends Parse and Describe in an earlier round trip, so what holds after a write that sent nothing is that no statement ran.
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changed?
Commitas one batch, which Postgres runs and commits as one implicit transaction.UpdateWorkflowExecutiononto it forUpdateCurrentandIgnoreCurrentupdates that create no new run and no new history branch. The history node inserts go in the same batch.ShardOwnershipLostErrorfor a stale range ID. The one difference is a missing shard row, which fails the range ID check and returnsShardOwnershipLostErrorinstead ofUnavailable, so the shard stops instead of reacquiring. The check errors carry only a fixed message, without the IDs and row details the sequential path adds. Any other batch error is converted as a failedCOMMITis today.lib/pqand the other SQL plugins. In simple protocol the batch is one query, which Postgres would log whole, values included, when a check fails. The path is picked before anything is sent. A batch that ran none of its statements falls back to the sequential path, and any other failed batch is not retried.Why?
Every workflow task and activity completion writes through
UpdateWorkflowExecution. Today each history node goes out as its own autocommit insert, and thenBEGIN, every statement andCOMMITeach take a round trip, while the shard and current execution row locks and a pooled connection stay held. The update pays for all of those round trips and waits for two WAL flushes. Pipelined, the update is one round trip once the connection has prepared its statements, and it waits for one WAL flush.How did you test it?
The persistence suites pass on Postgres 16.4 for pgx, pgx in simple protocol and
lib/pq. Under pgx in extended protocol the execution mutable state suite now runs through the pipeline, and a new plugin suite covers commits, each failed check, duplicate inserts, a canceled context and an error at commit. Both suites also run in the other three pgx exec modes that pipeline. The MySQL, SQLite and Cassandra persistence suites also pass.We also ran it under load on fixed hardware: three history pods at 2 cores each, a 4-core Postgres 16.4 primary, and three PgBouncer pods in transaction mode. About 6,000 workflows loop over short activities, and each activity costs about three
UpdateWorkflowExecutioncalls. Each column is a 12-minute steady-state window.UpdateWorkflowExecution/sUpdateWorkflowExecutionmean latencyPotential risks
A cancellation during the batch leaves both the history and the update unknown, so the shard reacquires. Today a cancellation during the history append returns
AppendHistoryTimeoutError, which the shard retries without reacquiring. A deadline that expires mid-statement reacquires on both paths.Postgres logs each failed check as an error with the check's statement, and by default without its values. Today these checks run in the client, and Postgres logs nothing for them.
A failed batch makes pgx drop its prepared statements for every query in the batch, so that connection's next update takes two extra round trips, one to deallocate them and one to prepare them again.