Native: drop locked statements whose answers the transition already has #1063

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 0014-t12-Write-a-run-s-summary-row-in-one-upsert.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: 16,841 bytes, SHA-256 1b4a258192fb0d63eb1db0b5ffbe28887c358648fcd70b633ef8b6ae2c25b479. This part: 16,841 bytes, SHA-256 1b4a258192fb0d63eb1db0b5ffbe28887c358648fcd70b633ef8b6ae2c25b479. 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 cd3f16a4..b751e2f7 100644
   --- a/rust/src/runtime/backend.rs
   +++ b/rust/src/runtime/backend.rs
   @@ -47,6 +47,8 @@ pub(super) trait Backend: Database {
        // over every retained wait of the run, as Workflow #941's projector stores.
        const WAIT_POSITION_MOVES: &str = "WITH clocks AS (SELECT id,wait_id,position,opened_at,EXTRACT(EPOCH FROM opened_at) AS seconds FROM workflow_run_waits WHERE workflow_run_id=$1), ordinals AS (SELECT id,position,ROW_NUMBER() OVER (ORDER BY CASE WHEN opened_at IS NULL THEN 1 ELSE 0 END,ROUND((CASE WHEN seconds<0 THEN 2*FLOOR(seconds)-seconds ELSE seconds END)*1000),wait_id)-1 AS ordinal FROM clocks) SELECT id,ordinal FROM ordinals WHERE position IS DISTINCT FROM ordinal ORDER BY ordinal LIMIT 100";
        const SUMMARY_HISTORY_SQL: &str = "SELECT id,sequence,event_type,CAST(payload AS TEXT) AS payload_json FROM workflow_history_events WHERE workflow_run_id=$1 AND sequence>$2 ORDER BY sequence LIMIT 16";
   +    /// Writes a run's projected summary row over any stored one.
   +    const RUN_SUMMARY_WRITE: &str = super::terminal_summary::RUN_SUMMARY_UPSERT;
        const OBSERVATION_FAILURE_EVENT: &str = "SELECT id,sequence,event_type,recorded_at FROM workflow_history_events WHERE workflow_run_id=$1 AND payload->>'failure_id'=$2 AND event_type IN ('ActivityFailed','ActivityTimedOut','ChildRunFailed','ChildRunCancelled','ChildRunTerminated','WorkflowFailed','WorkflowTimedOut','WorkflowCancelled','WorkflowTerminated','UpdateCompleted') ORDER BY sequence DESC LIMIT 1";
        const OBSERVATION_ATTRIBUTES: &str = "SELECT \"key\" AS attribute_key,type,value_string,value_keyword,value_int,value_float,CASE WHEN value_bool IS NULL THEN NULL WHEN value_bool THEN 1 ELSE 0 END AS value_bool,value_datetime,value_keyword_list FROM workflow_search_attributes WHERE workflow_run_id=$1 AND workflow_instance_id=$2 AND \"key\" IN ($3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22) ORDER BY \"key\" LIMIT 20";
        const ACTIVITY_RESULT_EVENT_SQL: &str = "SELECT sequence,event_type,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('ActivityScheduled','ActivityStarted','ActivityHeartbeatRecorded','ActivityRetryScheduled','ActivityCompleted','ActivityFailed','ActivityTimedOut','ActivityCancelled') AND (payload->>'activity_execution_id'=$2 OR payload->'activity'->>'id'=$3) ORDER BY sequence DESC LIMIT 1";
   @@ -99,6 +101,8 @@ pub(super) trait Backend: Database {
        /// Write a cache row whether or not its key exists, as `QUERY_CACHE_UPDATE`
        /// or else `QUERY_CACHE_INSERT` would, in one statement.
        const QUERY_CACHE_PUT: &str = "INSERT INTO dw_query_cache(\"key\",value,expiration) VALUES ($1,$2,$3) ON CONFLICT (\"key\") DO UPDATE SET value=excluded.value,expiration=excluded.expiration";
   +    /// `QUERY_CACHE_PUT` of two rows with different keys.
   +    const QUERY_CACHE_PUT_PAIR: &str = "INSERT INTO dw_query_cache(\"key\",value,expiration) VALUES ($1,$2,$3),($4,$5,$6) ON CONFLICT (\"key\") DO UPDATE SET value=excluded.value,expiration=excluded.expiration";
        /// `WorkflowTaskKindSelector`: create the queue's cursor at
        /// `update_validation` if it is missing, and lock it.
        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 CONFLICT (namespace,task_queue) DO UPDATE SET next_task_kind=workflow_task_poll_cursors.next_task_kind";
   @@ -572,6 +576,7 @@ impl Backend for MySql {
        // filesort of one large payload exceeds the default sort buffer and
        // fails the transaction ("Out of sort memory").
        const SUMMARY_HISTORY_SQL: &str = "SELECT id,sequence,event_type,CAST(payload AS CHAR) AS payload_json FROM workflow_history_events FORCE INDEX (workflow_history_events_workflow_run_id_sequence_unique) WHERE workflow_run_id=$1 AND sequence>$2 ORDER BY sequence LIMIT 16";
   +    const RUN_SUMMARY_WRITE: &str = super::terminal_summary::RUN_SUMMARY_REPLACE;
        const OBSERVATION_FAILURE_EVENT: &str = "SELECT id,sequence,event_type,recorded_at FROM workflow_history_events WHERE workflow_run_id=? AND CAST(JSON_UNQUOTE(JSON_EXTRACT(payload,'$.failure_id')) AS BINARY)=CAST(? AS BINARY) AND event_type IN ('ActivityFailed','ActivityTimedOut','ChildRunFailed','ChildRunCancelled','ChildRunTerminated','WorkflowFailed','WorkflowTimedOut','WorkflowCancelled','WorkflowTerminated','UpdateCompleted') ORDER BY sequence DESC LIMIT 1";
        const OBSERVATION_ATTRIBUTES: &str = "SELECT `key` AS attribute_key,type,value_string,value_keyword,value_int,value_float,CASE WHEN value_bool IS NULL THEN NULL WHEN value_bool THEN 1 ELSE 0 END AS value_bool,value_datetime,value_keyword_list FROM workflow_search_attributes WHERE workflow_run_id=$1 AND workflow_instance_id=$2 AND `key` IN ($3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22) ORDER BY `key` LIMIT 20";
        fn optional_float(row: &Self::Row, field: &str) -> Result<Option<f64>> {
   @@ -609,6 +614,7 @@ impl Backend for MySql {
        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 QUERY_CACHE_PUT: &str = "INSERT INTO dw_query_cache(`key`,value,expiration) VALUES ($1,$2,$3) ON DUPLICATE KEY UPDATE value=VALUES(value),expiration=VALUES(expiration)";
   +    const QUERY_CACHE_PUT_PAIR: &str = "INSERT INTO dw_query_cache(`key`,value,expiration) VALUES ($1,$2,$3),($4,$5,$6) ON DUPLICATE KEY UPDATE value=VALUES(value),expiration=VALUES(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";
        const TIMER_TASK_SQL: &str = "SELECT id,payload FROM workflow_tasks WHERE workflow_run_id=? AND task_type='timer' AND status='ready' AND JSON_UNQUOTE(JSON_EXTRACT(payload,'$.timer_id'))=? AND BINARY namespace=BINARY ?";
   diff --git a/rust/src/runtime/terminal_summary.rs b/rust/src/runtime/terminal_summary.rs
   index bdec7b3c..a3838b26 100644
   --- a/rust/src/runtime/terminal_summary.rs
   +++ b/rust/src/runtime/terminal_summary.rs
   @@ -15,6 +15,25 @@ use std::collections::BTreeSet;
    /// The events a page of `Backend::SUMMARY_HISTORY_SQL` holds at most (its `LIMIT`).
    const SUMMARY_HISTORY_PAGE: usize = 16;
    
   +/// The summary row a projection writes, from its run and instance.
   +macro_rules! run_summary_row {
   +    () => {
   +        "workflow_run_summaries(id,workflow_instance_id,run_number,is_current_run,engine_source,projection_schema_version,class,workflow_type,namespace,compatibility,declared_entry_mode,declared_contract_source,business_key,visibility_labels,status,status_bucket,closed_reason,connection,queue,started_at,sort_timestamp,sort_key,closed_at,archived_at,archive_command_id,archive_reason,duration_ms,liveness_state,liveness_reason,repair_blocked_reason,repair_attention,task_problem,exception_count,history_event_count,history_size_bytes,history_fan_out,continue_as_new_recommended,history_budget_pressure,created_at,updated_at,wait_kind,wait_reason,wait_started_at,wait_deadline_at,open_wait_id,resume_source_kind,resume_source_id,next_task_at,next_task_id,next_task_type,next_task_status,next_task_lease_expires_at) SELECT r.id,r.workflow_instance_id,r.run_number,CASE WHEN i.current_run_id=r.id THEN TRUE ELSE FALSE END,'v2',2,r.workflow_class,r.workflow_type,COALESCE(r.namespace,i.namespace),r.compatibility,$1,$2,COALESCE(r.business_key,i.business_key),COALESCE(r.visibility_labels,i.visibility_labels),r.status,CASE WHEN r.status='completed' THEN 'completed' WHEN r.status IN ('failed','cancelled','terminated') THEN 'failed' ELSE 'running' END,r.closed_reason,r.connection,r.queue,r.started_at,$3,$4,r.closed_at,r.archived_at,r.archive_command_id,r.archive_reason,CASE WHEN r.started_at IS NULL OR r.closed_at IS NULL THEN NULL ELSE $5 END,$6,$7,CASE WHEN i.current_run_id=r.id THEN $8 ELSE 'selected_run_not_current' END,CASE WHEN $9=1 THEN TRUE ELSE FALSE END,CASE WHEN $10=1 THEN TRUE ELSE FALSE END,$11,$12,$13,$14,CASE WHEN $15='continue_as_new_recommended' THEN TRUE ELSE FALSE END,$16,r.created_at,COALESCE(r.closed_at,r.last_progress_at,r.updated_at),$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28 FROM workflow_runs r JOIN workflow_instances i ON i.id=r.workflow_instance_id WHERE r.id=$29"
   +    };
   +}
   +
   +/// `Backend::RUN_SUMMARY_WRITE`: the projection's row over any stored one,
   +/// in one statement, leaving what deleting the stored row and inserting
   +/// this one would. Every column is written.
   +pub(super) const RUN_SUMMARY_UPSERT: &str = concat!(
   +    "INSERT INTO ",
   +    run_summary_row!(),
   +    " ON CONFLICT (id) DO UPDATE SET workflow_instance_id=excluded.workflow_instance_id,run_number=excluded.run_number,is_current_run=excluded.is_current_run,engine_source=excluded.engine_source,projection_schema_version=excluded.projection_schema_version,class=excluded.class,workflow_type=excluded.workflow_type,namespace=excluded.namespace,compatibility=excluded.compatibility,declared_entry_mode=excluded.declared_entry_mode,declared_contract_source=excluded.declared_contract_source,business_key=excluded.business_key,visibility_labels=excluded.visibility_labels,status=excluded.status,status_bucket=excluded.status_bucket,closed_reason=excluded.closed_reason,connection=excluded.connection,queue=excluded.queue,started_at=excluded.started_at,sort_timestamp=excluded.sort_timestamp,sort_key=excluded.sort_key,closed_at=excluded.closed_at,archived_at=excluded.archived_at,archive_command_id=excluded.archive_command_id,archive_reason=excluded.archive_reason,duration_ms=excluded.duration_ms,liveness_state=excluded.liveness_state,liveness_reason=excluded.liveness_reason,repair_blocked_reason=excluded.repair_blocked_reason,repair_attention=excluded.repair_attention,task_problem=excluded.task_problem,exception_count=excluded.exception_count,history_event_count=excluded.history_event_count,history_size_bytes=excluded.history_size_bytes,history_fan_out=excluded.history_fan_out,continue_as_new_recommended=excluded.continue_as_new_recommended,history_budget_pressure=excluded.history_budget_pressure,created_at=excluded.created_at,updated_at=excluded.updated_at,wait_kind=excluded.wait_kind,wait_reason=excluded.wait_reason,wait_started_at=excluded.wait_started_at,wait_deadline_at=excluded.wait_deadline_at,open_wait_id=excluded.open_wait_id,resume_source_kind=excluded.resume_source_kind,resume_source_id=excluded.resume_source_id,next_task_at=excluded.next_task_at,next_task_id=excluded.next_task_id,next_task_type=excluded.next_task_type,next_task_status=excluded.next_task_status,next_task_lease_expires_at=excluded.next_task_lease_expires_at"
   +);
   +/// MySQL's `RUN_SUMMARY_WRITE`. The primary key is the table's only unique
   +/// key, so InnoDB applies the replacement as an update of the stored row.
   +pub(super) const RUN_SUMMARY_REPLACE: &str = concat!("REPLACE INTO ", run_summary_row!());
   +
    impl<DB: Backend> Store<DB>
    where
        for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>,
   @@ -48,13 +67,9 @@ where
            }
            // A state outside the qualified producer must not retain an earlier
            // healthy snapshot. Reads still never create or repair this row.
   -        Self::query("DELETE FROM workflow_run_summaries WHERE id=$1")
   -            .bind(run_id)
   -            .execute(&mut **tx)
   -            .await?;
            let last = DB::number(&run, "last_history_sequence")?;
            let Some(history) = Self::summary_history(tx, run_id, closed, last).await? else {
   -            return Ok(());
   +            return Self::drop_run_summary(tx, run_id).await;
            };
            let SummaryHistory {
                count,
   @@ -75,7 +90,7 @@ where
                None
            } else {
                if !active_history_supported {
   -                return Ok(());
   +                return Self::drop_run_summary(tx, run_id).await;
                }
                match Self::active_task_summary(
                    tx,
   @@ -90,7 +105,7 @@ where
                .await?
                {
                    Some(summary) => Some(summary),
   -                None => return Ok(()),
   +                None => return Self::drop_run_summary(tx, run_id).await,
                }
            };
            let mut after = String::new();
   @@ -133,7 +148,7 @@ where
                .map_or(Some("run_closed"), |summary| summary.repair_blocked);
            // An active run's wait and next task go in with its row; a closed run has neither.
            let wait = active.as_ref().and_then(|summary| summary.wait.as_ref());
   -        Self::query("INSERT INTO workflow_run_summaries(id,workflow_instance_id,run_number,is_current_run,engine_source,projection_schema_version,class,workflow_type,namespace,compatibility,declared_entry_mode,declared_contract_source,business_key,visibility_labels,status,status_bucket,closed_reason,connection,queue,started_at,sort_timestamp,sort_key,closed_at,archived_at,archive_command_id,archive_reason,duration_ms,liveness_state,liveness_reason,repair_blocked_reason,repair_attention,task_problem,exception_count,history_event_count,history_size_bytes,history_fan_out,continue_as_new_recommended,history_budget_pressure,created_at,updated_at,wait_kind,wait_reason,wait_started_at,wait_deadline_at,open_wait_id,resume_source_kind,resume_source_id,next_task_at,next_task_id,next_task_type,next_task_status,next_task_lease_expires_at) SELECT r.id,r.workflow_instance_id,r.run_number,CASE WHEN i.current_run_id=r.id THEN TRUE ELSE FALSE END,'v2',2,r.workflow_class,r.workflow_type,COALESCE(r.namespace,i.namespace),r.compatibility,$1,$2,COALESCE(r.business_key,i.business_key),COALESCE(r.visibility_labels,i.visibility_labels),r.status,CASE WHEN r.status='completed' THEN 'completed' WHEN r.status IN ('failed','cancelled','terminated') THEN 'failed' ELSE 'running' END,r.closed_reason,r.connection,r.queue,r.started_at,$3,$4,r.closed_at,r.archived_at,r.archive_command_id,r.archive_reason,CASE WHEN r.started_at IS NULL OR r.closed_at IS NULL THEN NULL ELSE $5 END,$6,$7,CASE WHEN i.current_run_id=r.id THEN $8 ELSE 'selected_run_not_current' END,CASE WHEN $9=1 THEN TRUE ELSE FALSE END,CASE WHEN $10=1 THEN TRUE ELSE FALSE END,$11,$12,$13,$14,CASE WHEN $15='continue_as_new_recommended' THEN TRUE ELSE FALSE END,$16,r.created_at,COALESCE(r.closed_at,r.last_progress_at,r.updated_at),$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28 FROM workflow_runs r JOIN workflow_instances i ON i.id=r.workflow_instance_id WHERE r.id=$29")
   +        let written = Self::query(DB::RUN_SUMMARY_WRITE)
                .bind(contract["entry_mode"].as_str()).bind(contract["source"].as_str()).bind(DB::optional_time(sort_time)).bind(sort_key.as_deref()).bind(duration).bind(state).bind(reason).bind(repair_blocked)
                .bind(i64::from(repair_blocked == Some("unsupported_history"))).bind(i64::from(active.as_ref().is_some_and(|summary| summary.problem)))
                .bind(failures.len() as i64).bind(count).bind(bytes).bind(fan_out).bind(pressure).bind(pressure)
   @@ -144,6 +159,9 @@ where
                .bind(active.as_ref().and_then(|summary| summary.task_type.as_deref())).bind(active.as_ref().and_then(|summary| summary.task_status.as_deref()))
                .bind(DB::optional_time(active.as_ref().and_then(|summary| summary.lease_expires_at)))
                .bind(run_id).execute(&mut **tx).await?;
   +        if DB::affected(written) == 0 {
   +            return Self::drop_run_summary(tx, run_id).await;
   +        }
            if embedded_import {
                Self::query("UPDATE workflow_run_summaries SET engine_source=$1 WHERE id=$2")
                    .bind(super::workflow_import::ENGINE_SOURCE)
   @@ -153,6 +171,13 @@ where
            }
            Ok(())
        }
   +    async fn drop_run_summary(tx: &mut Transaction<'_, DB>, run_id: &str) -> Result<()> {
   +        Self::query("DELETE FROM workflow_run_summaries WHERE id=$1")
   +            .bind(run_id)
   +            .execute(&mut **tx)
   +            .await?;
   +        Ok(())
   +    }
        pub(super) async fn workflow_history_budget(
            tx: &mut Transaction<'_, DB>,
            run_id: &str,
3. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0015-t12-Write-a-poll-s-binding-and-result-cache-rows-togethe.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: 1,249 bytes, SHA-256 8cd4d67fb3f4b4244d2cc2c66d0a63c3529a4d0d8039c49319bc8567a98e5136. This part: 1,249 bytes, SHA-256 8cd4d67fb3f4b4244d2cc2c66d0a63c3529a4d0d8039c49319bc8567a98e5136. Parts concatenate in order; a part boundary is between complete file diffs.

   diff --git a/rust/src/runtime/workflow_poll_requests.rs b/rust/src/runtime/workflow_poll_requests.rs
   index c5a25d42..bf15e9c8 100644
   --- a/rust/src/runtime/workflow_poll_requests.rs
   +++ b/rust/src/runtime/workflow_poll_requests.rs
   @@ -297,9 +297,13 @@ store_methods! {
        /// binding replays for the whole live lease.
        pub(super) async fn remember_workflow_poll(tx: &mut Transaction<'_, DB>, request: &WorkflowPollRequest, response: &Value) -> Result<()> {
            let ttl = result_ttl(response, request.pending_ttl);
   -        Self::put_workflow_poll_row(tx, &request.keys.binding, &json!({"task_kinds": request.kinds}), ttl).await?;
            let kept = if response["task"].is_object() && task_kind(response) == "workflow" { 3600 } else { ttl };
   -        Self::put_workflow_poll_row(tx, &request.keys.result, response, kept).await
   +        Self::query(DB::QUERY_CACHE_PUT_PAIR)
   +            .bind(&request.keys.binding).bind(json!({"task_kinds": request.kinds}).to_string()).bind(epoch() + ttl)
   +            .bind(&request.keys.result).bind(response.to_string()).bind(epoch() + kept)
   +            .execute(&mut **tx)
   +            .await?;
   +        Ok(())
        }
    
        /// `replay_workflow_poll` without the lock, when the replay would write
4. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0016-t12-Count-the-projection-s-waits-and-tasks-in-one-pass-e.patch, part 1/1.

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

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

   diff --git a/rust/src/runtime/active_summary.rs b/rust/src/runtime/active_summary.rs
   index 0244eac3..7aae3e44 100644
   --- a/rust/src/runtime/active_summary.rs
   +++ b/rust/src/runtime/active_summary.rs
   @@ -68,9 +68,10 @@ where
            tx: &mut Transaction<'_, DB>,
            run_id: &str,
        ) -> Result<ProjectionCensus> {
   -        let row = Self::query("SELECT (SELECT COUNT(*) FROM workflow_run_waits WHERE workflow_run_id=$1 AND status='open' AND kind NOT IN ('condition','signal','timer','activity','child','update')) AS other_waits,(SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id=$2 AND task_type='workflow' AND repair_count>0) AS repairs,(SELECT COUNT(*) FROM workflow_tasks WHERE workflow_run_id=$3 AND task_type='workflow' AND status='failed') AS failed_tasks,(SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_run_id=$4 AND status='received') AS received_signals,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$5 AND (w.status='open' OR (w.status='resolved' AND w.source_status='timed_out')) AND w.kind IN ('condition','signal')) AS external_waits,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$6 AND w.kind='timer' AND w.status='open') AS timers,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$7 AND w.kind='activity' AND w.status='open') AS activities,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$8 AND w.kind='child') AS children,(SELECT COUNT(*) FROM workflow_run_waits w WHERE w.workflow_run_id=$9 AND w.kind='update') AS updates,(SELECT COUNT(*) FROM workflow_failures WHERE workflow_run_id=$10) AS failures")
   -            .bind(run_id).bind(run_id).bind(run_id).bind(run_id).bind(run_id)
   -            .bind(run_id).bind(run_id).bind(run_id).bind(run_id).bind(run_id)
   +        // One pass over the run's waits and one over its tasks; each count is
   +        // the rows of that pass its condition holds for.
   +        let row = Self::query("SELECT w.other_waits,t.repairs,t.failed_tasks,(SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_run_id=$1 AND status='received') AS received_signals,w.external_waits,w.timers,w.activities,w.children,w.updates,(SELECT COUNT(*) FROM workflow_failures WHERE workflow_run_id=$2) AS failures FROM (SELECT COUNT(CASE WHEN status='open' AND kind NOT IN ('condition','signal','timer','activity','child','update') THEN 1 END) AS other_waits,COUNT(CASE WHEN (status='open' OR (status='resolved' AND source_status='timed_out')) AND kind IN ('condition','signal') THEN 1 END) AS external_waits,COUNT(CASE WHEN kind='timer' AND status='open' THEN 1 END) AS timers,COUNT(CASE WHEN kind='activity' AND status='open' THEN 1 END) AS activities,COUNT(CASE WHEN kind='child' THEN 1 END) AS children,COUNT(CASE WHEN kind='update' THEN 1 END) AS updates FROM workflow_run_waits WHERE workflow_run_id=$3) w CROSS JOIN (SELECT COUNT(CASE WHEN task_type='workflow' AND repair_count>0 THEN 1 END) AS repairs,COUNT(CASE WHEN task_type='workflow' AND status='failed' THEN 1 END) AS failed_tasks FROM workflow_tasks WHERE workflow_run_id=$4) t")
   +            .bind(run_id).bind(run_id).bind(run_id).bind(run_id)
                .fetch_one(&mut **tx).await?;
            Ok(ProjectionCensus {
                other_waits: DB::number(&row, "other_waits")?,
5. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0017-t12-Read-a-run-s-authored-sequences-with-its-row.patch, part 1/1.

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

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

   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index c14b5e80..2aea1209 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -1273,24 +1273,15 @@ where
            // Public control command order and deterministic authored call order
            // are different counters in PHP. Signals must never shift a replayed
            // activity/timer/wait's authored position.
   -        let activities = Self::query(
   -            "SELECT sequence FROM activity_executions WHERE workflow_run_id=$1 ORDER BY sequence DESC LIMIT 1",
   -        )
   -        .bind(run_id)
   -        .fetch_optional(&mut **tx)
   -        .await?
   -        .map(|row| DB::number(&row, "sequence"))
   -        .transpose()?
   -        .unwrap_or(0);
   -        let timers = Self::query(
   -            "SELECT sequence FROM workflow_run_timers WHERE workflow_run_id=$1 ORDER BY sequence DESC LIMIT 1",
   -        )
   -        .bind(run_id)
   -        .fetch_optional(&mut **tx)
   -        .await?
   -        .map(|row| DB::number(&row, "sequence"))
   -        .transpose()?
   -        .unwrap_or(0);
   +        // The run's row and each authored table's largest sequence, in one
   +        // statement; a table with no rows for the run counts as 0.
   +        let run = Self::query("SELECT r.cancellation_delivery_sequence,r.run_number,(SELECT sequence FROM activity_executions WHERE workflow_run_id=$1 ORDER BY sequence DESC LIMIT 1) AS activities,(SELECT sequence FROM workflow_run_timers WHERE workflow_run_id=$2 ORDER BY sequence DESC LIMIT 1) AS timers,(SELECT sequence FROM workflow_child_calls WHERE parent_workflow_run_id=$3 ORDER BY sequence DESC LIMIT 1) AS children FROM workflow_runs r WHERE r.id=$4")
   +            .bind(run_id).bind(run_id).bind(run_id).bind(run_id)
   +            .fetch_one(&mut **tx)
   +            .await?;
   +        let activities = DB::optional_number(&run, "activities")?.unwrap_or(0);
   +        let timers = DB::optional_number(&run, "timers")?.unwrap_or(0);
   +        let children = DB::optional_number(&run, "children")?.unwrap_or(0);
            let wait = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('SignalWaitOpened','ConditionWaitOpened','VersionMarkerRecorded','MemoUpserted','SideEffectRecorded','SearchAttributesUpserted') ORDER BY sequence DESC LIMIT 1")
                .bind(run_id).fetch_optional(&mut **tx).await?;
            let wait_sequence = wait
   @@ -1298,8 +1289,6 @@ where
                .transpose()?
                .and_then(|value| value["sequence"].as_i64())
                .unwrap_or(0);
   -        let children = Self::query("SELECT sequence FROM workflow_child_calls WHERE parent_workflow_run_id=$1 ORDER BY sequence DESC LIMIT 1")
   -            .bind(run_id).fetch_optional(&mut **tx).await?.map(|row| DB::number(&row,"sequence")).transpose()?.unwrap_or(0);
            // `nextDurableCommandSequence`: a scope opening consumes its sequence,
            // and a delivered scope boundary its whole span, as a delivered run
            // request does.
   @@ -1313,11 +1302,6 @@ where
                    scoped = scoped.max(span.and_then(|span| sequence.checked_add(span - 1)).unwrap_or(sequence));
                }
            }
   -        let run =
   -            Self::query("SELECT cancellation_delivery_sequence,run_number FROM workflow_runs WHERE id=$1")
   -                .bind(run_id)
   -                .fetch_one(&mut **tx)
   -                .await?;
            // A redriven run replays the activities before its resume step from
            // reused completions, which have no execution rows.
            let mut reused = 0;
6. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0018-t12-Read-a-completed-task-s-run-status-before-marking-it.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: 1,866 bytes, SHA-256 0d26bdf3ad8b7d0303c7799912b6365fff536a1f423a4f3f91a5f5fd23967ebe. This part: 1,866 bytes, SHA-256 0d26bdf3ad8b7d0303c7799912b6365fff536a1f423a4f3f91a5f5fd23967ebe. Parts concatenate in order; a part boundary is between complete file diffs.

   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index 2aea1209..d5c5e214 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -3409,15 +3409,27 @@ where
            Self::record_completion(&mut 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.
   -        Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1 AND status IN ('pending','running')")
   -            .bind(&run_id)
   -            .execute(&mut *tx)
   -            .await?;
   +        // A run already waiting or closed is read and left as it is.
            let current = Self::query("SELECT status FROM workflow_runs WHERE id=$1")
                .bind(&run_id)
                .fetch_one(&mut *tx)
                .await?;
   -        let run_status = DB::string(&current, "status")?;
   +        let mut run_status = DB::string(&current, "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)
   +                .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)
   +                    .await?;
   +                DB::string(&current, "status")?
   +            };
   +        }
            if let Some(worker) = &worker {
                Self::attribute_worker_terminal(&mut tx, &run_id, task_id, worker).await?;
            }
7. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0019-t12-Read-a-workflow-poll-s-binding-and-result-cache-rows.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: 8,846 bytes, SHA-256 310bae18f59ce4bc467471607644a8b7a0427b7923d7bd320ac5767612bc1c8e. This part: 8,846 bytes, SHA-256 310bae18f59ce4bc467471607644a8b7a0427b7923d7bd320ac5767612bc1c8e. 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 b751e2f7..d28a4bcc 100644
   --- a/rust/src/runtime/backend.rs
   +++ b/rust/src/runtime/backend.rs
   @@ -80,6 +80,9 @@ pub(super) trait Backend: Database {
        const QUERY_CACHE_SCAN: &str = "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND value::jsonb->>'namespace'=$1 ORDER BY \"key\" LIMIT 65";
        const QUERY_CACHE_GET: &str =
            "SELECT value,expiration FROM dw_query_cache WHERE \"key\"=$1";
   +    /// Two keys' rows, in one read.
   +    const QUERY_CACHE_GET_PAIR: &str =
   +        "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\" IN ($1,$2)";
        const QUERY_CACHE_INSERT: &str =
            "INSERT INTO dw_query_cache(\"key\",value,expiration) VALUES ($1,$2,$3)";
        const QUERY_CACHE_UPDATE: &str =
   @@ -603,6 +606,8 @@ impl Backend for MySql {
        type EncodedDocument = Json<Value>;
        const QUERY_CACHE_GET: &str =
            "SELECT value,expiration FROM dw_query_cache WHERE `key`=$1";
   +    const QUERY_CACHE_GET_PAIR: &str =
   +        "SELECT `key`,value,expiration FROM dw_query_cache WHERE `key` IN ($1,$2)";
        const QUERY_CACHE_INSERT: &str =
            "INSERT INTO dw_query_cache(`key`,value,expiration) VALUES ($1,$2,$3)";
        const QUERY_CACHE_UPDATE: &str =
   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index d5c5e214..ad3656f2 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -2259,8 +2259,8 @@ where
            if let Some(request) = poll_request
                && !request.replayed_in(tx)
            {
   -            Self::bind_workflow_poll(tx, request).await?;
   -            if let Some(answer) = self.deliverable_workflow_lease(tx, request, worker_id, queue).await?
   +            let result = Self::bind_workflow_poll(tx, request).await?;
   +            if let Some(answer) = self.deliverable_workflow_lease(tx, request, result, worker_id, queue).await?
                    && answer["task"]["task_kind"] != "update_validation"
                {
                    return Ok(Some(answer["task"].clone()));
   diff --git a/rust/src/runtime/update_validation.rs b/rust/src/runtime/update_validation.rs
   index 604c61b4..0b0ec511 100644
   --- a/rust/src/runtime/update_validation.rs
   +++ b/rust/src/runtime/update_validation.rs
   @@ -1045,8 +1045,8 @@ where
            // A concurrent duplicate may have leased since this poll began: the
            // binding row, locked first, makes it wait for that one's answer.
            if let Some(request) = request {
   -            Self::bind_workflow_poll(&mut tx, request).await?;
   -            if let Some(answer) = self.deliverable_workflow_lease(&mut tx, request, worker_id, task_queue).await? {
   +            let result = Self::bind_workflow_poll(&mut tx, request).await?;
   +            if let Some(answer) = self.deliverable_workflow_lease(&mut tx, request, result, worker_id, task_queue).await? {
                    tx.commit().await?;
                    return Ok(Some(Probe::Replayed(answer)));
                }
   diff --git a/rust/src/runtime/workflow_poll_requests.rs b/rust/src/runtime/workflow_poll_requests.rs
   index bf15e9c8..7e1b538a 100644
   --- a/rust/src/runtime/workflow_poll_requests.rs
   +++ b/rust/src/runtime/workflow_poll_requests.rs
   @@ -167,8 +167,8 @@ store_methods! {
            if self.sweeps.workflow_poll_rows.take() {
                Self::sweep_workflow_poll_rows(tx).await?;
            }
   -        Self::bind_workflow_poll(tx, request).await?;
   -        let replay = self.deliverable_workflow_poll(tx, request, worker_id, queue).await?;
   +        let result = Self::bind_workflow_poll(tx, request).await?;
   +        let replay = self.deliverable_workflow_poll(tx, request, result, worker_id, queue).await?;
            if replay.is_none() {
                tx.replayed_poll = Some(request.keys.binding.clone());
            }
   @@ -178,8 +178,9 @@ store_methods! {
        /// `bindTaskKinds`: a live binding to another set refuses the poll; an
        /// absent or expired one is written for the pending window. The binding
        /// row stays locked to the end of the transaction, so a duplicate's
   -    /// claim waits and then sees this one's answer.
   -    pub(super) async fn bind_workflow_poll(tx: &mut Transaction<'_, DB>, request: &WorkflowPollRequest) -> Result<()> {
   +    /// claim waits and then sees this one's answer. The request's answer,
   +    /// read with the binding under its lock, for `deliverable_workflow_poll`.
   +    pub(super) async fn bind_workflow_poll(tx: &mut Transaction<'_, DB>, request: &WorkflowPollRequest) -> Result<Option<Value>> {
            let binding = json!({"task_kinds": request.kinds});
            Self::query(DB::QUERY_CACHE_LOCK)
                .bind(&request.keys.binding)
   @@ -187,25 +188,36 @@ store_methods! {
                .bind(epoch() + request.pending_ttl)
                .execute(&mut **tx)
                .await?;
   -        match Self::workflow_poll_row(tx, &request.keys.binding).await? {
   -            Some(bound) if bound["task_kinds"] != binding["task_kinds"] => Err(kinds_conflict(request, bound["task_kinds"].clone())),
   -            Some(_) => Ok(()),
   -            None => Self::put_workflow_poll_row(tx, &request.keys.binding, &binding, request.pending_ttl).await,
   +        let (mut bound, mut answer) = (None, None);
   +        for row in Self::query(DB::QUERY_CACHE_GET_PAIR).bind(&request.keys.binding).bind(&request.keys.result).fetch_all(&mut **tx).await? {
   +            if DB::string(&row, "key")? == request.keys.binding {
   +                bound = Some(row);
   +            } else {
   +                answer = Some(row);
   +            }
            }
   +        match Self::workflow_poll_row(tx, &request.keys.binding, bound).await? {
   +            Some(bound) if bound["task_kinds"] != binding["task_kinds"] => return Err(kinds_conflict(request, bound["task_kinds"].clone())),
   +            Some(_) => {}
   +            None => Self::put_workflow_poll_row(tx, &request.keys.binding, &binding, request.pending_ttl).await?,
   +        }
   +        Self::workflow_poll_row(tx, &request.keys.result, answer).await
        }
    
        /// The cached answer if its task can still be delivered; a stale answer
        /// is deleted. A workflow lease shows the task's current lease, as PHP
        /// rebuilds it; a validation lease replays as it was delivered. An
        /// answer of a kind this request did not ask for is not replayed.
   +    /// `result` is the answer `bind_workflow_poll` read.
        pub(super) async fn deliverable_workflow_poll(
            &self,
            tx: &mut Transaction<'_, DB>,
            request: &WorkflowPollRequest,
   +        result: Option<Value>,
            worker_id: &str,
            queue: &str,
        ) -> Result<Option<Value>> {
   -        let Some(mut result) = Self::workflow_poll_row(tx, &request.keys.result).await? else {
   +        let Some(mut result) = result else {
                return Ok(None);
            };
            if !result["task"].is_object() {
   @@ -235,10 +247,11 @@ store_methods! {
            &self,
            tx: &mut Transaction<'_, DB>,
            request: &WorkflowPollRequest,
   +        result: Option<Value>,
            worker_id: &str,
            queue: &str,
        ) -> Result<Option<Value>> {
   -        Ok(self.deliverable_workflow_poll(tx, request, worker_id, queue).await?.filter(|answer| answer["task"].is_object()))
   +        Ok(self.deliverable_workflow_poll(tx, request, result, worker_id, queue).await?.filter(|answer| answer["task"].is_object()))
        }
    
        /// `cachedTaskStillDeliverable` for a workflow task: leased on this
   @@ -371,8 +384,8 @@ store_methods! {
            #[cfg(debug_assertions)]
            super::poll_pauses::stop_at(&request.id, super::poll_pauses::PollPoint::Terminal).await;
            let mut tx = self.begin().await?;
   -        Self::bind_workflow_poll(&mut tx, request).await?;
   -        let lease = self.deliverable_workflow_lease(&mut tx, request, worker_id, queue).await?;
   +        let result = Self::bind_workflow_poll(&mut tx, request).await?;
   +        let lease = self.deliverable_workflow_lease(&mut tx, request, result, worker_id, queue).await?;
            if lease.is_none() {
                Self::remember_workflow_poll(&mut tx, request, &json!({"task": null, "poll_status": poll_status})).await?;
            }
   @@ -380,8 +393,10 @@ store_methods! {
            Ok(lease)
        }
    
   -    async fn workflow_poll_row(tx: &mut Transaction<'_, DB>, key: &str) -> Result<Option<Value>> {
   -        let Some(row) = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&mut **tx).await? else {
   +    /// The value of a cache row this transaction read; an expired row is
   +    /// deleted and has none.
   +    async fn workflow_poll_row(tx: &mut Transaction<'_, DB>, key: &str, row: Option<DB::Row>) -> Result<Option<Value>> {
   +        let Some(row) = row else {
                return Ok(None);
            };
            if DB::number(&row, "expiration")? <= epoch() {
8. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0020-t12-Let-a-claim-use-the-registration-row-its-fence-just-.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: 10,243 bytes, SHA-256 0998f68703f4b37116717e01dc97313ee3d0c028673467d8fd563192cd3193fa. This part: 10,243 bytes, SHA-256 0998f68703f4b37116717e01dc97313ee3d0c028673467d8fd563192cd3193fa. Parts concatenate in order; a part boundary is between complete file diffs.

   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index ad3656f2..32b12288 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -2168,12 +2168,14 @@ where
            fence: Option<&super::worker_poll_fence::WorkerPollFence>,
        ) -> Result<Option<Value>> {
            let mut tx = self.begin().await?;
   -        if let Some(fence) = fence
   -            && !self.worker_poll_fence_holds_in(&mut tx, fence, text(body, "worker_id")?).await?
   -        {
   -            return Err(super::worker_poll_fence::stale_claim());
   -        }
   -        let claimed = self.claim_in(&mut tx, body, kind, protocol_version, page, poll_request).await?;
   +        let registration = match fence {
   +            Some(fence) => match self.worker_poll_fence_registration_in(&mut tx, fence, text(body, "worker_id")?).await? {
   +                Some(registration) => Some(registration),
   +                None => return Err(super::worker_poll_fence::stale_claim()),
   +            },
   +            None => None,
   +        };
   +        let claimed = self.claim_in(&mut tx, body, kind, protocol_version, page, poll_request, registration).await?;
            tx.commit().await?;
            Ok(claimed)
        }
   @@ -2198,13 +2200,17 @@ where
                tx.commit().await?;
                return Ok(FirstProbe::Replayed(replay));
            }
   -        if let Some(fence) = fence
   -            && !self.worker_poll_fence_holds_in(&mut tx, fence, worker_id).await?
   -        {
   -            tx.commit().await?;
   -            return Ok(FirstProbe::Claimed(Err(super::worker_poll_fence::stale_claim())));
   -        }
   -        match self.claim_in(&mut tx, body, kind, protocol_version, page, Some(request)).await {
   +        let registration = match fence {
   +            Some(fence) => match self.worker_poll_fence_registration_in(&mut tx, fence, worker_id).await? {
   +                Some(registration) => Some(registration),
   +                None => {
   +                    tx.commit().await?;
   +                    return Ok(FirstProbe::Claimed(Err(super::worker_poll_fence::stale_claim())));
   +                }
   +            },
   +            None => None,
   +        };
   +        match self.claim_in(&mut tx, body, kind, protocol_version, page, Some(request), registration).await {
                Ok(claimed) => {
                    tx.commit().await?;
                    Ok(FirstProbe::Claimed(Ok(claimed)))
   @@ -2221,7 +2227,9 @@ where
    
        /// One claim inside the caller's transaction, which commits it: a
        /// multiplexed probe claims, turns the queue's cursor and records the
   -    /// poll request's answer as one unit.
   +    /// poll request's answer as one unit. `registration` is the worker's
   +    /// registration row, when the caller has just read it whole.
   +    #[allow(clippy::too_many_arguments)]
        pub(super) async fn claim_in(
            &self,
            tx: &mut super::transaction::Transaction<'_, DB>,
   @@ -2230,17 +2238,21 @@ where
            protocol_version: &str,
            page: Option<i64>,
            poll_request: Option<&super::workflow_poll_requests::WorkflowPollRequest>,
   +        registration: Option<DB::Row>,
        ) -> Result<Option<Value>> {
            let worker_id = text(body, "worker_id")?;
            let queue = text(body, "task_queue")?;
   -        let worker = Self::query(
   -            "SELECT * FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2",
   -        )
   -        .bind(worker_id)
   -        .bind(self.namespace.as_ref())
   -        .fetch_optional(&mut **tx)
   -        .await?
   -        .ok_or_else(|| refuse(StatusCode::CONFLICT, "worker_not_registered"))?;
   +        let worker = match registration {
   +            Some(worker) => worker,
   +            None => Self::query(
   +                "SELECT * FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2",
   +            )
   +            .bind(worker_id)
   +            .bind(self.namespace.as_ref())
   +            .fetch_optional(&mut **tx)
   +            .await?
   +            .ok_or_else(|| refuse(StatusCode::CONFLICT, "worker_not_registered"))?,
   +        };
            if DB::string(&worker, "task_queue")? != queue {
                return Err(refuse(StatusCode::CONFLICT, "task_queue_mismatch"));
            }
   @@ -3490,11 +3502,11 @@ where
            fence: &super::worker_poll_fence::WorkerPollFence,
            protocol: &str,
        ) -> Result<Option<bool>> {
   -        if !self.worker_poll_fence_holds_in(tx, fence, text(poll, "worker_id")?).await? {
   +        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).await {
   +        match self.claim_in(tx, poll, "activity", protocol, None, None, Some(registration)).await {
                Ok(claimed) => Ok(Some(claimed.is_some())),
                Err(_) => {
                    sqlx::raw_sql("ROLLBACK TO SAVEPOINT activity_poll").execute(&mut **tx).await?;
   @@ -4123,11 +4135,11 @@ where
        /// 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? {
   +        let Some(registration) = self.worker_poll_fence_registration_in(tx, &poll.fence, text(poll.body, "worker_id")?).await? else {
                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 {
   +        match self.claim_in(tx, poll.body, "workflow", poll.protocol, poll.page, Some(&poll.request), Some(registration)).await {
                Ok(claimed) => Ok(Some(claimed.is_some())),
                Err(_) => {
                    sqlx::raw_sql("ROLLBACK TO SAVEPOINT workflow_poll").execute(&mut **tx).await?;
   diff --git a/rust/src/runtime/update_validation.rs b/rust/src/runtime/update_validation.rs
   index 0b0ec511..0d26d807 100644
   --- a/rust/src/runtime/update_validation.rs
   +++ b/rust/src/runtime/update_validation.rs
   @@ -1057,7 +1057,7 @@ where
            order.sort_by_key(|kind| *kind != first);
            for kind in order {
                let leased = if kind == "workflow" {
   -                self.claim_in(&mut tx, body, "workflow", protocol_version, page, request).await?.map(|task| {
   +                self.claim_in(&mut tx, body, "workflow", protocol_version, page, request, None).await?.map(|task| {
                        Leased::Workflow(json!({"task":task,"poll_status":"leased","server_capabilities":super::store::capabilities()}))
                    })
                } else if let Some(claim) = self.claim_update_validation_in(&mut tx, worker_id, task_queue).await? {
   diff --git a/rust/src/runtime/worker_poll_fence.rs b/rust/src/runtime/worker_poll_fence.rs
   index 6b2a41e2..8d3ee111 100644
   --- a/rust/src/runtime/worker_poll_fence.rs
   +++ b/rust/src/runtime/worker_poll_fence.rs
   @@ -32,8 +32,17 @@ use std::collections::BTreeMap;
    
    pub(super) const STALE: &str = "stale_worker_registration";
    
   +/// A fence's column for the registration's active incarnation token.
   +macro_rules! incarnation_token {
   +    () => {
   +        "(SELECT dw_worker_registration_incarnations.token FROM dw_worker_registration_incarnations WHERE dw_worker_registration_incarnations.namespace=workflow_worker_registrations.namespace AND dw_worker_registration_incarnations.worker_id=workflow_worker_registrations.worker_id AND dw_worker_registration_incarnations.status='active' ORDER BY dw_worker_registration_incarnations.created_at DESC LIMIT 1) AS incarnation_token"
   +    };
   +}
   +
    /// The registration a fence reads.
   -const FENCE_ROW: &str = "SELECT status,task_queue,build_id,supported_workflow_types,workflow_definition_fingerprints,workflow_command_contracts,supported_activity_types,capabilities,process_metrics,updated_at,last_heartbeat_at,(SELECT dw_worker_registration_incarnations.token FROM dw_worker_registration_incarnations WHERE dw_worker_registration_incarnations.namespace=workflow_worker_registrations.namespace AND dw_worker_registration_incarnations.worker_id=workflow_worker_registrations.worker_id AND dw_worker_registration_incarnations.status='active' ORDER BY dw_worker_registration_incarnations.created_at DESC LIMIT 1) AS incarnation_token FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2";
   +const FENCE_ROW: &str = concat!("SELECT status,task_queue,build_id,supported_workflow_types,workflow_definition_fingerprints,workflow_command_contracts,supported_activity_types,capabilities,process_metrics,updated_at,last_heartbeat_at,", incarnation_token!(), " FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2");
   +/// `FENCE_ROW` with the rest of the registration, for a claim that reads it next.
   +const FENCE_REGISTRATION: &str = concat!("SELECT workflow_worker_registrations.*,", incarnation_token!(), " FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2");
    
    /// How a poll is admitted.
    pub(super) enum PollAdmission {
   @@ -232,6 +241,19 @@ store_methods! {
            let row = Self::query(FENCE_ROW).bind(self.namespace.as_ref()).bind(worker_id).fetch_optional(&mut **tx).await?;
            fence.holds::<DB>(row.as_ref(), self.worker_stale_after_seconds)
        }
   +
   +    /// `worker_poll_fence_holds_in` before a claim: the whole registration
   +    /// row the fence holds on, which the claim reads instead of reading it
   +    /// again, or `None` when the fence no longer holds.
   +    pub(super) async fn worker_poll_fence_registration_in(
   +        &self,
   +        tx: &mut super::transaction::Transaction<'_, DB>,
   +        fence: &WorkerPollFence,
   +        worker_id: &str,
   +    ) -> Result<Option<DB::Row>> {
   +        let row = Self::query(FENCE_REGISTRATION).bind(self.namespace.as_ref()).bind(worker_id).fetch_optional(&mut **tx).await?;
   +        Ok(if fence.holds::<DB>(row.as_ref(), self.worker_stale_after_seconds)? { row } else { None })
   +    }
    }
    
    #[cfg(test)]
9. rmcdaniel commented on Nov 6, 2026

   

   rmcdaniel

   on Nov 6, 2026

   MemberAuthor

   More actions

   Sanitized reference diff 0021-t12-Keep-a-run-s-tags-from-the-run-row-a-transaction-alr.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: 12,625 bytes, SHA-256 5107f116242ef169daa18c71e6ac69fa571946a4cce8a6d6b39ac41e7f6bc42c. This part: 12,625 bytes, SHA-256 5107f116242ef169daa18c71e6ac69fa571946a4cce8a6d6b39ac41e7f6bc42c. Parts concatenate in order; a part boundary is between complete file diffs.

   diff --git a/rust/src/runtime/activity_failures.rs b/rust/src/runtime/activity_failures.rs
   index c4d748c8..7c13c7d2 100644
   --- a/rust/src/runtime/activity_failures.rs
   +++ b/rust/src/runtime/activity_failures.rs
   @@ -431,10 +431,13 @@ where
                        "protocol_version":super::WORKER_PROTOCOL_VERSION,"server_capabilities":super::store::capabilities()}),
                });
            }
   -        Self::closed_run_outcome(&mut tx, &task, &body, reported).await?;
   +        let read = Self::closed_run_outcome(&mut tx, &task, &body, reported).await?;
            Self::fence(&task, &body)?;
            let run_id = DB::string(&task, "workflow_run_id")?;
   -        Self::active_run(&mut tx, &run_id).await?;
   +        match read {
   +            Some(run) => Self::active(&mut tx, run)?,
   +            None => Self::active_run(&mut tx, &run_id).await?,
   +        };
            let activity_id = DB::string(&attempt, "activity_execution_id")?;
            let activity =
                Self::query("SELECT * FROM activity_executions WHERE id=$1 AND workflow_run_id=$2")
   diff --git a/rust/src/runtime/standalone_activities.rs b/rust/src/runtime/standalone_activities.rs
   index c5d25dd5..7c2de98b 100644
   --- a/rust/src/runtime/standalone_activities.rs
   +++ b/rust/src/runtime/standalone_activities.rs
   @@ -356,11 +356,17 @@ where
            run_id: &str,
            close: HostClose<'_>,
        ) -> Result<bool> {
   -        let run = Self::query("SELECT workflow_type,workflow_class FROM workflow_runs WHERE id=$1")
   -            .bind(run_id)
   -            .fetch_one(&mut **tx)
   -            .await?;
   -        if DB::string(&run, "workflow_type")? != HOST_TYPE && DB::string(&run, "workflow_class")? != HOST_TYPE {
   +        let host = match tx.run_tags.get(run_id) {
   +            Some(tags) => tags.workflow_type == HOST_TYPE || tags.workflow_class == HOST_TYPE,
   +            None => {
   +                let run = Self::query("SELECT workflow_type,workflow_class FROM workflow_runs WHERE id=$1")
   +                    .bind(run_id)
   +                    .fetch_one(&mut **tx)
   +                    .await?;
   +                DB::string(&run, "workflow_type")? == HOST_TYPE || DB::string(&run, "workflow_class")? == HOST_TYPE
   +            }
   +        };
   +        if !host {
                return Ok(false);
            }
            let at = now();
   diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
   index 32b12288..11f23f5e 100644
   --- a/rust/src/runtime/store.rs
   +++ b/rust/src/runtime/store.rs
   @@ -3164,11 +3164,13 @@ where
            self.guard_worker_session_commands(&mut tx, &task, task_id, &body, protocol, commands).await?;
            // `MessageStreamService::validateCompletion`, before anything changes.
            let streams = super::message_streams::Completion::of(&body);
   -        let stream_instance = Self::query("SELECT workflow_instance_id FROM workflow_runs WHERE id=$1")
   +        // The run's whole row, which nothing below changes before it is
   +        // checked as the active run.
   +        let stream_run = Self::query("SELECT * FROM workflow_runs WHERE id=$1")
                .bind(DB::string(&task, "workflow_run_id")?)
                .fetch_one(&mut *tx)
                .await?;
   -        let stream_instance = DB::string(&stream_instance, "workflow_instance_id")?;
   +        let stream_instance = DB::string(&stream_run, "workflow_instance_id")?;
            if !streams.is_empty() {
                self.validate_stream_completion(&mut tx, &stream_instance, &DB::string(&task, "workflow_run_id")?, &streams)
                    .await?;
   @@ -3226,9 +3228,7 @@ where
                return Err(refuse(StatusCode::CONFLICT, "invalid_commands"));
            }
            let run_id = DB::string(&task, "workflow_run_id")?;
   -        let run = Self::active_run(&mut tx, &run_id)
   -            .await
   -            .map_err(|error| Self::not_recorded(&task, &body, error))?;
   +        let run = Self::active(&mut tx, stream_run).map_err(|error| Self::not_recorded(&task, &body, error))?;
            // `timeoutIfDeadlineExpired`: published PHP applies a run's deadlines
            // when one of its workflow tasks completes, and closes the run.
            let mut tx = self.close_expired_run(tx, &run, &task, &body["workflow_task_attempt"]).await?;
   @@ -3510,6 +3510,7 @@ where
                Ok(claimed) => Ok(Some(claimed.is_some())),
                Err(_) => {
                    sqlx::raw_sql("ROLLBACK TO SAVEPOINT activity_poll").execute(&mut **tx).await?;
   +                tx.run_tags.clear();
                    Ok(None)
                }
            }
   @@ -4013,10 +4014,13 @@ where
                    response: receipt,
                });
            }
   -        Self::closed_run_outcome(&mut tx, &task, &body, "completed").await?;
   +        let read = Self::closed_run_outcome(&mut tx, &task, &body, "completed").await?;
            Self::fence(&task, &body)?;
            let run_id = DB::string(&task, "workflow_run_id")?;
   -        let run = Self::active_run(&mut tx, &run_id).await?;
   +        let run = match read {
   +            Some(run) => Self::active(&mut tx, run)?,
   +            None => Self::active_run(&mut tx, &run_id).await?,
   +        };
            let activity_id = DB::string(&attempt, "activity_execution_id")?;
            let activity =
                Self::query("SELECT * FROM activity_executions WHERE id=$1 AND workflow_run_id=$2")
   @@ -4143,6 +4147,7 @@ where
                Ok(claimed) => Ok(Some(claimed.is_some())),
                Err(_) => {
                    sqlx::raw_sql("ROLLBACK TO SAVEPOINT workflow_poll").execute(&mut **tx).await?;
   +                tx.run_tags.clear();
                    Ok(None)
                }
            }
   @@ -4560,24 +4565,25 @@ where
    
        /// An activity result for a run that completed or failed while the
        /// activity was open: PHP's `run_closed` outcome, recorded nowhere.
   -    /// Cancelled and terminated runs keep their own receipts.
   +    /// Cancelled and terminated runs keep their own receipts. The run's
   +    /// whole row, when it read it.
        pub(super) async fn closed_run_outcome(
            tx: &mut Transaction<'_, DB>,
            task: &DB::Row,
            body: &Value,
            outcome: &str,
   -    ) -> Result<()> {
   +    ) -> Result<Option<DB::Row>> {
            if DB::string(task, "status")? != "leased"
                || DB::optional_string(task, "lease_owner")?.as_deref() != Some(text(body, "lease_owner")?)
            {
   -            return Ok(());
   +            return Ok(None);
            }
   -        let run = Self::query("SELECT status FROM workflow_runs WHERE id=$1")
   +        let run = Self::query("SELECT * FROM workflow_runs WHERE id=$1")
                .bind(DB::string(task, "workflow_run_id")?)
                .fetch_one(&mut **tx)
                .await?;
            if !matches!(DB::string(&run, "status")?.as_str(), "completed" | "failed") {
   -            return Ok(());
   +            return Ok(Some(run));
            }
            Err(super::RuntimeError::Protocol {
                status: StatusCode::CONFLICT,
   @@ -4591,6 +4597,13 @@ where
                .bind(run_id)
                .fetch_one(&mut **tx)
                .await?;
   +        Self::active(tx, run)
   +    }
   +
   +    /// `active_run` of the run's whole row, which this transaction has
   +    /// just read and not changed since.
   +    pub(super) fn active(tx: &mut Transaction<'_, DB>, run: DB::Row) -> Result<DB::Row> {
   +        Self::keep_run_tags(tx, &run);
            if !matches!(
                DB::string(&run, "status")?.as_str(),
                "pending" | "running" | "waiting"
   @@ -4609,6 +4622,26 @@ where
            Ok(run)
        }
    
   +    /// Keeps a whole run row's tags for the rest of the transaction. A row
   +    /// missing any of them is not kept, so a reader of its tags queries
   +    /// them and fails as it would have.
   +    pub(super) fn keep_run_tags(tx: &mut Transaction<'_, DB>, run: &DB::Row) {
   +        let tags = || -> Result<(String, super::transaction::RunTags)> {
   +            Ok((DB::string(run, "id")?, super::transaction::RunTags {
   +                namespace: DB::string(run, "namespace")?,
   +                priority: DB::number(run, "priority")?,
   +                fairness_key: DB::optional_string(run, "fairness_key")?,
   +                fairness_weight: DB::number(run, "fairness_weight")?,
   +                compatibility: DB::optional_string(run, "compatibility")?,
   +                workflow_type: DB::string(run, "workflow_type")?,
   +                workflow_class: DB::string(run, "workflow_class")?,
   +            }))
   +        };
   +        if let Ok((id, tags)) = tags() {
   +            tx.run_tags.insert(id, tags);
   +        }
   +    }
   +
        pub(super) async fn create_task(
            tx: &mut Transaction<'_, DB>,
            run_id: &str,
   @@ -4632,19 +4665,24 @@ where
            tx.summary_changed(run_id);
            let task_id = id();
            // A task carries its run's priority and fairness tags, as in PHP.
   -        let run = Self::query("SELECT namespace,priority,fairness_key,fairness_weight,compatibility FROM workflow_runs WHERE id=$1")
   -            .bind(run_id)
   -            .fetch_one(&mut **tx)
   -            .await?;
   -        let namespace = DB::string(&run, "namespace")?;
   +        let (namespace, priority, fairness_key, fairness_weight, compatibility) = match tx.run_tags.get(run_id) {
   +            Some(tags) => (tags.namespace.clone(), tags.priority, tags.fairness_key.clone(), tags.fairness_weight, tags.compatibility.clone()),
   +            None => {
   +                let run = Self::query("SELECT namespace,priority,fairness_key,fairness_weight,compatibility FROM workflow_runs WHERE id=$1")
   +                    .bind(run_id)
   +                    .fetch_one(&mut **tx)
   +                    .await?;
   +                (DB::string(&run, "namespace")?, DB::number(&run, "priority")?, DB::optional_string(&run, "fairness_key")?,
   +                    DB::number(&run, "fairness_weight")?, DB::optional_string(&run, "compatibility")?)
   +            }
   +        };
            let created = now();
            let dispatched_at = dispatched.then_some(created);
   -        let fairness_key = DB::optional_string(&run, "fairness_key")?;
            Self::query("INSERT INTO workflow_tasks(id,workflow_run_id,task_type,status,payload,queue,available_at,namespace,created_at,updated_at,last_dispatched_at,last_dispatch_attempt_at,priority,fairness_key,fairness_weight,compatibility) VALUES ($1,$2,$3,'ready',$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)")
            .bind(&task_id).bind(run_id).bind(kind).bind(DB::document(&payload)).bind(queue).bind(DB::bind_time(created)).bind(&namespace)
            .bind(DB::bind_time(created)).bind(DB::bind_time(created)).bind(DB::optional_time(dispatched_at)).bind(DB::optional_time(dispatched_at))
   -        .bind(DB::number(&run, "priority")?).bind(fairness_key.as_deref()).bind(DB::number(&run, "fairness_weight")?)
   -        .bind(DB::optional_string(&run, "compatibility")?.as_deref())
   +        .bind(priority).bind(fairness_key.as_deref()).bind(fairness_weight)
   +        .bind(compatibility.as_deref())
            .execute(&mut **tx).await?;
            Ok(task_id)
        }
   diff --git a/rust/src/runtime/transaction.rs b/rust/src/runtime/transaction.rs
   index fd251b83..720c652b 100644
   --- a/rust/src/runtime/transaction.rs
   +++ b/rust/src/runtime/transaction.rs
   @@ -2,10 +2,21 @@
    use super::{Result, backend::Backend, store::Store};
    use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type};
    use std::{
   -    collections::BTreeSet,
   +    collections::{BTreeSet, HashMap},
        ops::{Deref, DerefMut},
    };
    
   +/// A run's columns that no transition changes once the run is stored.
   +pub(super) struct RunTags {
   +    pub(super) namespace: String,
   +    pub(super) priority: i64,
   +    pub(super) fairness_key: Option<String>,
   +    pub(super) fairness_weight: i64,
   +    pub(super) compatibility: Option<String>,
   +    pub(super) workflow_type: String,
   +    pub(super) workflow_class: 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`
   @@ -20,6 +31,9 @@ pub(super) struct Transaction<'a, DB: Backend> {
        /// The binding key of a workflow poll request this transaction has
        /// already bound and found no answer to replay for.
        pub(super) replayed_poll: Option<String>,
   +    /// The tags of the runs this transaction has read whole, by run id
   +    /// (`Store::keep_run_tags`).
   +    pub(super) run_tags: HashMap<String, RunTags>,
    }
    
    impl<'a, DB: Backend> Transaction<'a, DB> {
   @@ -35,6 +49,7 @@ impl<'a, DB: Backend> Transaction<'a, DB> {
                meter: None,
                scope_history: None,
                replayed_poll: None,
   +            run_tags: HashMap::new(),
            }
        }
    
10. rmcdaniel commented on Nov 6, 2026

    

    rmcdaniel

    on Nov 6, 2026

    MemberAuthor

    More actions

    Sanitized rebased reference for report topic 12: 0017-t12-Read-a-run-s-authored-sequences-with-its-row.patch.

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

    Sanitized patch: 3,569 bytes, SHA-256 9a5cd5b265d2474b5c7fcab5393abb5f208f60f666f83ee56321732e241bb561. Original-base sanitized identity: fbb211a4e75583b3358c011a64050d533392722a98feea3c4d28153034aa62e6. No patch was applied or executed during this update.

    diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
    index 66cc0684..c5c3868f 100644
    --- a/rust/src/runtime/store.rs
    +++ b/rust/src/runtime/store.rs
    @@ -1281,24 +1281,15 @@ where
             // Public control command order and deterministic authored call order
             // are different counters in PHP. Signals must never shift a replayed
             // activity/timer/wait's authored position.
    -        let activities = Self::query(
    -            "SELECT sequence FROM activity_executions WHERE workflow_run_id=$1 ORDER BY sequence DESC LIMIT 1",
    -        )
    -        .bind(run_id)
    -        .fetch_optional(&mut **tx)
    -        .await?
    -        .map(|row| DB::number(&row, "sequence"))
    -        .transpose()?
    -        .unwrap_or(0);
    -        let timers = Self::query(
    -            "SELECT sequence FROM workflow_run_timers WHERE workflow_run_id=$1 ORDER BY sequence DESC LIMIT 1",
    -        )
    -        .bind(run_id)
    -        .fetch_optional(&mut **tx)
    -        .await?
    -        .map(|row| DB::number(&row, "sequence"))
    -        .transpose()?
    -        .unwrap_or(0);
    +        // The run's row and each authored table's largest sequence, in one
    +        // statement; a table with no rows for the run counts as 0.
    +        let run = Self::query("SELECT r.cancellation_delivery_sequence,r.run_number,(SELECT sequence FROM activity_executions WHERE workflow_run_id=$1 ORDER BY sequence DESC LIMIT 1) AS activities,(SELECT sequence FROM workflow_run_timers WHERE workflow_run_id=$2 ORDER BY sequence DESC LIMIT 1) AS timers,(SELECT sequence FROM workflow_child_calls WHERE parent_workflow_run_id=$3 ORDER BY sequence DESC LIMIT 1) AS children FROM workflow_runs r WHERE r.id=$4")
    +            .bind(run_id).bind(run_id).bind(run_id).bind(run_id)
    +            .fetch_one(&mut **tx)
    +            .await?;
    +        let activities = DB::optional_number(&run, "activities")?.unwrap_or(0);
    +        let timers = DB::optional_number(&run, "timers")?.unwrap_or(0);
    +        let children = DB::optional_number(&run, "children")?.unwrap_or(0);
             let wait = Self::query("SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type IN ('SignalWaitOpened','ConditionWaitOpened','VersionMarkerRecorded','MemoUpserted','SideEffectRecorded','SearchAttributesUpserted') ORDER BY sequence DESC LIMIT 1")
                 .bind(run_id).fetch_optional(&mut **tx).await?;
             let wait_sequence = wait
    @@ -1306,8 +1297,6 @@ where
                 .transpose()?
                 .and_then(|value| value["sequence"].as_i64())
                 .unwrap_or(0);
    -        let children = Self::query("SELECT sequence FROM workflow_child_calls WHERE parent_workflow_run_id=$1 ORDER BY sequence DESC LIMIT 1")
    -            .bind(run_id).fetch_optional(&mut **tx).await?.map(|row| DB::number(&row,"sequence")).transpose()?.unwrap_or(0);
             // `nextDurableCommandSequence`: a scope opening consumes its sequence,
             // and a delivered scope boundary its whole span, as a delivered run
             // request does.
    @@ -1321,11 +1310,6 @@ where
                     scoped = scoped.max(span.and_then(|span| sequence.checked_add(span - 1)).unwrap_or(sequence));
                 }
             }
    -        let run =
    -            Self::query("SELECT cancellation_delivery_sequence,run_number FROM workflow_runs WHERE id=$1")
    -                .bind(run_id)
    -                .fetch_one(&mut **tx)
    -                .await?;
             // A redriven run replays the activities before its resume step from
             // reused completions, which have no execution rows.
             let mut reused = 0;
11. rmcdaniel commented on Nov 6, 2026

    

    rmcdaniel

    on Nov 6, 2026

    MemberAuthor

    More actions

    Sanitized rebased reference for report topic 12: 0018-t12-Read-a-completed-task-s-run-status-before-marking-it.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: 1,866 bytes, SHA-256 9f0ec14c9efad65dd34ceba07d321ada27bfa21c295220feccbf31ff71b73290. Original-base sanitized identity: 0d26bdf3ad8b7d0303c7799912b6365fff536a1f423a4f3f91a5f5fd23967ebe. No patch was applied or executed during this update.

    diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
    index c5c3868f..bb468d50 100644
    --- a/rust/src/runtime/store.rs
    +++ b/rust/src/runtime/store.rs
    @@ -3422,15 +3422,27 @@ where
             Self::record_completion(&mut 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.
    -        Self::query("UPDATE workflow_runs SET status='waiting' WHERE id=$1 AND status IN ('pending','running')")
    -            .bind(&run_id)
    -            .execute(&mut *tx)
    -            .await?;
    +        // A run already waiting or closed is read and left as it is.
             let current = Self::query("SELECT status FROM workflow_runs WHERE id=$1")
                 .bind(&run_id)
                 .fetch_one(&mut *tx)
                 .await?;
    -        let run_status = DB::string(&current, "status")?;
    +        let mut run_status = DB::string(&current, "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)
    +                .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)
    +                    .await?;
    +                DB::string(&current, "status")?
    +            };
    +        }
             if let Some(worker) = &worker {
                 Self::attribute_worker_terminal(&mut tx, &run_id, task_id, worker).await?;
             }
12. rmcdaniel commented on Nov 6, 2026

    

    rmcdaniel

    on Nov 6, 2026

    MemberAuthor

    More actions

    Sanitized rebased reference for report topic 12: 0019-t12-Read-a-workflow-poll-s-binding-and-result-cache-rows.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: 8,846 bytes, SHA-256 2a7dd8f79d1e1d10f159863923fd9cecc538e560a16f83c07ce042ae12fba7c4. Original-base sanitized identity: 310bae18f59ce4bc467471607644a8b7a0427b7923d7bd320ac5767612bc1c8e. No patch was applied or executed during this update.

    diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs
    index b751e2f7..d28a4bcc 100644
    --- a/rust/src/runtime/backend.rs
    +++ b/rust/src/runtime/backend.rs
    @@ -80,6 +80,9 @@ pub(super) trait Backend: Database {
         const QUERY_CACHE_SCAN: &str = "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\" LIKE 'dw-native-query:%' AND value::jsonb->>'namespace'=$1 ORDER BY \"key\" LIMIT 65";
         const QUERY_CACHE_GET: &str =
             "SELECT value,expiration FROM dw_query_cache WHERE \"key\"=$1";
    +    /// Two keys' rows, in one read.
    +    const QUERY_CACHE_GET_PAIR: &str =
    +        "SELECT \"key\",value,expiration FROM dw_query_cache WHERE \"key\" IN ($1,$2)";
         const QUERY_CACHE_INSERT: &str =
             "INSERT INTO dw_query_cache(\"key\",value,expiration) VALUES ($1,$2,$3)";
         const QUERY_CACHE_UPDATE: &str =
    @@ -603,6 +606,8 @@ impl Backend for MySql {
         type EncodedDocument = Json<Value>;
         const QUERY_CACHE_GET: &str =
             "SELECT value,expiration FROM dw_query_cache WHERE `key`=$1";
    +    const QUERY_CACHE_GET_PAIR: &str =
    +        "SELECT `key`,value,expiration FROM dw_query_cache WHERE `key` IN ($1,$2)";
         const QUERY_CACHE_INSERT: &str =
             "INSERT INTO dw_query_cache(`key`,value,expiration) VALUES ($1,$2,$3)";
         const QUERY_CACHE_UPDATE: &str =
    diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
    index bb468d50..e5be949d 100644
    --- a/rust/src/runtime/store.rs
    +++ b/rust/src/runtime/store.rs
    @@ -2267,8 +2267,8 @@ where
             if let Some(request) = poll_request
                 && !request.replayed_in(tx)
             {
    -            Self::bind_workflow_poll(tx, request).await?;
    -            if let Some(answer) = self.deliverable_workflow_lease(tx, request, worker_id, queue).await?
    +            let result = Self::bind_workflow_poll(tx, request).await?;
    +            if let Some(answer) = self.deliverable_workflow_lease(tx, request, result, worker_id, queue).await?
                     && answer["task"]["task_kind"] != "update_validation"
                 {
                     return Ok(Some(answer["task"].clone()));
    diff --git a/rust/src/runtime/update_validation.rs b/rust/src/runtime/update_validation.rs
    index 604c61b4..0b0ec511 100644
    --- a/rust/src/runtime/update_validation.rs
    +++ b/rust/src/runtime/update_validation.rs
    @@ -1045,8 +1045,8 @@ where
             // A concurrent duplicate may have leased since this poll began: the
             // binding row, locked first, makes it wait for that one's answer.
             if let Some(request) = request {
    -            Self::bind_workflow_poll(&mut tx, request).await?;
    -            if let Some(answer) = self.deliverable_workflow_lease(&mut tx, request, worker_id, task_queue).await? {
    +            let result = Self::bind_workflow_poll(&mut tx, request).await?;
    +            if let Some(answer) = self.deliverable_workflow_lease(&mut tx, request, result, worker_id, task_queue).await? {
                     tx.commit().await?;
                     return Ok(Some(Probe::Replayed(answer)));
                 }
    diff --git a/rust/src/runtime/workflow_poll_requests.rs b/rust/src/runtime/workflow_poll_requests.rs
    index bf15e9c8..7e1b538a 100644
    --- a/rust/src/runtime/workflow_poll_requests.rs
    +++ b/rust/src/runtime/workflow_poll_requests.rs
    @@ -167,8 +167,8 @@ store_methods! {
             if self.sweeps.workflow_poll_rows.take() {
                 Self::sweep_workflow_poll_rows(tx).await?;
             }
    -        Self::bind_workflow_poll(tx, request).await?;
    -        let replay = self.deliverable_workflow_poll(tx, request, worker_id, queue).await?;
    +        let result = Self::bind_workflow_poll(tx, request).await?;
    +        let replay = self.deliverable_workflow_poll(tx, request, result, worker_id, queue).await?;
             if replay.is_none() {
                 tx.replayed_poll = Some(request.keys.binding.clone());
             }
    @@ -178,8 +178,9 @@ store_methods! {
         /// `bindTaskKinds`: a live binding to another set refuses the poll; an
         /// absent or expired one is written for the pending window. The binding
         /// row stays locked to the end of the transaction, so a duplicate's
    -    /// claim waits and then sees this one's answer.
    -    pub(super) async fn bind_workflow_poll(tx: &mut Transaction<'_, DB>, request: &WorkflowPollRequest) -> Result<()> {
    +    /// claim waits and then sees this one's answer. The request's answer,
    +    /// read with the binding under its lock, for `deliverable_workflow_poll`.
    +    pub(super) async fn bind_workflow_poll(tx: &mut Transaction<'_, DB>, request: &WorkflowPollRequest) -> Result<Option<Value>> {
             let binding = json!({"task_kinds": request.kinds});
             Self::query(DB::QUERY_CACHE_LOCK)
                 .bind(&request.keys.binding)
    @@ -187,25 +188,36 @@ store_methods! {
                 .bind(epoch() + request.pending_ttl)
                 .execute(&mut **tx)
                 .await?;
    -        match Self::workflow_poll_row(tx, &request.keys.binding).await? {
    -            Some(bound) if bound["task_kinds"] != binding["task_kinds"] => Err(kinds_conflict(request, bound["task_kinds"].clone())),
    -            Some(_) => Ok(()),
    -            None => Self::put_workflow_poll_row(tx, &request.keys.binding, &binding, request.pending_ttl).await,
    +        let (mut bound, mut answer) = (None, None);
    +        for row in Self::query(DB::QUERY_CACHE_GET_PAIR).bind(&request.keys.binding).bind(&request.keys.result).fetch_all(&mut **tx).await? {
    +            if DB::string(&row, "key")? == request.keys.binding {
    +                bound = Some(row);
    +            } else {
    +                answer = Some(row);
    +            }
             }
    +        match Self::workflow_poll_row(tx, &request.keys.binding, bound).await? {
    +            Some(bound) if bound["task_kinds"] != binding["task_kinds"] => return Err(kinds_conflict(request, bound["task_kinds"].clone())),
    +            Some(_) => {}
    +            None => Self::put_workflow_poll_row(tx, &request.keys.binding, &binding, request.pending_ttl).await?,
    +        }
    +        Self::workflow_poll_row(tx, &request.keys.result, answer).await
         }
     
         /// The cached answer if its task can still be delivered; a stale answer
         /// is deleted. A workflow lease shows the task's current lease, as PHP
         /// rebuilds it; a validation lease replays as it was delivered. An
         /// answer of a kind this request did not ask for is not replayed.
    +    /// `result` is the answer `bind_workflow_poll` read.
         pub(super) async fn deliverable_workflow_poll(
             &self,
             tx: &mut Transaction<'_, DB>,
             request: &WorkflowPollRequest,
    +        result: Option<Value>,
             worker_id: &str,
             queue: &str,
         ) -> Result<Option<Value>> {
    -        let Some(mut result) = Self::workflow_poll_row(tx, &request.keys.result).await? else {
    +        let Some(mut result) = result else {
                 return Ok(None);
             };
             if !result["task"].is_object() {
    @@ -235,10 +247,11 @@ store_methods! {
             &self,
             tx: &mut Transaction<'_, DB>,
             request: &WorkflowPollRequest,
    +        result: Option<Value>,
             worker_id: &str,
             queue: &str,
         ) -> Result<Option<Value>> {
    -        Ok(self.deliverable_workflow_poll(tx, request, worker_id, queue).await?.filter(|answer| answer["task"].is_object()))
    +        Ok(self.deliverable_workflow_poll(tx, request, result, worker_id, queue).await?.filter(|answer| answer["task"].is_object()))
         }
     
         /// `cachedTaskStillDeliverable` for a workflow task: leased on this
    @@ -371,8 +384,8 @@ store_methods! {
             #[cfg(debug_assertions)]
             super::poll_pauses::stop_at(&request.id, super::poll_pauses::PollPoint::Terminal).await;
             let mut tx = self.begin().await?;
    -        Self::bind_workflow_poll(&mut tx, request).await?;
    -        let lease = self.deliverable_workflow_lease(&mut tx, request, worker_id, queue).await?;
    +        let result = Self::bind_workflow_poll(&mut tx, request).await?;
    +        let lease = self.deliverable_workflow_lease(&mut tx, request, result, worker_id, queue).await?;
             if lease.is_none() {
                 Self::remember_workflow_poll(&mut tx, request, &json!({"task": null, "poll_status": poll_status})).await?;
             }
    @@ -380,8 +393,10 @@ store_methods! {
             Ok(lease)
         }
     
    -    async fn workflow_poll_row(tx: &mut Transaction<'_, DB>, key: &str) -> Result<Option<Value>> {
    -        let Some(row) = Self::query(DB::QUERY_CACHE_GET).bind(key).fetch_optional(&mut **tx).await? else {
    +    /// The value of a cache row this transaction read; an expired row is
    +    /// deleted and has none.
    +    async fn workflow_poll_row(tx: &mut Transaction<'_, DB>, key: &str, row: Option<DB::Row>) -> Result<Option<Value>> {
    +        let Some(row) = row else {
                 return Ok(None);
             };
             if DB::number(&row, "expiration")? <= epoch() {
13. rmcdaniel commented on Nov 6, 2026

    

    rmcdaniel

    on Nov 6, 2026

    MemberAuthor

    More actions

    Sanitized rebased reference for report topic 12: 0020-t12-Let-a-claim-use-the-registration-row-its-fence-just-.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: 11,084 bytes, SHA-256 89a13457fffaeebf438e024b29d05236dd559149a11fba12e12fe42e80379328. Original-base sanitized identity: 0998f68703f4b37116717e01dc97313ee3d0c028673467d8fd563192cd3193fa. No patch was applied or executed during this update.

    diff --git a/rust/src/runtime/ready_age_tests.rs b/rust/src/runtime/ready_age_tests.rs
    index 2b3c3f71..798718e6 100644
    --- a/rust/src/runtime/ready_age_tests.rs
    +++ b/rust/src/runtime/ready_age_tests.rs
    @@ -226,7 +226,7 @@ store_methods! {
                 let request=json!({"worker_id":"ready-age-worker","task_queue":"ready-age","timeout_seconds":0,
                     "poll_request_id":format!("rollback-{}",ulid::Ulid::new())});
                 let mut tx=self.begin().await.unwrap();
    -            let claim=self.claim_in(&mut tx,&request,kind,"1.20",None,None).await.unwrap().unwrap();
    +            let claim=self.claim_in(&mut tx,&request,kind,"1.20",None,None,None).await.unwrap().unwrap();
                 assert_eq!(claim["task_id"],task_id);
                 tx.rollback().await.unwrap();
                 let row=self.ready_age_row(&task_id).await;
    diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
    index e5be949d..76ab4e7b 100644
    --- a/rust/src/runtime/store.rs
    +++ b/rust/src/runtime/store.rs
    @@ -2176,12 +2176,14 @@ where
             fence: Option<&super::worker_poll_fence::WorkerPollFence>,
         ) -> Result<Option<Value>> {
             let mut tx = self.begin().await?;
    -        if let Some(fence) = fence
    -            && !self.worker_poll_fence_holds_in(&mut tx, fence, text(body, "worker_id")?).await?
    -        {
    -            return Err(super::worker_poll_fence::stale_claim());
    -        }
    -        let claimed = self.claim_in(&mut tx, body, kind, protocol_version, page, poll_request).await?;
    +        let registration = match fence {
    +            Some(fence) => match self.worker_poll_fence_registration_in(&mut tx, fence, text(body, "worker_id")?).await? {
    +                Some(registration) => Some(registration),
    +                None => return Err(super::worker_poll_fence::stale_claim()),
    +            },
    +            None => None,
    +        };
    +        let claimed = self.claim_in(&mut tx, body, kind, protocol_version, page, poll_request, registration).await?;
             tx.commit().await?;
             Ok(claimed)
         }
    @@ -2206,13 +2208,17 @@ where
                 tx.commit().await?;
                 return Ok(FirstProbe::Replayed(replay));
             }
    -        if let Some(fence) = fence
    -            && !self.worker_poll_fence_holds_in(&mut tx, fence, worker_id).await?
    -        {
    -            tx.commit().await?;
    -            return Ok(FirstProbe::Claimed(Err(super::worker_poll_fence::stale_claim())));
    -        }
    -        match self.claim_in(&mut tx, body, kind, protocol_version, page, Some(request)).await {
    +        let registration = match fence {
    +            Some(fence) => match self.worker_poll_fence_registration_in(&mut tx, fence, worker_id).await? {
    +                Some(registration) => Some(registration),
    +                None => {
    +                    tx.commit().await?;
    +                    return Ok(FirstProbe::Claimed(Err(super::worker_poll_fence::stale_claim())));
    +                }
    +            },
    +            None => None,
    +        };
    +        match self.claim_in(&mut tx, body, kind, protocol_version, page, Some(request), registration).await {
                 Ok(claimed) => {
                     tx.commit().await?;
                     Ok(FirstProbe::Claimed(Ok(claimed)))
    @@ -2229,7 +2235,9 @@ where
     
         /// One claim inside the caller's transaction, which commits it: a
         /// multiplexed probe claims, turns the queue's cursor and records the
    -    /// poll request's answer as one unit.
    +    /// poll request's answer as one unit. `registration` is the worker's
    +    /// registration row, when the caller has just read it whole.
    +    #[allow(clippy::too_many_arguments)]
         pub(super) async fn claim_in(
             &self,
             tx: &mut super::transaction::Transaction<'_, DB>,
    @@ -2238,17 +2246,21 @@ where
             protocol_version: &str,
             page: Option<i64>,
             poll_request: Option<&super::workflow_poll_requests::WorkflowPollRequest>,
    +        registration: Option<DB::Row>,
         ) -> Result<Option<Value>> {
             let worker_id = text(body, "worker_id")?;
             let queue = text(body, "task_queue")?;
    -        let worker = Self::query(
    -            "SELECT * FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2",
    -        )
    -        .bind(worker_id)
    -        .bind(self.namespace.as_ref())
    -        .fetch_optional(&mut **tx)
    -        .await?
    -        .ok_or_else(|| refuse(StatusCode::CONFLICT, "worker_not_registered"))?;
    +        let worker = match registration {
    +            Some(worker) => worker,
    +            None => Self::query(
    +                "SELECT * FROM workflow_worker_registrations WHERE worker_id=$1 AND namespace=$2",
    +            )
    +            .bind(worker_id)
    +            .bind(self.namespace.as_ref())
    +            .fetch_optional(&mut **tx)
    +            .await?
    +            .ok_or_else(|| refuse(StatusCode::CONFLICT, "worker_not_registered"))?,
    +        };
             if DB::string(&worker, "task_queue")? != queue {
                 return Err(refuse(StatusCode::CONFLICT, "task_queue_mismatch"));
             }
    @@ -3503,11 +3515,11 @@ where
             fence: &super::worker_poll_fence::WorkerPollFence,
             protocol: &str,
         ) -> Result<Option<bool>> {
    -        if !self.worker_poll_fence_holds_in(tx, fence, text(poll, "worker_id")?).await? {
    +        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).await {
    +        match self.claim_in(tx, poll, "activity", protocol, None, None, Some(registration)).await {
                 Ok(claimed) => Ok(Some(claimed.is_some())),
                 Err(_) => {
                     sqlx::raw_sql("ROLLBACK TO SAVEPOINT activity_poll").execute(&mut **tx).await?;
    @@ -4136,11 +4148,11 @@ where
         /// 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? {
    +        let Some(registration) = self.worker_poll_fence_registration_in(tx, &poll.fence, text(poll.body, "worker_id")?).await? else {
                 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 {
    +        match self.claim_in(tx, poll.body, "workflow", poll.protocol, poll.page, Some(&poll.request), Some(registration)).await {
                 Ok(claimed) => Ok(Some(claimed.is_some())),
                 Err(_) => {
                     sqlx::raw_sql("ROLLBACK TO SAVEPOINT workflow_poll").execute(&mut **tx).await?;
    diff --git a/rust/src/runtime/update_validation.rs b/rust/src/runtime/update_validation.rs
    index 0b0ec511..0d26d807 100644
    --- a/rust/src/runtime/update_validation.rs
    +++ b/rust/src/runtime/update_validation.rs
    @@ -1057,7 +1057,7 @@ where
             order.sort_by_key(|kind| *kind != first);
             for kind in order {
                 let leased = if kind == "workflow" {
    -                self.claim_in(&mut tx, body, "workflow", protocol_version, page, request).await?.map(|task| {
    +                self.claim_in(&mut tx, body, "workflow", protocol_version, page, request, None).await?.map(|task| {
                         Leased::Workflow(json!({"task":task,"poll_status":"leased","server_capabilities":super::store::capabilities()}))
                     })
                 } else if let Some(claim) = self.claim_update_validation_in(&mut tx, worker_id, task_queue).await? {
    diff --git a/rust/src/runtime/worker_poll_fence.rs b/rust/src/runtime/worker_poll_fence.rs
    index 6b2a41e2..8d3ee111 100644
    --- a/rust/src/runtime/worker_poll_fence.rs
    +++ b/rust/src/runtime/worker_poll_fence.rs
    @@ -32,8 +32,17 @@ use std::collections::BTreeMap;
     
     pub(super) const STALE: &str = "stale_worker_registration";
     
    +/// A fence's column for the registration's active incarnation token.
    +macro_rules! incarnation_token {
    +    () => {
    +        "(SELECT dw_worker_registration_incarnations.token FROM dw_worker_registration_incarnations WHERE dw_worker_registration_incarnations.namespace=workflow_worker_registrations.namespace AND dw_worker_registration_incarnations.worker_id=workflow_worker_registrations.worker_id AND dw_worker_registration_incarnations.status='active' ORDER BY dw_worker_registration_incarnations.created_at DESC LIMIT 1) AS incarnation_token"
    +    };
    +}
    +
     /// The registration a fence reads.
    -const FENCE_ROW: &str = "SELECT status,task_queue,build_id,supported_workflow_types,workflow_definition_fingerprints,workflow_command_contracts,supported_activity_types,capabilities,process_metrics,updated_at,last_heartbeat_at,(SELECT dw_worker_registration_incarnations.token FROM dw_worker_registration_incarnations WHERE dw_worker_registration_incarnations.namespace=workflow_worker_registrations.namespace AND dw_worker_registration_incarnations.worker_id=workflow_worker_registrations.worker_id AND dw_worker_registration_incarnations.status='active' ORDER BY dw_worker_registration_incarnations.created_at DESC LIMIT 1) AS incarnation_token FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2";
    +const FENCE_ROW: &str = concat!("SELECT status,task_queue,build_id,supported_workflow_types,workflow_definition_fingerprints,workflow_command_contracts,supported_activity_types,capabilities,process_metrics,updated_at,last_heartbeat_at,", incarnation_token!(), " FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2");
    +/// `FENCE_ROW` with the rest of the registration, for a claim that reads it next.
    +const FENCE_REGISTRATION: &str = concat!("SELECT workflow_worker_registrations.*,", incarnation_token!(), " FROM workflow_worker_registrations WHERE namespace=$1 AND worker_id=$2");
     
     /// How a poll is admitted.
     pub(super) enum PollAdmission {
    @@ -232,6 +241,19 @@ store_methods! {
             let row = Self::query(FENCE_ROW).bind(self.namespace.as_ref()).bind(worker_id).fetch_optional(&mut **tx).await?;
             fence.holds::<DB>(row.as_ref(), self.worker_stale_after_seconds)
         }
    +
    +    /// `worker_poll_fence_holds_in` before a claim: the whole registration
    +    /// row the fence holds on, which the claim reads instead of reading it
    +    /// again, or `None` when the fence no longer holds.
    +    pub(super) async fn worker_poll_fence_registration_in(
    +        &self,
    +        tx: &mut super::transaction::Transaction<'_, DB>,
    +        fence: &WorkerPollFence,
    +        worker_id: &str,
    +    ) -> Result<Option<DB::Row>> {
    +        let row = Self::query(FENCE_REGISTRATION).bind(self.namespace.as_ref()).bind(worker_id).fetch_optional(&mut **tx).await?;
    +        Ok(if fence.holds::<DB>(row.as_ref(), self.worker_stale_after_seconds)? { row } else { None })
    +    }
     }
     
     #[cfg(test)]
14. rmcdaniel commented on Nov 6, 2026

    

    rmcdaniel

    on Nov 6, 2026

    MemberAuthor

    More actions

    Sanitized rebased reference for report topic 12: 0021-t12-Keep-a-run-s-tags-from-the-run-row-a-transaction-alr.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: 12,732 bytes, SHA-256 125dba25dfab135c7e6a3ba983fb259a7c95a6c12b2ae6874fdc776c070a8c2d. Original-base sanitized identity: 5107f116242ef169daa18c71e6ac69fa571946a4cce8a6d6b39ac41e7f6bc42c. No patch was applied or executed during this update.

    diff --git a/rust/src/runtime/activity_failures.rs b/rust/src/runtime/activity_failures.rs
    index 19e9058f..8cfd698a 100644
    --- a/rust/src/runtime/activity_failures.rs
    +++ b/rust/src/runtime/activity_failures.rs
    @@ -431,10 +431,13 @@ where
                         "protocol_version":super::WORKER_PROTOCOL_VERSION,"server_capabilities":super::store::capabilities()}),
                 });
             }
    -        Self::closed_run_outcome(&mut tx, &task, &body, reported).await?;
    +        let read = Self::closed_run_outcome(&mut tx, &task, &body, reported).await?;
             Self::fence(&task, &body)?;
             let run_id = DB::string(&task, "workflow_run_id")?;
    -        Self::active_run(&mut tx, &run_id).await?;
    +        match read {
    +            Some(run) => Self::active(&mut tx, run)?,
    +            None => Self::active_run(&mut tx, &run_id).await?,
    +        };
             let activity_id = DB::string(&attempt, "activity_execution_id")?;
             let activity =
                 Self::query("SELECT * FROM activity_executions WHERE id=$1 AND workflow_run_id=$2")
    diff --git a/rust/src/runtime/standalone_activities.rs b/rust/src/runtime/standalone_activities.rs
    index c5d25dd5..7c2de98b 100644
    --- a/rust/src/runtime/standalone_activities.rs
    +++ b/rust/src/runtime/standalone_activities.rs
    @@ -356,11 +356,17 @@ where
             run_id: &str,
             close: HostClose<'_>,
         ) -> Result<bool> {
    -        let run = Self::query("SELECT workflow_type,workflow_class FROM workflow_runs WHERE id=$1")
    -            .bind(run_id)
    -            .fetch_one(&mut **tx)
    -            .await?;
    -        if DB::string(&run, "workflow_type")? != HOST_TYPE && DB::string(&run, "workflow_class")? != HOST_TYPE {
    +        let host = match tx.run_tags.get(run_id) {
    +            Some(tags) => tags.workflow_type == HOST_TYPE || tags.workflow_class == HOST_TYPE,
    +            None => {
    +                let run = Self::query("SELECT workflow_type,workflow_class FROM workflow_runs WHERE id=$1")
    +                    .bind(run_id)
    +                    .fetch_one(&mut **tx)
    +                    .await?;
    +                DB::string(&run, "workflow_type")? == HOST_TYPE || DB::string(&run, "workflow_class")? == HOST_TYPE
    +            }
    +        };
    +        if !host {
                 return Ok(false);
             }
             let at = now();
    diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
    index 76ab4e7b..bb0b67c9 100644
    --- a/rust/src/runtime/store.rs
    +++ b/rust/src/runtime/store.rs
    @@ -3177,11 +3177,13 @@ where
             self.guard_worker_session_commands(&mut tx, &task, task_id, &body, protocol, commands).await?;
             // `MessageStreamService::validateCompletion`, before anything changes.
             let streams = super::message_streams::Completion::of(&body);
    -        let stream_instance = Self::query("SELECT workflow_instance_id FROM workflow_runs WHERE id=$1")
    +        // The run's whole row, which nothing below changes before it is
    +        // checked as the active run.
    +        let stream_run = Self::query("SELECT * FROM workflow_runs WHERE id=$1")
                 .bind(DB::string(&task, "workflow_run_id")?)
                 .fetch_one(&mut *tx)
                 .await?;
    -        let stream_instance = DB::string(&stream_instance, "workflow_instance_id")?;
    +        let stream_instance = DB::string(&stream_run, "workflow_instance_id")?;
             if !streams.is_empty() {
                 self.validate_stream_completion(&mut tx, &stream_instance, &DB::string(&task, "workflow_run_id")?, &streams)
                     .await?;
    @@ -3239,9 +3241,7 @@ where
                 return Err(refuse(StatusCode::CONFLICT, "invalid_commands"));
             }
             let run_id = DB::string(&task, "workflow_run_id")?;
    -        let run = Self::active_run(&mut tx, &run_id)
    -            .await
    -            .map_err(|error| Self::not_recorded(&task, &body, error))?;
    +        let run = Self::active(&mut tx, stream_run).map_err(|error| Self::not_recorded(&task, &body, error))?;
             // `timeoutIfDeadlineExpired`: published PHP applies a run's deadlines
             // when one of its workflow tasks completes, and closes the run.
             let mut tx = self.close_expired_run(tx, &run, &task, &body["workflow_task_attempt"]).await?;
    @@ -3523,6 +3523,7 @@ where
                 Ok(claimed) => Ok(Some(claimed.is_some())),
                 Err(_) => {
                     sqlx::raw_sql("ROLLBACK TO SAVEPOINT activity_poll").execute(&mut **tx).await?;
    +                tx.run_tags.clear();
                     Ok(None)
                 }
             }
    @@ -4026,10 +4027,13 @@ where
                     response: receipt,
                 });
             }
    -        Self::closed_run_outcome(&mut tx, &task, &body, "completed").await?;
    +        let read = Self::closed_run_outcome(&mut tx, &task, &body, "completed").await?;
             Self::fence(&task, &body)?;
             let run_id = DB::string(&task, "workflow_run_id")?;
    -        let run = Self::active_run(&mut tx, &run_id).await?;
    +        let run = match read {
    +            Some(run) => Self::active(&mut tx, run)?,
    +            None => Self::active_run(&mut tx, &run_id).await?,
    +        };
             let activity_id = DB::string(&attempt, "activity_execution_id")?;
             let activity =
                 Self::query("SELECT * FROM activity_executions WHERE id=$1 AND workflow_run_id=$2")
    @@ -4156,6 +4160,7 @@ where
                 Ok(claimed) => Ok(Some(claimed.is_some())),
                 Err(_) => {
                     sqlx::raw_sql("ROLLBACK TO SAVEPOINT workflow_poll").execute(&mut **tx).await?;
    +                tx.run_tags.clear();
                     Ok(None)
                 }
             }
    @@ -4573,24 +4578,25 @@ where
     
         /// An activity result for a run that completed or failed while the
         /// activity was open: PHP's `run_closed` outcome, recorded nowhere.
    -    /// Cancelled and terminated runs keep their own receipts.
    +    /// Cancelled and terminated runs keep their own receipts. The run's
    +    /// whole row, when it read it.
         pub(super) async fn closed_run_outcome(
             tx: &mut Transaction<'_, DB>,
             task: &DB::Row,
             body: &Value,
             outcome: &str,
    -    ) -> Result<()> {
    +    ) -> Result<Option<DB::Row>> {
             if DB::string(task, "status")? != "leased"
                 || DB::optional_string(task, "lease_owner")?.as_deref() != Some(text(body, "lease_owner")?)
             {
    -            return Ok(());
    +            return Ok(None);
             }
    -        let run = Self::query("SELECT status FROM workflow_runs WHERE id=$1")
    +        let run = Self::query("SELECT * FROM workflow_runs WHERE id=$1")
                 .bind(DB::string(task, "workflow_run_id")?)
                 .fetch_one(&mut **tx)
                 .await?;
             if !matches!(DB::string(&run, "status")?.as_str(), "completed" | "failed") {
    -            return Ok(());
    +            return Ok(Some(run));
             }
             Err(super::RuntimeError::Protocol {
                 status: StatusCode::CONFLICT,
    @@ -4604,6 +4610,13 @@ where
                 .bind(run_id)
                 .fetch_one(&mut **tx)
                 .await?;
    +        Self::active(tx, run)
    +    }
    +
    +    /// `active_run` of the run's whole row, which this transaction has
    +    /// just read and not changed since.
    +    pub(super) fn active(tx: &mut Transaction<'_, DB>, run: DB::Row) -> Result<DB::Row> {
    +        Self::keep_run_tags(tx, &run);
             if !matches!(
                 DB::string(&run, "status")?.as_str(),
                 "pending" | "running" | "waiting"
    @@ -4622,6 +4635,26 @@ where
             Ok(run)
         }
     
    +    /// Keeps a whole run row's tags for the rest of the transaction. A row
    +    /// missing any of them is not kept, so a reader of its tags queries
    +    /// them and fails as it would have.
    +    pub(super) fn keep_run_tags(tx: &mut Transaction<'_, DB>, run: &DB::Row) {
    +        let tags = || -> Result<(String, super::transaction::RunTags)> {
    +            Ok((DB::string(run, "id")?, super::transaction::RunTags {
    +                namespace: DB::string(run, "namespace")?,
    +                priority: DB::number(run, "priority")?,
    +                fairness_key: DB::optional_string(run, "fairness_key")?,
    +                fairness_weight: DB::number(run, "fairness_weight")?,
    +                compatibility: DB::optional_string(run, "compatibility")?,
    +                workflow_type: DB::string(run, "workflow_type")?,
    +                workflow_class: DB::string(run, "workflow_class")?,
    +            }))
    +        };
    +        if let Ok((id, tags)) = tags() {
    +            tx.run_tags.insert(id, tags);
    +        }
    +    }
    +
         pub(super) async fn create_task(
             tx: &mut Transaction<'_, DB>,
             run_id: &str,
    @@ -4645,19 +4678,24 @@ where
             tx.summary_changed(run_id);
             let task_id = id();
             // A task carries its run's priority and fairness tags, as in PHP.
    -        let run = Self::query("SELECT namespace,priority,fairness_key,fairness_weight,compatibility FROM workflow_runs WHERE id=$1")
    -            .bind(run_id)
    -            .fetch_one(&mut **tx)
    -            .await?;
    -        let namespace = DB::string(&run, "namespace")?;
    +        let (namespace, priority, fairness_key, fairness_weight, compatibility) = match tx.run_tags.get(run_id) {
    +            Some(tags) => (tags.namespace.clone(), tags.priority, tags.fairness_key.clone(), tags.fairness_weight, tags.compatibility.clone()),
    +            None => {
    +                let run = Self::query("SELECT namespace,priority,fairness_key,fairness_weight,compatibility FROM workflow_runs WHERE id=$1")
    +                    .bind(run_id)
    +                    .fetch_one(&mut **tx)
    +                    .await?;
    +                (DB::string(&run, "namespace")?, DB::number(&run, "priority")?, DB::optional_string(&run, "fairness_key")?,
    +                    DB::number(&run, "fairness_weight")?, DB::optional_string(&run, "compatibility")?)
    +            }
    +        };
             let created = now();
             let dispatched_at = dispatched.then_some(created);
    -        let fairness_key = DB::optional_string(&run, "fairness_key")?;
             Self::query("INSERT INTO workflow_tasks(id,workflow_run_id,task_type,status,payload,queue,available_at,namespace,created_at,updated_at,last_dispatched_at,last_dispatch_attempt_at,priority,fairness_key,fairness_weight,compatibility,ready_origin_at) VALUES ($1,$2,$3,'ready',$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16)")
             .bind(&task_id).bind(run_id).bind(kind).bind(DB::document(&payload)).bind(queue).bind(DB::bind_time(created)).bind(&namespace)
             .bind(DB::bind_time(created)).bind(DB::bind_time(created)).bind(DB::optional_time(dispatched_at)).bind(DB::optional_time(dispatched_at))
    -        .bind(DB::number(&run, "priority")?).bind(fairness_key.as_deref()).bind(DB::number(&run, "fairness_weight")?)
    -        .bind(DB::optional_string(&run, "compatibility")?.as_deref())
    +        .bind(priority).bind(fairness_key.as_deref()).bind(fairness_weight)
    +        .bind(compatibility.as_deref())
             .bind(DB::optional_time(matches!(kind, "workflow" | "activity").then_some(created)))
             .execute(&mut **tx).await?;
             Ok(task_id)
    diff --git a/rust/src/runtime/transaction.rs b/rust/src/runtime/transaction.rs
    index fd251b83..720c652b 100644
    --- a/rust/src/runtime/transaction.rs
    +++ b/rust/src/runtime/transaction.rs
    @@ -2,10 +2,21 @@
     use super::{Result, backend::Backend, store::Store};
     use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type};
     use std::{
    -    collections::BTreeSet,
    +    collections::{BTreeSet, HashMap},
         ops::{Deref, DerefMut},
     };
     
    +/// A run's columns that no transition changes once the run is stored.
    +pub(super) struct RunTags {
    +    pub(super) namespace: String,
    +    pub(super) priority: i64,
    +    pub(super) fairness_key: Option<String>,
    +    pub(super) fairness_weight: i64,
    +    pub(super) compatibility: Option<String>,
    +    pub(super) workflow_type: String,
    +    pub(super) workflow_class: 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`
    @@ -20,6 +31,9 @@ pub(super) struct Transaction<'a, DB: Backend> {
         /// The binding key of a workflow poll request this transaction has
         /// already bound and found no answer to replay for.
         pub(super) replayed_poll: Option<String>,
    +    /// The tags of the runs this transaction has read whole, by run id
    +    /// (`Store::keep_run_tags`).
    +    pub(super) run_tags: HashMap<String, RunTags>,
     }
     
     impl<'a, DB: Backend> Transaction<'a, DB> {
    @@ -35,6 +49,7 @@ impl<'a, DB: Backend> Transaction<'a, DB> {
                 meter: None,
                 scope_history: None,
                 replayed_poll: None,
    +            run_tags: HashMap::new(),
             }
         }
     
15. rmcdaniel commented on Nov 7, 2026

    

    rmcdaniel

    on Nov 7, 2026

    MemberAuthor

    More actions

    Three checklist items now have independent, unpublished Source preparations while the current P0 #1088 queue/main train finishes:

    • Summary-row upsert: local779118f7. It removes the early DELETE for valid rows, preserves all 52 columns and stale-row cleanup, and retains rollback/retention/import exceptions. The accepted missing-run cleanup is explicit; the former fetch-one error is not mislabeled as the same answer. Actual SQLite SQL controls compare insert/conflict/nullable clearing/rollback/empty-join cases. Full Rust/backend qualification is pending.
    • Binding/result pair write: local958820d4. One two-row upsert keeps both exact keys/JSON values and independent absolute expirations. Prepared common-backend controls compare zero/one/two existing rows, real caller retention cases and rollback against the two-write reference. Actual SQLite SQL comparisons pass. Full Rust/backend qualification is pending.
    • Binding/result pair read: local141ea82026c3e88dbec9f5b006bca02569fac9dc / tree5268fd4508708aa03c57476abb792d1980770fc6. One bounded statement returns integer-tagged rows from the two original key lookups, preserving backend key equality and order-independent roles. The locked binding is still validated before result JSON decoding, so malformed result data cannot mask a kinds conflict. Expired rows are still deleted in transitions; existing read-only snapshots remain read-only. Prepared all-backend controls cover raw bytes/expiry, all four presence states, missing/expired binding, malformed result, kind conflict and expiration deletion, and measure the statement drop. Actual SQLite3.46.1 executes the source SQL and preserves all four raw-row/expiry cases with two reads → one. No repository compilation or PostgreSQL/MySQL result is claimed.

    These are separate local candidate commits on the accepted current-main basis, not merged or released checklist completions. They will be refreshed onto actual maintained main before their ordinary PRs. All eight acceptance items remain open until delivered. The next Source window first belongs to the already prepared runner-only native3.2 qualifier prerequisite after #1088 actual-main acceptance/cleanup; no overlapping CI, new artifact publication or runtime is started here. The remaining census, authored-sequence, waiting-status, fenced-registration and savepoint-aware immutable-tag items are still unimplemented. Reported benchmark timings remain unverified.
16. rmcdaniel commented on Nov 7, 2026

    

    rmcdaniel

    on Nov 7, 2026

    MemberAuthor

    More actions

    The fourth independent local checklist preparation is f86e49f / tree197ec26abb6673f18d714856b39bc77cf3e1088f, on the current accepted-main basis. The census query now counts all six wait categories in one aggregate pass and both task categories in one aggregate pass. Its signal/failure subqueries and all ten answers remain unchanged; empty aggregate tables still return one zero-count row. Bindings drop10→4. This removes repeated table passes, not an extra client SQL statement: the census remains one statement, and I am not reporting1→0.

    Actual SQLite3.46.1 execution of both source SQL strings preserves all ten counts for empty, mixed, foreign-run and missing-run data. Actual EXPLAIN QUERY PLAN shows wait/task accesses(6,2)→(1,1). The prepared common SQLite/PostgreSQL/MySQL/MariaDB test adds all wait kind/status/source combinations and workflow/activity task status/repair combinations, compares every count against the retained exact old query, checks physical table accesses, and then retains the existing full stored-summary byte comparisons. These Rust tests are written but not yet repository-compiled or executed; no other-backend or runtime pass is claimed.

    This is unpublished local preparation, not a completed checkbox. PR1088's queue original now has the separately recorded historical-PHP corrected-marker observation failure; no retry/re-entry or next Source push is started while its remaining original jobs and finite cause/disposition are unresolved. All prior local preparations stay separate. The authored-sequence, waiting-status, fenced-registration and savepoint-aware run-tag items remain unimplemented.
17. rmcdaniel commented on Nov 7, 2026

    

    rmcdaniel

    on Nov 7, 2026

    MemberAuthor

    More actions

    The remaining four checklist items now have separate unpublished source preparations on accepted main6df80b35, alongside the four earlier preparations. No checkbox is marked complete and no additional CI train has started.

    • Authored counters:833d1225/tree4554abd2 reads the run and three physical counters in one statement. Latest logical wait history, scope spans, redrive and cancellation-delivery precedence retain the exact predecessor reader. The common backend tests compare both readers and require three fewer statements. Their synthetic history rows are explicitly reader controls, not writer-admission examples. Actual SQLite3.46.1 executes the source SQL over eight complete-column cases plus foreign/missing rows and confirms4→1 SELECTs.
    • Waiting status:13bb465a/tree8ac70f9c moves the already-required final status read before the conditional waiting write. Pending/running still receives that write; already-waiting and terminal statuses skip it. Common controls cover terminal completion, activity scheduling, empty completion and an already-waiting run, with returned/stored status and statement counts.
    • Fenced registration:038440ca/treebae9e1f7 reuses only the registration validated inside the current claim transaction, including carried and multiplexed workflow claims. Admission/idle reads remain byte-identical. The unfenced path retains its original lookup, and re-registration still invalidates the original token. Actual SQLite SQL preserves every fence and claim column, namespace exclusion and active-token ordering, with2→1 SELECTs across four registration states. Common tests retain the same-worker, slot, error and savepoint controls and add actual hot-path counts and replacement/fallback checks.
    • Transaction tags:4b8ac18a/tree97dfa79a caches namespace/priority/fairness/compatibility from authoritative run rows within one transaction. Savepoint rollback clears the cached view. Failed optional seeding removes an older cached value and leaves malformed-tag decoding at its original task-creation boundary. Common controls cover exact stored task tags/payload, cold reads, cross-run separation, rollback, a fresh transaction, missing rows and SQLite malformed storage.

    The Rust common-backend controls above are written but have not been repository-compiled or executed yet. These are source candidates, not backend/performance/release acceptance. Full maintained-toolchain builds, actual SQLite/PostgreSQL/MySQL/MariaDB qualification, review and integration remain required. The current public train is the separately narrowed #1069 PR1088, whose own63 checks passed before its first queue original; #1064/#1066 also retain separate source preparations. I retain the next integration and release work without overlapping CI.
18. rmcdaniel commented on Nov 8, 2026

    

    rmcdaniel

    on Nov 8, 2026

    MemberAuthor

    More actions

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

    

    rmcdaniel

    on Nov 8, 2026

    MemberAuthor

    More actions

    Draft PR #1099 is open, combining all eight source preparations on current main 6f279f9b. Head 50047cbb, tree 067f3bb2; diff 104,586 B / 15f93010…. The four test-only conflicts were resolved as unions, and the original WIP refs are kept unchanged. Its PR CI is the first compile and qualification; General selected that train. Next action: Server watches that train and reports the results to General. Owner: Server.
20. rmcdaniel

    added

    agent:general

    and removed

    agent:server

    on Nov 8, 2026