Native: start the next transition while the last one commits

Delivered in Native3.3.0

Implemented by #1084; qualified and published at source 2a15179f29801eec9f2a2d4e6a9c5f93233f104e / index sha256:cb4b11bd6235a551cf51e7dc6342504fbbcef19959618b376fb5648ca00fc388. Acceptance and scope limits. General reconciled this delivered item while the Server agent is Off. No benchmark speedup is claimed.

Original requirement and evidence

External performance report, topic 8. The reported timings and proposed fixes have not been independently reproduced. The supplied patches are reference drafts for review and current-source qualification.

Problem

The next queued transition waits for the current one's commit fsync to
finish before it even begins. PostgreSQL's advisory lock and MySQL's row
lock already pass to a waiter as the holder's commit releases them.

Each transition also makes round trips it doesn't need:

PostgreSQL's BEGIN and its lock are two round trips;
the completion reads the run's authored sequence twice;
every history event's sequence is a read plus a write.

Proposal

On PostgreSQL and MySQL, release the in-process queue (report topic 7) when
the commit starts. The next transition then begins and waits on the
database lock while this one commits. SQLite keeps the queue through the
commit, because its busy handler polls.
Send PostgreSQL's BEGIN and its advisory lock in one simple query.
Read the authored sequence once per completion.
Allocate a history sequence in one statement: UPDATE ... RETURNING on
SQLite and PostgreSQL, LAST_INSERT_ID(expr) on MySQL. A missing run,
or a sequence at its maximum, takes the old path and refuses as before.

Reported benchmark evidence

Reporter-supplied cumulative measurements, not an isolated-effect or independently verified performance claim: 200 sequentially started workflows, five sequential 20ms activities or five local activities each, six PHP worker processes, mean of three repetitions, one machine, fresh SQLite/MySQL8.0/PostgreSQL16 databases with default durability. "Five" and "local" below mean last-close wall time in seconds; S/M/P means SQLite/MySQL/PostgreSQL. Preserve regressions and noise in the reported results.

	five	local
slowest backend	22.60 → 21.48	5.47 → 5.71

Acceptance

The next transition never reads uncommitted state. It still waits on
the database lock.
The exhaustion refusal is unchanged. The existing test covers it.

Dependencies

Native: queue transitions in process before they take a pool connection #1058

Related existing work: #325.

Draft patches and validation boundary

Current main was independently read as b8f4dd13623ce645de77ba5bd89ae8dd8eea2f7c. The supplied draft series targets 3d4d4afaf2b24aaa5443eac154e253b011924738; patches are cumulative reference material, not a current-main application or test result. Review and rebase on maintained source, preserve all existing tests and published compatibility, then establish causal controls and actual conformance before shipping. No patch was applied or executed during intake.

Known draft gap / required control: Overlapping commit must preserve database-lock ordering, committed visibility, durability and exhaustion refusal. Do not weaken fsync, transaction isolation or cross-process fencing to reproduce a timing result.

The following sanitized unified diffs identify proposed reference material:

0007-t08-Start-the-next-transition-while-the-last-one-commits.patch — 3,508 bytes, SHA-256 dd6462fce4c781cd96fa2f7dbacae7cf62ecbd82d7dec1b55e82f25f01135dd6.
0008-t08-Read-the-authored-sequence-once-per-completion.patch — 3,238 bytes, SHA-256 cba2c41331743c5f612994c29a17924a2a2cb4fd2a5aaee9f38f99cf6a40b8d5.
0009-t08-Allocate-a-history-sequence-in-one-statement.patch — 5,791 bytes, SHA-256 bae07e98419dc06defe99fdff6d51c0df98b484198cafc5401bcb8cb86d435bb.

Current implementation: PR #1084, exact 6a8aacd89ad41391ad113ca3e4ea54ecbbc82d7c, is based on the accepted #1058 main 6bdb2c90. PostgreSQL/MySQL release the process queue after derived writes at COMMIT entry, while SQLite retains it. PostgreSQL combines BEGIN and its advisory lock in one simple query. Scope validation reuses its locked authored sequence. Ordinary history counters use one guarded update; each backend's retained column width and the original missing/exhausted/noncanonical path remain explicit.

New consumed backend controls cover committed visibility, cancellation before COMMIT, admission behind the actual database lock, fast versus original counter outcomes at boundaries, rollback, missing runs, malformed SQLite storage, and an actual completion's statement counts/contiguous history. git diff --check passes. Source qualification now passes all six original current-head workflows,62 successful checks/one declared skip. The new controls explicitly pass on SQLite, PostgreSQL16/17, MySQL8.0/8.4 and MariaDB10.11, with provider/download digest and tested-tree verification. Source-owner review has no remaining finding. The initial fixture compilation defect and superseded runs remain attributed in the PR; no performance or installed result is claimed.

Merged source: the original queue passed and merged #1084 as 4fb4cffc463d1d96bfce86231c12819bc2ad6b2a at 04:25:38Z. All48 original Shared queue jobs and the queue Feature suite passed. All five original automatic actual-main workflows passed at this exact merge. The main acceptance receipt preserves the original queue and source results. The exact merged remote head independently returns404; the clean owned worktree/topic is removed after ancestry verification.

Next owner/action: Server drives #1065 in PR #1085, then #1069, and prepares the exact-source native publication/qualification packet. The preceding source/main runs are terminal before this next CI train. Release/published acceptance remains open. No imported reference patch was applied.

Sanitized reference diff 0007-t08-Start-the-next-transition-while-the-last-one-commits.patch, part 1/1.

This is an unverified, cumulative draft from the external report. Mail author/date/commit-message metadata was removed; the unified-diff hunks were retained. No patch was applied or executed, and no build/test or compatibility result is implied. The known gaps and acceptance criteria in this issue remain mandatory.

Complete sanitized patch: 3,508 bytes, SHA-256 dd6462fce4c781cd96fa2f7dbacae7cf62ecbd82d7dec1b55e82f25f01135dd6. This part: 3,508 bytes, SHA-256 dd6462fce4c781cd96fa2f7dbacae7cf62ecbd82d7dec1b55e82f25f01135dd6. Parts concatenate in order; a part boundary is between complete file diffs.

diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
index 03d88f58..f6a210a2 100644
--- a/rust/src/runtime/backend.rs
+++ b/rust/src/runtime/backend.rs
@@ -38,6 +38,10 @@ pub(super) trait Backend: Database {
     // Multi-statement reads that must observe one committed state. SQLite's
     // deferred read transaction already keeps one WAL snapshot.
     const READ_SNAPSHOT_BEGIN: &str = "BEGIN";
+    // Whether a transaction waiting on the transition lock is granted it as
+    // the holder's commit releases it, as an advisory or row lock is.
+    // SQLite's busy handler retries a held lock on a sleep schedule instead.
+    const LOCK_GRANTS_WAITERS: bool = true;
     const ACTIVITY_METADATA_HISTORY: &str = "SELECT sequence,event_type,CASE WHEN projection_bytes<=8388608 THEN payload ELSE NULL END AS payload,workflow_task_id,recorded_at,projection_count FROM (SELECT sequence,event_type,payload,workflow_task_id,recorded_at,COUNT(*) OVER () AS projection_count,SUM(octet_length(CAST(payload AS TEXT))) OVER (ORDER BY sequence) AS projection_bytes FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled') ORDER BY sequence LIMIT 10001) AS page ORDER BY sequence";
     // Waits whose stored position differs from their chronological ordinal
     // over every retained wait of the run, as Workflow #941's projector stores.
@@ -198,6 +202,7 @@ fn sqlite_instant(value: String) -> Result<DateTime<Utc>> {
 }
 
 impl Backend for Sqlite {
+    const LOCK_GRANTS_WAITERS: bool = false;
     type EncodedInteger = Option<i64>;
     type EncodedFloat = Option<f64>;
     type EncodedFlag = Option<bool>;
@@ -486,14 +491,12 @@ impl Backend for Postgres {
             .map(|value| value.0))
     }
     async fn begin(pool: &Pool<Self>) -> Result<Transaction<'static, Self>> {
-        let mut tx = pool.begin().await?;
         // The first single-namespace slice serializes its durable transitions
         // across independent pools/processes. Held long polls wait without it
         // (Store::task_poll_ready). Throughput/idle cost remain unqualified.
-        sqlx::query("SELECT pg_advisory_xact_lock(4924501473315676163)")
-            .execute(&mut *tx)
-            .await?;
-        Ok(tx)
+        // One simple query begins the transaction and takes the lock in one
+        // round trip; a lock that fails rolls the transaction back.
+        Ok(pool.begin_with("BEGIN; SELECT pg_advisory_xact_lock(4924501473315676163)").await?)
     }
     async fn history_ready(pool: &Pool<Self>) -> bool {
         let Ok(mut connection) = pool.acquire().await else {
diff --git a/rust/src/runtime/transaction.rs b/rust/src/runtime/transaction.rs
index fd5a4c2d..fd251b83 100644
--- a/rust/src/runtime/transaction.rs
+++ b/rust/src/runtime/transaction.rs
@@ -107,6 +107,12 @@ where
             Store::<DB>::order_waits(&mut self, &run_id).await?;
             Store::<DB>::project_run_summary(&mut self, &run_id).await?;
         }
+        // On a database that grants its lock to the next waiter, the next
+        // transition in this server's queue starts and waits on the lock
+        // while this one commits, so the lock passes to it at once.
+        if DB::LOCK_GRANTS_WAITERS {
+            self.queued.take();
+        }
         self.inner.commit().await?;
         Ok(())
     }

Sanitized reference diff 0008-t08-Read-the-authored-sequence-once-per-completion.patch, part 1/1.

This is an unverified, cumulative draft from the external report. Mail author/date/commit-message metadata was removed; the unified-diff hunks were retained. No patch was applied or executed, and no build/test or compatibility result is implied. The known gaps and acceptance criteria in this issue remain mandatory.

Complete sanitized patch: 3,238 bytes, SHA-256 cba2c41331743c5f612994c29a17924a2a2cb4fd2a5aaee9f38f99cf6a40b8d5. This part: 3,238 bytes, SHA-256 cba2c41331743c5f612994c29a17924a2a2cb4fd2a5aaee9f38f99cf6a40b8d5. Parts concatenate in order; a part boundary is between complete file diffs.

diff --git a/rust/src/runtime/cancellation_scopes.rs b/rust/src/runtime/cancellation_scopes.rs
index eb6c37e7..94e4009e 100644
--- a/rust/src/runtime/cancellation_scopes.rs
+++ b/rust/src/runtime/cancellation_scopes.rs
@@ -2129,13 +2129,14 @@ store_methods! {
     }
 
     /// `validateCancellationScopeMembership` inside the bridge's completion,
-    /// over the batch's non-terminal commands.
-    pub(super) async fn cancellation_scope_completion_refusal(tx: &mut Transaction<'_, DB>, run_id: &str, commands: &[&Value])
+    /// over the batch's non-terminal commands. `authored` is the run's
+    /// `Store::authored_sequence` in `tx`.
+    pub(super) async fn cancellation_scope_completion_refusal(tx: &mut Transaction<'_, DB>, run_id: &str, commands: &[&Value], authored: i64)
         -> Result<Option<&'static str>>
     {
         let events = Self::scope_events(tx, run_id).await?;
         let scopes = scopes(&events, run_id)?;
-        let next = Self::authored_sequence(tx, run_id).await? + 1;
+        let next = authored + 1;
         membership_refusal(&events, &scopes, commands, next)
     }
 }
diff --git a/rust/src/runtime/scope_checkpoint.rs b/rust/src/runtime/scope_checkpoint.rs
index d24d1eca..2d846445 100644
--- a/rust/src/runtime/scope_checkpoint.rs
+++ b/rust/src/runtime/scope_checkpoint.rs
@@ -254,7 +254,7 @@ store_methods! {
             return refusal("cancellation_scope_checkpoint_sequence_mismatch");
         }
         let members: Vec<&Value> = commands.iter().collect();
-        if let Some(reason) = Self::cancellation_scope_completion_refusal(&mut tx, &run_id, &members).await? {
+        if let Some(reason) = Self::cancellation_scope_completion_refusal(&mut tx, &run_id, &members, sequence).await? {
             tx.rollback().await?;
             return refusal(reason);
         }
diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
index 9fa98f6e..d5e89e25 100644
--- a/rust/src/runtime/store.rs
+++ b/rust/src/runtime/store.rs
@@ -3260,7 +3260,8 @@ where
         // prepared scope admits no new member, and an undelivered preparation
         // reserves its boundary for every command.
         let non_terminal: Vec<&Value> = commands.iter().filter(|command| !is_terminal(command)).collect();
-        if let Some(reason) = Self::cancellation_scope_completion_refusal(&mut tx, &run_id, &non_terminal).await? {
+        let mut sequence = Self::authored_sequence(&mut tx, &run_id).await?;
+        if let Some(reason) = Self::cancellation_scope_completion_refusal(&mut tx, &run_id, &non_terminal, sequence).await? {
             return Err(super::RuntimeError::Protocol {
                 status: StatusCode::CONFLICT,
                 response: super::cancellation_scopes::enveloped(json!({"task_id": task_id,
@@ -3269,7 +3270,6 @@ where
             });
         }
         Self::validate_update_commands(&mut tx, &run_id, &task, commands).await?;
-        let mut sequence = Self::authored_sequence(&mut tx, &run_id).await?;
         // The tasks this batch's commands create, in order, as PHP reports them.
         let mut created = Vec::new();
         Self::apply_task_signal(&mut tx, &run_id, task_id, &task, transition_codec, false, commands, signal).await?;

Sanitized reference diff 0009-t08-Allocate-a-history-sequence-in-one-statement.patch, part 1/1.

This is an unverified, cumulative draft from the external report. Mail author/date/commit-message metadata was removed; the unified-diff hunks were retained. No patch was applied or executed, and no build/test or compatibility result is implied. The known gaps and acceptance criteria in this issue remain mandatory.

Complete sanitized patch: 5,791 bytes, SHA-256 bae07e98419dc06defe99fdff6d51c0df98b484198cafc5401bcb8cb86d435bb. This part: 5,791 bytes, SHA-256 bae07e98419dc06defe99fdff6d51c0df98b484198cafc5401bcb8cb86d435bb. Parts concatenate in order; a part boundary is between complete file diffs.

diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
index f6a210a2..cd3f16a4 100644
--- a/rust/src/runtime/backend.rs
+++ b/rust/src/runtime/backend.rs
@@ -175,6 +175,10 @@ pub(super) trait Backend: Database {
     fn document_row(row: &Self::Row, field: &str) -> Result<Value>;
     fn optional_document_row(row: &Self::Row, field: &str) -> Result<Option<Value>>;
     fn begin(pool: &Pool<Self>) -> impl Future<Output = Result<Transaction<'static, Self>>> + Send;
+    /// The run's next history sequence, allocated in one statement under the
+    /// transition lock, or `None` when it allocated none: no such run, or a
+    /// sequence already at its largest.
+    fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> impl Future<Output = Result<Option<i64>>> + Send;
     fn history_ready(pool: &Pool<Self>) -> impl Future<Output = bool> + Send;
 }
 
@@ -378,6 +382,14 @@ DELETE FROM dw_poll_receipts WHERE (namespace,worker_id,kind,request_id) IN (
     async fn begin(pool: &Pool<Self>) -> Result<Transaction<'static, Self>> {
         Ok(pool.begin_with("BEGIN IMMEDIATE").await?)
     }
+    async fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> Result<Option<i64>> {
+        sqlx::query("UPDATE workflow_runs SET last_history_sequence=last_history_sequence+1 WHERE id=$1 AND last_history_sequence<9223372036854775807 RETURNING last_history_sequence")
+            .bind(run_id)
+            .fetch_optional(connection)
+            .await?
+            .map(|row| Self::number(&row, "last_history_sequence"))
+            .transpose()
+    }
     async fn history_ready(pool: &Pool<Self>) -> bool {
         let markers: std::result::Result<Vec<(String, i64)>, _> =
             sqlx::query_as("SELECT engine,version FROM dw_server_schema")
@@ -498,6 +510,14 @@ impl Backend for Postgres {
         // round trip; a lock that fails rolls the transaction back.
         Ok(pool.begin_with("BEGIN; SELECT pg_advisory_xact_lock(4924501473315676163)").await?)
     }
+    async fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> Result<Option<i64>> {
+        sqlx::query("UPDATE workflow_runs SET last_history_sequence=last_history_sequence+1 WHERE id=$1 AND last_history_sequence<9223372036854775807 RETURNING last_history_sequence")
+            .bind(run_id)
+            .fetch_optional(connection)
+            .await?
+            .map(|row| Self::number(&row, "last_history_sequence"))
+            .transpose()
+    }
     async fn history_ready(pool: &Pool<Self>) -> bool {
         let Ok(mut connection) = pool.acquire().await else {
             return false;
@@ -717,6 +737,15 @@ impl Backend for MySql {
             .bind(mysql::VERSION).fetch_one(&mut *tx).await?;
         Ok(tx)
     }
+    async fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> Result<Option<i64>> {
+        // MySQL has no UPDATE RETURNING: `LAST_INSERT_ID(expr)` returns the
+        // incremented counter in the statement's own result.
+        let result = sqlx::query("UPDATE workflow_runs SET last_history_sequence=LAST_INSERT_ID(last_history_sequence+1) WHERE id=? AND last_history_sequence<9223372036854775807")
+            .bind(run_id)
+            .execute(connection)
+            .await?;
+        Ok((result.rows_affected() == 1).then(|| result.last_insert_id() as i64))
+    }
     async fn history_ready(pool: &Pool<Self>) -> bool {
         let Ok(mut connection) = pool.acquire().await else {
             return false;
diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
index d5e89e25..a5c4d037 100644
--- a/rust/src/runtime/store.rs
+++ b/rust/src/runtime/store.rs
@@ -4548,20 +4548,27 @@ where
         recorded_at: Option<DateTime<Utc>>,
         command_snapshot: bool,
     ) -> Result<String> {
-        // The backend transition lock protects this read and write together.
-        // MySQL has no UPDATE RETURNING; all adapters use this same allocation.
-        let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1")
-            .bind(run_id)
-            .fetch_one(&mut **tx)
-            .await?;
-        let sequence = DB::number(&row, "last_history_sequence")?
-            .checked_add(1)
-            .ok_or_else(|| refuse(StatusCode::CONFLICT, "history_sequence_exhausted"))?;
-        Self::query("UPDATE workflow_runs SET last_history_sequence=$1 WHERE id=$2")
-            .bind(sequence)
-            .bind(run_id)
-            .execute(&mut **tx)
-            .await?;
+        // The backend transition lock protects the allocation. One that
+        // allocated nothing fails as this read of a missing or exhausted
+        // sequence does.
+        let sequence = match DB::next_history_sequence(&mut **tx, run_id).await? {
+            Some(sequence) => sequence,
+            None => {
+                let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1")
+                    .bind(run_id)
+                    .fetch_one(&mut **tx)
+                    .await?;
+                let sequence = DB::number(&row, "last_history_sequence")?
+                    .checked_add(1)
+                    .ok_or_else(|| refuse(StatusCode::CONFLICT, "history_sequence_exhausted"))?;
+                Self::query("UPDATE workflow_runs SET last_history_sequence=$1 WHERE id=$2")
+                    .bind(sequence)
+                    .bind(run_id)
+                    .execute(&mut **tx)
+                    .await?;
+                sequence
+            }
+        };
         let recorded_at = recorded_at.unwrap_or_else(now);
         let mut payload = payload;
         // `WorkflowHistoryEvent::record`: an event PHP records with its task

Sanitized rebased reference for report topic 8: 0008-t08-Read-the-authored-sequence-once-per-completion.patch.

The supplied revision targets Server b8f4dd13623ce645de77ba5bd89ae8dd8eea2f7c. It is cumulative external reference material, not independently qualified against today's maintained source. Mail headers and commit-message/application instructions were removed; the unified diff is retained. The reporter now claims every rebased commit builds and the tip passes SQLite/MySQL/PostgreSQL tests. Those are attributed claims, not a new maintainer test result. Existing acceptance criteria, security findings and completed implementation receipts remain authoritative. Earlier attached patches are retained as historical references.

Sanitized patch: 3,238 bytes, SHA-256 8acaf6c9311fbddb02029a1476e23e82376e8e1a40dcc4550ba5952050dd8dd9. Original-base sanitized identity: cba2c41331743c5f612994c29a17924a2a2cb4fd2a5aaee9f38f99cf6a40b8d5. No patch was applied or executed during this update.

diff --git a/rust/src/runtime/cancellation_scopes.rs b/rust/src/runtime/cancellation_scopes.rs
index eb6c37e7..94e4009e 100644
--- a/rust/src/runtime/cancellation_scopes.rs
+++ b/rust/src/runtime/cancellation_scopes.rs
@@ -2129,13 +2129,14 @@ store_methods! {
     }
 
     /// `validateCancellationScopeMembership` inside the bridge's completion,
-    /// over the batch's non-terminal commands.
-    pub(super) async fn cancellation_scope_completion_refusal(tx: &mut Transaction<'_, DB>, run_id: &str, commands: &[&Value])
+    /// over the batch's non-terminal commands. `authored` is the run's
+    /// `Store::authored_sequence` in `tx`.
+    pub(super) async fn cancellation_scope_completion_refusal(tx: &mut Transaction<'_, DB>, run_id: &str, commands: &[&Value], authored: i64)
         -> Result<Option<&'static str>>
     {
         let events = Self::scope_events(tx, run_id).await?;
         let scopes = scopes(&events, run_id)?;
-        let next = Self::authored_sequence(tx, run_id).await? + 1;
+        let next = authored + 1;
         membership_refusal(&events, &scopes, commands, next)
     }
 }
diff --git a/rust/src/runtime/scope_checkpoint.rs b/rust/src/runtime/scope_checkpoint.rs
index d24d1eca..2d846445 100644
--- a/rust/src/runtime/scope_checkpoint.rs
+++ b/rust/src/runtime/scope_checkpoint.rs
@@ -254,7 +254,7 @@ store_methods! {
             return refusal("cancellation_scope_checkpoint_sequence_mismatch");
         }
         let members: Vec<&Value> = commands.iter().collect();
-        if let Some(reason) = Self::cancellation_scope_completion_refusal(&mut tx, &run_id, &members).await? {
+        if let Some(reason) = Self::cancellation_scope_completion_refusal(&mut tx, &run_id, &members, sequence).await? {
             tx.rollback().await?;
             return refusal(reason);
         }
diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
index 4d81e7dd..cb8a5c82 100644
--- a/rust/src/runtime/store.rs
+++ b/rust/src/runtime/store.rs
@@ -3273,7 +3273,8 @@ where
         // prepared scope admits no new member, and an undelivered preparation
         // reserves its boundary for every command.
         let non_terminal: Vec<&Value> = commands.iter().filter(|command| !is_terminal(command)).collect();
-        if let Some(reason) = Self::cancellation_scope_completion_refusal(&mut tx, &run_id, &non_terminal).await? {
+        let mut sequence = Self::authored_sequence(&mut tx, &run_id).await?;
+        if let Some(reason) = Self::cancellation_scope_completion_refusal(&mut tx, &run_id, &non_terminal, sequence).await? {
             return Err(super::RuntimeError::Protocol {
                 status: StatusCode::CONFLICT,
                 response: super::cancellation_scopes::enveloped(json!({"task_id": task_id,
@@ -3282,7 +3283,6 @@ where
             });
         }
         Self::validate_update_commands(&mut tx, &run_id, &task, commands).await?;
-        let mut sequence = Self::authored_sequence(&mut tx, &run_id).await?;
         // The tasks this batch's commands create, in order, as PHP reports them.
         let mut created = Vec::new();
         Self::apply_task_signal(&mut tx, &run_id, task_id, &task, transition_codec, false, commands, signal).await?;

Sanitized rebased reference for report topic 8: 0009-t08-Allocate-a-history-sequence-in-one-statement.patch.

The supplied revision targets Server b8f4dd13623ce645de77ba5bd89ae8dd8eea2f7c. It is cumulative external reference material, not independently qualified against today's maintained source. Mail headers and commit-message/application instructions were removed; the unified diff is retained. The reporter now claims every rebased commit builds and the tip passes SQLite/MySQL/PostgreSQL tests. Those are attributed claims, not a new maintainer test result. Existing acceptance criteria, security findings and completed implementation receipts remain authoritative. Earlier attached patches are retained as historical references.

Sanitized patch: 5,791 bytes, SHA-256 6a9e950e03056f001c4e6d202db4595bfd8eee05e2bf65e2653e8d2e9997d7b5. Original-base sanitized identity: bae07e98419dc06defe99fdff6d51c0df98b484198cafc5401bcb8cb86d435bb. No patch was applied or executed during this update.

diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
index f6a210a2..cd3f16a4 100644
--- a/rust/src/runtime/backend.rs
+++ b/rust/src/runtime/backend.rs
@@ -175,6 +175,10 @@ pub(super) trait Backend: Database {
     fn document_row(row: &Self::Row, field: &str) -> Result<Value>;
     fn optional_document_row(row: &Self::Row, field: &str) -> Result<Option<Value>>;
     fn begin(pool: &Pool<Self>) -> impl Future<Output = Result<Transaction<'static, Self>>> + Send;
+    /// The run's next history sequence, allocated in one statement under the
+    /// transition lock, or `None` when it allocated none: no such run, or a
+    /// sequence already at its largest.
+    fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> impl Future<Output = Result<Option<i64>>> + Send;
     fn history_ready(pool: &Pool<Self>) -> impl Future<Output = bool> + Send;
 }
 
@@ -378,6 +382,14 @@ DELETE FROM dw_poll_receipts WHERE (namespace,worker_id,kind,request_id) IN (
     async fn begin(pool: &Pool<Self>) -> Result<Transaction<'static, Self>> {
         Ok(pool.begin_with("BEGIN IMMEDIATE").await?)
     }
+    async fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> Result<Option<i64>> {
+        sqlx::query("UPDATE workflow_runs SET last_history_sequence=last_history_sequence+1 WHERE id=$1 AND last_history_sequence<9223372036854775807 RETURNING last_history_sequence")
+            .bind(run_id)
+            .fetch_optional(connection)
+            .await?
+            .map(|row| Self::number(&row, "last_history_sequence"))
+            .transpose()
+    }
     async fn history_ready(pool: &Pool<Self>) -> bool {
         let markers: std::result::Result<Vec<(String, i64)>, _> =
             sqlx::query_as("SELECT engine,version FROM dw_server_schema")
@@ -498,6 +510,14 @@ impl Backend for Postgres {
         // round trip; a lock that fails rolls the transaction back.
         Ok(pool.begin_with("BEGIN; SELECT pg_advisory_xact_lock(4924501473315676163)").await?)
     }
+    async fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> Result<Option<i64>> {
+        sqlx::query("UPDATE workflow_runs SET last_history_sequence=last_history_sequence+1 WHERE id=$1 AND last_history_sequence<9223372036854775807 RETURNING last_history_sequence")
+            .bind(run_id)
+            .fetch_optional(connection)
+            .await?
+            .map(|row| Self::number(&row, "last_history_sequence"))
+            .transpose()
+    }
     async fn history_ready(pool: &Pool<Self>) -> bool {
         let Ok(mut connection) = pool.acquire().await else {
             return false;
@@ -717,6 +737,15 @@ impl Backend for MySql {
             .bind(mysql::VERSION).fetch_one(&mut *tx).await?;
         Ok(tx)
     }
+    async fn next_history_sequence(connection: &mut Self::Connection, run_id: &str) -> Result<Option<i64>> {
+        // MySQL has no UPDATE RETURNING: `LAST_INSERT_ID(expr)` returns the
+        // incremented counter in the statement's own result.
+        let result = sqlx::query("UPDATE workflow_runs SET last_history_sequence=LAST_INSERT_ID(last_history_sequence+1) WHERE id=? AND last_history_sequence<9223372036854775807")
+            .bind(run_id)
+            .execute(connection)
+            .await?;
+        Ok((result.rows_affected() == 1).then(|| result.last_insert_id() as i64))
+    }
     async fn history_ready(pool: &Pool<Self>) -> bool {
         let Ok(mut connection) = pool.acquire().await else {
             return false;
diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
index cb8a5c82..fccbe10d 100644
--- a/rust/src/runtime/store.rs
+++ b/rust/src/runtime/store.rs
@@ -4562,20 +4562,27 @@ where
         recorded_at: Option<DateTime<Utc>>,
         command_snapshot: bool,
     ) -> Result<String> {
-        // The backend transition lock protects this read and write together.
-        // MySQL has no UPDATE RETURNING; all adapters use this same allocation.
-        let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1")
-            .bind(run_id)
-            .fetch_one(&mut **tx)
-            .await?;
-        let sequence = DB::number(&row, "last_history_sequence")?
-            .checked_add(1)
-            .ok_or_else(|| refuse(StatusCode::CONFLICT, "history_sequence_exhausted"))?;
-        Self::query("UPDATE workflow_runs SET last_history_sequence=$1 WHERE id=$2")
-            .bind(sequence)
-            .bind(run_id)
-            .execute(&mut **tx)
-            .await?;
+        // The backend transition lock protects the allocation. One that
+        // allocated nothing fails as this read of a missing or exhausted
+        // sequence does.
+        let sequence = match DB::next_history_sequence(&mut **tx, run_id).await? {
+            Some(sequence) => sequence,
+            None => {
+                let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1")
+                    .bind(run_id)
+                    .fetch_one(&mut **tx)
+                    .await?;
+                let sequence = DB::number(&row, "last_history_sequence")?
+                    .checked_add(1)
+                    .ok_or_else(|| refuse(StatusCode::CONFLICT, "history_sequence_exhausted"))?;
+                Self::query("UPDATE workflow_runs SET last_history_sequence=$1 WHERE id=$2")
+                    .bind(sequence)
+                    .bind(run_id)
+                    .execute(&mut **tx)
+                    .await?;
+                sequence
+            }
+        };
         let recorded_at = recorded_at.unwrap_or_else(now);
         let mut payload = payload;
         // `WorkflowHistoryEvent::record`: an event PHP records with its task

The maintained #1059 implementation is now draft PR #1084, exact a4ded60, tree 413142eb4546a54225c5893102983b2d27c04a9a on accepted main6bdb2c90. It implements all four source proposals with retained counter-width/fallback semantics and real backend visibility/cancellation/allocation/completion controls. The original #1083 main runs were terminal before this sole new PR train started. Hosted qualification is pending. Server owns fixes, ordinary integration/main and later release acceptance; #1065/#1069 follow in the user's order.

Implemented by #1084 and delivered in Native3.3.0 at protected source 2a15179f29801eec9f2a2d4e6a9c5f93233f104e / index sha256:cb4b11bd6235a551cf51e7dc6342504fbbcef19959618b376fb5648ca00fc388. The accepted source head 6a8aacd89ad41391ad113ca3e4ea54ecbbc82d7c independently still shows62successful checks and1intended skip. Its merge is an ancestor of the release. The later integrated source head 5f937ec8f385c71dc781875f2f38026b63277faa has a byte-identical rust/ tree to that release and passed the registered all-targets/ignored suites across the backend lanes.

The next transition can begin admission while the prior commit finishes, but still waits on the database lock. Registered all-backend controls prove no uncommitted read, commit/cancellation visibility, and unchanged history-sequence exhaustion/refusal behavior.

Release provenance, published SQLite corpus, four-cell MySQL Compose acceptance, default-channel acceptance and the release's three-backend startup/identity smokes establish publication follow-through. The issue's functional/source controls and release delivery are complete. Closing with no benchmark reproduction, isolated speedup or installed contention-benchmark claim. Earlier NONPASSs remain attributed. This does not close the remaining P0 work or #1099.

Navigation Menu

Repository navigation

Native: start the next transition while the last one commits #1059

Delivered in Native3.3.0

Original requirement and evidence

Problem

Proposal

Reported benchmark evidence

Acceptance

Dependencies

Draft patches and validation boundary

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Native: start the next transition while the last one commits #1059

Description

Delivered in Native3.3.0

Original requirement and evidence

Problem

Proposal

Reported benchmark evidence

Acceptance

Dependencies

Draft patches and validation boundary

Activity

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 7, 2026

rmcdaniel commented on Nov 8, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions