Native: include server_time in leased workflow poll responses #1053

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 0001-t03+t02a-Keep-the-poll-path-s-queries-off-the-size-of-the-tab.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: 17,026 bytes, SHA-256 78a6762fdaaf4e9622b1aace4b93aff98d13eee6ccff9081e4b852a443a690e1. This part: 17,026 bytes, SHA-256 78a6762fdaaf4e9622b1aace4b93aff98d13eee6ccff9081e4b852a443a690e1. Parts concatenate in order; a part boundary is between complete file diffs.

   diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
   index 02735d70..aaf916f0 100644
   --- a/rust/src/runtime/backend.rs
   +++ b/rust/src/runtime/backend.rs
   @@ -84,8 +84,8 @@ pub(super) trait Backend: Database {
        // cutover of 3.0.1 query rows (Server #964).
        const QUERY_CACHE_PAGE: &str = "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\">$1 ORDER BY \"key\" LIMIT 64";
        // A bounded batch of expired per-request query poll rows.
   -    const QUERY_POLL_EXPIRED: &str = "SELECT \"key\" FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query-poll:%' AND expiration<=$1 ORDER BY \"key\" LIMIT 32";
   -    const WORKFLOW_POLL_EXPIRED: &str = "SELECT \"key\" FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-workflow-poll:%' AND expiration<=$1 ORDER BY \"key\" LIMIT 32";
   +    const QUERY_POLL_EXPIRED: &str = "SELECT \"key\" FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query-poll:%' AND expiration<=$1 ORDER BY \"key\" LIMIT 256";
   +    const WORKFLOW_POLL_EXPIRED: &str = "SELECT \"key\" FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-workflow-poll:%' AND expiration<=$1 ORDER BY \"key\" LIMIT 256";
        /// Add a cache row, or lock the existing one with a no-op update, so
        /// transactions over one poll request ID run one at a time.
        /// A worker session's running activity attempts held by its owner,
   @@ -243,7 +243,7 @@ WITH utc_tasks AS (
        SELECT t.id,t.namespace,t.queue,t.workflow_run_id,t.status,t.available_at,
            replace(replace(replace(t.available_at,'T',' '),'Z',''),'+00:00','') AS available_utc,
            replace(replace(replace(t.lease_expires_at,'T',' '),'Z',''),'+00:00','') AS lease_utc
   -    FROM workflow_tasks t WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type!=$3 AND t.task_type IN ('workflow','activity')
   +    FROM workflow_tasks t WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type!=$3 AND t.task_type IN ('workflow','activity') AND t.status IN ('ready','leased')
    ), ordered_tasks AS (
        SELECT *,
            rtrim(rtrim(CASE WHEN length(available_utc)=19 THEN available_utc||'.0' ELSE available_utc END,'0'),'.') AS available_order,
   @@ -296,7 +296,7 @@ WITH utc_tasks AS (
        SELECT t.*,
            replace(replace(replace(t.available_at,'T',' '),'Z',''),'+00:00','') AS _dw_available_utc,
            replace(replace(replace(t.lease_expires_at,'T',' '),'Z',''),'+00:00','') AS _dw_lease_utc
   -    FROM workflow_tasks t WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type=$3
   +    FROM workflow_tasks t WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type=$3 AND t.status IN ('ready','leased')
    ), ordered_tasks AS (
        SELECT *,
            rtrim(rtrim(CASE WHEN length(_dw_available_utc)=19 THEN _dw_available_utc||'.0' ELSE _dw_available_utc END,'0'),'.') AS _dw_available_order,
   @@ -579,8 +579,8 @@ impl Backend for MySql {
        const QUERY_CACHE_DELETE: &str = "DELETE FROM dw_query_cache WHERE `key`=$1";
        const QUERY_CACHE_DELETE_EXPIRED: &str = "DELETE FROM dw_query_cache WHERE `key`=$1 AND expiration<=$2";
        const QUERY_CACHE_PAGE: &str = "SELECT `key`,value,expiration FROM dw_query_cache WHERE `key`>$1 ORDER BY `key` LIMIT 64";
   -    const QUERY_POLL_EXPIRED: &str = "SELECT `key` FROM dw_query_cache WHERE `key` LIKE 'dw-native-query-poll:%' AND expiration<=$1 ORDER BY `key` LIMIT 32";
   -    const WORKFLOW_POLL_EXPIRED: &str = "SELECT `key` FROM dw_query_cache WHERE `key` LIKE 'dw-native-workflow-poll:%' AND expiration<=$1 ORDER BY `key` LIMIT 32";
   +    const QUERY_POLL_EXPIRED: &str = "SELECT `key` FROM dw_query_cache WHERE `key` LIKE 'dw-native-query-poll:%' AND expiration<=$1 ORDER BY `key` LIMIT 256";
   +    const WORKFLOW_POLL_EXPIRED: &str = "SELECT `key` FROM dw_query_cache WHERE `key` LIKE 'dw-native-workflow-poll:%' AND expiration<=$1 ORDER BY `key` LIMIT 256";
        const QUERY_CACHE_LOCK: &str = "INSERT INTO dw_query_cache(`key`,value,expiration) VALUES ($1,$2,$3) ON DUPLICATE KEY UPDATE expiration=expiration";
        const SESSION_RUNNING_ATTEMPTS: &str = "SELECT COUNT(*) FROM activity_attempts p JOIN activity_executions e ON e.id=p.activity_execution_id JOIN workflow_tasks t ON t.id=p.workflow_task_id WHERE t.namespace=? AND p.status='running' AND p.lease_expires_at>? AND p.lease_owner=? AND JSON_UNQUOTE(JSON_EXTRACT(e.activity_options,'$.worker_session.session_id'))=? AND p.workflow_task_id<>?";
        const TASK_KIND_CURSOR_LOCK: &str = "INSERT INTO workflow_task_poll_cursors(namespace,task_queue,next_task_kind,created_at,updated_at) VALUES ($1,$2,'update_validation',$3,$4) ON DUPLICATE KEY UPDATE next_task_kind=next_task_kind";
   diff --git a/rust/src/runtime/cross_kind_poll.rs b/rust/src/runtime/cross_kind_poll.rs
   index 2db96437..0a42230d 100644
   --- a/rust/src/runtime/cross_kind_poll.rs
   +++ b/rust/src/runtime/cross_kind_poll.rs
   @@ -8,7 +8,7 @@ pub(super) struct PollWakeSnapshot {
        ready: Vec<String>,
    }
    
   -pub(super) const TASKS_SQL: &str = "SELECT t.id,t.namespace,t.queue FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type!=$3 AND t.task_type IN ('workflow','activity') AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) ORDER BY t.available_at,t.id LIMIT 100";
   +pub(super) const TASKS_SQL: &str = "SELECT t.id,t.namespace,t.queue FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type!=$3 AND t.task_type IN ('workflow','activity') AND t.status IN ('ready','leased') AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) ORDER BY t.available_at,t.id LIMIT 100";
    
    impl<DB: Backend> Store<DB>
    where
   diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs
   index 01ccecb1..be36228b 100644
   --- a/rust/src/runtime/execution.rs
   +++ b/rust/src/runtime/execution.rs
   @@ -1229,9 +1229,12 @@ impl Runtime {
            let kinds = super::update_validation::task_kinds(&body)?;
            let pages = self.history_pages();
            let wait = self.worker_poll_wait();
   +        // A leased answer carries `server_time`, as an activity poll does, so
   +        // a worker can tell how much of the lease is left without renewing.
            let mark = |mut answer: Value| {
                if let Some(task) = answer["task"].as_object_mut() {
                    task.insert("task_kind".into(), json!("workflow"));
   +                answer["server_time"] = json!(super::activity_lease_renewal::micros(super::store::now()));
                }
                answer
            };
   diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs
   index d98a1882..5031bc5a 100644
   --- a/rust/src/runtime/mod.rs
   +++ b/rust/src/runtime/mod.rs
   @@ -155,6 +155,7 @@ mod start_request;
    mod message_streams;
    mod storage_test;
    mod store;
   +mod sweep_throttle;
    mod system_metrics;
    mod system_health;
    mod task_failures;
   diff --git a/rust/src/runtime/query_poll_requests.rs b/rust/src/runtime/query_poll_requests.rs
   index 00a381ff..1ec35415 100644
   --- a/rust/src/runtime/query_poll_requests.rs
   +++ b/rust/src/runtime/query_poll_requests.rs
   @@ -119,7 +119,7 @@ where
        }
    
        // Workers send a fresh request ID with every poll, so retire a bounded
   -    // batch of expired poll rows whenever a new request arrives.
   +    // batch of expired poll rows at most once a second as new requests arrive.
        async fn sweep_poll_rows(tx: &mut Transaction<'_, DB>) -> Result<()> {
            let expired = Self::query(DB::QUERY_POLL_EXPIRED)
                .bind(now())
   @@ -165,7 +165,9 @@ where
        // of an older request never takes that place back.
        async fn mark_current_if_fresh(&self, keys: &PollKeys, config: &QueryTaskConfig) -> Result<()> {
            let mut tx = self.begin().await?;
   -        Self::sweep_poll_rows(&mut tx).await?;
   +        if self.sweeps.query_poll_rows.take() {
   +            Self::sweep_poll_rows(&mut tx).await?;
   +        }
            if Self::poll_row(&mut tx, &keys.pending).await?.is_none()
                && Self::poll_row(&mut tx, &keys.result).await?.is_none()
            {
   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index a99bdaa0..54088ee2 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -93,7 +93,7 @@ pub(super) struct SignalWrite<'a> {
    }
    /// The runtime-owned message-stream delivery signal (`WorkflowStub::MESSAGE_STREAM_RUNTIME_SIGNAL`).
    const MESSAGE_STREAM_SIGNAL: &str = super::message_streams::INTERNAL_SIGNAL;
   -pub(super) const TASK_CANDIDATES_SQL: &str = "SELECT t.*,r.workflow_type FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type=$3 AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) AND (COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,'')) IS NULL OR COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,''))=$6) ORDER BY t.priority,t.available_at,t.id LIMIT 100";
   +pub(super) const TASK_CANDIDATES_SQL: &str = "SELECT t.*,r.workflow_type FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type=$3 AND t.status IN ('ready','leased') AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) AND (COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,'')) IS NULL OR COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,''))=$6) ORDER BY t.priority,t.available_at,t.id LIMIT 100";
    /// A signal record's place in `PendingMessageTask::signalFollows`: its
    /// command sequence (unsequenced last), receipt and creation times, id.
    struct SignalOrder {
   @@ -167,6 +167,8 @@ pub(super) struct Store<DB: Backend> {
        /// When this server last ran a task repair pass, shared by its namespaces
        /// as PHP shares its loop throttle key.
        pub(super) repair_pass_at: Arc<std::sync::Mutex<Option<std::time::Instant>>>,
   +    /// When this server next retires expired poll rows and receipts.
   +    pub(super) sweeps: Arc<super::sweep_throttle::Sweeps>,
    }
    /// The `closed_reason` a terminal history event gives its run, or `None` for
    /// an event that does not close the run.
   @@ -314,6 +316,7 @@ where
                workflow_task_recovery_scan_limit: 5,
                repair_codec: Arc::new(Semaphore::new(1)),
                repair_pass_at: Arc::default(),
   +            sweeps: Arc::default(),
            }
        }
        pub(super) fn in_namespace(&self, namespace: &str) -> Self {
   @@ -332,6 +335,7 @@ where
                workflow_task_recovery_scan_limit: self.workflow_task_recovery_scan_limit,
                repair_codec: self.repair_codec.clone(),
                repair_pass_at: self.repair_pass_at.clone(),
   +            sweeps: self.sweeps.clone(),
            }
        }
        pub(super) async fn database_live(&self) -> bool {
   @@ -2169,11 +2173,13 @@ where
            if DB::string(&worker, "task_queue")? != queue {
                return Err(refuse(StatusCode::CONFLICT, "task_queue_mismatch"));
            }
   -        Self::query(DB::POLL_RECEIPT_CLEANUP_SQL)
   -            .bind(DB::bind_time(now()))
   -            .bind(DB::bind_time(now()))
   -            .execute(&mut **tx)
   -            .await?;
   +        if self.sweeps.poll_receipts.take() {
   +            Self::query(DB::POLL_RECEIPT_CLEANUP_SQL)
   +                .bind(DB::bind_time(now()))
   +                .bind(DB::bind_time(now()))
   +                .execute(&mut **tx)
   +                .await?;
   +        }
            // A concurrent duplicate may have leased since this poll began: the
            // binding row, locked here, makes it wait for that one's answer. Only
            // a workflow answer replays as a workflow claim; a multiplexed probe
   @@ -2196,12 +2202,22 @@ where
            if let Some(request_id) = request_id
                && let Some(receipt) = Self::query("SELECT p.*,t.status,t.attempt_count,t.lease_owner,t.lease_expires_at FROM dw_poll_receipts p JOIN workflow_tasks t ON t.id=p.task_id WHERE p.worker_id=$1 AND p.kind=$2 AND p.request_id=$3 AND p.namespace=$4")
                    .bind(worker_id).bind(kind).bind(request_id).bind(self.namespace.as_ref()).fetch_optional(&mut **tx).await? {
   -                if DB::string(&receipt,"status")? == "leased" && DB::number(&receipt,"attempt")? == DB::number(&receipt,"attempt_count")?
   +            let at = now();
   +            let leased = DB::string(&receipt,"status")? == "leased";
   +            let lease_expires_at = DB::optional_instant(&receipt,"lease_expires_at")?;
   +            // The receipt cleanup's condition: a receipt it would have deleted
   +            // is absent, whether or not this poll ran the cleanup.
   +            if DB::instant(&receipt,"expires_at")? <= at && (!leased || lease_expires_at.is_some_and(|expires| expires <= at)) {
   +                Self::query("DELETE FROM dw_poll_receipts WHERE worker_id=$1 AND kind=$2 AND request_id=$3 AND namespace=$4")
   +                    .bind(worker_id).bind(kind).bind(request_id).bind(self.namespace.as_ref()).execute(&mut **tx).await?;
   +            } else {
   +                if leased && DB::number(&receipt,"attempt")? == DB::number(&receipt,"attempt_count")?
                        && DB::optional_string(&receipt,"lease_owner")?.as_deref() == Some(worker_id)
   -                    && DB::optional_instant(&receipt,"lease_expires_at")?.is_some_and(|expires| expires > now()) {
   +                    && lease_expires_at.is_some_and(|expires| expires > now()) {
                        return Ok(Some(DB::document_row(&receipt,"response")?));
                    }
                    return Err(refuse(StatusCode::CONFLICT,"poll_cached_task_expired"));
   +            }
            }
            let types: Value = DB::document_row(
                &worker,
   diff --git a/rust/src/runtime/sweep_throttle.rs b/rust/src/runtime/sweep_throttle.rs
   new file mode 100644
   index 00000000..a0bb3546
   --- /dev/null
   +++ b/rust/src/runtime/sweep_throttle.rs
   @@ -0,0 +1,42 @@
   +//! Housekeeping deletes that once ran on every poll, now at most once a
   +//! second per server. Each retires rows its readers already treat as absent:
   +//! an expired poll row reads as missing, and the claim forgets a receipt the
   +//! cleanup would delete. Skipping a sweep only leaves dead rows a little
   +//! longer, while each sweep that runs scans under the transition lock.
   +use std::sync::atomic::{AtomicI64, Ordering};
   +
   +#[derive(Default)]
   +pub(super) struct Sweeps {
   +    pub(super) workflow_poll_rows: Due,
   +    pub(super) query_poll_rows: Due,
   +    pub(super) poll_receipts: Due,
   +}
   +
   +/// When a sweep is next due, in epoch milliseconds.
   +#[derive(Default)]
   +pub(super) struct Due(AtomicI64);
   +
   +const INTERVAL_MILLIS: i64 = 1000;
   +
   +impl Due {
   +    /// Whether this caller runs the sweep: the first to ask once it is due.
   +    pub(super) fn take(&self) -> bool {
   +        let now = chrono::Utc::now().timestamp_millis();
   +        let due = self.0.load(Ordering::Relaxed);
   +        now >= due && self.0.compare_exchange(due, now + INTERVAL_MILLIS, Ordering::Relaxed, Ordering::Relaxed).is_ok()
   +    }
   +}
   +
   +#[cfg(test)]
   +mod tests {
   +    use super::*;
   +
   +    #[test]
   +    fn one_caller_a_second_sweeps() {
   +        let due = Due::default();
   +        assert!(due.take());
   +        assert!(!due.take());
   +        due.0.store(chrono::Utc::now().timestamp_millis(), Ordering::Relaxed);
   +        assert!(due.take());
   +    }
   +}
   diff --git a/rust/src/runtime/workflow_poll_requests.rs b/rust/src/runtime/workflow_poll_requests.rs
   index 2f290218..4eb5d1a5 100644
   --- a/rust/src/runtime/workflow_poll_requests.rs
   +++ b/rust/src/runtime/workflow_poll_requests.rs
   @@ -140,7 +140,9 @@ store_methods! {
            queue: &str,
        ) -> Result<Option<Value>> {
            let mut tx = self.begin().await?;
   -        Self::sweep_workflow_poll_rows(&mut tx).await?;
   +        if self.sweeps.workflow_poll_rows.take() {
   +            Self::sweep_workflow_poll_rows(&mut tx).await?;
   +        }
            Self::bind_workflow_poll(&mut tx, request).await?;
            let replay = self.deliverable_workflow_poll(&mut tx, request, worker_id, queue).await?;
            tx.commit().await?;
   @@ -326,7 +328,7 @@ store_methods! {
        }
    
        // Workers send a fresh request ID with most polls, so retire a bounded
   -    // batch of expired rows whenever a request arrives.
   +    // batch of expired rows at most once a second as requests arrive.
        async fn sweep_workflow_poll_rows(tx: &mut Transaction<'_, DB>) -> Result<()> {
            for row in Self::query(DB::WORKFLOW_POLL_EXPIRED).bind(epoch()).fetch_all(&mut **tx).await? {
                Self::query(DB::QUERY_CACHE_DELETE).bind(DB::string(&row, "key")?).execute(&mut **tx).await?;
3. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0012-t02a-Let-a-replayed-workflow-poll-carry-the-clock-of-its-.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: 2,934 bytes, SHA-256 a6c90a60b637d44652ef7cd6b8ade4b74d7d03ebab621dc3594706990dc4e204. This part: 2,934 bytes, SHA-256 a6c90a60b637d44652ef7cd6b8ade4b74d7d03ebab621dc3594706990dc4e204. Parts concatenate in order; a part boundary is between complete file diffs.

   diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs
   index e8e7046d..aec2c319 100644
   --- a/rust/tests/execution.rs
   +++ b/rust/tests/execution.rs
   @@ -1103,6 +1103,18 @@ async fn request(app: &Router, method: &str, path: &str, body: Value) -> (Status
        request_protocol(app, method, path, body, "1.19").await
    }
    
   +/// A repeated poll replays its first answer. Only the observation clock,
   +/// `server_time`, may differ, and it does not run backwards.
   +fn assert_poll_replay(first: &Value, repeated: &Value, message: &str) {
   +    let clock = |answer: &Value| {
   +        chrono::DateTime::parse_from_rfc3339(answer["server_time"].as_str().unwrap()).unwrap()
   +    };
   +    assert!(clock(repeated) >= clock(first), "{message}");
   +    let mut replay = repeated.clone();
   +    replay["server_time"] = first["server_time"].clone();
   +    assert_eq!(replay, *first, "{message}");
   +}
   +
    async fn request_protocol(
        app: &Router,
        method: &str,
   @@ -5223,9 +5235,10 @@ async fn repeated_poll_retains_the_original_claim_and_worker_history_paginates()
        .await;
        let b = request(&app_b, "POST", "/api/worker/workflow-tasks/poll", body).await;
        assert_eq!(a.0, StatusCode::OK);
   -    assert_eq!(
   -        a.1, b.1,
   -        "retry on another process obtains the original task/snapshot/fence"
   +    assert_poll_replay(
   +        &a.1,
   +        &b.1,
   +        "retry on another process obtains the original task/snapshot/fence",
        );
        let task = &a.1["task"];
        assert_eq!(task["history_events"].as_array().unwrap().len(), 1);
   diff --git a/rust/tests/execution/history_budget.rs b/rust/tests/execution/history_budget.rs
   index 9c17076c..0abb681c 100644
   --- a/rust/tests/execution/history_budget.rs
   +++ b/rust/tests/execution/history_budget.rs
   @@ -197,17 +197,16 @@ async fn history_budget_profile_child() {
                task["history_budget_pressure_dimensions"],
                initial_dimensions
            );
   -        assert_eq!(
   -            namespaces::scoped(
   -                &app,
   -                namespace,
   -                "POST",
   -                "/api/worker/workflow-tasks/poll",
   -                body
   -            )
   -            .await,
   -            claim
   -        );
   +        let replayed = namespaces::scoped(
   +            &app,
   +            namespace,
   +            "POST",
   +            "/api/worker/workflow-tasks/poll",
   +            body,
   +        )
   +        .await;
   +        assert_eq!(replayed.0, claim.0);
   +        assert_poll_replay(&claim.1, &replayed.1, "a repeated poll replays its claim");
            let next_page = namespaces::scoped(&app,namespace,"POST",&format!("/api/worker/workflow-tasks/{}/history",task["task_id"].as_str().unwrap()),json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"],"next_history_page_token":task["next_history_page_token"]})).await;
            assert_eq!(next_page.0, StatusCode::OK, "{}", next_page.1);
            assert_eq!(
4. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   The P0 source work starts with this field-only change on maintained b8f4dd13. The current response wrapper is shared by fresh, cached and multiplexed workflow leases, so it can stamp server_time after polling without modifying the cached claim or adding a transaction. The new controls preserve every non-clock response field, including cold replay and paginated history. OpenAPI stays additive and protocol 1.20 is unchanged.

   The sanitized cumulative references were decoded and their declared byte counts/SHA-256 verified before review. No cumulative patch was applied. The query and sweep work in patch 0001 remains separate under #1054. Normal current-head backend/compatibility qualification follows; the reported benchmark figures remain unverified.
5. rmcdaniel

   closed this as completedin #1070on Nov 6, 2026
6. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Delivered through #1070 and accepted protected main 866a4067d052ae70f22060fd656b6154d0debc36. Exact source/merge, original 48-job queue qualification, automatic-main results, response-field handoff and cleanup are recorded at #1070 (comment) . The clock remains separate from #1054's query/sweep work; no benchmark timing or installed/release result is claimed.
7. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized rebased reference for report topic 2a: 0001-t03+t02a-Keep-the-poll-path-s-queries-off-the-size-of-the-tab.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: 16,849 bytes, SHA-256 5f018f020e3f3d3c2c8030132f8d2fb0bc3fa63462464709edb924822cb67e21. Original-base sanitized identity: 78a6762fdaaf4e9622b1aace4b93aff98d13eee6ccff9081e4b852a443a690e1. No patch was applied or executed during this update.

   diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
   index 02735d70..aaf916f0 100644
   --- a/rust/src/runtime/backend.rs
   +++ b/rust/src/runtime/backend.rs
   @@ -84,8 +84,8 @@ pub(super) trait Backend: Database {
        // cutover of 3.0.1 query rows (Server #964).
        const QUERY_CACHE_PAGE: &str = "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\">$1 ORDER BY \"key\" LIMIT 64";
        // A bounded batch of expired per-request query poll rows.
   -    const QUERY_POLL_EXPIRED: &str = "SELECT \"key\" FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query-poll:%' AND expiration<=$1 ORDER BY \"key\" LIMIT 32";
   -    const WORKFLOW_POLL_EXPIRED: &str = "SELECT \"key\" FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-workflow-poll:%' AND expiration<=$1 ORDER BY \"key\" LIMIT 32";
   +    const QUERY_POLL_EXPIRED: &str = "SELECT \"key\" FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query-poll:%' AND expiration<=$1 ORDER BY \"key\" LIMIT 256";
   +    const WORKFLOW_POLL_EXPIRED: &str = "SELECT \"key\" FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-workflow-poll:%' AND expiration<=$1 ORDER BY \"key\" LIMIT 256";
        /// Add a cache row, or lock the existing one with a no-op update, so
        /// transactions over one poll request ID run one at a time.
        /// A worker session's running activity attempts held by its owner,
   @@ -243,7 +243,7 @@ WITH utc_tasks AS (
        SELECT t.id,t.namespace,t.queue,t.workflow_run_id,t.status,t.available_at,
            replace(replace(replace(t.available_at,'T',' '),'Z',''),'+00:00','') AS available_utc,
            replace(replace(replace(t.lease_expires_at,'T',' '),'Z',''),'+00:00','') AS lease_utc
   -    FROM workflow_tasks t WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type!=$3 AND t.task_type IN ('workflow','activity')
   +    FROM workflow_tasks t WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type!=$3 AND t.task_type IN ('workflow','activity') AND t.status IN ('ready','leased')
    ), ordered_tasks AS (
        SELECT *,
            rtrim(rtrim(CASE WHEN length(available_utc)=19 THEN available_utc||'.0' ELSE available_utc END,'0'),'.') AS available_order,
   @@ -296,7 +296,7 @@ WITH utc_tasks AS (
        SELECT t.*,
            replace(replace(replace(t.available_at,'T',' '),'Z',''),'+00:00','') AS _dw_available_utc,
            replace(replace(replace(t.lease_expires_at,'T',' '),'Z',''),'+00:00','') AS _dw_lease_utc
   -    FROM workflow_tasks t WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type=$3
   +    FROM workflow_tasks t WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type=$3 AND t.status IN ('ready','leased')
    ), ordered_tasks AS (
        SELECT *,
            rtrim(rtrim(CASE WHEN length(_dw_available_utc)=19 THEN _dw_available_utc||'.0' ELSE _dw_available_utc END,'0'),'.') AS _dw_available_order,
   @@ -579,8 +579,8 @@ impl Backend for MySql {
        const QUERY_CACHE_DELETE: &str = "DELETE FROM dw_query_cache WHERE `key`=$1";
        const QUERY_CACHE_DELETE_EXPIRED: &str = "DELETE FROM dw_query_cache WHERE `key`=$1 AND expiration<=$2";
        const QUERY_CACHE_PAGE: &str = "SELECT `key`,value,expiration FROM dw_query_cache WHERE `key`>$1 ORDER BY `key` LIMIT 64";
   -    const QUERY_POLL_EXPIRED: &str = "SELECT `key` FROM dw_query_cache WHERE `key` LIKE 'dw-native-query-poll:%' AND expiration<=$1 ORDER BY `key` LIMIT 32";
   -    const WORKFLOW_POLL_EXPIRED: &str = "SELECT `key` FROM dw_query_cache WHERE `key` LIKE 'dw-native-workflow-poll:%' AND expiration<=$1 ORDER BY `key` LIMIT 32";
   +    const QUERY_POLL_EXPIRED: &str = "SELECT `key` FROM dw_query_cache WHERE `key` LIKE 'dw-native-query-poll:%' AND expiration<=$1 ORDER BY `key` LIMIT 256";
   +    const WORKFLOW_POLL_EXPIRED: &str = "SELECT `key` FROM dw_query_cache WHERE `key` LIKE 'dw-native-workflow-poll:%' AND expiration<=$1 ORDER BY `key` LIMIT 256";
        const QUERY_CACHE_LOCK: &str = "INSERT INTO dw_query_cache(`key`,value,expiration) VALUES ($1,$2,$3) ON DUPLICATE KEY UPDATE expiration=expiration";
        const SESSION_RUNNING_ATTEMPTS: &str = "SELECT COUNT(*) FROM activity_attempts p JOIN activity_executions e ON e.id=p.activity_execution_id JOIN workflow_tasks t ON t.id=p.workflow_task_id WHERE t.namespace=? AND p.status='running' AND p.lease_expires_at>? AND p.lease_owner=? AND JSON_UNQUOTE(JSON_EXTRACT(e.activity_options,'$.worker_session.session_id'))=? AND p.workflow_task_id<>?";
        const TASK_KIND_CURSOR_LOCK: &str = "INSERT INTO workflow_task_poll_cursors(namespace,task_queue,next_task_kind,created_at,updated_at) VALUES ($1,$2,'update_validation',$3,$4) ON DUPLICATE KEY UPDATE next_task_kind=next_task_kind";
   diff --git a/rust/src/runtime/cross_kind_poll.rs b/rust/src/runtime/cross_kind_poll.rs
   index 2db96437..0a42230d 100644
   --- a/rust/src/runtime/cross_kind_poll.rs
   +++ b/rust/src/runtime/cross_kind_poll.rs
   @@ -8,7 +8,7 @@ pub(super) struct PollWakeSnapshot {
        ready: Vec<String>,
    }
    
   -pub(super) const TASKS_SQL: &str = "SELECT t.id,t.namespace,t.queue FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type!=$3 AND t.task_type IN ('workflow','activity') AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) ORDER BY t.available_at,t.id LIMIT 100";
   +pub(super) const TASKS_SQL: &str = "SELECT t.id,t.namespace,t.queue FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type!=$3 AND t.task_type IN ('workflow','activity') AND t.status IN ('ready','leased') AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) ORDER BY t.available_at,t.id LIMIT 100";
    
    impl<DB: Backend> Store<DB>
    where
   diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs
   index 01ccecb1..be36228b 100644
   --- a/rust/src/runtime/execution.rs
   +++ b/rust/src/runtime/execution.rs
   @@ -1229,9 +1229,12 @@ impl Runtime {
            let kinds = super::update_validation::task_kinds(&body)?;
            let pages = self.history_pages();
            let wait = self.worker_poll_wait();
   +        // A leased answer carries `server_time`, as an activity poll does, so
   +        // a worker can tell how much of the lease is left without renewing.
            let mark = |mut answer: Value| {
                if let Some(task) = answer["task"].as_object_mut() {
                    task.insert("task_kind".into(), json!("workflow"));
   +                answer["server_time"] = json!(super::activity_lease_renewal::micros(super::store::now()));
                }
                answer
            };
   diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs
   index 7ada5bb5..09eeda16 100644
   --- a/rust/src/runtime/mod.rs
   +++ b/rust/src/runtime/mod.rs
   @@ -158,6 +158,7 @@ mod start_request;
    mod message_streams;
    mod storage_test;
    mod store;
   +mod sweep_throttle;
    mod system_metrics;
    mod system_health;
    mod task_failures;
   diff --git a/rust/src/runtime/query_poll_requests.rs b/rust/src/runtime/query_poll_requests.rs
   index 00a381ff..1ec35415 100644
   --- a/rust/src/runtime/query_poll_requests.rs
   +++ b/rust/src/runtime/query_poll_requests.rs
   @@ -119,7 +119,7 @@ where
        }
    
        // Workers send a fresh request ID with every poll, so retire a bounded
   -    // batch of expired poll rows whenever a new request arrives.
   +    // batch of expired poll rows at most once a second as new requests arrive.
        async fn sweep_poll_rows(tx: &mut Transaction<'_, DB>) -> Result<()> {
            let expired = Self::query(DB::QUERY_POLL_EXPIRED)
                .bind(now())
   @@ -165,7 +165,9 @@ where
        // of an older request never takes that place back.
        async fn mark_current_if_fresh(&self, keys: &PollKeys, config: &QueryTaskConfig) -> Result<()> {
            let mut tx = self.begin().await?;
   -        Self::sweep_poll_rows(&mut tx).await?;
   +        if self.sweeps.query_poll_rows.take() {
   +            Self::sweep_poll_rows(&mut tx).await?;
   +        }
            if Self::poll_row(&mut tx, &keys.pending).await?.is_none()
                && Self::poll_row(&mut tx, &keys.result).await?.is_none()
            {
   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index 970a8ee9..fd1c91c0 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -93,7 +93,7 @@ pub(super) struct SignalWrite<'a> {
    }
    /// The runtime-owned message-stream delivery signal (`WorkflowStub::MESSAGE_STREAM_RUNTIME_SIGNAL`).
    const MESSAGE_STREAM_SIGNAL: &str = super::message_streams::INTERNAL_SIGNAL;
   -pub(super) const TASK_CANDIDATES_SQL: &str = "SELECT t.*,r.workflow_type FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type=$3 AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) AND (COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,'')) IS NULL OR COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,''))=$6) ORDER BY t.priority,t.available_at,t.id LIMIT 100";
   +pub(super) const TASK_CANDIDATES_SQL: &str = "SELECT t.*,r.workflow_type FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type=$3 AND t.status IN ('ready','leased') AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) AND (COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,'')) IS NULL OR COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,''))=$6) ORDER BY t.priority,t.available_at,t.id LIMIT 100";
    /// A signal record's place in `PendingMessageTask::signalFollows`: its
    /// command sequence (unsequenced last), receipt and creation times, id.
    struct SignalOrder {
   @@ -171,6 +171,8 @@ pub(super) struct Store<DB: Backend> {
        // production claims continue to sample the existing canonical clock.
        #[cfg(test)]
        pub(super) test_claim_at: Option<DateTime<Utc>>,
   +    /// When this server next retires expired poll rows and receipts.
   +    pub(super) sweeps: Arc<super::sweep_throttle::Sweeps>,
    }
    /// The `closed_reason` a terminal history event gives its run, or `None` for
    /// an event that does not close the run.
   @@ -320,6 +322,7 @@ where
                repair_pass_at: Arc::default(),
                #[cfg(test)]
                test_claim_at: None,
   +            sweeps: Arc::default(),
            }
        }
        pub(super) fn in_namespace(&self, namespace: &str) -> Self {
   @@ -340,6 +343,7 @@ where
                repair_pass_at: self.repair_pass_at.clone(),
                #[cfg(test)]
                test_claim_at: self.test_claim_at,
   +            sweeps: self.sweeps.clone(),
            }
        }
        pub(super) async fn database_live(&self) -> bool {
   @@ -2177,11 +2181,13 @@ where
            if DB::string(&worker, "task_queue")? != queue {
                return Err(refuse(StatusCode::CONFLICT, "task_queue_mismatch"));
            }
   -        Self::query(DB::POLL_RECEIPT_CLEANUP_SQL)
   -            .bind(DB::bind_time(now()))
   -            .bind(DB::bind_time(now()))
   -            .execute(&mut **tx)
   -            .await?;
   +        if self.sweeps.poll_receipts.take() {
   +            Self::query(DB::POLL_RECEIPT_CLEANUP_SQL)
   +                .bind(DB::bind_time(now()))
   +                .bind(DB::bind_time(now()))
   +                .execute(&mut **tx)
   +                .await?;
   +        }
            // A concurrent duplicate may have leased since this poll began: the
            // binding row, locked here, makes it wait for that one's answer. Only
            // a workflow answer replays as a workflow claim; a multiplexed probe
   @@ -2204,12 +2210,22 @@ where
            if let Some(request_id) = request_id
                && let Some(receipt) = Self::query("SELECT p.*,t.status,t.attempt_count,t.lease_owner,t.lease_expires_at FROM dw_poll_receipts p JOIN workflow_tasks t ON t.id=p.task_id WHERE p.worker_id=$1 AND p.kind=$2 AND p.request_id=$3 AND p.namespace=$4")
                    .bind(worker_id).bind(kind).bind(request_id).bind(self.namespace.as_ref()).fetch_optional(&mut **tx).await? {
   -                if DB::string(&receipt,"status")? == "leased" && DB::number(&receipt,"attempt")? == DB::number(&receipt,"attempt_count")?
   +            let at = now();
   +            let leased = DB::string(&receipt,"status")? == "leased";
   +            let lease_expires_at = DB::optional_instant(&receipt,"lease_expires_at")?;
   +            // The receipt cleanup's condition: a receipt it would have deleted
   +            // is absent, whether or not this poll ran the cleanup.
   +            if DB::instant(&receipt,"expires_at")? <= at && (!leased || lease_expires_at.is_some_and(|expires| expires <= at)) {
   +                Self::query("DELETE FROM dw_poll_receipts WHERE worker_id=$1 AND kind=$2 AND request_id=$3 AND namespace=$4")
   +                    .bind(worker_id).bind(kind).bind(request_id).bind(self.namespace.as_ref()).execute(&mut **tx).await?;
   +            } else {
   +                if leased && DB::number(&receipt,"attempt")? == DB::number(&receipt,"attempt_count")?
                        && DB::optional_string(&receipt,"lease_owner")?.as_deref() == Some(worker_id)
   -                    && DB::optional_instant(&receipt,"lease_expires_at")?.is_some_and(|expires| expires > now()) {
   +                    && lease_expires_at.is_some_and(|expires| expires > now()) {
                        return Ok(Some(DB::document_row(&receipt,"response")?));
                    }
                    return Err(refuse(StatusCode::CONFLICT,"poll_cached_task_expired"));
   +            }
            }
            let types: Value = DB::document_row(
                &worker,
   diff --git a/rust/src/runtime/sweep_throttle.rs b/rust/src/runtime/sweep_throttle.rs
   new file mode 100644
   index 00000000..a0bb3546
   --- /dev/null
   +++ b/rust/src/runtime/sweep_throttle.rs
   @@ -0,0 +1,42 @@
   +//! Housekeeping deletes that once ran on every poll, now at most once a
   +//! second per server. Each retires rows its readers already treat as absent:
   +//! an expired poll row reads as missing, and the claim forgets a receipt the
   +//! cleanup would delete. Skipping a sweep only leaves dead rows a little
   +//! longer, while each sweep that runs scans under the transition lock.
   +use std::sync::atomic::{AtomicI64, Ordering};
   +
   +#[derive(Default)]
   +pub(super) struct Sweeps {
   +    pub(super) workflow_poll_rows: Due,
   +    pub(super) query_poll_rows: Due,
   +    pub(super) poll_receipts: Due,
   +}
   +
   +/// When a sweep is next due, in epoch milliseconds.
   +#[derive(Default)]
   +pub(super) struct Due(AtomicI64);
   +
   +const INTERVAL_MILLIS: i64 = 1000;
   +
   +impl Due {
   +    /// Whether this caller runs the sweep: the first to ask once it is due.
   +    pub(super) fn take(&self) -> bool {
   +        let now = chrono::Utc::now().timestamp_millis();
   +        let due = self.0.load(Ordering::Relaxed);
   +        now >= due && self.0.compare_exchange(due, now + INTERVAL_MILLIS, Ordering::Relaxed, Ordering::Relaxed).is_ok()
   +    }
   +}
   +
   +#[cfg(test)]
   +mod tests {
   +    use super::*;
   +
   +    #[test]
   +    fn one_caller_a_second_sweeps() {
   +        let due = Due::default();
   +        assert!(due.take());
   +        assert!(!due.take());
   +        due.0.store(chrono::Utc::now().timestamp_millis(), Ordering::Relaxed);
   +        assert!(due.take());
   +    }
   +}
   diff --git a/rust/src/runtime/workflow_poll_requests.rs b/rust/src/runtime/workflow_poll_requests.rs
   index 2f290218..4eb5d1a5 100644
   --- a/rust/src/runtime/workflow_poll_requests.rs
   +++ b/rust/src/runtime/workflow_poll_requests.rs
   @@ -140,7 +140,9 @@ store_methods! {
            queue: &str,
        ) -> Result<Option<Value>> {
            let mut tx = self.begin().await?;
   -        Self::sweep_workflow_poll_rows(&mut tx).await?;
   +        if self.sweeps.workflow_poll_rows.take() {
   +            Self::sweep_workflow_poll_rows(&mut tx).await?;
   +        }
            Self::bind_workflow_poll(&mut tx, request).await?;
            let replay = self.deliverable_workflow_poll(&mut tx, request, worker_id, queue).await?;
            tx.commit().await?;
   @@ -326,7 +328,7 @@ store_methods! {
        }
    
        // Workers send a fresh request ID with most polls, so retire a bounded
   -    // batch of expired rows whenever a request arrives.
   +    // batch of expired rows at most once a second as requests arrive.
        async fn sweep_workflow_poll_rows(tx: &mut Transaction<'_, DB>) -> Result<()> {
            for row in Self::query(DB::WORKFLOW_POLL_EXPIRED).bind(epoch()).fetch_all(&mut **tx).await? {
                Self::query(DB::QUERY_CACHE_DELETE).bind(DB::string(&row, "key")?).execute(&mut **tx).await?;
8. rmcdaniel commented on Nov 7, 2026

   

   rmcdaniel

   on Nov 7, 2026

   MemberAuthor

   More actions

   Finite native maintenance source assessment for SDK #327: the field-only backport is feasible independently of the carried-poll/3.2 train. Server owns the next isolated source-authoring checkpoint after the active #1073 qualification/review work; no competing branch or PR has been opened by this assessment.

   The published baseline is annotated tag 3.0.5 / tag object 685f57ed19127a9e53ea6da28fddf310bfb16c3a / source commit bc24589db02161a21febbb30211309537fd1e24e, independently re-read from GitHub. Latest final release still resolves to3.0.5; no3.0.6 tag exists at this read. A new maintenance number remains a later release choice, not a selected tag here.

   The three production lines from accepted #1070 / merge 866a4067d052ae70f22060fd656b6154d0debc36 dry-apply cleanly to this exact baseline. Its existing Runtime::poll_workflow_tasks already shares the marker closure between fresh, replayed and multiplexed workflow leases. Both store::now() and activity_lease_renewal::micros() already exist. The patch samples only a response observation clock after polling, renders the existing millisecond clock with six fractional digits, and changes no lease, history, attempt, receipt, database schema or capability. No Cargo/Composer graph change is required; main's additional serde_json features are unrelated and must not be imported as a clock dependency.

   The full source merge must not be cherry-picked wholesale. The tag's OpenAPI document is revision56, whereas #1070 was authored on58 and moved it to59. The combined production/schema dry-check correctly refuses that header context. Port only the optional WorkerServerTime semantics and plain/multiplexed workflow-claim properties onto the maintained baseline document, with an explicitly identified maintenance schema blob/revision. Preserve protocol1.20, older-producer omission, validation-task/empty shapes and all existing capabilities. No carried-poll, ready-age or other3.2 contract is needed.

   Qualification must include actual maintenance-head normal checks on SQLite, PostgreSQL and MySQL-family backends; real HTTP fresh/cached/multiplexed/cold responses with every non-clock authority/history field unchanged; and the unchanged empty response. Published acceptance then needs one exact supported native maintenance image + published SDK3.9.5 + runner tuple, with a positive fresh-clock path and the actual older3.0.5 fallback criterion. Source tests or a main image cannot replace that published pair.

   Two release-path facts need explicit ownership before publication. The3.0.5 workflow still contains direct rolling-alias publication, while the maintained path separates exact publication, published-image qualification and protected alias promotion; do not silently reuse the older alias sequence or import a whole3.2 runtime to obtain newer guards. Also, inherited Sonar triggers target main and the known FREE-plan non-main restriction remains General's quality/account responsibility. A maintenance PR analyzed as main is not maintenance-branch acceptance. Server will return the concrete narrow release-workflow/quality input graph with the source candidate; General retains the existing quality/publication disposition.

   This is a read-only source/dependency/path assessment and dry-check of our already-reviewed production hunk. No source patch was applied to a branch, no Rust/backend or installed SDK ran, and no tag, build, dispatch, registry alias, runtime, PHP Server release or frozen/private operating tuple was selected.
9. rmcdaniel

   reopened this on Nov 7, 2026
10. rmcdaniel commented on Nov 7, 2026

    

    rmcdaniel

    on Nov 7, 2026

    MemberAuthor

    More actions

    General release follow-through: reopen this existing producer issue for its remaining published-product qualification. The accepted field-only Source in PR1070 /866a4067 remains accepted; this is not a request to rewrite or relabel that Source result.

    SDK3.9.5 is now published at immutable57562b53 and its normal Packagist/installed-package proof passed SDK337/6699764564. The operator's explicit SDK release order selected closure of the delivered conditional SDK implementation after package verification. Remaining end-to-end producer work stays here rather than being falsely called installed conformance: qualify an exact supported published Native Server containing this clock field, plus a real older-Server tuple proving the SDK heartbeat fallback. Latest published native3.0.5/bc24589d lacks the field; PHP2.11.8 also lacks it and is not a positive substitute. No new PHP Server release is allowed.

    General owns the publication/tuple decision and real producer/consumer follow-through. The active Server owner has been asked for a finite3.0.5-based field-only maintenance Source assessment, preserving app/protocol/support/lease/receipt/history bytes and the additive OpenAPI59 contract, independently of the unready3.2/carried-poll train. Acknowledgement/assessment is still requested; no tag, build, runtime or release version is selected by this comment. The new Cloud runtime security P0#656 currently preempts General's release authoring, while the bounded assessment can proceed.

    Acceptance remaining: exact public producer artifact identity and unchanged compatibility; fresh/replayed/plain/multiplexed clock behavior and original claim/attempt/lease equality; actual released SDK3.9.5 on that tuple; real old-producer fallback; bounded scenario/resource/cleanup evidence. Fixture and benchmark results remain separately attributed. Keep this issue open until the user-visible producer/consumer outcome is verified.
11. 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: ordinary and cached workflow responses refresh the outer server_time while retaining the original task/attempt/lease/history. assert_poll_replay requires a nondecreasing clock and full equality after replacing only that field; the history-budget replay uses the same assertion. OpenAPI's WorkerServerTime, WorkflowTaskClaim and multiplexed claim document UTC microseconds and the original-lease semantics. This item is additive and does not change worker protocol 1.20.

    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.
12. rmcdaniel

    closed this as completedon Nov 9, 2026
13. rmcdaniel

    added

    agent:general

    and removed

    agent:server

    on Nov 9, 2026