Native: claim the woken workflow task for a poll carried on the activity completion #1061

Activity

1. rmcdaniel

   added

   kind:featureA public product capability or experience is requestedA public product capability or experience is requested

   priority:P0Immediate release or data-integrity riskImmediate release or data-integrity risk

   repo:serverOwned by the standalone server repositoryOwned by the standalone server repository

   agent:server

   on Nov 6, 2026
2. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0011-t10-Carry-the-next-workflow-poll-on-the-activity-complet.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: 13,995 bytes, SHA-256 db7c915cd8bb2df185db3a7f03fc983f8748fb9420999f8d5d3c36673fbda92d. This part: 13,995 bytes, SHA-256 db7c915cd8bb2df185db3a7f03fc983f8748fb9420999f8d5d3c36673fbda92d. Parts concatenate in order; a part boundary is between complete file diffs.

   diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs
   index 39affb00..bf713fa4 100644
   --- a/rust/src/runtime/execution.rs
   +++ b/rust/src/runtime/execution.rs
   @@ -1781,6 +1781,16 @@ impl Runtime {
            }
        }
        pub(crate) async fn complete_activity(&self, task_id: &str, body: Value) -> Result<Value> {
   +        self.complete_activity_with_workflow_poll(task_id, body, None).await
   +    }
   +    /// `complete_activity` with the worker's next workflow poll and its
   +    /// protocol version, which the completion's transaction claims for.
   +    pub(crate) async fn complete_activity_with_workflow_poll(
   +        &self,
   +        task_id: &str,
   +        body: Value,
   +        workflow_poll: Option<(Value, &str)>,
   +    ) -> Result<Value> {
            let external = self.prepare_external(&body["result"]).await?;
            // `ActivityOutcomeRecorder` measures the result it would store.
            let stored = match &external {
   @@ -1797,10 +1807,16 @@ impl Runtime {
                    Storage::MySql(store) => store.fail_activity(task_id, body, failure, blob, limit).await,
                };
            }
   +        let workflow_poll = workflow_poll.as_ref().map(|(poll, protocol)| super::store::CarriedWorkflowPoll {
   +            body: poll,
   +            protocol,
   +            pages: self.history_pages(),
   +            default_wait: self.worker_poll_wait(),
   +        });
            match &*self.storage {
   -            Storage::Sqlite(store) => store.complete_activity(task_id, body, external).await,
   -            Storage::Postgres(store) => store.complete_activity(task_id, body, external).await,
   -            Storage::MySql(store) => store.complete_activity(task_id, body, external).await,
   +            Storage::Sqlite(store) => store.complete_activity(task_id, body, external, workflow_poll).await,
   +            Storage::Postgres(store) => store.complete_activity(task_id, body, external, workflow_poll).await,
   +            Storage::MySql(store) => store.complete_activity(task_id, body, external, workflow_poll).await,
            }
        }
        delegate!(heartbeat_activity(task_id: &str, body: Value) -> Result<Value>);
   diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs
   index 1259f739..31ed8749 100644
   --- a/rust/src/runtime/http.rs
   +++ b/rust/src/runtime/http.rs
   @@ -2545,11 +2545,25 @@ async fn fail_workflow_task(
    async fn complete_activity(
        Extension(runtime): Extension<Runtime>,
        Path(task_id): Path<String>,
   -    Json(body): Json<Value>,
   +    headers: HeaderMap,
   +    Json(mut body): Json<Value>,
    ) -> Result<Json<Value>> {
   +    // The worker's next workflow poll, carried by the completion so its
   +    // transaction claims the workflow task the result makes ready. As with
   +    // the workflow task completion's activity poll, a poll the rules refuse,
   +    // or one for other task kinds, is left to the worker's own poll.
   +    let workflow_poll = body
   +        .as_object_mut()
   +        .and_then(|body| body.remove("workflow_poll"))
   +        .filter(|poll| {
   +            poll.is_object()
   +                && poll_rules(poll, WORKFLOW_POLL_RULES).is_ok()
   +                && super::update_validation::task_kinds(poll).is_ok_and(|kinds| kinds == ["workflow"])
   +        });
   +    let protocol = headers.get("x-durable-workflow-protocol-version").and_then(|value| value.to_str().ok());
        let attempt = body["activity_attempt_id"].clone();
        runtime
   -        .complete_activity(&task_id, body)
   +        .complete_activity_with_workflow_poll(&task_id, body, workflow_poll.zip(protocol))
            .await
            .map_err(|error| {
                completion_refusal(
   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index fa00a0d4..5f61260a 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -1976,6 +1976,13 @@ where
                }
                _ => None,
            };
   +        // A request whose answer an activity completion claimed, or a retry,
   +        // replays before the lock when the replay would write nothing.
   +        if let Some(request) = &poll_request
   +            && let Some(replay) = self.replay_workflow_poll_unlocked(request, text(&body, "worker_id")?, text(&body, "task_queue")?).await?
   +        {
   +            return replayed_poll(replay, encoding);
   +        }
            let mut cross_kind = if timeout > 0 {
                self.poll_wake_snapshot(&body, kind).await?
            } else {
   @@ -3922,12 +3929,17 @@ where
            task_id: &str,
            body: Value,
            external: Option<super::external_resolution::PreparedExternal>,
   +        workflow_poll: Option<CarriedWorkflowPoll<'_>>,
        ) -> Result<Value> {
            let result = match &external {
                Some(prepared) => prepared.stored(),
                None => envelope(&body, "result")?,
            };
            let attempt_id = text(&body, "activity_attempt_id")?;
   +        let workflow_poll = match workflow_poll {
   +            Some(poll) => self.admit_carried_workflow_poll(poll).await,
   +            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" {
   @@ -4062,11 +4074,54 @@ where
                )
                .await?
            };
   +        let workflow_poll = match &workflow_poll {
   +            Some(poll) => self.claim_carried_workflow(&mut tx, poll).await?,
   +            None => None,
   +        };
            tx.commit().await?;
            self.wake.notify_waiters();
   -        Ok(
   -            json!({"task_id": task_id, "activity_attempt_id": attempt_id, "outcome": "completed", "recorded": true, "reason": null, "next_task_id": next_task_id}),
   -        )
   +        let mut response = json!({"task_id": task_id, "activity_attempt_id": attempt_id, "outcome": "completed", "recorded": true, "reason": null, "next_task_id": next_task_id});
   +        if let Some(leased) = workflow_poll {
   +            response["workflow_poll"] = json!({"leased": leased});
   +        }
   +        Ok(response)
   +    }
   +
   +    /// 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
   +    /// not claimed for, and the worker's own poll answers it.
   +    async fn admit_carried_workflow_poll<'a>(&self, poll: CarriedWorkflowPoll<'a>) -> Option<AdmittedWorkflowPoll<'a>> {
   +        let worker_id = text(poll.body, "worker_id").ok()?;
   +        let queue = text(poll.body, "task_queue").ok()?;
   +        let id = super::workflow_poll_requests::request_id(poll.body)?;
   +        self.refuse_draining_worker(worker_id, queue).await.ok()?;
   +        let requested = super::history_pages::requested_size(poll.body).ok()?;
   +        let encoding = super::history_pages::requested(poll.body).ok()?;
   +        let fence = match self.admit_worker_poll(worker_id, queue).await.ok()? {
   +            super::worker_poll_fence::PollAdmission::Fenced(fence) => fence,
   +            _ => return None,
   +        };
   +        let request = self.workflow_poll_request(worker_id, queue, id, poll.default_wait, &["workflow"]).await.ok()?;
   +        Some(AdmittedWorkflowPoll { body: poll.body, protocol: poll.protocol, page: poll.pages.poll(requested, encoding), fence, request })
   +    }
   +
   +    /// The carried workflow poll's claim, in the activity completion's
   +    /// transaction, under the poll's request ID: the worker's poll with that
   +    /// ID replays the lease. A claim that fails is rolled back to before it,
   +    /// and the completion commits without it.
   +    async fn claim_carried_workflow(&self, tx: &mut Transaction<'_, DB>, poll: &AdmittedWorkflowPoll<'_>) -> Result<Option<bool>> {
   +        if !self.worker_poll_fence_holds_in(tx, &poll.fence, text(poll.body, "worker_id")?).await? {
   +            return Ok(None);
   +        }
   +        sqlx::raw_sql("SAVEPOINT workflow_poll").execute(&mut **tx).await?;
   +        match self.claim_in(tx, poll.body, "workflow", poll.protocol, poll.page, Some(&poll.request)).await {
   +            Ok(claimed) => Ok(Some(claimed.is_some())),
   +            Err(_) => {
   +                sqlx::raw_sql("ROLLBACK TO SAVEPOINT workflow_poll").execute(&mut **tx).await?;
   +                Ok(None)
   +            }
   +        }
        }
        pub(crate) async fn fire_due_timers(&self) -> Result<()> {
            // Expired root authority wins before any ordinary due timer fires.
   @@ -5383,6 +5438,24 @@ pub(super) fn reject_fields(body: &Value, allowed: &[&str]) -> Result<()> {
        Ok(())
    }
    
   +/// A workflow poll an activity completion carries: the poll's body and
   +/// protocol version, and the configuration its own poll would use.
   +pub(crate) struct CarriedWorkflowPoll<'a> {
   +    pub(crate) body: &'a Value,
   +    pub(crate) protocol: &'a str,
   +    pub(crate) pages: super::history_pages::HistoryPages,
   +    pub(crate) default_wait: u64,
   +}
   +
   +/// A carried workflow poll admitted before the completion's transaction.
   +struct AdmittedWorkflowPoll<'a> {
   +    body: &'a Value,
   +    protocol: &'a str,
   +    page: Option<i64>,
   +    fence: super::worker_poll_fence::WorkerPollFence,
   +    request: super::workflow_poll_requests::WorkflowPollRequest,
   +}
   +
    /// What a workflow poll's first probe under a request ID found.
    enum FirstProbe {
        /// The request's last answer, still deliverable.
   @@ -5414,7 +5487,7 @@ pub(crate) fn capabilities() -> Value {
            "workflow_updates": true, "query_tasks": true, "query_task_authority":true, "query_task_poll_request_idempotency": true,
            "activity_lease_renewal": true, "activity_cancellation_acknowledgement": true,
            "child_workflow_start_turn": true, "condition_wait_idempotent_reopen": true,
   -        "workflow_task_completion_activity_poll": true});
   +        "workflow_task_completion_activity_poll": true, "activity_completion_workflow_poll": 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/workflow_poll_requests.rs b/rust/src/runtime/workflow_poll_requests.rs
   index 95388f14..c5a25d42 100644
   --- a/rust/src/runtime/workflow_poll_requests.rs
   +++ b/rust/src/runtime/workflow_poll_requests.rs
   @@ -24,6 +24,9 @@ use sha2::{Digest, Sha256};
    // Distinct from the query poll prefix, so each sweep sees only its own rows.
    const PREFIX: &str = "dw-native-workflow-poll:";
    
   +// The task row a cached workflow lease is delivered from.
   +const WORKFLOW_LEASE: &str = "SELECT status,queue,lease_owner,lease_expires_at,attempt_count FROM workflow_tasks WHERE id=$1 AND task_type='workflow'";
   +
    pub(super) struct WorkflowPollKeys {
        binding: String,
        result: String,
   @@ -245,10 +248,12 @@ store_methods! {
            let Some(task_id) = task["task_id"].as_str().map(str::to_owned) else {
                return Ok(false);
            };
   -        let row = Self::query("SELECT status,queue,lease_owner,lease_expires_at,attempt_count FROM workflow_tasks WHERE id=$1 AND task_type='workflow'")
   -            .bind(&task_id)
   -            .fetch_optional(&mut **tx)
   -            .await?;
   +        let row = Self::query(WORKFLOW_LEASE).bind(&task_id).fetch_optional(&mut **tx).await?;
   +        Self::lease_row_deliverable(row, task, worker_id, queue)
   +    }
   +
   +    /// `workflow_lease_deliverable`'s judgment of the task row it read.
   +    fn lease_row_deliverable(row: Option<DB::Row>, task: &mut Value, worker_id: &str, queue: &str) -> Result<bool> {
            if let Some(row) = row
                && DB::string(&row, "status")? == "leased"
                && DB::optional_string(&row, "queue")?.as_deref() == Some(queue)
   @@ -297,6 +302,49 @@ store_methods! {
            Self::put_workflow_poll_row(tx, &request.keys.result, response, kept).await
        }
    
   +    /// `replay_workflow_poll` without the lock, when the replay would write
   +    /// nothing: the binding is live and holds this request's kinds, and the
   +    /// answer is live and deliverable. In any other state the replay binds,
   +    /// refuses or forgets, and `None` leaves that to the locked replay.
   +    pub(super) async fn replay_workflow_poll_unlocked(
   +        &self,
   +        request: &WorkflowPollRequest,
   +        worker_id: &str,
   +        queue: &str,
   +    ) -> Result<Option<Value>> {
   +        let Some(bound) = self.live_workflow_poll_row(&request.keys.binding).await? else {
   +            return Ok(None);
   +        };
   +        if bound["task_kinds"] != json!(request.kinds) {
   +            return Ok(None);
   +        }
   +        let Some(mut result) = self.live_workflow_poll_row(&request.keys.result).await? else {
   +            return Ok(None);
   +        };
   +        if !result["task"].is_object() {
   +            return Ok(Some(result));
   +        }
   +        if task_kind(&result) != "workflow" || !request.kinds.contains(&"workflow") {
   +            return Ok(None);
   +        }
   +        let Some(task_id) = result["task"]["task_id"].as_str().map(str::to_owned) else {
   +            return Ok(None);
   +        };
   +        let row = Self::query(WORKFLOW_LEASE).bind(&task_id).fetch_optional(&self.pool).await?;
   +        Ok(Self::lease_row_deliverable(row, &mut result["task"], worker_id, queue)?.then_some(result))
   +    }
   +
   +    /// A cache row read without the lock, if it has not expired.
   +    async fn live_workflow_poll_row(&self, key: &str) -> Result<Option<Value>> {
   +        let Some(row) = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&self.pool).await? else {
   +            return Ok(None);
   +        };
   +        if DB::number(&row, "expiration")? <= epoch() {
   +            return Ok(None);
   +        }
   +        Ok(Some(serde_json::from_str(&DB::string(&row, "value")?)?))
   +    }
   +
        /// Whether a concurrent duplicate already left an answer to replay.
        pub(super) async fn workflow_poll_answered(&self, keys: &WorkflowPollKeys) -> Result<bool> {
            Ok(match Self::query(DB::QUERY_CACHE_GET).bind(&keys.result).fetch_optional(&self.pool).await? {
3. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Security static review found two blocking controls missing from the reference implementation.

   1. The activity completion proves ownership with body.lease_owner, but admit_carried_workflow_poll independently trusts workflow_poll.worker_id. Worker credentials are tenant/role scoped rather than cryptographically bound to one worker ID. Worker A can therefore complete its activity while causing the same transaction to lease a workflow task and cache a replay answer for registered worker B. Require workflow_poll.worker_id == completion.lease_owner before admission and revalidate it inside the transaction with the task/fence. Add a cross-worker refusal control proving no foreign lease, binding, or result-cache row is written.

   2. claim_carried_workflow catches every Err from claim_in, rolls back the savepoint, and returns Ok(None). That includes unexpected backend, decoding, stored-shape, budget, and invariant failures, not only the documented soft refusals. An internal fault must not be reported as a successful completion with the carried poll silently omitted. Downgrade only an explicit, closed refusal class; propagate or explicitly surface unexpected failures. Add causal database-error and malformed-stored-state controls while preserving the contract that a normal poll refusal does not erase an already valid completion.

   This is a static finding only. The supplied patch was not applied or executed.
4. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized rebased reference for report topic 10a: 0011-t10-Carry-the-next-workflow-poll-on-the-activity-complet.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: 14,027 bytes, SHA-256 7d671325c50584b126ead2038390419748362d70a2709648da97d3543d98b6a3. Original-base sanitized identity: db7c915cd8bb2df185db3a7f03fc983f8748fb9420999f8d5d3c36673fbda92d. No patch was applied or executed during this update.

   diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs
   index 39affb00..bf713fa4 100644
   --- a/rust/src/runtime/execution.rs
   +++ b/rust/src/runtime/execution.rs
   @@ -1781,6 +1781,16 @@ impl Runtime {
            }
        }
        pub(crate) async fn complete_activity(&self, task_id: &str, body: Value) -> Result<Value> {
   +        self.complete_activity_with_workflow_poll(task_id, body, None).await
   +    }
   +    /// `complete_activity` with the worker's next workflow poll and its
   +    /// protocol version, which the completion's transaction claims for.
   +    pub(crate) async fn complete_activity_with_workflow_poll(
   +        &self,
   +        task_id: &str,
   +        body: Value,
   +        workflow_poll: Option<(Value, &str)>,
   +    ) -> Result<Value> {
            let external = self.prepare_external(&body["result"]).await?;
            // `ActivityOutcomeRecorder` measures the result it would store.
            let stored = match &external {
   @@ -1797,10 +1807,16 @@ impl Runtime {
                    Storage::MySql(store) => store.fail_activity(task_id, body, failure, blob, limit).await,
                };
            }
   +        let workflow_poll = workflow_poll.as_ref().map(|(poll, protocol)| super::store::CarriedWorkflowPoll {
   +            body: poll,
   +            protocol,
   +            pages: self.history_pages(),
   +            default_wait: self.worker_poll_wait(),
   +        });
            match &*self.storage {
   -            Storage::Sqlite(store) => store.complete_activity(task_id, body, external).await,
   -            Storage::Postgres(store) => store.complete_activity(task_id, body, external).await,
   -            Storage::MySql(store) => store.complete_activity(task_id, body, external).await,
   +            Storage::Sqlite(store) => store.complete_activity(task_id, body, external, workflow_poll).await,
   +            Storage::Postgres(store) => store.complete_activity(task_id, body, external, workflow_poll).await,
   +            Storage::MySql(store) => store.complete_activity(task_id, body, external, workflow_poll).await,
            }
        }
        delegate!(heartbeat_activity(task_id: &str, body: Value) -> Result<Value>);
   diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs
   index 1259f739..31ed8749 100644
   --- a/rust/src/runtime/http.rs
   +++ b/rust/src/runtime/http.rs
   @@ -2545,11 +2545,25 @@ async fn fail_workflow_task(
    async fn complete_activity(
        Extension(runtime): Extension<Runtime>,
        Path(task_id): Path<String>,
   -    Json(body): Json<Value>,
   +    headers: HeaderMap,
   +    Json(mut body): Json<Value>,
    ) -> Result<Json<Value>> {
   +    // The worker's next workflow poll, carried by the completion so its
   +    // transaction claims the workflow task the result makes ready. As with
   +    // the workflow task completion's activity poll, a poll the rules refuse,
   +    // or one for other task kinds, is left to the worker's own poll.
   +    let workflow_poll = body
   +        .as_object_mut()
   +        .and_then(|body| body.remove("workflow_poll"))
   +        .filter(|poll| {
   +            poll.is_object()
   +                && poll_rules(poll, WORKFLOW_POLL_RULES).is_ok()
   +                && super::update_validation::task_kinds(poll).is_ok_and(|kinds| kinds == ["workflow"])
   +        });
   +    let protocol = headers.get("x-durable-workflow-protocol-version").and_then(|value| value.to_str().ok());
        let attempt = body["activity_attempt_id"].clone();
        runtime
   -        .complete_activity(&task_id, body)
   +        .complete_activity_with_workflow_poll(&task_id, body, workflow_poll.zip(protocol))
            .await
            .map_err(|error| {
                completion_refusal(
   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index a6d3c211..0624bbb6 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -1984,6 +1984,13 @@ where
                }
                _ => None,
            };
   +        // A request whose answer an activity completion claimed, or a retry,
   +        // replays before the lock when the replay would write nothing.
   +        if let Some(request) = &poll_request
   +            && let Some(replay) = self.replay_workflow_poll_unlocked(request, text(&body, "worker_id")?, text(&body, "task_queue")?).await?
   +        {
   +            return replayed_poll(replay, encoding);
   +        }
            let mut cross_kind = if timeout > 0 {
                self.poll_wake_snapshot(&body, kind).await?
            } else {
   @@ -3935,12 +3942,17 @@ where
            task_id: &str,
            body: Value,
            external: Option<super::external_resolution::PreparedExternal>,
   +        workflow_poll: Option<CarriedWorkflowPoll<'_>>,
        ) -> Result<Value> {
            let result = match &external {
                Some(prepared) => prepared.stored(),
                None => envelope(&body, "result")?,
            };
            let attempt_id = text(&body, "activity_attempt_id")?;
   +        let workflow_poll = match workflow_poll {
   +            Some(poll) => self.admit_carried_workflow_poll(poll).await,
   +            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" {
   @@ -4075,11 +4087,54 @@ where
                )
                .await?
            };
   +        let workflow_poll = match &workflow_poll {
   +            Some(poll) => self.claim_carried_workflow(&mut tx, poll).await?,
   +            None => None,
   +        };
            tx.commit().await?;
            self.wake.notify_waiters();
   -        Ok(
   -            json!({"task_id": task_id, "activity_attempt_id": attempt_id, "outcome": "completed", "recorded": true, "reason": null, "next_task_id": next_task_id}),
   -        )
   +        let mut response = json!({"task_id": task_id, "activity_attempt_id": attempt_id, "outcome": "completed", "recorded": true, "reason": null, "next_task_id": next_task_id});
   +        if let Some(leased) = workflow_poll {
   +            response["workflow_poll"] = json!({"leased": leased});
   +        }
   +        Ok(response)
   +    }
   +
   +    /// 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
   +    /// not claimed for, and the worker's own poll answers it.
   +    async fn admit_carried_workflow_poll<'a>(&self, poll: CarriedWorkflowPoll<'a>) -> Option<AdmittedWorkflowPoll<'a>> {
   +        let worker_id = text(poll.body, "worker_id").ok()?;
   +        let queue = text(poll.body, "task_queue").ok()?;
   +        let id = super::workflow_poll_requests::request_id(poll.body)?;
   +        self.refuse_draining_worker(worker_id, queue).await.ok()?;
   +        let requested = super::history_pages::requested_size(poll.body).ok()?;
   +        let encoding = super::history_pages::requested(poll.body).ok()?;
   +        let fence = match self.admit_worker_poll(worker_id, queue).await.ok()? {
   +            super::worker_poll_fence::PollAdmission::Fenced(fence) => fence,
   +            _ => return None,
   +        };
   +        let request = self.workflow_poll_request(worker_id, queue, id, poll.default_wait, &["workflow"]).await.ok()?;
   +        Some(AdmittedWorkflowPoll { body: poll.body, protocol: poll.protocol, page: poll.pages.poll(requested, encoding), fence, request })
   +    }
   +
   +    /// The carried workflow poll's claim, in the activity completion's
   +    /// transaction, under the poll's request ID: the worker's poll with that
   +    /// ID replays the lease. A claim that fails is rolled back to before it,
   +    /// and the completion commits without it.
   +    async fn claim_carried_workflow(&self, tx: &mut Transaction<'_, DB>, poll: &AdmittedWorkflowPoll<'_>) -> Result<Option<bool>> {
   +        if !self.worker_poll_fence_holds_in(tx, &poll.fence, text(poll.body, "worker_id")?).await? {
   +            return Ok(None);
   +        }
   +        sqlx::raw_sql("SAVEPOINT workflow_poll").execute(&mut **tx).await?;
   +        match self.claim_in(tx, poll.body, "workflow", poll.protocol, poll.page, Some(&poll.request)).await {
   +            Ok(claimed) => Ok(Some(claimed.is_some())),
   +            Err(_) => {
   +                sqlx::raw_sql("ROLLBACK TO SAVEPOINT workflow_poll").execute(&mut **tx).await?;
   +                Ok(None)
   +            }
   +        }
        }
        pub(crate) async fn fire_due_timers(&self) -> Result<()> {
            // Expired root authority wins before any ordinary due timer fires.
   @@ -5397,6 +5452,24 @@ pub(super) fn reject_fields(body: &Value, allowed: &[&str]) -> Result<()> {
        Ok(())
    }
    
   +/// A workflow poll an activity completion carries: the poll's body and
   +/// protocol version, and the configuration its own poll would use.
   +pub(crate) struct CarriedWorkflowPoll<'a> {
   +    pub(crate) body: &'a Value,
   +    pub(crate) protocol: &'a str,
   +    pub(crate) pages: super::history_pages::HistoryPages,
   +    pub(crate) default_wait: u64,
   +}
   +
   +/// A carried workflow poll admitted before the completion's transaction.
   +struct AdmittedWorkflowPoll<'a> {
   +    body: &'a Value,
   +    protocol: &'a str,
   +    page: Option<i64>,
   +    fence: super::worker_poll_fence::WorkerPollFence,
   +    request: super::workflow_poll_requests::WorkflowPollRequest,
   +}
   +
    /// What a workflow poll's first probe under a request ID found.
    enum FirstProbe {
        /// The request's last answer, still deliverable.
   @@ -5428,7 +5501,7 @@ pub(crate) fn capabilities() -> Value {
            "workflow_updates": true, "query_tasks": true, "query_task_authority":true, "query_task_poll_request_idempotency": true,
            "claimed_task_ready_age": true, "activity_lease_renewal": true, "activity_cancellation_acknowledgement": true,
            "child_workflow_start_turn": true, "condition_wait_idempotent_reopen": true,
   -        "workflow_task_completion_activity_poll": true});
   +        "workflow_task_completion_activity_poll": true, "activity_completion_workflow_poll": 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/workflow_poll_requests.rs b/rust/src/runtime/workflow_poll_requests.rs
   index 95388f14..c5a25d42 100644
   --- a/rust/src/runtime/workflow_poll_requests.rs
   +++ b/rust/src/runtime/workflow_poll_requests.rs
   @@ -24,6 +24,9 @@ use sha2::{Digest, Sha256};
    // Distinct from the query poll prefix, so each sweep sees only its own rows.
    const PREFIX: &str = "dw-native-workflow-poll:";
    
   +// The task row a cached workflow lease is delivered from.
   +const WORKFLOW_LEASE: &str = "SELECT status,queue,lease_owner,lease_expires_at,attempt_count FROM workflow_tasks WHERE id=$1 AND task_type='workflow'";
   +
    pub(super) struct WorkflowPollKeys {
        binding: String,
        result: String,
   @@ -245,10 +248,12 @@ store_methods! {
            let Some(task_id) = task["task_id"].as_str().map(str::to_owned) else {
                return Ok(false);
            };
   -        let row = Self::query("SELECT status,queue,lease_owner,lease_expires_at,attempt_count FROM workflow_tasks WHERE id=$1 AND task_type='workflow'")
   -            .bind(&task_id)
   -            .fetch_optional(&mut **tx)
   -            .await?;
   +        let row = Self::query(WORKFLOW_LEASE).bind(&task_id).fetch_optional(&mut **tx).await?;
   +        Self::lease_row_deliverable(row, task, worker_id, queue)
   +    }
   +
   +    /// `workflow_lease_deliverable`'s judgment of the task row it read.
   +    fn lease_row_deliverable(row: Option<DB::Row>, task: &mut Value, worker_id: &str, queue: &str) -> Result<bool> {
            if let Some(row) = row
                && DB::string(&row, "status")? == "leased"
                && DB::optional_string(&row, "queue")?.as_deref() == Some(queue)
   @@ -297,6 +302,49 @@ store_methods! {
            Self::put_workflow_poll_row(tx, &request.keys.result, response, kept).await
        }
    
   +    /// `replay_workflow_poll` without the lock, when the replay would write
   +    /// nothing: the binding is live and holds this request's kinds, and the
   +    /// answer is live and deliverable. In any other state the replay binds,
   +    /// refuses or forgets, and `None` leaves that to the locked replay.
   +    pub(super) async fn replay_workflow_poll_unlocked(
   +        &self,
   +        request: &WorkflowPollRequest,
   +        worker_id: &str,
   +        queue: &str,
   +    ) -> Result<Option<Value>> {
   +        let Some(bound) = self.live_workflow_poll_row(&request.keys.binding).await? else {
   +            return Ok(None);
   +        };
   +        if bound["task_kinds"] != json!(request.kinds) {
   +            return Ok(None);
   +        }
   +        let Some(mut result) = self.live_workflow_poll_row(&request.keys.result).await? else {
   +            return Ok(None);
   +        };
   +        if !result["task"].is_object() {
   +            return Ok(Some(result));
   +        }
   +        if task_kind(&result) != "workflow" || !request.kinds.contains(&"workflow") {
   +            return Ok(None);
   +        }
   +        let Some(task_id) = result["task"]["task_id"].as_str().map(str::to_owned) else {
   +            return Ok(None);
   +        };
   +        let row = Self::query(WORKFLOW_LEASE).bind(&task_id).fetch_optional(&self.pool).await?;
   +        Ok(Self::lease_row_deliverable(row, &mut result["task"], worker_id, queue)?.then_some(result))
   +    }
   +
   +    /// A cache row read without the lock, if it has not expired.
   +    async fn live_workflow_poll_row(&self, key: &str) -> Result<Option<Value>> {
   +        let Some(row) = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&self.pool).await? else {
   +            return Ok(None);
   +        };
   +        if DB::number(&row, "expiration")? <= epoch() {
   +            return Ok(None);
   +        }
   +        Ok(Some(serde_json::from_str(&DB::string(&row, "value")?)?))
   +    }
   +
        /// Whether a concurrent duplicate already left an answer to replay.
        pub(super) async fn workflow_poll_answered(&self, keys: &WorkflowPollKeys) -> Result<bool> {
            Ok(match Self::query(DB::QUERY_CACHE_GET).bind(&keys.result).fetch_optional(&self.pool).await? {
5. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Security follow-up on the updated rebased reference 0011-t10-Carry-the-next-workflow-poll-on-the-activity-complet.patch (14,027 B / SHA-256 7d671325c50584b126ead2038390419748362d70a2709648da97d3543d98b6a3): both blockers in 6692716653 remain.

   The rebase updates surrounding maintained capability context, but admit_carried_workflow_poll still trusts workflow_poll.worker_id independently of the completed activity's lease_owner; no equality is revalidated with the task/fence inside the transaction. claim_carried_workflow also still catches every Err(_) from claim_in and returns Ok(None), so unexpected database, decoding, stored-state, budget and invariant faults can still masquerade as a successful completion with an omitted carried result.

   The updated reference therefore still needs the cross-worker no-lease/no-binding/no-cache control and a closed documented soft-refusal class with unexpected faults propagated or explicitly surfaced. Static review only; no patch was applied or executed.
6. rmcdaniel commented on Nov 7, 2026

   

   rmcdaniel

   on Nov 7, 2026

   MemberAuthor

   More actions

   Accepted producer checkpoint: PR1073, accepted head 511bd6fd, normally merged 54af4640. Its contract is included in the accepted 4fb runtime and now the immutable published native 3.2.0 source/tag 3c6f1eda9c6d2ebc0ee5bec8ccf431900052be9e. The earlier reference-draft same-worker/error findings are superseded by that implementation and its actual SQLite/PostgreSQL/MySQL/MariaDB Source controls. They are not remaining implementation holds on the accepted producer.

   The authoritative OpenAPI schema 63 and closed replay vectors retain protocol 1.20. Schema SHA-256 is d6c726a7f97417494d9897ae1101ac7300ba6414174041dea875ae01e5052391; vectors SHA-256 12ca5db29d96b5e0f03852b698ca1d15bd796c07486a1af85484d4e4dfc45e9b.

   Capability activity_completion_workflow_poll admits the optional workflow_poll on activity completion, under that completion's owner and retained registration fence. A live activity slot with known or unknown expiry declines the opposite-kind claim. Malformed/unsupported admission and the closed soft-refusal table are separate from unexpected internal/backend errors, which remain visible and cannot masquerade as an empty claim. Duplicate and uncertain completion recovery retains original completion and poll identities.

   The actual publisher input is now available: owner terminal PASS, independent Security publication PASS, index ghcr.io/durable-workflow/server@sha256:6916f6f3bb1ee58d071f10f888068fd321cd1576fb70dc97ce8e9e61da493d62, exact producer/native runner 3c6. This issue stays open for actual published qualification. Server next owns the existing corpus/four-cell Compose receipt packet 325/6705614909. General next binds the actual published SDK package and SDK runner plus the bounded invocation to this producer; SDK-PHP owns its existing consumer/harness execution and result. The shared consumer handoff is SDK293/6705418888. Installed clock/transit/buffer-age, missing-key capacity, lost-answer/original-ID recovery, expiry/refence and selected load acceptance remain open. No tentative/withdrawn head, Source fixture or older tuple supplies that result.
7. rmcdaniel commented on Nov 9, 2026

   

   rmcdaniel

   on Nov 9, 2026

   Last edited by rmcdaniel

   MemberAuthor

   More actions

   General reconciled the carried-poll and workflow-clock delivery against the original acceptance criteria while both repository agents are Off.

   The producer is published Native Server 3.3.0, source 2a15179f29801eec9f2a2d4e6a9c5f93233f104e, index sha256:cb4b11bd6235a551cf51e7dc6342504fbbcef19959618b376fb5648ca00fc388. The consumer is normal published PHP SDK 3.11.0, source fbc64ef98c27be467d75a29a29bfb0ee4e80dba7.

   Feature-specific acceptance is covered by the maintained source controls. The carried-poll runtime tests, HTTP execution tests, workflow replay/history-budget tests and versioned OpenAPI are byte-identical between the qualified integrated source 5f937ec8 (run 42448138571) and the 3.3.0 cut. The source lanes execute the registered controls on SQLite, MySQL, MariaDB and PostgreSQL; the accepted #1073 source also has 62 successful checks and its declared skip.

   Published compatibility is now verified with the unchanged full PHP replay matrix: 42531701411, 13 executed scenarios passed, zero skips, exact Server 3.3.0 and normal Packagist SDK 3.11.0, distinct clients, restarted worker, completed-history replay/query, in-flight signal/restart and nondeterminism refusal. Captured audit. Its runner bootstrap fix has complete ordinary main custody: signed reviewed tree, 71 successful main checks/eight expected skips and topic absence verified. The published product bytes and feature-specific controls are unchanged.

   The source controls establish the transaction/receipt/answer invariants. The published matrix establishes producer/consumer compatibility; it is not an installed SQL-count or benchmark experiment. No speedup is claimed. #1073's earlier archive-acquisition NONPASSs and the first published replay's interpreter-bootstrap NONPASS remain attributed.

   Acceptance: qualify_carried_workflow_polls checks the completion's workflow claim and the same receipt/task/attempt/history/ready-age on ordinary replay without another transition. Refused carriers preserve a valid recorded completion and the ready task. The optional field, capability and response are documented. Published SDK 3.11.0 contains the accepted #329 consumer.

   The separate condition-wait durable-workflow/sdk-php#269 final-audit C2/C5 follow-up remains open; it is not a carried-poll or workflow-clock acceptance claim. The original reporter benchmark remains unverified.
8. rmcdaniel

   closed this as completedon Nov 9, 2026
9. rmcdaniel

   added

   agent:general

   and removed

   agent:server

   on Nov 9, 2026