Native: reuse merged-transition state only with a guarded history-sequence invariant

Current owner and next action — November 8, 2026

General owns this item after the operator switched the Server agent Off. Its WIP refs remain preserved; no former session is resumed. Keep this dependent on #1078. Once its producer transaction/savepoint contract is accepted, prove remembered history sequence and registration authority with a bypass negative and rollback/concurrency controls before removing reads. The current 3.3.1 preparation contains only the seven accepted #1099 reductions, not this unfinished item. No benchmark or delivery result is claimed here.

Original requirement and evidence

External report topic 25, accepted for P0 intake at the user's direction. The proposal, timings and supplied test results are unverified reference evidence. No patch was applied or executed during intake.

Implementation owner: general. The owning agent drives source, release and applicable published-artifact verification. Existing release, operating and credential gates remain in force.

Problem and proposed scope

After an accepted merged activity/workflow transition, reuse already-authoritative history sequence, cancellation, registration and owner-freshness data instead of four extra reads. This depends on #1078 and its exact transaction/savepoint semantics.

Acceptance

Prove that every normal and local-activity history write updates the transaction's remembered sequence, and that savepoint rollback rewinds it with all other speculative state.
Add a genuine bypass negative (or an independently read debug assertion) that detects a history insert outside the tracked write path. Preserve the guard when later writers are added.
Bind reused registration/freshness to the same worker, lease/fence, snapshot and admitted lifetime. Prove the exclusion of other writers; the assertion that no transition changes registration is not sufficient by itself.
Preserve cancellation, result/history identity, pagination, whole-response budgets, idempotency, errors and old non-merged behavior. Keep related first-page sequence work in server#1066 scoped to its existing claim context.
Pass causal and rollback/concurrency tests on SQLite, MySQL and PostgreSQL, then measure the reduction on the actual merged path. The reporter's 56→52 statements and 10.38→10.17-second MySQL-only result are unverified; no fresh all-backend benchmark or clippy result was supplied for this stage.

Reported workload and evidence limits

The report describes 200 workflows, six PHP worker processes, five sequential 20 ms activities or five local activities, last-close wall time, and three or four interleaved A/B rounds. MySQL session-to-session drift is reported as high as 11%. Stock database durability, network placement and the absence of deferred projections are reporter conditions, not independently verified facts. The headline run covers topics 1–24; topic 25 has only a MySQL five-steps A/B. No speedup, production readiness or suite pass is accepted from this intake.

Sanitized reference data

The supplied Server series targets reported base d4c6de8d; the SDK reference targets 96dc061. Maintained heads were independently read as Server c3bbfa5 and SDK c87eae37. These cumulative references are not current-main patches or maintained qualification.

0008-Skip-four-reads-the-merged-transition-already-holds.patch — 12,475 bytes / SHA-256 b15902790a96f7c1583abd7a5439fd7d3569776d3c6b9c73a0b5217bcc4af16e.

Bench-only Server0001 is retained as diagnostic context, not a separate production change. Complete sanitized diff references, shared bundle identity and source-review findings are linked in the intake index and the issue comments. Original mail headers, author/date metadata, application commands and runner instructions are omitted.

Sanitized external reference for topic 25: 0008-Skip-four-reads-the-merged-transition-already-holds.patch, part 1/1. Complete sanitized diff: 12,475 bytes / SHA-256 b15902790a96f7c1583abd7a5439fd7d3569776d3c6b9c73a0b5217bcc4af16e. Ordered parts preserve the complete diff bytes; none was applied or executed. The patch is a cumulative older-source draft, not maintained implementation or qualification. Mail headers/author/date/message/application instructions were removed.

