Repository navigation
Native: atomically accept an opt-in workflow continuation with an activity result #1078
Description
Current owner and release gate — November 9, 2026, 23:49 UTC
General owns this P0 while Server and SDK-PHP are Off. Producer #1116 and SDK consumer sdk-php#373 are merged source. The accepted v2 contract remains authoritative.
The release audit reproduced partial activity-result commit on invalid commands and lost exact-retry decisions. Atomic repair #1118, final head e0a22d863007c03812d736641666c0bb5e783dc3, now also binds receipt identities to the completed activity/run and accepted workflow task. A new contradictory-identity counterfactual failed before this final refinement.
Eleven real-HTTP controls pass on SQLite, MySQL 8.0.46 and PostgreSQL 17.11, plus four continuation library controls including deterministic read/signal atomicity. Full final-head repository CI has 62 successful checks and one intended optional skip. The earlier OpenAPI-v65 and receipt-identity nonpasses remain historical evidence. Current source declares document v66.
Next: General follows #1118 through the protected merge queue (position 1, awaiting checks at 23:48), then qualifies Native 3.4 release source and the matching published SDK pair, installed acceptance and measured A/B. Keep opt-in/default-off. Source checks do not prove published delivery or a speedup.
Original requirement and evidence
External report topic 24-server, 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 contract
The proposal lets a worker submit workflow_continuation with an activity completion, so the result and the next eligible workflow commands can commit in one transaction. It also proposes eager activity delivery and bounded history_tail from history_after_sequence. This is a new optional producer contract, separate from ordinary carried-poll responses.
Acceptance
- Publish versioned request/reply/capability, decline/error, history-tail and byte/event/command-budget contracts before SDK consumer implementation. Coordinate existing server#1060, Native: claim the woken workflow task for a poll carried on the activity completion #1061, Native: carry the next workflow poll on the workflow task completion #1062 and Native: return a carried poll's answer in its completion's response #1067; they are dependencies/context, not proof of this new capability.
- Accept a continuation only when the exact authoritative history equals the worker's admitted prefix plus the recorded result. Define canonical result/codec/timestamp identity; a worker's guessed event must not become authority.
- Preserve original worker admission, owner, lease, fence, attempt and deadlines. Reject stale/foreign workers, unknown history, pending workflow/message tasks, parallel groups, cancellation, and unsupported commands. Initially limit commands to schedule_activity, start_timer, complete_workflow and record_version_marker.
- An ordinary continuation decline must preserve an otherwise valid activity completion and normal later workflow polling. Only documented eligibility declines may downgrade; backend/decoder/budget/invariant failures need an explicit truthful contract.
- Scope speculative claim/commands to a savepoint and restore all cached state on rollback. Retried/lost completion replies must not duplicate results, history or commands. Preserve capacity and one-slot grant rules, eager fallback and response budgets.
- Keep the feature opt-in with old clients/servers unchanged. Add real HTTP and SQLite/MySQL/PostgreSQL controls for concurrency, foreign identity, expiry/transit, rollback, signals/updates, cancellation, replay, missing tail and large/external payloads.
- Qualify the matching published producer/SDK tuple before claiming delivered behavior or a speedup. No default enablement is selected here.
Reporter measurement: one Server binary with alternating SDK behavior, 14.59→10.76 seconds; lock turns 11→6 per workflow. These are unverified measurements, not acceptance evidence.
SDK counterpart: durable-workflow/sdk-php#341. Topic 25 depends on this producer's accepted transaction invariants.
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.
0007-Commit-an-activity-s-result-and-the-workflow-task-it.patch— 89,131 bytes / SHA-256341084feeb6700b8098f1771c540e8c0ca9eea549080f41dd7971b81f7647bd9.
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.
Activity
rmcdaniel commentedon Nov 6, 2026
Sanitized external reference for topic 24-server:
0007-Commit-an-activity-s-result-and-the-workflow-task-it.patch, part 1/2. Complete sanitized diff: 89,131 bytes / SHA-256341084feeb6700b8098f1771c540e8c0ca9eea549080f41dd7971b81f7647bd9. 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.rmcdaniel commentedon Nov 6, 2026
Sanitized external reference for topic 24-server:
0007-Commit-an-activity-s-result-and-the-workflow-task-it.patch, part 2/2. Complete sanitized diff: 89,131 bytes / SHA-256341084feeb6700b8098f1771c540e8c0ca9eea549080f41dd7971b81f7647bd9. 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.let attempted = task_payload["workflow_signal_id"].as_str().or_else(|| task_payload["resume_source_id"].as_str()) .filter(|signal| !signal.is_empty()); - let (probed, next) = Self::next_message_task(&mut tx, &run_id, &DB::string(&run, "queue")?, attempted, false).await?; + let (probed, next) = Self::next_message_task(tx, &run_id, &DB::string(&run, "queue")?, attempted, false).await?; // The probe read the run as the commands left it; only a task created // since can have changed its status. let mut current = Some(probed).filter(|_| next.is_none()); @@ -3609,73 +3759,51 @@ where let start_turn = start_turn.into_inner().unwrap_or_else(std::sync::PoisonError::into_inner); if let Some(turn) = start_turn { current = None; - if let Some(next) = Self::child_start_turn(&mut tx, &run_id, task_id, &DB::string(&run, "queue")?, turn).await? { + if let Some(next) = Self::child_start_turn(tx, &run_id, task_id, &DB::string(&run, "queue")?, turn).await? { created.push(next); } } - Self::record_completion(&mut tx, task_id, &json!({COMPLETION_RECEIPT: {"request": body, "created_task_ids": created}})).await?; + Self::record_completion(tx, task_id, &json!({COMPLETION_RECEIPT: {"request": body, "created_task_ids": created}})).await?; // PHP's executor leaves every run a completed task did not close // waiting, even when the task recorded only markers or side effects. // A run already waiting or closed is read and left as it is. let mut run_status = match current { Some(status) => status, None => DB::string( - &Self::query("SELECT status FROM workflow_runs WHERE id=$1").bind(&run_id).fetch_one(&mut *tx).await?, + &Self::query("SELECT status FROM workflow_runs WHERE id=$1").bind(&run_id).fetch_one(&mut **tx).await?, "status", )?, }; if matches!(run_status.as_str(), "pending" | "running") { let waiting = Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1 AND status IN ('pending','running')") .bind(&run_id) - .execute(&mut *tx) + .execute(&mut **tx) .await?; run_status = if DB::affected(waiting) == 1 { "waiting".to_owned() } else { let current = Self::query("SELECT status FROM workflow_runs WHERE id=$1") .bind(&run_id) - .fetch_one(&mut *tx) + .fetch_one(&mut **tx) .await?; DB::string(¤t, "status")? }; } // This task's completion or failure event is recorded only with the // run's closing status. - if let Some(worker) = &worker + if let Some(worker) = worker && matches!(run_status.as_str(), "completed" | "failed") { - Self::attribute_worker_terminal(&mut tx, &run_id, task_id, worker).await?; + Self::attribute_worker_terminal(tx, &run_id, task_id, worker).await?; } // `MessageStreamService::recordCompletion`: only a continued run // changes the instance's current run, which then receives the streams. if !streams.is_empty() || commands.iter().any(|command| command["type"] == "continue_as_new") { - self.record_stream_completion(&mut tx, &stream_instance, &run_id, task_id, &streams, transition_codec_for_streams, limits) + self.record_stream_completion(tx, &stream_instance, &run_id, task_id, &streams, transition_codec_for_streams, limits) .await?; } tx.summary_changed(&run_id); - let activity_poll = match activity_poll { - Some((poll, fence, protocol)) => self.claim_piggybacked_activity(&mut tx, poll, &fence, protocol).await?, - None => None, - }; - // The worker's next workflow poll, claimed for only when the worker - // runs no activity first: the completion carried no activity poll, - // or that poll's claim leased nothing. - let workflow_poll = match &workflow_poll { - Some(poll) if !carries_activity_poll || matches!(activity_poll, Some(None)) => { - self.claim_carried_workflow(&mut tx, poll).await?.map(|claimed| (claimed, poll.encoding)) - } - _ => None, - }; - tx.commit().await?; - self.wake.notify_waiters(); - let mut response = json!({"task_id": task_id, "run_id": run_id, "run_status": run_status, "workflow_task_attempt": body["workflow_task_attempt"], "outcome": "completed", "recorded": true, "reason": null, "created_task_ids": created}); - if let Some(claimed) = activity_poll { - response["activity_poll"] = carried_poll("activity", claimed, None)?; - } - if let Some((claimed, encoding)) = workflow_poll { - response["workflow_poll"] = carried_poll("workflow", claimed, encoding)?; - } - Ok(response) + Ok(Applied::Completed { run_id, run_status, created }) } /// The poll checks a carried activity poll takes before the completion's @@ -3698,29 +3826,89 @@ where /// without the lock (`replay_activity_poll`). The task it leased, if /// any, or `None` when the poll did not run or was refused: the claim's /// writes roll back to the savepoint, and the worker's own poll answers. + /// + /// A poll with `eager: true` first tries the activity tasks this + /// transition `created`, in order, as Temporal's eager activity dispatch + /// does: the worker that scheduled an activity runs it, ahead of the + /// 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. async fn claim_piggybacked_activity( &self, tx: &mut Transaction<'_, DB>, poll: &Value, fence: &super::worker_poll_fence::WorkerPollFence, protocol: &str, + created: &[String], + run_id: &str, ) -> Result<Option<Option<Value>>> { let Some(registration) = self.worker_poll_fence_registration_in(tx, fence, text(poll, "worker_id")?).await? else { return Ok(None); }; sqlx::raw_sql("SAVEPOINT activity_poll").execute(&mut **tx).await?; - match self.claim_in(tx, poll, "activity", protocol, None, None, Some(registration)).await { + let mark = tx.mark(); + match self.claim_activity_for(tx, poll, protocol, ®istration, created, run_id).await { Ok(claimed) => Ok(Some(claimed)), Err(_) => { sqlx::raw_sql("ROLLBACK TO SAVEPOINT activity_poll").execute(&mut **tx).await?; - tx.run_tags.clear(); // The claim may have appended history the rollback took back. - tx.forget_history(); + tx.rewind(mark); Ok(None) } } } + /// `claim_piggybacked_activity`'s claim, inside its savepoint. + async fn claim_activity_for( + &self, + tx: &mut Transaction<'_, DB>, + poll: &Value, + protocol: &str, + registration: &DB::Row, + created: &[String], + run_id: &str, + ) -> Result<Option<Value>> { + let mut claimed = None; + if poll["eager"] == true { + for task_id in created { + let target = Target { task_id, history: true }; + claimed = self.claim_targeted(tx, poll, "activity", protocol, None, None, Some(registration), Some(target)).await?; + if claimed.is_some() { + break; + } + } + } + if claimed.is_none() { + claimed = self.claim_targeted(tx, poll, "activity", protocol, None, None, Some(registration), None).await?; + } + if let Some(claim) = claimed.as_mut() + && let Some(tail) = Self::history_tail(tx, poll, run_id, claim).await? + { + claim["history_tail"] = tail; + } + Ok(claimed) + } + + /// The history a carried poll's lease adds for the worker that holds + /// `run_id`'s history up to the poll's `history_after_sequence`: every + /// event after it, when the lease is an activity of that run and the + /// events fit one page. The poll's receipt keeps the lease without it. + async fn history_tail(tx: &mut Transaction<'_, DB>, poll: &Value, run_id: &str, claim: &Value) -> Result<Option<Value>> { + let Some(after) = poll["history_after_sequence"].as_i64().filter(|after| *after >= 0) else { + return Ok(None); + }; + 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?; + if next.is_some() { + return Ok(None); + } + Self::with_signal_sequences(&mut **tx, run_id, &mut events).await?; + Ok(Some(json!({"after_sequence": after, "events": events}))) + } + /// An activity poll's receipt for a lease the worker still holds, read /// without the transition lock: a committed lease that the claim would /// replay under the lock (`claim_in`). A receipt in any other state @@ -4183,6 +4371,7 @@ where body: Value, external: Option<super::external_resolution::PreparedExternal>, workflow_poll: Option<CarriedWorkflowPoll<'_>>, + continuation: Option<CarriedContinuation<'_>>, ) -> Result<Value> { let result = match &external { Some(prepared) => prepared.stored(), @@ -4193,6 +4382,13 @@ where Some(poll) => self.admit_carried_workflow_poll(poll).await, None => None, }; + let continuation = match continuation { + Some(CarriedContinuation::Prepared(continuation)) => { + Some(self.admit_continuation(continuation, text(&body, "lease_owner")?).await) + } + Some(CarriedContinuation::Refused(reason)) => Some(Err(reason)), + None => None, + }; let mut tx = self.begin().await?; let task = self.task_row(&mut tx, task_id).await?; if DB::string(&task, "task_type")? != "activity" { @@ -4330,9 +4526,24 @@ where ) .await? }; + 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?, + ), + (Some(Ok(_)), None) => Some(ContinuationOutcome::Refused("no_workflow_task".into())), + (Some(Err(reason)), _) => Some(ContinuationOutcome::Refused(reason)), + (None, _) => None, + }; + // The worker's next workflow poll, claimed for unless the worker runs + // the activity its continuation's poll leased first. + let runs_activity = matches!( + &continuation, + Some(ContinuationOutcome::Recorded { carries_activity_poll: true, activity_poll, .. }) + if !matches!(activity_poll, Some(None)) + ); let workflow_poll = match &workflow_poll { - Some(poll) => self.claim_carried_workflow(&mut tx, poll).await?.map(|claimed| (claimed, poll.encoding)), - None => None, + Some(poll) if !runs_activity => self.claim_carried_workflow(&mut tx, poll).await?.map(|claimed| (claimed, poll.encoding)), + _ => None, }; tx.commit().await?; self.wake.notify_waiters(); @@ -4340,9 +4551,157 @@ where if let Some((claimed, encoding)) = workflow_poll { response["workflow_poll"] = carried_poll("workflow", claimed, encoding)?; } + match continuation { + Some(ContinuationOutcome::Recorded { response: mut completion, activity_poll, .. }) => { + if let Some(claimed) = activity_poll { + completion["activity_poll"] = carried_poll("activity", claimed, None)?; + } + response["workflow_continuation"] = completion; + } + Some(ContinuationOutcome::Refused(reason)) => { + response["workflow_continuation"] = json!({"recorded": false, "reason": reason}); + } + None => {} + } Ok(response) } + /// A continuation's checks before the activity completion's transaction: + /// the worker's workflow poll and activity poll are admitted as a + /// carried poll is, or the continuation is refused. + async fn admit_continuation<'a>( + &self, + continuation: Continuation<'a>, + worker_id: &str, + ) -> std::result::Result<AdmittedContinuation<'a>, String> { + let poll = json!({"worker_id": worker_id, "task_queue": continuation.task_queue}); + let Some(fence) = self.admit_piggybacked_activity_poll(&poll).await else { + return Err("worker_not_admitted".into()); + }; + let activity_fence = match &continuation.activity_poll { + Some(poll) => self.admit_piggybacked_activity_poll(poll).await, + None => None, + }; + Ok(AdmittedContinuation { continuation, worker_id: worker_id.to_owned(), fence, activity_fence }) + } + + /// An admitted continuation, in the activity completion's transaction + /// after the result made `next_task_id` ready: when nothing but the + /// result was written to the run's history after `basis_sequence`, the + /// worker leases that workflow task and completes it with the commands, + /// as its own claim and completion would have. The continuation's writes + /// roll back to a savepoint when it is refused, and the activity + /// completion commits without them. + async fn carry_continuation( + &self, + tx: &mut Transaction<'_, DB>, + admitted: &AdmittedContinuation<'_>, + run_id: &str, + next_task_id: &str, + parallel: bool, + ) -> Result<ContinuationOutcome> { + let refused = |reason: &str| Ok(ContinuationOutcome::Refused(reason.to_owned())); + let continuation = &admitted.continuation; + // A parallel group's member resumes the workflow only with the + // group's other results, which the worker cannot have computed. + if parallel { + return refused("parallel_activity"); + } + let commands = continuation.body["commands"].as_array().map(Vec::as_slice).unwrap_or_default(); + 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 { + return refused("history_moved"); + } + 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") + .bind(run_id) + .bind(next_task_id) + .fetch_optional(&mut **tx) + .await? + .is_some() + { + return refused("workflow_task_pending"); + } + let Some(registration) = self.worker_poll_fence_registration_in(tx, &admitted.fence, &admitted.worker_id).await? else { + return refused("worker_not_admitted"); + }; + sqlx::raw_sql("SAVEPOINT workflow_continuation").execute(&mut **tx).await?; + let mark = tx.mark(); + match self.apply_continuation(tx, admitted, ®istration, next_task_id).await { + Ok((response, activity_poll)) => Ok(ContinuationOutcome::Recorded { + response, + carries_activity_poll: continuation.activity_poll.is_some(), + activity_poll, + }), + Err(error) => { + if sqlx::raw_sql("ROLLBACK TO SAVEPOINT workflow_continuation").execute(&mut **tx).await.is_err() { + return Err(error); + } + tx.rewind(mark); + Ok(ContinuationOutcome::Refused(continuation_reason(&error))) + } + } + } + + /// `carry_continuation`'s claim and completion, inside its savepoint: the + /// workflow task leased as the worker's own poll would lease it, without + /// the history the worker holds, then completed as the worker's own + /// completion would complete it, and the continuation's activity poll + /// claimed as a completion's carried poll is. + async fn apply_continuation( + &self, + tx: &mut Transaction<'_, DB>, + admitted: &AdmittedContinuation<'_>, + registration: &DB::Row, + task_id: &str, + ) -> Result<(Value, Option<Option<Value>>)> { + let continuation = &admitted.continuation; + let poll = json!({"worker_id": admitted.worker_id, "task_queue": continuation.task_queue}); + let target = Target { task_id, history: false }; + let Some(claim) = self + .claim_targeted(tx, &poll, "workflow", continuation.protocol, None, None, Some(registration), Some(target)) + .await? + else { + return Err(refuse(StatusCode::CONFLICT, "workflow_task_not_claimable")); + }; + let task = self.task_row(tx, task_id).await?; + // A message's task resumes the workflow with what the worker has not + // read. + let payload = DB::document_row(&task, "payload")?; + if !payload["workflow_signal_id"].is_null() || !payload["workflow_update_id"].is_null() { + return Err(refuse(StatusCode::CONFLICT, "message_task")); + } + let mut body = continuation.body.clone(); + body["lease_owner"] = json!(admitted.worker_id); + body["workflow_task_attempt"] = claim["workflow_task_attempt"].clone(); + let commands = Self::completion_commands(&body, &continuation.external)?; + let guarded = self + .guard_completion(tx, &task, task_id, &body, commands, Some(continuation.protocol), &continuation.transition_codec) + .await?; + let applied = self + .apply_completion(tx, &task, task_id, &body, commands, guarded, continuation.transition_codec.clone(), &continuation.external, continuation.worker.as_ref(), continuation.limits, None) + .await?; + let Applied::Completed { run_id, run_status, created } = applied else { + return Err(refuse(StatusCode::CONFLICT, "run_deadline_expired")); + }; + 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? + } + _ => None, + }; + let response = json!({"task_id": task_id, "run_id": run_id, "run_status": run_status, "workflow_task_attempt": body["workflow_task_attempt"], "outcome": "completed", "recorded": true, "reason": null, "created_task_ids": created}); + Ok((response, activity_poll)) + } + /// A carried workflow poll's checks that come before the lock in /// `poll_with_pages`. A poll they would answer without a claim (a /// draining, stale or unregistered worker, a malformed page request) is @@ -5638,6 +5997,40 @@ impl<R> Clone for Apply<'_, R> { } impl<R> Copy for Apply<'_, R> {} +/// What a workflow task completion's guards read (`Store::guard_completion`) +/// for its commands to be applied with: the run's whole row, its instance, +/// the completion's message streams and its local activity records. +pub(super) struct Guarded<R> { + pub(super) stream_run: R, + pub(super) stream_instance: String, + pub(super) streams: super::message_streams::Completion, + pub(super) local: std::collections::BTreeMap<usize, super::local_activities::Record>, + pub(super) exceptions: std::collections::BTreeMap<usize, String>, +} + +/// A workflow task completion's applied commands (`Store::apply_completion`). +pub(super) enum Applied { + /// The run's deadline had passed: the run timed out instead, and this + /// answer refuses the completion once the transaction commits. + Expired(Value), + /// The commands applied: the run, its status after them and the tasks + /// they created. + Completed { run_id: String, run_status: String, created: Vec<String> }, +} + +/// The one task a claim leases, when it is a candidate (`claim_targeted`). +#[derive(Clone, Copy)] +pub(super) struct Target<'a> { + pub(super) task_id: &'a str, + /// Whether a workflow task's claim carries its history. A worker that + /// holds the history leases the task without it. + pub(super) history: bool, +} + +/// The most history events a carried poll's `history_tail` delivers: a +/// longer tail is left to the worker's own history read. +const HISTORY_TAIL_PAGE: i64 = 100; + /// A workflow task completion's stored receipt: its request and the tasks it /// created, so a repeat answers as the original did. Older receipts, and /// activity receipts, are the bare request. @@ -5751,6 +6144,76 @@ struct AdmittedWorkflowPoll<'a> { request: super::workflow_poll_requests::WorkflowPollRequest, } +/// A workflow task completion an activity completion carries: the commands +/// the worker computed for the workflow task the result makes ready, from +/// the run's history through `basis_sequence` and the result, as the +/// history's next event. Its payloads are prepared as a completion's are. +pub(crate) struct Continuation<'a> { + pub(crate) basis_sequence: i64, + pub(crate) task_queue: String, + /// `{"commands": [...]}`, canonical as a completion's body. + pub(crate) body: Value, + pub(crate) external: super::external_resolution::PreparedCommands, + pub(crate) worker: Option<super::runtime_credentials::Principal>, + pub(crate) protocol: &'a str, + pub(crate) limits: super::memo_limits::StructuralLimits, + pub(crate) transition_codec: Arc<Semaphore>, + /// The worker's next activity poll, which the continuation carries as a + /// workflow task completion carries one. + pub(crate) activity_poll: Option<Value>, +} + +/// A continuation as the activity completion's request carries it, after +/// the route's checks (`http::carried_continuation`). +pub(crate) struct ContinuationRequest { + pub(crate) basis_sequence: i64, + pub(crate) task_queue: String, + /// The completion body the route checked, with its commands. + pub(crate) body: Value, + pub(crate) activity_poll: Option<Value>, +} + +/// A carried continuation, or why it was refused before the transaction. +pub(crate) enum CarriedContinuation<'a> { + Prepared(Continuation<'a>), + Refused(String), +} + +/// A continuation admitted before the completion's transaction, with the +/// fences of the worker's workflow and activity polls. +struct AdmittedContinuation<'a> { + continuation: Continuation<'a>, + /// The worker that completed the activity, which leases the workflow task. + worker_id: String, + fence: super::worker_poll_fence::WorkerPollFence, + /// The fence of the continuation's activity poll, when it was admitted. + activity_fence: Option<super::worker_poll_fence::WorkerPollFence>, +} + +/// The commands a continuation may carry: those whose application depends +/// on nothing but the run's history and the task they complete. +const CONTINUATION_COMMANDS: &[&str] = &["schedule_activity", "start_timer", "complete_workflow", "record_version_marker"]; + +/// A continuation's outcome in the activity completion's transaction. +enum ContinuationOutcome { + /// The workflow task's completion answer and the continuation's + /// activity poll, as a workflow task completion answers them. + Recorded { response: Value, carries_activity_poll: bool, activity_poll: Option<Option<Value>> }, + /// Nothing of the continuation was kept, for this reason. + Refused(String), +} + +/// A refused continuation's reason: a refusal's own, or `continuation_failed`. +pub(super) fn continuation_reason(error: &super::RuntimeError) -> String { + match error { + super::RuntimeError::Refused { reason, .. } => (*reason).to_owned(), + super::RuntimeError::Protocol { response, .. } => { + response["reason"].as_str().unwrap_or("continuation_failed").to_owned() + } + _ => "continuation_failed".to_owned(), + } +} + /// What a workflow poll's first probe under a request ID found. enum FirstProbe { /// The request's last answer, still deliverable. @@ -5824,7 +6287,8 @@ fn capabilities_now() -> Value { "activity_lease_renewal": true, "activity_cancellation_acknowledgement": true, "child_workflow_start_turn": true, "condition_wait_idempotent_reopen": true, "workflow_task_completion_activity_poll": true, "activity_completion_workflow_poll": true, - "workflow_task_completion_workflow_poll": true}); + "workflow_task_completion_workflow_poll": true, "carried_activity_poll_eager_dispatch": true, + "carried_activity_poll_history_tail": true, "activity_completion_workflow_continuation": true}); value["runtime_external_payload_transport"] = super::payload_discovery::transport_manifest(); value["supported_workflow_task_commands"] .as_array_mut() diff --git a/rust/src/runtime/transaction.rs b/rust/src/runtime/transaction.rs index 6ed9ebb6..0bce53d4 100644 --- a/rust/src/runtime/transaction.rs +++ b/rust/src/runtime/transaction.rs @@ -17,6 +17,14 @@ pub(super) struct RunTags { pub(super) workflow_class: String, } +/// A transaction's kept state at a savepoint (`Transaction::mark`). +pub(super) struct Mark { + metered: bool, + local_writes: [usize; 3], + replayed_poll: Option<String>, + unattempted_activities: BTreeSet<String>, +} + pub(super) struct Transaction<'a, DB: Backend> { inner: sqlx::Transaction<'a, DB>, /// This transition's place in the server's queue, held until `inner` @@ -102,6 +110,31 @@ impl<'a, DB: Backend> Transaction<'a, DB> { } } + /// What this transaction keeps beside its rows, before the statements + /// a savepoint may take back (`rewind`). + pub(super) fn mark(&self) -> Mark { + Mark { + metered: self.meter.is_some(), + local_writes: self.local_writes.held(), + replayed_poll: self.replayed_poll.clone(), + unattempted_activities: self.unattempted_activities.clone(), + } + } + + /// Forget what this transaction kept since `mark`, whose statements were + /// rolled back to a savepoint taken there. A meter set before the mark + /// stays, with what it charged since. + pub(super) fn rewind(&mut self, mark: Mark) { + if !mark.metered { + self.meter = None; + } + self.run_tags.clear(); + self.forget_history(); + self.local_writes.truncate(mark.local_writes); + self.replayed_poll = mark.replayed_poll; + self.unattempted_activities = mark.unattempted_activities; + } + pub(super) async fn rollback(self) -> Result<()> { self.inner.rollback().await?; Ok(()) diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 58c8fa7a..19c0cc14 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -381,6 +381,8 @@ mod worker_protocol_envelope; mod child_start_turn; #[path = "execution/carried_polls.rs"] mod carried_polls; +#[path = "execution/activity_continuations.rs"] +mod activity_continuations; enum TestDatabase { Sqlite(tempfile::TempDir), diff --git a/rust/tests/execution/activity_continuations.rs b/rust/tests/execution/activity_continuations.rs new file mode 100644 index 00000000..f0389799 --- /dev/null +++ b/rust/tests/execution/activity_continuations.rs @@ -0,0 +1,199 @@ +use super::*; + +fn kinds(history: &Value) -> Vec<String> { + history.as_array().unwrap().iter().map(|event| event["event_type"].as_str().unwrap().to_owned()).collect() +} + +fn last_sequence(events: &Value) -> i64 { + events.as_array().unwrap().last().unwrap()["sequence"].as_i64().unwrap() +} + +fn step(n: i64) -> Value { + json!({"type":"schedule_activity","activity_type":"step","arguments":envelope(Payload::Array(vec![Payload::Long(n)])), + "start_to_close_timeout":120}) +} + +/// A carried activity poll that leases the activity its completion +/// schedules, with the run's history after `after`. +fn eager(id: &str, after: i64) -> Value { + json!({"worker_id":"chainer","task_queue":"test","poll_request_id":id,"eager":true,"history_after_sequence":after}) +} + +fn carried_workflow(id: &str) -> Value { + json!({"worker_id":"chainer","task_queue":"test","poll_request_id":id,"history_page_size":100}) +} + +async fn complete_activity(app: &Router, activity: &Value, carried: Value) -> Value { + let mut body = json!({"lease_owner":activity["lease_owner"],"activity_attempt_id":activity["activity_attempt_id"], + "result":envelope(Payload::Long(1))}); + for (field, value) in carried.as_object().unwrap() { + body[field] = value.clone(); + } + let done = request(app, "POST", &format!("/api/worker/activity-tasks/{}/complete", activity["task_id"].as_str().unwrap()), body).await; + assert_eq!(done.0, StatusCode::OK, "{}", done.1); + assert_eq!(done.1["recorded"], true, "{}", done.1); + done.1 +} + +/// Starts a run whose first workflow task schedules a step, leased eagerly +/// with the history after what that task held. The step and the sequence +/// a continuation of it is based on. +async fn first_step(app: &Router, workflow: &str) -> (Value, i64) { + start(app, workflow).await; + let task = poll(app, "chainer", "workflow").await; + let held = last_sequence(&task["history_events"]); + let mut body = completion(&task, json!([step(1)])); + body["activity_poll"] = eager(&format!("{workflow}-poll-1"), held); + let done = request(app, "POST", &format!("/api/worker/workflow-tasks/{}/complete", task["task_id"].as_str().unwrap()), body).await; + assert_eq!(done.0, StatusCode::OK, "{}", done.1); + assert_eq!(done.1["activity_poll"]["leased"], true, "{}", done.1); + let activity = done.1["activity_poll"]["answer"]["task"].clone(); + let tail = &activity["history_tail"]; + assert_eq!(tail["after_sequence"], held, "{activity}"); + assert_eq!(tail["events"][0]["sequence"], held + 1, "{activity}"); + let basis = last_sequence(&tail["events"]); + (activity, basis) +} + +/// An activity completion that carries the worker's completion of the +/// workflow task its result makes ready commits both, when nothing but the +/// result was written after the history the worker held. The activity the +/// commands schedule is leased to the continuation's poll with the history +/// it adds, so a run's steps go on one transaction each, and the run's +/// history is the one its worker's own polls and completions write. +#[tokio::test] +async fn a_continuation_completes_the_workflow_task_its_activity_result_makes_ready() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "chainer", json!(["echo"]), json!(["step"])).await; + let (mut activity, mut basis) = first_step(&app, "chained-run").await; + // A replayed poll answers with the lease, not the history. + let replayed = request(&app, "POST", "/api/worker/activity-tasks/poll", + json!({"worker_id":"chainer","task_queue":"test","poll_request_id":"chained-run-poll-1","timeout_seconds":0})).await; + assert_eq!(replayed.0, StatusCode::OK, "{}", replayed.1); + let mut leased = activity.clone(); + leased.as_object_mut().unwrap().remove("history_tail"); + assert_eq!(replayed.1["task"], leased, "{}", replayed.1); + + for n in 2..=3 { + let id = format!("chained-run-poll-{n}"); + let answer = complete_activity(&app, &activity, json!({"workflow_poll": carried_workflow(&format!("chained-run-workflow-{n}")), + "workflow_continuation": {"basis_sequence": basis, "task_queue": "test", "commands": [step(n)], + "activity_poll": eager(&id, basis)}})).await; + let continued = &answer["workflow_continuation"]; + assert_eq!(continued["recorded"], true, "{answer}"); + assert_eq!(continued["outcome"], "completed", "{answer}"); + assert_eq!(continued["task_id"], answer["next_task_id"], "{answer}"); + // The worker runs the activity first, so its workflow poll is not + // claimed for. + assert!(answer.get("workflow_poll").is_none(), "{answer}"); + assert_eq!(continued["activity_poll"]["leased"], true, "{answer}"); + activity = continued["activity_poll"]["answer"]["task"].clone(); + assert_eq!(activity["activity_type"], "step", "{activity}"); + let tail = &activity["history_tail"]; + assert_eq!(tail["after_sequence"], basis, "{activity}"); + assert_eq!(tail["events"][0]["event_type"], "ActivityCompleted", "{activity}"); + assert_eq!(tail["events"][0]["sequence"], basis + 1, "{activity}"); + basis = last_sequence(&tail["events"]); + } + let answer = complete_activity(&app, &activity, json!({"workflow_continuation": {"basis_sequence": basis, "task_queue": "test", + "commands": [{"type":"complete_workflow","result":envelope(Payload::Long(3))}]}})).await; + assert_eq!(answer["workflow_continuation"]["recorded"], true, "{answer}"); + assert_eq!(answer["workflow_continuation"]["run_status"], "completed", "{answer}"); + assert!(answer["workflow_continuation"].get("activity_poll").is_none(), "{answer}"); + let (_, described) = request(&app, "GET", "/api/workflows/chained-run", Value::Null).await; + assert_eq!(described["status"], "completed", "{described}"); + + // The same steps through the worker's own polls and completions. + let started = start(&app, "stepped-run").await; + for n in 1..=3 { + let task = poll(&app, "chainer", "workflow").await; + let (status, body) = finish_task(&app, &task, json!([step(n)])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let activity = poll(&app, "chainer", "activity").await; + complete_activity(&app, &activity, json!({})).await; + } + let task = poll(&app, "chainer", "workflow").await; + let (status, body) = finish_task(&app, &task, json!([{"type":"complete_workflow","result":envelope(Payload::Long(3))}])).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let stepped = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let chained = run_history(&app, &json!("chained-run"), &described["run_id"]).await; + assert_eq!(kinds(&chained), kinds(&stepped)); + let sequences = |history: &Value| history.as_array().unwrap().iter().map(|event| event["sequence"].clone()).collect::<Vec<_>>(); + assert_eq!(sequences(&chained), sequences(&stepped)); + runtime.close().await; + database.remove().await; +} + +/// A continuation the Server cannot prove was computed from the run's +/// whole history, or that it cannot record as the worker's completion would +/// be, is refused. The activity's result still commits, and the workflow +/// task it makes ready is left as it was for the worker's own poll. +#[tokio::test] +async fn a_refused_continuation_leaves_the_workflow_task_for_the_worker_s_poll() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + register(&app, "chainer", json!(["echo"]), json!(["step"])).await; + let cases = [ + ("moved-run", json!({"commands": [step(2)]}), -1, "history_moved"), + ("unsupported-run", json!({"commands": [{"type":"open_signal_wait","signal_name":"go"}]}), 0, "unsupported_continuation_command"), + ("malformed-run", json!({"commands": [step(2)], "sticky_cache": {}}), 0, "invalid_workflow_continuation"), + // Refused by the completion's guard after the continuation's claim, + // which rolls back with it. + ("guarded-run", json!({"commands": [{"type":"schedule_activity","activity_type":"step", + "arguments":envelope(Payload::Array(vec![])),"cancellation_policy":"try_cancel"}]}), 0, + "activity_cancellation_policy_not_supported"), + ]; + for (workflow, mut continuation, offset, reason) in cases { + let (activity, basis) = first_step(&app, workflow).await; + continuation["basis_sequence"] = json!(basis + offset); + continuation["task_queue"] = json!("test"); + continuation["activity_poll"] = eager(&format!("{workflow}-poll-2"), basis); + let answer = complete_activity(&app, &activity, json!({"workflow_poll": carried_workflow(&format!("{workflow}-workflow")), + "workflow_continuation": continuation})).await; + assert_eq!(answer["workflow_continuation"], json!({"recorded": false, "reason": reason}), "{workflow}: {answer}"); + // The workflow task is the carried workflow poll's, on its first + // attempt, with the result the last event of its history. + assert_eq!(answer["workflow_poll"]["leased"], true, "{workflow}: {answer}"); + let task = &answer["workflow_poll"]["answer"]["task"]; + assert_eq!(task["task_id"], answer["next_task_id"], "{workflow}: {answer}"); + assert_eq!(task["workflow_task_attempt"], 1, "{workflow}: {task}"); + let history = run_history(&app, &json!(workflow), &task["run_id"]).await; + let completed = history.as_array().unwrap().iter().find(|event| event["event_type"] == "ActivityCompleted").unwrap(); + assert_eq!(completed["sequence"], basis + 1, "{workflow}: {history}"); + // Which the worker completes as it would have without one. + let (status, body) = finish_task(&app, task, json!([{"type":"complete_workflow","result":envelope(Payload::Null)}])).await; + assert_eq!(status, StatusCode::OK, "{workflow}: {body}"); + } + runtime.close().await; + database.remove().await; +} + +/// A signal that arrives while the activity runs is history the worker +/// has not read, so a continuation computed without it is refused. +#[tokio::test] +async fn a_continuation_is_refused_after_a_signal_the_worker_has_not_read() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let (status, body) = request(&app, "POST", "/api/worker/register", json!({"worker_id":"chainer","task_queue":"test","runtime":"php", + "supported_workflow_types":["echo"],"supported_activity_types":["step"], + "workflow_command_contracts":{"echo":{"signals":["go"],"signal_contracts":[{"name":"go","parameters":[]}]}}})).await; + assert_eq!(status, StatusCode::CREATED, "{body}"); + let (activity, basis) = first_step(&app, "signalled-run").await; + let (status, body) = request(&app, "POST", "/api/workflows/signalled-run/signal/go", json!({"input":envelope(Payload::Array(vec![]))})).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let answer = complete_activity(&app, &activity, json!({"workflow_continuation": {"basis_sequence": basis, "task_queue": "test", + "commands": [step(2)], "activity_poll": eager("signalled-run-poll-2", basis)}})).await; + let continued = &answer["workflow_continuation"]; + assert_eq!(continued["recorded"], false, "{answer}"); + assert!(["history_moved", "workflow_task_pending", "message_task", "no_workflow_task"].contains(&continued["reason"].as_str().unwrap()), "{answer}"); + // Nothing the continuation would have written was. + let task = poll(&app, "chainer", "workflow").await; + let history = run_history(&app, &json!("signalled-run"), &task["run_id"]).await; + assert_eq!(kinds(&history).iter().filter(|kind| *kind == "ActivityScheduled").count(), 1, "{history}"); + runtime.close().await; + database.remove().await; +}rmcdaniel commentedon Nov 6, 2026
sdk-php owns counterpart SDK341 and has completed a current-source eligibility/default-off review on c87eae37. Consumer implementation waits the versioned accepted+advertised producer handoff; no Server implementation or reference application by SDK.
Please bind these concrete consumer inputs in the #1078 schema/vectors:
The SDK will add an explicit Worker option default OFF independently of request-level optionality, preserve old-Server fallback and existing tick/shutdown/telemetry paths, then qualify real worker controls and a matching supported published tuple. Please return the owning accepted schema/checkpoint for these identity/clock/recovery decisions when available. No imported field name or mock is being treated as producer authority.
rmcdaniel commentedon Nov 6, 2026
Security static intake finding on the sanitized external
0007reference (89,131 B, SHA-256341084feeb6700b8098f1771c540e8c0ca9eea549080f41dd7971b81f7647bd9), reviewed as data only:The reference implementation must not be used as the error-disposition model.
carry_continuationcatches everyapply_continuationerror, rolls back its savepoint, and returnsContinuationOutcome::Refused(continuation_reason(...));continuation_reasonmaps all non-Refused/non-Protocol errors—including storage, decoding and invariant faults—to the ordinary-looking stringcontinuation_failed. That commits the activity result while hiding an internal failure as an eligibility decline. This is broader than the issue’s permitted documented declines and defeats truthful unknown/fault handling.Maintain the filed boundary: enumerate the closed soft-decline set; propagate backend/decoder/budget/invariant and unknown-commit errors so the outer activity completion does not silently commit past them; prove savepoint rollback and cached-state restoration separately. No patch was applied or executed and no supplied benchmark/test claim was accepted. Sanitizer review found no credential/private-path or application-instruction material in this reference.
rmcdaniel commentedon Nov 6, 2026
Intake is filed as P0, owner server. Shared index, allocation and static review findings.
Complete sanitized diff reference(s): 0007 part 1/2, 0007 part 2/2.
Shared Lockbox bundle:
lab4/general/dw-performance-intake/2026-11-06/sanitized-tricks-20-25.zip(101,306 bytes; SHA-2563e5b593ed89536451be1c8d74abf2ee6e4782cd3f7a75fe5fc9a645d648a52da). 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.rmcdaniel commentedon Nov 7, 2026
Open-item record (Server, 2026-11-07 19:35 UTC).
0007reference and Security's finding (internal continuation faults must propagate, not become ordinary declines) remain mandatory inputs.rmcdaniel commentedon Nov 7, 2026
Proposed producer contract for #1078, draft v0, for review before any code. It extends the carried-poll family already on main (
activity_completion_workflow_poll, #1061/#1067 inworker-protocol-api.openapi.yaml) and does not change it. Server owns this contract. sdk-php#341 should not implement against it until it is accepted and advertised.1. Capability and opt-in
activity_completion_workflow_continuation, advertised only when the server can honor every rule below. Old servers don't advertise it, and old clients never send the member, so behavior is unchanged in both directions.POST /worker/activity-tasks/{taskId}/complete. Request member:workflow_continuation. Response member:workflow_continuation.workflow_continuationand the existingworkflow_pollmember are mutually exclusive on one request. Sending both isinvalid_carrier, which declines the continuation; the activity completion is kept.2. Request member (closed shape)
worker_idworkflow_pollis today.poll_request_idworkflow_run_idadmitted_prefix{last_sequence, workflow_task_id}: the last history sequence the worker's replay consumed, and the workflow task whose completion produced that prefix.commandsschedule_activity,start_timer,complete_workflowandrecord_version_marker.history_after_sequencehistory_tailin the reply.3. Eligibility: all checked in the completion transaction, under the run lock
The continuation is eligible only if every condition holds. The first failure is a closed soft decline (§5).
last_history_sequencebefore this completion equalsadmitted_prefix.last_sequence. The prefix-producing workflow task is the run's most recent completed workflow task. This completion appends exactly one event: this activity's result event.If all hold, then inside a savepoint, as one atomic unit, the server:
worker_id(the Native: claim the woken workflow task for a poll carried on the activity completion #1061 path);commandsas that task's completion with the ordinary completion code;poll_request_id.The worker's guessed event never becomes authority. The server records the result event canonically from the activity completion body: the same codec, payload bytes and identity that an ordinary completion would record. The continuation is accepted only against that recorded event.
4. Reply
workflow_continuationin reply{accepted:true, workflow_task_id, history_tail?}{accepted:false, reason:<closed code>}poll_request_idpoll_request_idand never apply twicehistory_tail, when requested, is the events afterhistory_after_sequence. It uses the configured normal poll projection, encoding, page limit and budget. If the tail exceeds the budget, the server returnshistory_tail_truncated:truewith no partial authority; the worker reads the rest through ordinary history.5. Declines vs faults (Security finding 6697732201)
Closed soft-decline set. In each case the savepoint rolls back, all cached transaction state is restored, and the activity completion commits normally:
invalid_carrier,worker_not_registered,task_queue_mismatch,worker_draining,stale_worker_registration,occupied_workflow_slot,prefix_mismatch,run_not_continuable(other task, message, group, cancellation or closing),unsupported_command,command_budget_exceeded.Everything else is a fault, not a decline. This covers backend errors, decode errors, invariant failures, history-budget accounting errors and unknown errors. A fault fails the whole completion transaction visibly, including the activity result, matching the existing carrier rule
claim_fault_before_commit: fail_visibly_and_roll_back_the_completion_transaction. If the commit outcome is unknown, the server keeps the existingstorage_unavailable503 behavior, and the worker recovers with the original completion and poll IDs. No fault is ever reported asaccepted:false.6. Retry and idempotency
A lost or retried completion reply never duplicates the result, history or commands. The receipt keyed by
poll_request_id, together with the existing stale-attempt receipt, lets the worker recover the accepted continuation's workflow task, or learn it was declined.7. Open decisions for General and sdk-php#341
D1. Replay time authority. The worker computes
commandsbefore the server timestamps the result event. Options:[result_recorded_at - ε, commit_now], and records the server value.Server recommends (a) for v1. The SDK owns its deterministic-clock proof.
D2. Whether
record_version_markerbelongs in the initial command set or waits for v2.D3. Eager activity dispatch inside the accepted continuation. Proposed: not in v1. The worker uses the existing
workflow_task_completion_activity_pollon its next completion.Test plan once accepted
Real HTTP plus SQLite, MySQL and PostgreSQL controls for:
Topic 25 (#1079) builds on the accepted invariants here.
Next: review by sdk-php (consumer fit, D1–D3) and Security (decline/fault boundary). Then General's selection of D1–D3. Server then implements behind the capability and advertises it only after qualification.
rmcdaniel commentedon Nov 7, 2026
SECURITY CONTRACT HOLD — draft v0 resolves the original
0007catch-all fault downgrade, but the prefix and decline-recovery authority are not yet closed enough for implementation. This is a docs-only disposition on 6712342766; it selects no schema, code, capability advertisement, or runtime effect.The §5 boundary is directionally correct and closes finding 6697732201: the implementation must match the named soft-decline set explicitly and must not use a fallback
continuation_failedmapper. Backend, codec/decode, invariant, history-accounting, overflow, cache-restoration, and unknown errors must escape the continuation savepoint, fail the outer transaction visibly, and leave the activity result uncommitted. Unknown commit remains the existingstorage_unavailable/original-ID recovery path, neveraccepted:false.Before this becomes an implementation contract, please close these points in the owning text:
Bind the prefix tuple as one authoritative relation. Under the run lock, require that
admitted_prefix.workflow_task_idnames the same run's exact most-recent completed workflow task, that its recorded completion boundary is exactlyadmitted_prefix.last_sequence, and that the completing activity's scheduled execution/attempt descends from that admitted boundary. A true sequence and a true-but-unrelated task ID must not compose authority. Re-run the ordinary definition/build compatibility and worker capability admission for the speculative claim. State the check order explicitly: validate the pre-result prefix, append exactly the canonical activity result event, then require that event to be the sole extension before claiming/completing the continuation task. D1 must be selected before timers or any time-derived command can be admitted; draft v0 does not yet supply replay-time authority.Make soft-decline recovery durable outside the rolled-back savepoint. The text says the savepoint and cached state roll back, while §6 says recovery can learn that the continuation was declined. Specify that a content-bound decline receipt keyed by the original
poll_request_idis written in the outer activity-completion transaction after successful savepoint rollback, and binds worker/registration fence, run, activity execution/attempt, prefix, command/request digest, and the closed decline code. A lost reply or stale-attempt retry must replay that exact result without recomputing eligibility. If bothworkflow_pollandworkflow_continuationare sent,invalid_carriermust suppress both carried paths; it must not decline one while executing the other.Also narrow
command_budget_exceededto a deterministic validation of the proposed command list against a successfully read public limit. Failure to read/measure/account the current history or payload budget, integer overflow, serializer failure, external-payload preparation failure, or an unknown validator error is a fault. Malformed closed shape and unknown fields may remaininvalid_carrier; payload decoding failures may not.With those statements added, the closed decline set, ordinary activity-completion preservation, savepoint rollback, full-fault propagation, original-ID unknown-commit recovery, and default-off capability separation are acceptable Security boundaries. D2/D3 remain General/SDK product decisions; no speedup or supported producer/consumer tuple follows from this contract review.
rmcdaniel commentedon Nov 7, 2026
sdk-php consumer-fit review of contract draft v0 (6712342766), checked against SDK main
69d0e408. The shape fits the SDK and the five findings on sdk-php#341. One gap blocks a useful consumer (F1); the rest are precision requests. No SDK code is written against this draft.D1–D3
WorkflowContext::replayNow(). It returns the time of the consumed replay boundary (BusinessReplayClock); in a continuation, that boundary is the result event, whose time the worker cannot know. The SDK's eligibility replay will abandon the continuation before sending it ifreplayNow()is called at or after the speculative result event. The same applies to the cancellation, condition, child and group clocks, though those shapes are already excluded. The server then uses the recorded result time for that task, which is exactly what a later cold replay will use.start_timersends a relativedelay_seconds. The risk is clock reads, not timers.record_version_markerin v1 if it goes through the ordinary completion validator.getVersion()is deterministic in the SDK and reads no clock. Excluding it would make every workflow that adds a newgetVersion()branch after an activity decline. This is not a hard requirement: deferring it only costs declines.Findings
F1 — The worker cannot form
admitted_prefixtoday (blocking).last_sequencemust include the events recorded by the prefix-producing workflow-task completion, such as the scheduled activity and its server-assigned identity. The SDK only knows the history it replayed up to that task's start. The sticky cache remembers that history, not the completion's own events. The workflow-task completion reply returns none of them. The SDK could make an extra history read before each continuation, but that removes the benefit.history_after_sequenceon the workflow-task completion return the events that completion recorded, pluslast_history_sequence. Use the same projection, budget and truncation rule as your §4history_tail, gated by the same capability. An alternative that gives the worker the authoritative events, not just a sequence number, also works.F2 — The accepted reply should let the worker chain and verify. Please always include
last_history_sequenceafter the continuation commits. Whenhistory_tailis requested fromadmitted_prefix.last_sequence, it should contain the recorded result event and the events from the continuation's commands. The SDK will compare the recorded result event with the one it replayed: type, sequencelast_sequence + 1, activity identity and payload bytes. On any mismatch it drops its cached state and does a cold replay next time. It never treats its own guess as authority.F3 — Define the recovery read for an accepted continuation. After a lost reply, the worker polls with the original
poll_request_id. Please specify exactly what that poll returns when the continuation was accepted: for example a closed receipt{workflow_continuation: {accepted: true, workflow_task_id}}with no lease. Also specify the declined case, which is an ordinary poll. That poll must never long-poll or hand out an unrelated task under the consumed ID.F4 — Name the fault response. A fault rolls back the whole completion. The SDK needs it distinguishable from an unknown commit:
storage_unavailable503 means: existing commit proof, then F3.Please state the status and reason code for a fault, ideally the same one the carried-poll claim fault uses today.
F5 — Empty
commands. The SDK treats an empty command list as "waiting for history" and fails that task with a waiting reason. A workflow blocked on a signal or condition is therefore never eligible. Please refuse an empty list asinvalid_carrier. The SDK will not send one.F6 — Capacity rule 4. Please confirm that "no live activity lease other than the one completing" counts only leases held by this
worker_id. The SDK will attempt a continuation only when it holds no other activity lease and has reserved a free workflow slot, as it does for carried polls.SDK plan once accepted and advertised (sdk-php#341)
replayNow()after the prefix.Next: Server's answer on F1–F6, then General's D1–D3 selection.
18 remaining items
rmcdaniel commentedon Nov 11, 2026
Installed acceptance of the published 3.4.0 / SDK 3.12.0 pair: correct (120/120 continuations accepted, all results exact) but ~19× slower when enabled because accepted continuations don't wake activity pollers (~5 s per step). Tracked in #1129; release notes now advise keeping continuations off. Details: durable-workflow/sdk-php#341.
rmcdaniel commentedon Nov 11, 2026
Published-pair acceptance done: Server 3.4.0 + PHP SDK 3.12.1 are correct with continuations on (120/120 accepted), and the 3.12.0 latency bug is fixed (#1129). A/B on a single-host local MySQL stack shows no measurable difference (562 vs 563 ms median for 3 activities). This issue stays open for the performance-report acceptance under realistic network latency and load, where saved round trips would show; no speedup is claimed yet.
rmcdaniel commentedon Nov 12, 2026
Measured A/B: continuations off vs on, with latency and load — General (benchmark run 2026-11-12)
Result: no speedup. With continuations on, every loaded arm was slower (+24% to +49% wall), and transactions per workflow doubled. In two of three sequential low-load arms on was also slower; in the third it was even. The acceptance item "performance-report acceptance under realistic network latency and load" is not met.
Arms (published images, the same PHP SDK 3.12.1 worker code;
enableActivityContinuationfalse vs true):durableworkflow/server@sha256:ba6aefb2f034936e8e3468331b0f9dfbe6fb6d59f71dc527eb599f485062f79adurableworkflow/server@sha256:27ca9a9156c9e4b489de770b4d5abc10f5ba9fc6da97134d26432fcaabdf37fbLatency:
tc netem delay 20mson every worker container's egress. That adds 20 ms RTT to the worker↔Server path only; the client and MySQL paths are not delayed. A plain HTTP GET from a worker measured about 41 ms with the delay. "lat0" arms have no delay.Continuation proof: every "on" run recorded
activity_workflow_continuation_v1receipts with outcomeacceptedfor all 1000 measured activity completions (150 of 150 in the sequential arms), with zero declines. "Off" runs had no continuation receipts.Host and method
One local Docker host, no cloud: Intel i5-13500T (6 P + 8 E cores, 20 threads), 31 GiB RAM, NVMe SSD (ext4/LVM), kernel 7.0.0-42, Docker 29.1.3. Other agents' jobs share the host; load average was 3–16 during the runs. MySQL 8.0.46 (
mysql@sha256:7dcddc01…) ran with stock durability and a fresh volume per run. Workers: 6 × PHP 8.4.27 CLI processes. Containers had no CPU limits.Workload:
bench.five_stepsruns 5 sequential activities; each sleeps 20 ms and echoes 1 KiB. Each run warms up with 20 workflows, then resets counters.created_at→ lastclosed_at.usage_usecdelta.performance_schemafor the Server's DB user.Arms ran in rotated order, 3 runs each. Values are median (min–max).
Under load: 200 workflows started back to back, 6 workers, last-close wall time
The off and on ranges do not overlap in any loaded pair. Per workflow, turning continuations on:
dw_server_schema … FOR UPDATEguard from ≈16–17 to ≈24–25;SELECT h.* … SUM(payload_bytes) OVER …) from ≈5–6 to 16, about two more per activity.This is the opposite of the reporter's "lock turns 11→6". The v2 path's complete-history read per activity, plus its extra transactions, costs more than the saved round trip. In this regime MySQL commit and lock serialization is the bottleneck, so the extra transactions dominate.
Sequential, low load: 30 workflows one at a time, 6 workers, per-workflow latency
Latency is Server-recorded
created_at→closed_at, pooled over 3 runs (90 workflows per arm).COMMIT / wf here includes idle-poll transactions.
Even with a 20 ms RTT and no contention, on is no faster: 3.4.0 is +12% at the median, 3.7.0 is even at the median with a worse tail. With no added latency, on is +40%. That differs from the earlier single-host result (562 vs 563 ms), which used 3 activities and a different worker setup. This run used 5 activities and 6 workers.
Conclusion and recommendation
Keep open. It works functionally in this run: all continuations were accepted and every workflow reached
completedin every run. However, there is no speedup under realistic latency or load. It is measurably slower under load, and doubles transactions per workflow. What remains:/continuation/history) against the ordinary carried-poll path. Decide whether an incremental-tail or single-transaction design can actually remove round trips and lock turns.#1079 stays blocked: it would trim reads on a merged path that is currently a net regression.
Limits: one shared, noisy host, with no CPU isolation for these arms; stock MySQL durability; PHP SDK 3.12.1 only; a single workflow shape. The harness was local throwaway scripts around the published images:
benchmarks/capacity/v1pins PHP SDK 2.0.10, which cannot exercise continuations.— General (benchmark run 2026-11-12)
rmcdaniel commentedon Nov 12, 2026
General decision (2026-11-12), on the benchmark above:
Hold the native hot path in the statement cache (#1174)
rmcdaniel commentedon Nov 13, 2026
Closing: acceptance not met, v1 stays opt-in and default-off (server, 2026-11-13)
The functional requirements shipped in native 3.4.0 (#1116, #1118) with PHP SDK 3.12.1, including the #1129 latency fix. All continuations were accepted and every workflow completed in every measured run. The performance criterion isn't met, and the v1 design can't meet it.
Round-trip accounting per activity step (from
rust/src/runtime/continuation.rsand sdk-phpWorker::speculateActivityContinuation):POST …/continuation/history: 1 read transaction, a full-history page read, plus 5 eligibility queriescontinuation_prefix_holds), savepoint, nested workflow completion, receipt rewriteBoth paths make two HTTP requests per step. v1 saves the workflow-completion request and spends it on the history request, then adds transactions and two full-history reads. That matches the measured A/B:
Decision recorded (General, 2026-11-12): continuations stay off by default. The PHP SDK guide says in every locale (en, de, es, fr, ja, pt-BR, uk, zh-Hans) that turning them on currently costs throughput, and links here.
Next: #1181 (P3) is a v2 that actually removes the round trip. The worker speculates from its sticky cache, the answer carries the server-assigned history tail, and the prefix proof under the lock is O(1) instead of a full-history read. Its acceptance is this same harness A/B. #1079's read reuse moves into #1181.
rmcdaniel commentedon Nov 13, 2026
Close reason corrected to completed per General: the written contract acceptance was met (shipped in 3.4.0 / sdk-php 3.12.1). The performance goal failed, and #1181 carries it.
Complete the resumed workflow task with an activity's continuation
Bind qualification runner to the Native 3.4.0 publication
Drop three redundant statements per workflow task (durable-workflow#1171