diff --git a/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md new file mode 100644 index 000000000..63f110abb --- /dev/null +++ b/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. diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 93c299d12..d02b6ee5c 100644 --- a/rust/src/runtime/store.rs +++ b/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> { let mut events = Vec::new(); let mut after = 0; diff --git a/rust/src/runtime/worker_incarnations.rs b/rust/src/runtime/worker_incarnations.rs index 2a83a3117..5685a92be 100644 --- a/rust/src/runtime/worker_incarnations.rs +++ b/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) } diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index c3ce551e4..eabeeade2 100644 --- a/rust/tests/execution.rs +++ b/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"] diff --git a/rust/tests/execution/worker_history_presentation.rs b/rust/tests/execution/worker_history_presentation.rs new file mode 100644 index 000000000..a5cafa602 --- /dev/null +++ b/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 { + 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}"); +}