diff --git a/rust/src/runtime/local_activities.rs b/rust/src/runtime/local_activities.rs
index a5672f2f..80dace9d 100644
--- a/rust/src/runtime/local_activities.rs
+++ b/rust/src/runtime/local_activities.rs
@@ -1241,7 +1241,10 @@ store_methods! {
             let run_id = events[0].run_id.as_str();
             let count = events.len() as i64;
             let sequences: Vec<i64> = match DB::next_history_sequences(&mut **tx, run_id, count).await? {
-                Some(last) => (last - count + 1..=last).collect(),
+                Some(last) => {
+                    tx.history_appended(run_id, last);
+                    (last - count + 1..=last).collect()
+                }
                 None => {
                     let mut sequences = Vec::with_capacity(events.len());
                     for _ in events {
diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
index c1341c28..42b0a937 100644
--- a/rust/src/runtime/store.rs
+++ b/rust/src/runtime/store.rs
@@ -3433,7 +3433,7 @@ where
             Applied::Completed { run_id, run_status, created } => (run_id, run_status, created),
         };
         let activity_poll = match activity_poll {
-            Some((poll, fence, protocol)) => self.claim_piggybacked_activity(&mut tx, poll, &fence, protocol, &created, &run_id).await?,
+            Some((poll, fence, protocol)) => self.claim_piggybacked_activity(&mut tx, poll, &fence, protocol, &created, &run_id, None).await?,
             None => None,
         };
         // The worker's next workflow poll, claimed for only when the worker
@@ -3833,7 +3833,9 @@ where
     /// queue. A poll with `history_after_sequence` takes, when its lease is an
     /// activity of `run_id`, the run's history events after that sequence
     /// (`history_tail`), so the worker that holds the run's history up to
-    /// there holds all of it.
+    /// there holds all of it. `known` is a worker and the registration row
+    /// this transaction already read for it, which the poll's fence checks
+    /// instead of reading it again when the poll is that worker's.
     async fn claim_piggybacked_activity(
         &self,
         tx: &mut Transaction<'_, DB>,
@@ -3842,13 +3844,28 @@ where
         protocol: &str,
         created: &[String],
         run_id: &str,
+        known: Option<(&str, &DB::Row)>,
     ) -> Result<Option<Option<Value>>> {
-        let Some(registration) = self.worker_poll_fence_registration_in(tx, fence, text(poll, "worker_id")?).await? else {
-            return Ok(None);
+        let worker_id = text(poll, "worker_id")?;
+        let read;
+        let registration = match known {
+            Some((known_worker, row)) if known_worker == worker_id => {
+                if !self.worker_poll_fence_holds_row(fence, row)? {
+                    return Ok(None);
+                }
+                row
+            }
+            _ => {
+                let Some(row) = self.worker_poll_fence_registration_in(tx, fence, worker_id).await? else {
+                    return Ok(None);
+                };
+                read = row;
+                &read
+            }
         };
         sqlx::raw_sql("SAVEPOINT activity_poll").execute(&mut **tx).await?;
         let mark = tx.mark();
-        match self.claim_activity_for(tx, poll, protocol, &registration, created, run_id).await {
+        match self.claim_activity_for(tx, poll, protocol, registration, created, run_id).await {
             Ok(claimed) => Ok(Some(claimed)),
             Err(_) => {
                 sqlx::raw_sql("ROLLBACK TO SAVEPOINT activity_poll").execute(&mut **tx).await?;
@@ -3901,7 +3918,8 @@ where
         if claim["workflow_run_id"].as_str() != Some(run_id) {
             return Ok(None);
         }
-        let (mut events, next) = Self::history_page_known(&mut **tx, run_id, after, HISTORY_TAIL_PAGE, None, None).await?;
+        let last = tx.history_sequence(run_id);
+        let (mut events, next) = Self::history_page_known(&mut **tx, run_id, after, HISTORY_TAIL_PAGE, last, None).await?;
         if next.is_some() {
             return Ok(None);
         }
@@ -4528,7 +4546,7 @@ where
         };
         let continuation = match (continuation, &next_task_id) {
             (Some(Ok(continuation)), Some(next)) => Some(
-                self.carry_continuation(&mut tx, &continuation, &run_id, next, activity.parallel_group_path.as_ref().is_some_and(|path| !path.is_null())).await?,
+                self.carry_continuation(&mut tx, &continuation, &run, &run_id, next, activity.parallel_group_path.as_ref().is_some_and(|path| !path.is_null())).await?,
             ),
             (Some(Ok(_)), None) => Some(ContinuationOutcome::Refused("no_workflow_task".into())),
             (Some(Err(reason)), _) => Some(ContinuationOutcome::Refused(reason)),
@@ -4596,6 +4614,7 @@ where
         &self,
         tx: &mut Transaction<'_, DB>,
         admitted: &AdmittedContinuation<'_>,
+        run: &DB::Row,
         run_id: &str,
         next_task_id: &str,
         parallel: bool,
@@ -4611,14 +4630,16 @@ where
         if commands.iter().any(|command| !command["type"].as_str().is_some_and(|kind| CONTINUATION_COMMANDS.contains(&kind))) {
             return refused("unsupported_continuation_command");
         }
-        let run = Self::query("SELECT last_history_sequence,cancellation_request_command_id FROM workflow_runs WHERE id=$1")
-            .bind(run_id)
-            .fetch_one(&mut **tx)
-            .await?;
-        if DB::number(&run, "last_history_sequence")? != continuation.basis_sequence + 1 {
+        // `run` is the row the completion read under the transition lock;
+        // since then this transaction has changed only its history sequence.
+        let last_sequence = match tx.history_sequence(run_id) {
+            Some(sequence) => sequence,
+            None => DB::number(&Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1").bind(run_id).fetch_one(&mut **tx).await?, "last_history_sequence")?,
+        };
+        if last_sequence != continuation.basis_sequence + 1 {
             return refused("history_moved");
         }
-        if DB::optional_string(&run, "cancellation_request_command_id")?.is_some() {
+        if DB::optional_string(run, "cancellation_request_command_id")?.is_some() {
             return refused("cancellation_requested");
         }
         if Self::query("SELECT id FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='workflow' AND status IN ('ready','leased') AND id<>$2 LIMIT 1")
@@ -4694,7 +4715,7 @@ where
         };
         let activity_poll = match (&continuation.activity_poll, &admitted.activity_fence) {
             (Some(poll), Some(fence)) => {
-                self.claim_piggybacked_activity(tx, poll, fence, continuation.protocol, &created, &run_id).await?
+                self.claim_piggybacked_activity(tx, poll, fence, continuation.protocol, &created, &run_id, Some((&admitted.worker_id, registration))).await?
             }
             _ => None,
         };
@@ -5381,6 +5402,7 @@ where
     /// missing or exhausted sequence does.
     pub(super) async fn next_history_sequence(tx: &mut Transaction<'_, DB>, run_id: &str) -> Result<i64> {
         if let Some(sequence) = DB::next_history_sequences(&mut **tx, run_id, 1).await? {
+            tx.history_appended(run_id, sequence);
             return Ok(sequence);
         }
         let row = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1")
@@ -5395,6 +5417,7 @@ where
             .bind(run_id)
             .execute(&mut **tx)
             .await?;
+        tx.history_appended(run_id, sequence);
         Ok(sequence)
     }
 
diff --git a/rust/src/runtime/transaction.rs b/rust/src/runtime/transaction.rs
index 0bce53d4..a77f164d 100644
--- a/rust/src/runtime/transaction.rs
+++ b/rust/src/runtime/transaction.rs
@@ -23,6 +23,7 @@ pub(super) struct Mark {
     local_writes: [usize; 3],
     replayed_poll: Option<String>,
     unattempted_activities: BTreeSet<String>,
+    history_sequences: HashMap<String, i64>,
 }
 
 pub(super) struct Transaction<'a, DB: Backend> {
@@ -52,6 +53,14 @@ pub(super) struct Transaction<'a, DB: Backend> {
     /// The activities this transaction scheduled and has not yet started an
     /// attempt of: their ids are new, so they have no attempt to expire.
     pub(super) unattempted_activities: BTreeSet<String>,
+    /// Each run's `last_history_sequence` as this transaction last set it
+    /// (`Store::next_history_sequence`), which no other transition changes
+    /// while this one holds the lock.
+    history_sequences: HashMap<String, i64>,
+    /// The namespace and worker whose registration a poll's fence found
+    /// current in this transaction (`worker_poll_fence_registration_in`).
+    /// No transition changes a registration, so its owner stays fresh.
+    pub(super) fresh_owner: Option<(String, String)>,
 }
 
 impl<'a, DB: Backend> Transaction<'a, DB> {
@@ -72,9 +81,21 @@ impl<'a, DB: Backend> Transaction<'a, DB> {
             run_tags: HashMap::new(),
             local_writes: Default::default(),
             unattempted_activities: BTreeSet::new(),
+            history_sequences: HashMap::new(),
+            fresh_owner: None,
         }
     }
 
+    /// `run_id`'s history now ends at `sequence`.
+    pub(super) fn history_appended(&mut self, run_id: &str, sequence: i64) {
+        self.history_sequences.insert(run_id.to_owned(), sequence);
+    }
+
+    /// `run_id`'s `last_history_sequence`, when this transaction set it.
+    pub(super) fn history_sequence(&self, run_id: &str) -> Option<i64> {
+        self.history_sequences.get(run_id).copied()
+    }
+
     pub(super) fn summary_changed(&mut self, run_id: &str) {
         self.summaries.insert(run_id.to_owned());
     }
@@ -84,6 +105,7 @@ impl<'a, DB: Backend> Transaction<'a, DB> {
     pub(super) fn forget_history(&mut self) {
         self.scope_history = None;
         self.summary_scan = None;
+        self.history_sequences.clear();
     }
 
     /// Rows the next statement will write, charged to a metered completion.
@@ -118,6 +140,7 @@ impl<'a, DB: Backend> Transaction<'a, DB> {
             local_writes: self.local_writes.held(),
             replayed_poll: self.replayed_poll.clone(),
             unattempted_activities: self.unattempted_activities.clone(),
+            history_sequences: self.history_sequences.clone(),
         }
     }
 
@@ -133,6 +156,7 @@ impl<'a, DB: Backend> Transaction<'a, DB> {
         self.local_writes.truncate(mark.local_writes);
         self.replayed_poll = mark.replayed_poll;
         self.unattempted_activities = mark.unattempted_activities;
+        self.history_sequences = mark.history_sequences;
     }
 
     pub(super) async fn rollback(self) -> Result<()> {
diff --git a/rust/src/runtime/worker_poll_fence.rs b/rust/src/runtime/worker_poll_fence.rs
index 8d3ee111..90e98187 100644
--- a/rust/src/runtime/worker_poll_fence.rs
+++ b/rust/src/runtime/worker_poll_fence.rs
@@ -207,6 +207,9 @@ store_methods! {
         owner: &str,
         stale_after_seconds: i64,
     ) -> Result<bool> {
+        if tx.fresh_owner.as_ref().is_some_and(|(known, worker)| known == namespace && worker == owner) {
+            return Ok(true);
+        }
         let row = Self::query("SELECT status,last_heartbeat_at FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2")
             .bind(namespace).bind(owner).fetch_optional(&mut **tx).await?;
         owner_fresh::<DB>(row.as_ref(), stale_after_seconds)
@@ -252,7 +255,17 @@ store_methods! {
         worker_id: &str,
     ) -> Result<Option<DB::Row>> {
         let row = Self::query(FENCE_REGISTRATION).bind(self.namespace.as_ref()).bind(worker_id).fetch_optional(&mut **tx).await?;
-        Ok(if fence.holds::<DB>(row.as_ref(), self.worker_stale_after_seconds)? { row } else { None })
+        if !fence.holds::<DB>(row.as_ref(), self.worker_stale_after_seconds)? {
+            return Ok(None);
+        }
+        tx.fresh_owner = Some((self.namespace.to_string(), worker_id.to_owned()));
+        Ok(row)
+    }
+
+    /// `worker_poll_fence_registration_in` on a registration row this
+    /// transaction has read for the same worker.
+    pub(super) fn worker_poll_fence_holds_row(&self, fence: &WorkerPollFence, row: &DB::Row) -> Result<bool> {
+        fence.holds::<DB>(Some(row), self.worker_stale_after_seconds)
     }
 }

Security static intake finding on sanitized external 0008 (12,475 B, SHA-256 b15902790a96f7c1583abd7a5439fd7d3569776d3c6b9c73a0b5217bcc4af16e), reviewed as data only:

The proposed Transaction::fresh_owner optimization is not part of Mark and is not restored by rewind. A speculative operation can therefore populate (namespace, worker), roll back to a savepoint, and leave later owner_fresh_in calls returning true without another registration read. The cached pair also carries no observation timestamp/deadline, despite owner freshness being time-dependent and registration/heartbeat state not being proven immutable merely by the transition lock. This is exactly the kind of authority cache that must not outlive the state that established it.

Any maintained implementation needs the admitted registration/fence row plus its freshness observation bound to the same owner, snapshot and lifetime; savepoint marks must restore/clear it; expiry and concurrent re-registration/heartbeat/drain controls must force a real recheck. The same patch adds history-sequence caching without a causal bypass test, so retain the filed requirement for an independent guard that detects every history write outside the tracked append paths and proves rewind after speculative writes.

No patch was applied or executed, and no benchmark/test claim was accepted. Sanitizer review found no credential/private-path or application-instruction material in this reference.

Intake is filed as P0, owner server. Shared index, allocation and static review findings.

Complete sanitized diff reference(s): 0008 part 1/1.

Shared Lockbox bundle: lab4/general/dw-performance-intake/2026-11-06/sanitized-tricks-20-25.zip (101,306 bytes; SHA-256 3e5b593ed89536451be1c8d74abf2ee6e4782cd3f7a75fe5fc9a645d648a52da). Its roundtrip and member hashes are verified. The supplied patches and benchmark results remain external reference data; no patch was applied or executed during intake. Existing issue acceptance and subsequent owner review remain in force.

Operator nudge: this benchmax ticket (P0) has had no linked PR and no activity for about 29 h. Please pick it up next, after the current work on #1057. Suggested order for the stalled server tickets: #1063, #1068, #1075, #1077, #1078, #1079, then #1081 (P2); this one is 6 of 7. Open a draft PR as soon as work starts so the tracker shows it. If it is blocked, add the status:blocked label with a one-line reason.

Navigation Menu

Repository navigation

Native: reuse merged-transition state only with a guarded history-sequence invariant #1079

Current owner and next action — November 8, 2026

Original requirement and evidence

Problem and proposed scope

Acceptance

Reported workload and evidence limits

Sanitized reference data

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Native: reuse merged-transition state only with a guarded history-sequence invariant #1079

Description

Current owner and next action — November 8, 2026

Original requirement and evidence

Problem and proposed scope

Acceptance

Reported workload and evidence limits

Sanitized reference data

Activity

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 6, 2026

rmcdaniel commented on Nov 8, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions