Rust #325: worker history shows consumed signal sequences; full task snapshots

docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md

@@ -0,0 +1,48 @@
+# October 23, what a worker reads in history
+
+- **Found by.** Comparing the message-stream fixture's worker polls against
+  published Server 2.11.1.
+- **The gaps.**
+  - PHP's poll and history-page responses show a consumed signal's
+    `workflow_sequence` on its `SignalReceived` event, read from the signal's
+    record (`WorkflowTaskPoller::historyEventsWithSignalArgumentEnvelopes`).
+    Native showed the event as recorded.
+  - PHP's task snapshots in history keep every field that is set
+    (`WorkflowHistoryEvent::taskSnapshot`). Native's left out
+    `compatibility`, `last_dispatch_attempt_at`, `last_dispatched_at`,
+    `last_dispatch_error`, `last_claim_failed_at` and `last_claim_error`.
+- **What native does now.**
+  - Worker polls and history pages add `workflow_sequence` to a consumed
+    signal's `SignalReceived`, when the payload does not have one. The
+    run's own history route still shows the events as recorded, as PHP's
+    does.
+  - Task snapshots keep those fields when they are set.
+- **Which signals carry a sequence.** PHP's `applySignal` takes a sequence
+  only from the signal's own wait. A signal the run's contract declares is
+  applied without one even while a condition wait is open, and
+  `markConditionWaitSignalConsumed` then finds it applied, so its record keeps
+  no sequence. Only an undeclared signal takes the condition wait's sequence.
+  Native had recorded the condition wait's sequence for declared signals too.
+  That difference was invisible until worker polls showed the record, and the
+  published-SDK signal and query cohorts then caught it: their worker history
+  must equal the run's history prefix. Native now records a sequence only from
+  the signal's own wait, or from a condition wait for an undeclared signal.
+- **How it was checked.** A test runs on SQLite, PostgreSQL, MySQL and
+  MariaDB:
+  - a condition wait consumes an undeclared signal;
+  - the next poll and history page show that wait's sequence, and the
+    run's history does not;
+  - the completing task's snapshot has its dispatch stamps.
+
+  A second test applies a declared signal under a condition wait twice. The
+  worker's history shows no sequence on either `SignalReceived`, and it fails
+  without the record change.
+- **Not changed yet.** Some events PHP records with a task snapshot have
+  none in native, such as `ConditionWaitOpened` and `SignalWaitOpened`.
+  That audit is a #325 follow-up.
+- **Known difference now visible in snapshots.** PHP stamps a dispatch only
+  on tasks a control-plane call dispatches, such as a signal's task. Tasks a
+  workflow task completion creates, such as the next message task, carry
+  none. Native stamps every new task, so those snapshots show stamps where
+  PHP's do not. This is the cause of the known `tasks_dispatched_last_minute`
+  difference too; fixing it is a #325 follow-up.

rust/src/runtime/store.rs

@@ -1124,11 +1124,10 @@ where
         let wait = open.clone().filter(|wait| {
             wait.get("signal_name").is_none() || (wait["signal_name"] == name && wait["signal_wait_id"] == wait_id)
         });
-        if wait.is_none()
-            && !Self::declared_signals(tx, run_id)
-                .await?
-                .is_some_and(|names| names.contains(&name))
-        {
+        let declared = Self::declared_signals(tx, run_id)
+            .await?
+            .is_some_and(|names| names.contains(&name));
+        if wait.is_none() && !declared {
             let reason = if open.is_some() { "signal_wait_mismatch" } else { "signal_wait_not_open" };
             return Err(refuse(StatusCode::CONFLICT, reason));
         }
@@ -1156,8 +1155,13 @@ where
             .bind(&command_id)
             .execute(&mut **tx)
             .await?;
+        // PHP's `applySignal` takes a sequence only from the signal's own
+        // wait, so a declared signal applied under a condition wait keeps no
+        // workflow sequence. Only an undeclared signal takes the condition
+        // wait's (`markConditionWaitSignalConsumed`).
+        let record_sequence = if direct || !declared { sequence } else { None };
         Self::query("UPDATE workflow_signal_records SET status='applied',workflow_sequence=$1,applied_at=$2,closed_at=$3 WHERE id=$4")
-            .bind(DB::optional_integer(sequence)).bind(DB::bind_time(applied_at)).bind(DB::bind_time(applied_at)).bind(signal_id).execute(&mut **tx).await?;
+            .bind(DB::optional_integer(record_sequence)).bind(DB::bind_time(applied_at)).bind(DB::bind_time(applied_at)).bind(signal_id).execute(&mut **tx).await?;
         let run = Self::query(
             "SELECT workflow_instance_id,message_cursor_position FROM workflow_runs WHERE id=$1",
         )
@@ -1798,10 +1802,11 @@ where
                         }
                     }
                 }
-                let (events, next) = match page {
+                let (mut events, next) = match page {
                     Some(size) => Self::history_page_connection(&mut tx, &run_id, 0, size).await?,
                     None => (Self::whole_history(&mut tx, &run_id).await?, None),
                 };
+                Self::with_signal_sequences(&mut tx, &run_id, &mut events).await?;
                 let budget = Self::workflow_history_budget(
                     &mut tx,
                     &run_id,
@@ -1865,7 +1870,8 @@ where
         }
         Self::workflow_task_guard(&mut tx, &task, &body).await?;
         let run_id = DB::string(&task, "workflow_run_id")?;
-        let (events, next) = Self::history_page_connection(&mut tx, &run_id, after, size).await?;
+        let (mut events, next) = Self::history_page_connection(&mut tx, &run_id, after, size).await?;
+        Self::with_signal_sequences(&mut tx, &run_id, &mut events).await?;
         let run = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1")
             .bind(&run_id)
             .fetch_one(&mut *tx)
@@ -3565,6 +3571,33 @@ where
 
     /// A poll's whole history, read in bounded pages: published PHP sends
     /// every event unless the worker asks for pages.
+    /// `WorkflowTaskPoller::historyEventsWithSignalArgumentEnvelopes`: the
+    /// history a worker reads shows a consumed signal's workflow sequence on
+    /// its `SignalReceived`, from the signal's record. The run's own history
+    /// route shows the events as recorded.
+    async fn with_signal_sequences(connection: &mut DB::Connection, run_id: &str, events: &mut [Value]) -> Result<()> {
+        if !events.iter().any(|event| event["event_type"] == "SignalReceived") {
+            return Ok(());
+        }
+        let rows = Self::query("SELECT id,workflow_sequence FROM workflow_signal_records WHERE workflow_run_id=$1 AND workflow_sequence IS NOT NULL")
+            .bind(run_id)
+            .fetch_all(&mut *connection)
+            .await?;
+        let mut sequences = std::collections::HashMap::new();
+        for row in rows {
+            sequences.insert(DB::string(&row, "id")?, DB::number(&row, "workflow_sequence")?);
+        }
+        for event in events.iter_mut().filter(|event| event["event_type"] == "SignalReceived") {
+            let sequence = event["payload"]["signal_id"].as_str().and_then(|id| sequences.get(id)).copied();
+            if let (Some(sequence), Some(payload)) = (sequence, event["payload"].as_object_mut()) {
+                if payload.get("workflow_sequence").is_none_or(Value::is_null) {
+                    payload.insert("workflow_sequence".into(), json!(sequence));
+                }
+            }
+        }
+        Ok(())
+    }
+
     async fn whole_history(connection: &mut DB::Connection, run_id: &str) -> Result<Vec<Value>> {
         let mut events = Vec::new();
         let mut after = 0;

rust/src/runtime/worker_incarnations.rs

@@ -339,18 +339,24 @@ where
             "status":DB::string(task,"status")?,"attempt_count":DB::number(task,"attempt_count")?,
             "repair_count":DB::number(task,"repair_count")?,"queue":DB::string(task,"queue")?,
             "connection":DB::PHP_CONNECTION});
+        // `WorkflowHistoryEvent::taskSnapshot` keeps every field that is set.
         for field in [
             "available_at",
             "leased_at",
             "lease_expires_at",
+            "last_dispatch_attempt_at",
+            "last_dispatched_at",
+            "last_claim_failed_at",
             "repair_available_at",
         ] {
             if let Some(value) = DB::optional_instant(task, field)? {
                 snapshot[field] = json!(value);
             }
         }
-        if let Some(owner) = DB::optional_string(task, "lease_owner")? {
-            snapshot["lease_owner"] = json!(owner);
+        for field in ["lease_owner", "compatibility", "last_dispatch_error", "last_claim_error"] {
+            if let Some(value) = DB::optional_string(task, field)? {
+                snapshot[field] = json!(value);
+            }
         }
         Ok(snapshot)
     }

rust/tests/execution.rs

@@ -109,6 +109,8 @@ mod parent_close_policies;
 mod history_pages;
 #[path = "execution/declared_signals.rs"]
 mod declared_signals;
+#[path = "execution/worker_history_presentation.rs"]
+mod worker_history_presentation;
 #[path = "execution/poll_waits.rs"]
 mod poll_waits;
 #[path = "execution/child_cascade.rs"]

rust/tests/execution/worker_history_presentation.rs

@@ -0,0 +1,97 @@
+use super::*;
+
+// Server 2.11.1 `WorkflowTaskPoller::historyEventsWithSignalArgumentEnvelopes`
+// and `WorkflowHistoryEvent::taskSnapshot`: the history a worker reads shows a
+// consumed signal's workflow sequence, and task snapshots keep their dispatch
+// stamps.
+
+fn wait(key: &str) -> Value {
+    json!([{"type":"open_condition_wait","condition_key":key,
+        "condition_definition_fingerprint":format!("sha256:{}", "b".repeat(64))}])
+}
+
+fn received(history: &Value) -> Vec<Value> {
+    history.as_array().unwrap().iter().filter(|event| event["event_type"] == "SignalReceived")
+        .map(|event| event["payload"].clone()).collect()
+}
+
+#[tokio::test]
+async fn a_worker_reads_the_sequence_that_consumed_a_signal() {
+    let database = TestDatabase::new().await;
+    let app = router(database.open().await.unwrap());
+    // A run started before any worker declared a contract admits any signal.
+    let started = start(&app, "presented-signals").await;
+    let worker = json!({"worker_id":"presenter","task_queue":"test","runtime":"php","supported_workflow_types":["echo"],
+        "supported_activity_types":[]});
+    assert_eq!(request(&app, "POST", "/api/worker/register", worker).await.0, StatusCode::CREATED);
+    let task = poll(&app, "presenter", "workflow").await;
+    assert_eq!(finish_task(&app, &task, wait("first")).await.0, StatusCode::OK);
+    let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await;
+    let opened = history.as_array().unwrap().iter().find(|event| event["event_type"] == "ConditionWaitOpened").unwrap();
+    let sequence = opened["payload"]["sequence"].clone();
+
+    let (status, body) = request(&app, "POST", "/api/workflows/presented-signals/signal/go", json!([])).await;
+    assert_eq!(status, StatusCode::ACCEPTED, "{body}");
+    let task = poll(&app, "presenter", "workflow").await;
+    assert!(received(&task["history_events"])[0].get("workflow_sequence").is_none(), "not consumed yet: {task}");
+    assert_eq!(finish_task(&app, &task, wait("second")).await.0, StatusCode::OK);
+
+    let (status, body) = request(&app, "POST", "/api/workflows/presented-signals/signal/again", json!([])).await;
+    assert_eq!(status, StatusCode::ACCEPTED, "{body}");
+    let task = poll(&app, "presenter", "workflow").await;
+    let signals = received(&task["history_events"]);
+    assert_eq!(signals[0]["workflow_sequence"], sequence, "the consuming wait's sequence: {signals:?}");
+    assert!(signals[1].get("workflow_sequence").is_none(), "{signals:?}");
+    // A history page reads the same.
+    let (status, page) = request(&app, "POST", &format!("/api/worker/workflow-tasks/{}/history", task["task_id"].as_str().unwrap()),
+        json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"]})).await;
+    assert_eq!(status, StatusCode::OK, "{page}");
+    assert_eq!(received(&page["history_events"])[0]["workflow_sequence"], sequence, "{page}");
+    // The run's own history shows the event as recorded.
+    let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await;
+    assert!(received(&history)[0].get("workflow_sequence").is_none(), "{history}");
+
+    // The completing task was dispatched when the signal created it, and its
+    // snapshot keeps the dispatch stamps.
+    let done = json!([{"type":"complete_workflow","result":envelope(Payload::Null)}]);
+    assert_eq!(finish_task(&app, &task, done).await.0, StatusCode::OK);
+    let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await;
+    let completed = history.as_array().unwrap().iter().find(|event| event["event_type"] == "WorkflowCompleted").unwrap();
+    for stamp in ["last_dispatch_attempt_at", "last_dispatched_at"] {
+        assert!(completed["payload"]["task"][stamp].is_string(), "{stamp}: {completed}");
+    }
+    assert!(completed["payload"]["task"].get("last_claim_error").is_none(), "unset fields stay out: {completed}");
+}
+
+// Workflow 2.10.2 `DefaultWorkflowTaskBridge::applySignal` takes a workflow
+// sequence only from the signal's own wait. A signal the run's contract
+// declares is applied without one even while a condition wait is open, so
+// `markConditionWaitSignalConsumed` finds it applied and its record keeps no
+// sequence: the history a worker reads matches the run's history.
+#[tokio::test]
+async fn a_declared_signal_consumed_by_a_condition_wait_keeps_no_sequence() {
+    let database = TestDatabase::new().await;
+    let app = router(database.open().await.unwrap());
+    update_worker(&app, "declared-presenter").await;
+    let started = start(&app, "declared-signals").await;
+    let task = poll(&app, "declared-presenter", "workflow").await;
+    assert_eq!(finish_task(&app, &task, wait("first")).await.0, StatusCode::OK);
+
+    let finish = json!({"input":envelope(Payload::Array(vec![Payload::Long(7)]))});
+    let (status, body) = request(&app, "POST", "/api/workflows/declared-signals/signal/finish", finish.clone()).await;
+    assert_eq!(status, StatusCode::ACCEPTED, "{body}");
+    let task = poll(&app, "declared-presenter", "workflow").await;
+    assert_eq!(finish_task(&app, &task, wait("second")).await.0, StatusCode::OK);
+    let (status, body) = request(&app, "POST", "/api/workflows/declared-signals/signal/finish", finish).await;
+    assert_eq!(status, StatusCode::ACCEPTED, "{body}");
+    let task = poll(&app, "declared-presenter", "workflow").await;
+
+    let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await;
+    let polled = received(&task["history_events"]);
+    assert_eq!(polled.len(), 2, "{task}");
+    assert!(polled.iter().all(|signal| signal.get("workflow_sequence").is_none()), "{polled:?}");
+    let recorded = received(&history);
+    assert_eq!(polled[0]["signal_id"], recorded[0]["signal_id"]);
+    let (_, described) = request(&app, "GET", "/api/workflows/declared-signals", Value::Null).await;
+    assert_eq!(described["status"], "waiting", "{described}");
+}