From 0ec63aae91f51015ef143b4630f9a08fb37e31ed Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 20:49:49 +0000 Subject: [PATCH 1/4] Snapshot the task on every event PHP records with it Published PHP's WorkflowHistoryEvent::record adds the task's snapshot to each event it records with the task model. Native now adds it centrally for those event types, uses the full snapshot on TimerScheduled and the activity events, and leases a timer task to itself while it fires, as RunTimerTask does, so TimerFired carries the leased timer task. --- .../2026-10-23-task-snapshots.md | 36 ++++++++++ rust/src/runtime/activity_failures.rs | 4 +- rust/src/runtime/store.rs | 66 +++++++++++++++++-- .../execution/worker_history_presentation.rs | 59 +++++++++++++++++ 4 files changed, 157 insertions(+), 8 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md diff --git a/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md b/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md new file mode 100644 index 000000000..10c5a1ce1 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md @@ -0,0 +1,36 @@ +# October 23, task snapshots on history events + +- **The gap.** Published PHP's `WorkflowHistoryEvent::record` adds a `task` + snapshot to every event it records with the task model in hand. Native + added snapshots event by event and missed several. Driving one workflow + through a side effect, an activity, a timer, a child, a signal wait and a + condition wait against both servers showed that native had no snapshot on: + - `TimerFired`, `ChildWorkflowScheduled`, `ChildRunStarted`; + - `SignalWaitOpened`, `ConditionWaitOpened`, `ConditionWaitSatisfied`; + - `SignalApplied` and the instance `MessageCursorAdvanced`. + + Native's own snapshots on `TimerScheduled`, `ActivityStarted`, + `ActivityCompleted` and `ActivityFailed` also left out fields that PHP + keeps, such as `queue`, `connection`, `available_at` and `repair_count`. +- **What native does now.** + - When an event of a type PHP records with the task model is appended with + a task and no snapshot of its own, native adds the task's snapshot. The + list of those event types comes from the Workflow package's source. + - `TimerScheduled` uses the full snapshot. The activity events' snapshots + keep PHP's fields. + - As `RunTimerTask` does, a timer task is leased to itself while its timer + fires: status leased, `leased_at`, `lease_owner` set to the task id, a + workflow-task lease, and one more attempt. Its lease is cleared once the + timer has fired. `TimerFired` therefore carries the leased timer task. +- **How it was checked.** The same probe now finds the same events with + snapshots on both servers, with the same fields. A new integration test + checks the timer, child, signal-wait, signal and condition-wait events. +- **Left as they are.** + - PHP's command snapshots carry `request_route_name`, a name Laravel + generates at random for the deployment's cached routes. Native has no + such name. + - Native stamps a dispatch on every new task, so the snapshot of a task + created inside a completion keeps `last_dispatch_attempt_at` and + `last_dispatched_at`, which PHP's does not (Server #325 follow-up). + - A message stream's `SignalApplied` and the condition wait it satisfies + follow the message streams change (#709). diff --git a/rust/src/runtime/activity_failures.rs b/rust/src/runtime/activity_failures.rs index 062f7720a..e24c73bb7 100644 --- a/rust/src/runtime/activity_failures.rs +++ b/rust/src/runtime/activity_failures.rs @@ -540,7 +540,9 @@ where } payload["task"] = json!({"id":task_id,"type":"activity","status":"leased", "available_at":DB::optional_instant(&task,"available_at")?,"leased_at":DB::optional_instant(&task,"leased_at")?, - "lease_owner":DB::optional_string(&task,"lease_owner")?,"attempt_count":count}); + "lease_owner":DB::optional_string(&task,"lease_owner")?,"lease_expires_at":DB::optional_instant(&task,"lease_expires_at")?, + "attempt_count":count,"repair_count":DB::number(&task,"repair_count")?,"queue":DB::string(&task,"queue")?, + "connection":DB::PHP_CONNECTION}); let failed_event = payload.clone(); Self::append( &mut tx, diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 03dfa699a..2995d2b9a 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -15,6 +15,37 @@ pub(super) struct PreparedControlArguments { pub(super) value: String, pub(super) command: String, } +/// The events PHP's Workflow package records with the task model in hand, +/// which therefore carry the task's snapshot. +const TASK_SNAPSHOT_EVENTS: &[&str] = &[ + "ActivityCompleted", + "ActivityFailed", + "ActivityHeartbeatRecorded", + "ActivityRetryScheduled", + "ActivityScheduled", + "ActivityStarted", + "ActivityTimedOut", + "ChildRunStarted", + "ChildWorkflowScheduled", + "ConditionWaitOpened", + "ConditionWaitSatisfied", + "ConditionWaitTimedOut", + "FailureHandled", + "MemoUpserted", + "MessageCursorAdvanced", + "SearchAttributesUpserted", + "SelectionOperationCancelled", + "SideEffectRecorded", + "SignalApplied", + "SignalWaitOpened", + "TimerFired", + "TimerScheduled", + "VersionMarkerRecorded", + "WorkflowCompleted", + "WorkflowContinuedAsNew", + "WorkflowFailed", + "WorkflowTimedOut", +]; /// The runtime-owned message-stream delivery signal (`WorkflowStub::MESSAGE_STREAM_RUNTIME_SIGNAL`). const MESSAGE_STREAM_SIGNAL: &str = "__durable_workflow_message_stream"; pub(super) const TASK_CANDIDATES_SQL: &str = "SELECT t.*,r.workflow_type FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type=$3 AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) AND (COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,'')) IS NULL OR COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,''))=$6) ORDER BY t.priority,t.available_at,t.id LIMIT 100"; @@ -1709,7 +1740,8 @@ where payload["task"] = json!({"id":task_id,"type":"activity","status":"leased", "repair_count":DB::number(&task,"repair_count")?, "available_at":DB::optional_instant(&task,"available_at")?,"leased_at":leased_at, - "lease_owner":worker_id,"lease_expires_at":expires,"attempt_count":attempt,"queue":queue}); + "lease_owner":worker_id,"lease_expires_at":expires,"attempt_count":attempt,"queue":queue, + "connection":DB::PHP_CONNECTION}); Self::append_at( &mut tx, &run_id, @@ -2490,8 +2522,7 @@ where Self::query("INSERT INTO workflow_run_timers(id,workflow_run_id,sequence,status,delay_seconds,fire_at) VALUES ($1,$2,$3,'pending',$4,$5)") .bind(&timer_id).bind(&run_id).bind(sequence).bind(delay).bind(DB::bind_time(fire_at)).execute(&mut *tx).await?; let mut timer_payload = json!({"timer_id":timer_id,"sequence":sequence,"delay_seconds":delay,"fire_at":fire_at, - "task":{"id":task_id,"type":"workflow","status":"leased","lease_owner":DB::optional_string(&task,"lease_owner")?, - "lease_expires_at":DB::optional_instant(&task,"lease_expires_at")?,"attempt_count":DB::number(&task,"attempt_count")?}}); + "task":Self::task_snapshot(&task)?}); super::selections::merge(&mut timer_payload, &metadata); Self::append_at( &mut tx, @@ -2889,7 +2920,8 @@ where payload["task"] = json!({"id":task_id,"type":"activity","status":"leased", "repair_count":DB::number(&task,"repair_count")?, "available_at":DB::optional_instant(&task,"available_at")?,"leased_at":DB::optional_instant(&task,"leased_at")?, - "lease_owner":DB::optional_string(&task,"lease_owner")?,"attempt_count":DB::number(&task,"attempt_count")?}); + "lease_owner":DB::optional_string(&task,"lease_owner")?,"lease_expires_at":DB::optional_instant(&task,"lease_expires_at")?, + "attempt_count":DB::number(&task,"attempt_count")?,"queue":DB::string(&task,"queue")?,"connection":DB::PHP_CONNECTION}); Self::append( &mut tx, &run_id, @@ -3015,6 +3047,15 @@ where json!({"timer_id":timer_id}) }; super::selections::merge(&mut payload, &metadata); + // `RunTimerTask`: the timer task is leased to itself while its + // timer fires, so `TimerFired` carries the leased task. + Self::query("UPDATE workflow_tasks SET status='leased',leased_at=$1,lease_owner=$2,lease_expires_at=$3,attempt_count=attempt_count+1 WHERE id=$4") + .bind(DB::bind_time(fired_at)) + .bind(&task_id) + .bind(DB::bind_time(fired_at + chrono::Duration::seconds(self.workflow_task_lease_seconds))) + .bind(&task_id) + .execute(&mut *tx) + .await?; Self::append( &mut tx, &run_id, @@ -3038,7 +3079,7 @@ where .bind(&timer_id) .execute(&mut *tx) .await?; - Self::query("UPDATE workflow_tasks SET status='completed' WHERE id=$1") + Self::query("UPDATE workflow_tasks SET status='completed',lease_expires_at=NULL WHERE id=$1") .bind(&task_id) .execute(&mut *tx) .await?; @@ -3367,9 +3408,20 @@ where .execute(&mut **tx) .await?; let recorded_at = recorded_at.unwrap_or_else(now); - // `WorkflowHistoryEvent::record`: an event recorded with a command - // carries that command's snapshot. let mut payload = payload; + // `WorkflowHistoryEvent::record`: an event PHP records with its task + // model carries that task's snapshot. + if let Some(task_id) = task_id + && TASK_SNAPSHOT_EVENTS.contains(&event_type) + && payload.as_object().is_some_and(|fields| !fields.contains_key("task")) + && let Some(task) = Self::query("SELECT * FROM workflow_tasks WHERE id=$1") + .bind(task_id) + .fetch_optional(&mut **tx) + .await? + { + payload["task"] = Self::task_snapshot(&task)?; + } + // An event recorded with a command carries that command's snapshot. if let Some(command_id) = command_id.filter(|_| command_snapshot) && payload.as_object().is_some_and(|fields| !fields.contains_key("command")) && let Some(snapshot) = Self::command_snapshot(tx, command_id).await? diff --git a/rust/tests/execution/worker_history_presentation.rs b/rust/tests/execution/worker_history_presentation.rs index db76a3d75..401ac593f 100644 --- a/rust/tests/execution/worker_history_presentation.rs +++ b/rust/tests/execution/worker_history_presentation.rs @@ -62,3 +62,62 @@ async fn a_worker_reads_the_sequence_that_consumed_a_signal() { } assert!(completed["payload"]["task"].get("last_claim_error").is_none(), "unset fields stay out: {completed}"); } + +// `WorkflowHistoryEvent::record` with the task model: each event PHP records +// while handling a task carries that task's snapshot, and a fired timer +// carries its timer task, leased to itself. +#[tokio::test] +async fn events_recorded_with_a_task_carry_its_snapshot() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + let started = start(&app, "snapshot-events").await; + register(&app, "snapshotter", json!(["echo", "child"]), json!([])).await; + let events = || run_history(&app, &started["workflow_id"], &started["run_id"]); + let snapshot_of = |history: &Value, kind: &str| -> Value { + history.as_array().unwrap().iter().rev().find(|event| event["event_type"] == kind) + .map(|event| event["payload"]["task"].clone()).unwrap_or(Value::Null) + }; + + let task = poll(&app, "snapshotter", "workflow").await; + assert_eq!(finish_task(&app, &task, json!([{"type":"start_timer","delay_seconds":1}])).await.0, StatusCode::OK); + let scheduled = snapshot_of(&events().await, "TimerScheduled"); + assert_eq!((scheduled["id"].clone(), scheduled["type"].clone()), (task["task_id"].clone(), json!("workflow")), "{scheduled}"); + for field in ["queue", "connection", "available_at", "leased_at", "repair_count"] { + assert!(scheduled.get(field).is_some(), "{field}: {scheduled}"); + } + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15); + let fired = loop { + let fired = snapshot_of(&events().await, "TimerFired"); + if !fired.is_null() { + break fired; + } + assert!(tokio::time::Instant::now() < deadline, "the timer never fired"); + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + }; + assert_eq!((fired["type"].clone(), fired["status"].clone(), fired["attempt_count"].clone()), + (json!("timer"), json!("leased"), json!(1)), "{fired}"); + assert_eq!(fired["lease_owner"], fired["id"], "{fired}"); + assert!(fired["lease_expires_at"].is_string() && fired["leased_at"].is_string(), "{fired}"); + + let task = poll(&app, "snapshotter", "workflow").await; + assert_eq!(finish_task(&app, &task, json!([ + {"type":"start_child_workflow","workflow_type":"child","arguments":envelope(Payload::Array(vec![]))}, + {"type":"open_signal_wait","signal_name":"go"}])).await.0, StatusCode::OK); + let history = events().await; + for kind in ["ChildWorkflowScheduled", "ChildRunStarted", "SignalWaitOpened"] { + assert_eq!(snapshot_of(&history, kind)["id"], task["task_id"], "{kind}: {history}"); + } + let (status, body) = request(&app, "POST", "/api/workflows/snapshot-events/signal/go", json!([])).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let mut task = poll(&app, "snapshotter", "workflow").await; + // The child's first task may come first on the shared queue. + if task["workflow_type"] == "child" { + assert_eq!(finish_task(&app, &task, json!([{"type":"complete_workflow","result":envelope(Payload::Null)}])).await.0, StatusCode::OK); + task = poll(&app, "snapshotter", "workflow").await; + } + assert_eq!(finish_task(&app, &task, json!([{"type":"open_condition_wait","condition_key":"k", + "condition_definition_fingerprint":format!("sha256:{}", "c".repeat(64))}])).await.0, StatusCode::OK); + let history = events().await; + assert_eq!(snapshot_of(&history, "SignalApplied")["id"], task["task_id"], "{history}"); + assert_eq!(snapshot_of(&history, "ConditionWaitOpened")["id"], task["task_id"], "{history}"); +} From 05a6110cb4122c62cf18e8b52c9437c7bf08116a Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 21:04:16 +0000 Subject: [PATCH 2/4] Describe each command's workflow class and preview flag like PHP PHP's run description names the run's workflow class on every command and says whether the command's payload preview is omitted. Native left both out. Only a stored external reference is read to decide the flag. --- .../2026-10-23-task-snapshots.md | 10 +++++++++- rust/src/runtime/run_describe.rs | 2 +- rust/src/runtime/store.rs | 11 +++++++++-- 3 files changed, 19 insertions(+), 4 deletions(-) diff --git a/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md b/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md index 10c5a1ce1..87931cf15 100644 --- a/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md +++ b/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md @@ -1,4 +1,4 @@ -# October 23, task snapshots on history events +# October 23, task snapshots on history events and described commands - **The gap.** Published PHP's `WorkflowHistoryEvent::record` adds a `task` snapshot to every event it records with the task model in hand. Native @@ -25,6 +25,14 @@ - **How it was checked.** The same probe now finds the same events with snapshots on both servers, with the same fields. A new integration test checks the timer, child, signal-wait, signal and condition-wait events. +- **Described commands.** A field-by-field diff of the same workflow's + description on both servers found two more gaps. Each command PHP + describes names the run's `workflow_class` (`WorkflowCommand::record` keeps + it), and says whether its payload preview is omitted + (`payload_preview_omitted`, true only for a stored external payload over the + size limit). Native now sends both. With fresh workers on both sides, the + descriptions then differ only in `request_route_name` and the + `request_fingerprint` hashed from it. - **Left as they are.** - PHP's command snapshots carry `request_route_name`, a name Laravel generates at random for the deployment's cached routes. Native has no diff --git a/rust/src/runtime/run_describe.rs b/rust/src/runtime/run_describe.rs index 070e4e364..ae9e6ae3b 100644 --- a/rust/src/runtime/run_describe.rs +++ b/rust/src/runtime/run_describe.rs @@ -31,7 +31,7 @@ fn max_payload_bytes() -> i64 { } /// `oversizedPayloadPreview`: an external payload larger than the limit. -fn oversized(envelope: &Value) -> bool { +pub(super) fn oversized(envelope: &Value) -> bool { envelope["external_storage"]["size_bytes"].as_i64().is_some_and(|size| size > max_payload_bytes()) } diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 2995d2b9a..268e891b8 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -553,7 +553,9 @@ where // refused signals which intentionally have no workflow history event. // Select metadata only: large customer payloads are never decoded or // copied merely to display the command's target and admission result. - let command_rows = Self::query("SELECT c.id,c.command_sequence,c.command_type,c.target_scope,c.requested_workflow_run_id,c.resolved_workflow_run_id,c.source,c.status,c.outcome,c.rejection_reason,c.workflow_type,c.workflow_class,c.accepted_at,c.applied_at,c.rejected_at,c.context,s.signal_name FROM workflow_commands c LEFT JOIN workflow_signal_records s ON s.workflow_command_id=c.id WHERE c.workflow_run_id=$1 AND c.workflow_instance_id=$2 ORDER BY c.command_sequence,c.created_at,c.id") + // Only a stored external reference is read, to tell whether its preview + // is omitted; inline payloads stay unread. + let command_rows = Self::query("SELECT c.id,c.command_sequence,c.command_type,c.target_scope,c.requested_workflow_run_id,c.resolved_workflow_run_id,c.source,c.status,c.outcome,c.rejection_reason,c.workflow_type,c.workflow_class,c.accepted_at,c.applied_at,c.rejected_at,c.context,s.signal_name,CASE WHEN c.payload LIKE 'dw-%' THEN c.payload END AS stored_payload FROM workflow_commands c LEFT JOIN workflow_signal_records s ON s.workflow_command_id=c.id WHERE c.workflow_run_id=$1 AND c.workflow_instance_id=$2 ORDER BY c.command_sequence,c.created_at,c.id") .bind(DB::string(&row, "id")?).bind(workflow_id).fetch_all(&self.pool).await?; let mut commands = Vec::with_capacity(command_rows.len()); for command in command_rows { @@ -564,7 +566,12 @@ where "target_name":DB::optional_string(&command,"signal_name")?,"source":DB::string(&command,"source")?, "status":DB::string(&command,"status")?,"outcome":DB::optional_string(&command,"outcome")?, "reason":DB::optional_string(&command,"rejection_reason")?,"rejection_reason":DB::optional_string(&command,"rejection_reason")?, - "workflow_type":DB::optional_string(&command,"workflow_type")?,"workflow_class":DB::optional_string(&command,"workflow_class")?, + // `WorkflowCommand::record` keeps the run's class on each command. + "workflow_type":DB::optional_string(&command,"workflow_type")?, + "workflow_class":DB::optional_string(&command,"workflow_class")?.or(DB::optional_string(&row,"workflow_class")?), + "payload_preview_omitted":DB::optional_string(&command,"stored_payload")?.as_deref() + .and_then(super::external_resolution::stored_envelope) + .is_some_and(|envelope| super::run_describe::oversized(&envelope)), "accepted_at":DB::optional_instant(&command,"accepted_at")?,"applied_at":DB::optional_instant(&command,"applied_at")?, "rejected_at":DB::optional_instant(&command,"rejected_at")?}); let fields = public.as_object_mut().expect("command object"); From 65772abac8ecb65f7bde7f85c56b80bb77054777 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 24 Oct 2026 00:50:27 +0000 Subject: [PATCH 3/4] Rust #325: preview omission reads native stored references; exact prefix; run-bound task snapshots Security review of #718: - oversized() reads native's external_payload envelope as well as PHP's external_storage, so a stored reference over the payload limit omits the preview (and the target name, as PHP does). Run input and output previews use the same check. - The described-command query reads a payload only when its first 30 characters equal native's stored prefix, compared case-sensitively on each backend (COLLATE BINARY, binary cast, PostgreSQL text). - record() reads a task snapshot only from a task of the event's own run. Tests: a real stored reference at the 2 MiB limit and one byte above, upper-case/v2/malformed/large-inline counterfactuals, the query itself on SQLite, and a mismatched-run event with no snapshot. --- .../2026-10-23-task-snapshots.md | 27 ++++ rust/src/runtime/backend.rs | 8 ++ rust/src/runtime/run_describe.rs | 8 +- rust/src/runtime/store.rs | 37 ++++- rust/src/runtime/store_record_tests.rs | 128 ++++++++++++++++++ .../execution/external_payloads/commands.rs | 57 ++++++++ 6 files changed, 258 insertions(+), 7 deletions(-) create mode 100644 rust/src/runtime/store_record_tests.rs diff --git a/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md b/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md index 87931cf15..5ea9a10aa 100644 --- a/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md +++ b/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md @@ -33,6 +33,33 @@ size limit). Native now sends both. With fresh workers on both sides, the descriptions then differ only in `request_route_name` and the `request_fingerprint` hashed from it. +- **Security review corrections.** + - The size check read only PHP's `external_storage` envelope. Native's + stored envelope names its reference `external_payload`, so the flag was + always false. It now reads both, which also corrects the run's input and + output previews. As PHP does, an omitted preview also omits the + command's `target_name`. + - The query read any payload starting with `dw-`, and SQLite's `LIKE` and + MySQL's case-insensitive collation ignore case. It now reads a payload + only when its first 30 characters equal native's stored prefix exactly: + `COLLATE BINARY` on SQLite, a binary cast on MySQL and MariaDB, and + PostgreSQL's case-sensitive text comparison. On MySQL 8.0 and MariaDB + 10.11 the binary cast refuses the upper-case prefix that the column's + collation would match. + - A task snapshot is read only from a task of the event's own run. + - Tests: + - a real uploaded reference on a signal, then the same reference stored + at the 2 MiB limit (not omitted) and one byte over (omitted, with no + target name); the described command never includes the reference; + - counterfactuals that are not native references: an upper-case or v2 + prefix, the prefix with no valid reference after it, and a large inline + payload starting with `dw-`; + - on SQLite, the query itself returns the payload only for the exact + prefix; + - an event recorded on one run with another run's task carries no + snapshot. + + Each test fails without its correction. - **Left as they are.** - PHP's command snapshots carry `request_route_name`, a name Laravel generates at random for the deployment's cached routes. Native has no diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 722cd8f75..22eb45864 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -35,6 +35,9 @@ pub(super) trait Backend: Database { const TIMER_SCHEDULE_SQL: &str = "SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='TimerScheduled' AND payload->>'timer_id'=$2 ORDER BY sequence DESC LIMIT 1"; /// The history sequence of a signal's `SignalReceived` event. const SIGNAL_RECEIVED_SEQUENCE_SQL: &str = "SELECT sequence FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SignalReceived' AND payload->>'signal_id'=$2 ORDER BY sequence LIMIT 1"; + // PostgreSQL compares text case-sensitively. + const RUN_COMMANDS_SQL: &str = + super::store::run_commands_sql!("substr(c.payload,1,30)='dw-native-external-payload-v1:'"); const SELECTION_RESOLUTION_SQL: &str = "SELECT id,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SelectionResolved' AND payload->>'selection_group_id'=$2 ORDER BY sequence LIMIT 1"; // Frozen run command/task repair counters use each backend's actual width. const REPAIR_COUNTER_MAX: i64 = i32::MAX as i64; @@ -185,6 +188,8 @@ ORDER BY t.available_order,t.id LIMIT 100 const TIMER_TASK_SQL: &str = "SELECT id,payload FROM workflow_tasks WHERE workflow_run_id=$1 AND task_type='timer' AND status='ready' AND json_extract(payload,'$.timer_id')=$2 AND namespace=$3"; const TIMER_SCHEDULE_SQL: &str = "SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='TimerScheduled' AND json_extract(payload,'$.timer_id')=$2 ORDER BY sequence DESC LIMIT 1"; const SIGNAL_RECEIVED_SEQUENCE_SQL: &str = "SELECT sequence FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SignalReceived' AND json_extract(payload,'$.signal_id')=$2 ORDER BY sequence LIMIT 1"; + const RUN_COMMANDS_SQL: &str = + super::store::run_commands_sql!("substr(c.payload,1,30) COLLATE BINARY='dw-native-external-payload-v1:'"); const HISTORY_PAGE: &str = "SELECT * FROM (SELECT *,SUM(length(CAST(payload AS BLOB))) OVER (ORDER BY sequence) AS page_bytes FROM workflow_history_events WHERE workflow_run_id=$1 AND sequence>$2 ORDER BY sequence LIMIT $3) AS page WHERE page_bytes<=8388608 ORDER BY sequence"; // Normalize only the two supported UTC storage shapes. Removing trailing // fractional zeros preserves exact decimal ordering, including earlier @@ -457,6 +462,9 @@ impl Backend for MySql { 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 ?"; const TIMER_SCHEDULE_SQL: &str = "SELECT payload FROM workflow_history_events WHERE workflow_run_id=? AND event_type='TimerScheduled' AND JSON_UNQUOTE(JSON_EXTRACT(payload,'$.timer_id'))=? ORDER BY sequence DESC LIMIT 1"; const SIGNAL_RECEIVED_SEQUENCE_SQL: &str = "SELECT sequence FROM workflow_history_events WHERE workflow_run_id=? AND event_type='SignalReceived' AND JSON_UNQUOTE(JSON_EXTRACT(payload,'$.signal_id'))=? ORDER BY sequence LIMIT 1"; + // The payload column's collation ignores case; compare bytes. + const RUN_COMMANDS_SQL: &str = + super::store::run_commands_sql!("CAST(SUBSTRING(c.payload,1,30) AS BINARY)='dw-native-external-payload-v1:'"); // First use the preserved run/sequence index to materialize only IDs and // byte counts for this bounded page. Sorting JSON payloads in the window // exhausts MySQL's default sort buffer before the size rejection can run. diff --git a/rust/src/runtime/run_describe.rs b/rust/src/runtime/run_describe.rs index ae9e6ae3b..c18bf74d1 100644 --- a/rust/src/runtime/run_describe.rs +++ b/rust/src/runtime/run_describe.rs @@ -31,8 +31,12 @@ fn max_payload_bytes() -> i64 { } /// `oversizedPayloadPreview`: an external payload larger than the limit. +/// PHP's stored envelope names its reference `external_storage`; native's +/// names it `external_payload`. pub(super) fn oversized(envelope: &Value) -> bool { - envelope["external_storage"]["size_bytes"].as_i64().is_some_and(|size| size > max_payload_bytes()) + ["external_storage", "external_payload"] + .iter() + .any(|key| envelope[key]["size_bytes"].as_i64().is_some_and(|size| size > max_payload_bytes())) } /// Carbon's `toIso8601String()`, which PHP's run description uses. @@ -315,6 +319,8 @@ mod tests { assert!(oversized(&json!({"external_storage":{"size_bytes":2 * 1024 * 1024 + 1}}))); assert!(!oversized(&json!({"external_storage":{"size_bytes":2 * 1024 * 1024}}))); assert!(!oversized(&json!({"codec":"avro","blob":"AA=="}))); + assert!(oversized(&json!({"codec":"avro","external_payload":{"size_bytes":2 * 1024 * 1024 + 1}}))); + assert!(!oversized(&json!({"codec":"avro","external_payload":{"size_bytes":2 * 1024 * 1024}}))); assert_eq!(without_empty(json!({"a":null,"b":[],"c":{},"d":0})), json!({"c":{},"d":0})); let time = DateTime::parse_from_rfc3339("2026-10-21T07:38:51.123Z").unwrap().with_timezone(&Utc); assert_eq!(iso8601(Some(time)), json!("2026-10-21T07:38:51+00:00")); diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index e09c281fe..8b970ff41 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -46,6 +46,20 @@ const TASK_SNAPSHOT_EVENTS: &[&str] = &[ "WorkflowFailed", "WorkflowTimedOut", ]; +/// Describe's command journal. `$stored` is each backend's exact, +/// case-sensitive test for native's stored-reference prefix +/// (`external_resolution::STORED_PREFIX`), so a command's payload is read only +/// when it is a stored reference. +macro_rules! run_commands_sql { + ($stored:literal) => { + concat!( + "SELECT c.id,c.command_sequence,c.command_type,c.target_scope,c.requested_workflow_run_id,c.resolved_workflow_run_id,c.source,c.status,c.outcome,c.rejection_reason,c.workflow_type,c.workflow_class,c.accepted_at,c.applied_at,c.rejected_at,c.context,s.signal_name,CASE WHEN ", + $stored, + " THEN c.payload END AS stored_payload FROM workflow_commands c LEFT JOIN workflow_signal_records s ON s.workflow_command_id=c.id WHERE c.workflow_run_id=$1 AND c.workflow_instance_id=$2 ORDER BY c.command_sequence,c.created_at,c.id" + ) + }; +} +pub(super) use run_commands_sql; /// The runtime-owned message-stream delivery signal (`WorkflowStub::MESSAGE_STREAM_RUNTIME_SIGNAL`). const MESSAGE_STREAM_SIGNAL: &str = "__durable_workflow_message_stream"; pub(super) const TASK_CANDIDATES_SQL: &str = "SELECT t.*,r.workflow_type FROM workflow_tasks t JOIN workflow_runs r ON r.id=t.workflow_run_id WHERE t.namespace=$1 AND t.queue=$2 AND t.task_type=$3 AND r.status IN ('pending','running','waiting') AND (t.status='ready' OR (t.status='leased' AND t.task_type='workflow' AND t.lease_expires_at<=$4)) AND (t.available_at IS NULL OR t.available_at<=$5) AND (COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,'')) IS NULL OR COALESCE(NULLIF(t.compatibility,''),NULLIF(r.compatibility,''))=$6) ORDER BY t.priority,t.available_at,t.id LIMIT 100"; @@ -555,23 +569,28 @@ where // copied merely to display the command's target and admission result. // Only a stored external reference is read, to tell whether its preview // is omitted; inline payloads stay unread. - let command_rows = Self::query("SELECT c.id,c.command_sequence,c.command_type,c.target_scope,c.requested_workflow_run_id,c.resolved_workflow_run_id,c.source,c.status,c.outcome,c.rejection_reason,c.workflow_type,c.workflow_class,c.accepted_at,c.applied_at,c.rejected_at,c.context,s.signal_name,CASE WHEN c.payload LIKE 'dw-%' THEN c.payload END AS stored_payload FROM workflow_commands c LEFT JOIN workflow_signal_records s ON s.workflow_command_id=c.id WHERE c.workflow_run_id=$1 AND c.workflow_instance_id=$2 ORDER BY c.command_sequence,c.created_at,c.id") + let command_rows = Self::query(DB::RUN_COMMANDS_SQL) .bind(DB::string(&row, "id")?).bind(workflow_id).fetch_all(&self.pool).await?; let mut commands = Vec::with_capacity(command_rows.len()); for command in command_rows { + // `oversizedPayloadPreview` over the stored reference's envelope; + // an omitted preview also omits the target name, as PHP does. + let omitted = DB::optional_string(&command, "stored_payload")? + .as_deref() + .and_then(super::external_resolution::stored_envelope) + .is_some_and(|envelope| super::run_describe::oversized(&envelope)); + let target_name = DB::optional_string(&command, "signal_name")?.filter(|_| !omitted); let mut public = json!({"id":DB::string(&command,"id")?,"sequence":DB::optional_number(&command,"command_sequence")?, "type":DB::string(&command,"command_type")?,"target_scope":DB::string(&command,"target_scope")?, "requested_run_id":DB::optional_string(&command,"requested_workflow_run_id")?, "resolved_run_id":DB::optional_string(&command,"resolved_workflow_run_id")?, - "target_name":DB::optional_string(&command,"signal_name")?,"source":DB::string(&command,"source")?, + "target_name":target_name,"source":DB::string(&command,"source")?, "status":DB::string(&command,"status")?,"outcome":DB::optional_string(&command,"outcome")?, "reason":DB::optional_string(&command,"rejection_reason")?,"rejection_reason":DB::optional_string(&command,"rejection_reason")?, // `WorkflowCommand::record` keeps the run's class on each command. "workflow_type":DB::optional_string(&command,"workflow_type")?, "workflow_class":DB::optional_string(&command,"workflow_class")?.or(DB::optional_string(&row,"workflow_class")?), - "payload_preview_omitted":DB::optional_string(&command,"stored_payload")?.as_deref() - .and_then(super::external_resolution::stored_envelope) - .is_some_and(|envelope| super::run_describe::oversized(&envelope)), + "payload_preview_omitted":omitted, "accepted_at":DB::optional_instant(&command,"accepted_at")?,"applied_at":DB::optional_instant(&command,"applied_at")?, "rejected_at":DB::optional_instant(&command,"rejected_at")?}); let fields = public.as_object_mut().expect("command object"); @@ -3425,11 +3444,13 @@ where if let Some(task_id) = task_id && TASK_SNAPSHOT_EVENTS.contains(&event_type) && payload.as_object().is_some_and(|fields| !fields.contains_key("task")) - && let Some(task) = Self::query("SELECT * FROM workflow_tasks WHERE id=$1") + && let Some(task) = Self::query("SELECT * FROM workflow_tasks WHERE id=$1 AND workflow_run_id=$2") .bind(task_id) + .bind(run_id) .fetch_optional(&mut **tx) .await? { + // Only the run's own task: another run's is never copied. payload["task"] = Self::task_snapshot(&task)?; } // An event recorded with a command carries that command's snapshot. @@ -4082,3 +4103,7 @@ pub(crate) fn capabilities() -> Value { } value } + +#[cfg(test)] +#[path = "store_record_tests.rs"] +mod record_tests; diff --git a/rust/src/runtime/store_record_tests.rs b/rust/src/runtime/store_record_tests.rs new file mode 100644 index 000000000..1f312ffeb --- /dev/null +++ b/rust/src/runtime/store_record_tests.rs @@ -0,0 +1,128 @@ +//! History recording and described-command boundaries that a public route +//! cannot reach with mismatched identifiers. +use super::Store; +use crate::runtime::backend::Backend; +use crate::runtime::external_resolution::STORED_PREFIX; +use serde_json::{Value, json}; +use std::collections::BTreeMap; + +#[test] +fn described_commands_read_only_the_exact_native_stored_prefix() { + // Each backend compares the first 30 characters, byte for byte, with + // the whole prefix. + assert_eq!(STORED_PREFIX.len(), 30); + let quoted = format!("'{STORED_PREFIX}'"); + for (backend, sql, comparison) in [ + ("sqlite", ::RUN_COMMANDS_SQL, "substr(c.payload,1,30) COLLATE BINARY="), + ("postgres", ::RUN_COMMANDS_SQL, "substr(c.payload,1,30)="), + ("mysql", ::RUN_COMMANDS_SQL, "CAST(SUBSTRING(c.payload,1,30) AS BINARY)="), + ] { + assert!(sql.contains(&format!("CASE WHEN {comparison}{quoted} THEN c.payload END")), "{backend}: {sql}"); + assert!(!sql.contains("LIKE"), "{backend}: {sql}"); + } +} + +async fn recorded_task(store: &Store, run_id: &str) -> Value { + let row = Store::::query( + "SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SideEffectRecorded'", + ) + .bind(run_id) + .fetch_one(&store.pool) + .await + .unwrap(); + ::document_row(&row, "payload").unwrap()["task"].clone() +} + +#[tokio::test] +async fn a_task_snapshot_comes_only_from_the_events_own_run() { + let directory = tempfile::tempdir().unwrap(); + let store = Store::::new( + crate::runtime::sqlite::open(directory.path().join("record.sqlite").to_str().unwrap()) + .await + .unwrap(), + ); + store + .register( + json!({"worker_id":"record-worker","task_queue":"test","runtime":"php", + "supported_workflow_types":["echo"],"supported_activity_types":[]}), + &crate::runtime::QueryTaskConfig::default(), + None, + ) + .await + .unwrap(); + let mut runs = Vec::new(); + for workflow_id in ["record-own", "record-other"] { + let started = store + .start( + json!({"workflow_id":workflow_id,"workflow_type":"echo","task_queue":"test", + "input":{"codec":"avro","blob":"AA=="}}), + BTreeMap::new(), + None, + None, + ) + .await + .unwrap(); + runs.push(started["run_id"].as_str().unwrap().to_owned()); + } + let task = Store::::query("SELECT id FROM workflow_tasks WHERE workflow_run_id=$1") + .bind(&runs[0]) + .fetch_one(&store.pool) + .await + .unwrap(); + let task_id = ::string(&task, "id").unwrap(); + + let mut tx = store.begin().await.unwrap(); + for run_id in &runs { + Store::::record(&mut tx, run_id, "SideEffectRecorded", json!({"sequence":1}), + Some(&task_id), None, None, false) + .await + .unwrap(); + } + tx.commit().await.unwrap(); + + assert_eq!(recorded_task(&store, &runs[0]).await["id"], json!(task_id), "the run's own task"); + // Recorded on the other run with the first run's task, the event carries + // no snapshot at all. + assert!(recorded_task(&store, &runs[1]).await.is_null()); +} + +// The query itself, run on rows whose payloads begin like a stored +// reference: only native's exact prefix is selected. +#[tokio::test] +async fn described_commands_select_no_inline_payload() { + let directory = tempfile::tempdir().unwrap(); + let store = Store::::new( + crate::runtime::sqlite::open(directory.path().join("commands.sqlite").to_str().unwrap()) + .await + .unwrap(), + ); + let exact = format!("{STORED_PREFIX}{{}}"); + let rows = [ + ("c1", exact.clone(), true), + ("c2", exact.to_uppercase(), false), + ("c3", format!("dw-native-external-payload-v1{}", "A".repeat(4096)), false), + ("c4", format!("dw-{}", "A".repeat(4096)), false), + ("c5", "dw-native".to_owned(), false), + ]; + for (sequence, (id, payload, _)) in rows.iter().enumerate() { + Store::::query("INSERT INTO workflow_commands(id,workflow_instance_id,workflow_run_id,command_type,status,payload,command_sequence) VALUES ($1,'instance','run','signal','accepted',$2,$3)") + .bind(*id) + .bind(payload.as_str()) + .bind(sequence as i64 + 1) + .execute(&store.pool) + .await + .unwrap(); + } + let selected = Store::::query(::RUN_COMMANDS_SQL) + .bind("run") + .bind("instance") + .fetch_all(&store.pool) + .await + .unwrap(); + assert_eq!(selected.len(), rows.len()); + for (row, (id, payload, read)) in selected.iter().zip(&rows) { + assert_eq!(::string(row, "id").unwrap(), *id); + let stored = ::optional_string(row, "stored_payload").unwrap(); + assert_eq!(stored.as_deref(), read.then_some(payload.as_str()), "{id}"); + } +} diff --git a/rust/tests/execution/external_payloads/commands.rs b/rust/tests/execution/external_payloads/commands.rs index 82237259c..bb962993b 100644 --- a/rust/tests/execution/external_payloads/commands.rs +++ b/rust/tests/execution/external_payloads/commands.rs @@ -151,3 +151,60 @@ async fn signal_update_and_query_input_resolve_runtime_references_like_php() { runtime.close().await; database.remove().await; } + +/// The described signal command, which must show only metadata. +async fn described_signal(app: &Router, workflow_id: &str) -> Value { + let (status, body) = json_request(app, "GET", &format!("/api/workflows/{workflow_id}"), "operator-token", Value::Null).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let command = body["commands"].as_array().unwrap().iter().find(|command| command["type"] == "signal").unwrap().clone(); + for field in ["payload", "external_payload", "external_storage", "reference_id", "sha256", "size_bytes", "arguments"] { + assert!(command.get(field).is_none(), "{field}: {command}"); + } + assert!(!command.to_string().contains("ep_"), "no reference identity: {command}"); + command +} + +// `WorkflowController::oversizedPayloadPreview` over each command's stored +// envelope: only a stored external reference larger than the payload limit +// (2 MiB by default) omits the preview, and then the target name too. +#[tokio::test] +async fn described_commands_omit_only_an_oversized_stored_reference() { + let temp = tempfile::tempdir().unwrap(); + let (database, runtime, app) = setup(ExternalPayloadConfig::default(), &temp.path().join("objects")).await; + waiting_run(&app, "described-reference", json!({"worker_id":"describe-worker","task_queue":"described-reference", + "runtime":"php","supported_workflow_types":["echo"],"supported_activity_types":[]})).await; + let reference = upload(&app, &argument("signal")).await; + let (status, body) = json_request(&app, "POST", "/api/workflows/described-reference/signal/s", "operator-token", + json!({"input":{"codec":"avro","external_payload":reference}})).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let command = described_signal(&app, "described-reference").await; + assert_eq!((command["payload_preview_omitted"].clone(), command["target_name"].clone()), (json!(false), json!("s")), "{command}"); + let command_id = command["id"].as_str().unwrap().to_owned(); + + // The same stored reference, as native writes it, at sizes around the + // limit, and counterfactuals that are not native stored references. + let limit = 2 * 1024 * 1024_i64; + let stored = |size: i64| { + let mut reference = reference.clone(); + reference["size_bytes"] = json!(size); + format!("dw-native-external-payload-v1:{reference}") + }; + for (payload, omitted) in [ + (stored(limit), false), + (stored(limit + 1), true), + // Not native's exact prefix, so never read. + (stored(limit + 1).replace("dw-native-external-payload-v1:", "DW-NATIVE-EXTERNAL-PAYLOAD-V1:"), false), + (stored(limit + 1).replace("dw-native-external-payload-v1:", "dw-native-external-payload-v2:"), false), + // Native's prefix with no valid reference after it. + (format!("dw-native-external-payload-v1:{{\"size_bytes\":{}}}", limit + 1), false), + // A large inline payload. + (format!("dw-{}", "A".repeat(limit as usize + 1)), false), + ] { + database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_commands SET payload='{payload}' WHERE id='{command_id}'"))).await; + let command = described_signal(&app, "described-reference").await; + assert_eq!(command["payload_preview_omitted"], json!(omitted), "{}: {command}", &payload[..payload.len().min(48)]); + assert_eq!(command.get("target_name").is_none(), omitted, "{command}"); + } + runtime.close().await; + database.remove().await; +} From 38717d3372767b72cf17851b0bd16ca935515e81 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 24 Oct 2026 01:43:15 +0000 Subject: [PATCH 4/4] Rust #325: read a signal's stored reference from its signal record Security review of #718 at 65772abac: native keeps a signal's stored input reference in workflow_signal_records.arguments; workflow_commands.payload holds the encoded command. Read the exact native sentinel from s.arguments for signals and from c.payload for starts only, case-sensitively per backend. The describe test now proves the real signal row is the source, mutates that row around the limit with the same counterfactuals, shows an oversized sentinel in a signal's command row does not count, and covers the start command's row. --- .../2026-10-23-task-snapshots.md | 30 +++++---- rust/src/runtime/backend.rs | 15 ++++- rust/src/runtime/store.rs | 18 +++--- rust/src/runtime/store_record_tests.rs | 62 +++++++++++++------ rust/tests/execution/external_payloads.rs | 2 +- .../execution/external_payloads/commands.rs | 50 +++++++++++---- 6 files changed, 124 insertions(+), 53 deletions(-) diff --git a/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md b/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md index 5ea9a10aa..f139ee749 100644 --- a/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md +++ b/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md @@ -39,23 +39,31 @@ always false. It now reads both, which also corrects the run's input and output previews. As PHP does, an omitted preview also omits the command's `target_name`. - - The query read any payload starting with `dw-`, and SQLite's `LIKE` and - MySQL's case-insensitive collation ignore case. It now reads a payload - only when its first 30 characters equal native's stored prefix exactly: - `COLLATE BINARY` on SQLite, a binary cast on MySQL and MariaDB, and - PostgreSQL's case-sensitive text comparison. On MySQL 8.0 and MariaDB + - The query read any command payload starting with `dw-`, and SQLite's + `LIKE` and MySQL's case-insensitive collation ignore case. It also read the + wrong column for signals. Native keeps a signal's stored input reference + in its signal record (`workflow_signal_records.arguments`) and a start's + in the command row (`workflow_commands.payload`). Other commands keep + none. The query now reads only that column for each of the two command + types, and only when its first 30 characters equal native's stored prefix + exactly: `COLLATE BINARY` on SQLite, a binary cast on MySQL and MariaDB, + and PostgreSQL's case-sensitive text comparison. On MySQL 8.0 and MariaDB 10.11 the binary cast refuses the upper-case prefix that the column's collation would match. - A task snapshot is read only from a task of the event's own run. - Tests: - - a real uploaded reference on a signal, then the same reference stored - at the 2 MiB limit (not omitted) and one byte over (omitted, with no - target name); the described command never includes the reference; + - a real uploaded reference on a signal, found in the signal record and + not in the command row; + - that signal record rewritten with the same reference at the 2 MiB limit + (not omitted) and one byte over (omitted, with no target name). The + described command never includes the reference; - counterfactuals that are not native references: an upper-case or v2 prefix, the prefix with no valid reference after it, and a large inline - payload starting with `dw-`; - - on SQLite, the query itself returns the payload only for the exact - prefix; + payload starting with `dw-`. An oversized reference in the signal's + command row is not counted either; + - the same sizes and counterfactuals in a start command's row; + - on SQLite, the query itself returns only the exact prefix in the + command type's own column; - an event recorded on one run with another run's task carries no snapshot. diff --git a/rust/src/runtime/backend.rs b/rust/src/runtime/backend.rs index 22eb45864..67a823065 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -37,7 +37,10 @@ pub(super) trait Backend: Database { const SIGNAL_RECEIVED_SEQUENCE_SQL: &str = "SELECT sequence FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SignalReceived' AND payload->>'signal_id'=$2 ORDER BY sequence LIMIT 1"; // PostgreSQL compares text case-sensitively. const RUN_COMMANDS_SQL: &str = - super::store::run_commands_sql!("substr(c.payload,1,30)='dw-native-external-payload-v1:'"); + super::store::run_commands_sql!( + "substr(s.arguments,1,30)='dw-native-external-payload-v1:'", + "substr(c.payload,1,30)='dw-native-external-payload-v1:'" + ); const SELECTION_RESOLUTION_SQL: &str = "SELECT id,payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SelectionResolved' AND payload->>'selection_group_id'=$2 ORDER BY sequence LIMIT 1"; // Frozen run command/task repair counters use each backend's actual width. const REPAIR_COUNTER_MAX: i64 = i32::MAX as i64; @@ -189,7 +192,10 @@ ORDER BY t.available_order,t.id LIMIT 100 const TIMER_SCHEDULE_SQL: &str = "SELECT payload FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='TimerScheduled' AND json_extract(payload,'$.timer_id')=$2 ORDER BY sequence DESC LIMIT 1"; const SIGNAL_RECEIVED_SEQUENCE_SQL: &str = "SELECT sequence FROM workflow_history_events WHERE workflow_run_id=$1 AND event_type='SignalReceived' AND json_extract(payload,'$.signal_id')=$2 ORDER BY sequence LIMIT 1"; const RUN_COMMANDS_SQL: &str = - super::store::run_commands_sql!("substr(c.payload,1,30) COLLATE BINARY='dw-native-external-payload-v1:'"); + super::store::run_commands_sql!( + "substr(s.arguments,1,30) COLLATE BINARY='dw-native-external-payload-v1:'", + "substr(c.payload,1,30) COLLATE BINARY='dw-native-external-payload-v1:'" + ); const HISTORY_PAGE: &str = "SELECT * FROM (SELECT *,SUM(length(CAST(payload AS BLOB))) OVER (ORDER BY sequence) AS page_bytes FROM workflow_history_events WHERE workflow_run_id=$1 AND sequence>$2 ORDER BY sequence LIMIT $3) AS page WHERE page_bytes<=8388608 ORDER BY sequence"; // Normalize only the two supported UTC storage shapes. Removing trailing // fractional zeros preserves exact decimal ordering, including earlier @@ -464,7 +470,10 @@ impl Backend for MySql { const SIGNAL_RECEIVED_SEQUENCE_SQL: &str = "SELECT sequence FROM workflow_history_events WHERE workflow_run_id=? AND event_type='SignalReceived' AND JSON_UNQUOTE(JSON_EXTRACT(payload,'$.signal_id'))=? ORDER BY sequence LIMIT 1"; // The payload column's collation ignores case; compare bytes. const RUN_COMMANDS_SQL: &str = - super::store::run_commands_sql!("CAST(SUBSTRING(c.payload,1,30) AS BINARY)='dw-native-external-payload-v1:'"); + super::store::run_commands_sql!( + "CAST(SUBSTRING(s.arguments,1,30) AS BINARY)='dw-native-external-payload-v1:'", + "CAST(SUBSTRING(c.payload,1,30) AS BINARY)='dw-native-external-payload-v1:'" + ); // First use the preserved run/sequence index to materialize only IDs and // byte counts for this bounded page. Sorting JSON payloads in the window // exhausts MySQL's default sort buffer before the size rejection can run. diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 8b970ff41..8c0133c61 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -46,15 +46,19 @@ const TASK_SNAPSHOT_EVENTS: &[&str] = &[ "WorkflowFailed", "WorkflowTimedOut", ]; -/// Describe's command journal. `$stored` is each backend's exact, -/// case-sensitive test for native's stored-reference prefix -/// (`external_resolution::STORED_PREFIX`), so a command's payload is read only -/// when it is a stored reference. +/// Describe's command journal. Each backend passes its exact, case-sensitive +/// test for native's stored-reference prefix +/// (`external_resolution::STORED_PREFIX`) over the column that holds a +/// command's input: a signal's in its signal record (`s.arguments`), a start's +/// in the command row (`c.payload`). Other commands keep no stored reference. +/// Only a stored reference is read; inline payloads are not. macro_rules! run_commands_sql { - ($stored:literal) => { + ($signal:literal, $start:literal) => { concat!( - "SELECT c.id,c.command_sequence,c.command_type,c.target_scope,c.requested_workflow_run_id,c.resolved_workflow_run_id,c.source,c.status,c.outcome,c.rejection_reason,c.workflow_type,c.workflow_class,c.accepted_at,c.applied_at,c.rejected_at,c.context,s.signal_name,CASE WHEN ", - $stored, + "SELECT c.id,c.command_sequence,c.command_type,c.target_scope,c.requested_workflow_run_id,c.resolved_workflow_run_id,c.source,c.status,c.outcome,c.rejection_reason,c.workflow_type,c.workflow_class,c.accepted_at,c.applied_at,c.rejected_at,c.context,s.signal_name,CASE WHEN c.command_type='signal' AND ", + $signal, + " THEN s.arguments WHEN c.command_type='start' AND ", + $start, " THEN c.payload END AS stored_payload FROM workflow_commands c LEFT JOIN workflow_signal_records s ON s.workflow_command_id=c.id WHERE c.workflow_run_id=$1 AND c.workflow_instance_id=$2 ORDER BY c.command_sequence,c.created_at,c.id" ) }; diff --git a/rust/src/runtime/store_record_tests.rs b/rust/src/runtime/store_record_tests.rs index 1f312ffeb..9529c9473 100644 --- a/rust/src/runtime/store_record_tests.rs +++ b/rust/src/runtime/store_record_tests.rs @@ -8,16 +8,21 @@ use std::collections::BTreeMap; #[test] fn described_commands_read_only_the_exact_native_stored_prefix() { - // Each backend compares the first 30 characters, byte for byte, with - // the whole prefix. + // Each backend compares the first 30 characters, byte for byte, with the + // whole prefix: a signal's in its signal record, a start's in its command. assert_eq!(STORED_PREFIX.len(), 30); let quoted = format!("'{STORED_PREFIX}'"); - for (backend, sql, comparison) in [ - ("sqlite", ::RUN_COMMANDS_SQL, "substr(c.payload,1,30) COLLATE BINARY="), - ("postgres", ::RUN_COMMANDS_SQL, "substr(c.payload,1,30)="), - ("mysql", ::RUN_COMMANDS_SQL, "CAST(SUBSTRING(c.payload,1,30) AS BINARY)="), + for (backend, sql, signal, start) in [ + ("sqlite", ::RUN_COMMANDS_SQL, + "substr(s.arguments,1,30) COLLATE BINARY=", "substr(c.payload,1,30) COLLATE BINARY="), + ("postgres", ::RUN_COMMANDS_SQL, + "substr(s.arguments,1,30)=", "substr(c.payload,1,30)="), + ("mysql", ::RUN_COMMANDS_SQL, + "CAST(SUBSTRING(s.arguments,1,30) AS BINARY)=", "CAST(SUBSTRING(c.payload,1,30) AS BINARY)="), ] { - assert!(sql.contains(&format!("CASE WHEN {comparison}{quoted} THEN c.payload END")), "{backend}: {sql}"); + let expected = format!("CASE WHEN c.command_type='signal' AND {signal}{quoted} THEN s.arguments \ + WHEN c.command_type='start' AND {start}{quoted} THEN c.payload END"); + assert!(sql.contains(&expected), "{backend}: {sql}"); assert!(!sql.contains("LIKE"), "{backend}: {sql}"); } } @@ -86,8 +91,8 @@ async fn a_task_snapshot_comes_only_from_the_events_own_run() { assert!(recorded_task(&store, &runs[1]).await.is_null()); } -// The query itself, run on rows whose payloads begin like a stored -// reference: only native's exact prefix is selected. +// The query itself, run on rows whose inputs begin like a stored reference: +// only native's exact prefix in the command type's own input column is read. #[tokio::test] async fn described_commands_select_no_inline_payload() { let directory = tempfile::tempdir().unwrap(); @@ -97,21 +102,40 @@ async fn described_commands_select_no_inline_payload() { .unwrap(), ); let exact = format!("{STORED_PREFIX}{{}}"); + let large = format!("dw-{}", "A".repeat(4096)); + let near = format!("dw-native-external-payload-v1{}", "A".repeat(4096)); + let upper = exact.to_uppercase(); + // (id, type, command payload, signal arguments, read) let rows = [ - ("c1", exact.clone(), true), - ("c2", exact.to_uppercase(), false), - ("c3", format!("dw-native-external-payload-v1{}", "A".repeat(4096)), false), - ("c4", format!("dw-{}", "A".repeat(4096)), false), - ("c5", "dw-native".to_owned(), false), + ("c1", "signal", "AA==", Some(exact.as_str()), Some(exact.as_str())), + ("c2", "signal", "AA==", Some(upper.as_str()), None), + ("c3", "signal", "AA==", Some(near.as_str()), None), + ("c4", "signal", "AA==", Some(large.as_str()), None), + // A signal's command row is not its input. + ("c5", "signal", exact.as_str(), Some("AA=="), None), + ("c6", "start", exact.as_str(), None, Some(exact.as_str())), + ("c7", "start", large.as_str(), None, None), + // Other commands keep no stored reference. + ("c8", "update", exact.as_str(), None, None), ]; - for (sequence, (id, payload, _)) in rows.iter().enumerate() { - Store::::query("INSERT INTO workflow_commands(id,workflow_instance_id,workflow_run_id,command_type,status,payload,command_sequence) VALUES ($1,'instance','run','signal','accepted',$2,$3)") + for (sequence, (id, kind, payload, arguments, _)) in rows.iter().enumerate() { + Store::::query("INSERT INTO workflow_commands(id,workflow_instance_id,workflow_run_id,command_type,status,payload,command_sequence) VALUES ($1,'instance','run',$2,'accepted',$3,$4)") .bind(*id) - .bind(payload.as_str()) + .bind(*kind) + .bind(*payload) .bind(sequence as i64 + 1) .execute(&store.pool) .await .unwrap(); + if let Some(arguments) = arguments { + Store::::query("INSERT INTO workflow_signal_records(id,workflow_command_id,signal_name,status,arguments) VALUES ($1,$2,'s','received',$3)") + .bind(format!("s-{id}")) + .bind(*id) + .bind(*arguments) + .execute(&store.pool) + .await + .unwrap(); + } } let selected = Store::::query(::RUN_COMMANDS_SQL) .bind("run") @@ -120,9 +144,9 @@ async fn described_commands_select_no_inline_payload() { .await .unwrap(); assert_eq!(selected.len(), rows.len()); - for (row, (id, payload, read)) in selected.iter().zip(&rows) { + for (row, (id, _, _, _, read)) in selected.iter().zip(&rows) { assert_eq!(::string(row, "id").unwrap(), *id); let stored = ::optional_string(row, "stored_payload").unwrap(); - assert_eq!(stored.as_deref(), read.then_some(payload.as_str()), "{id}"); + assert_eq!(stored.as_deref(), *read, "{id}"); } } diff --git a/rust/tests/execution/external_payloads.rs b/rust/tests/execution/external_payloads.rs index c3771c42e..5acb9e25a 100644 --- a/rust/tests/execution/external_payloads.rs +++ b/rust/tests/execution/external_payloads.rs @@ -133,7 +133,7 @@ async fn unclaimed(database: &TestDatabase) -> i64 { } /// One integer the database answers. -async fn count(database: &TestDatabase, query: &'static str) -> i64 { +async fn count(database: &TestDatabase, query: impl sqlx::SqlSafeStr) -> i64 { match database { TestDatabase::Sqlite(dir) => { let pool = sqlx::SqlitePool::connect(&format!( diff --git a/rust/tests/execution/external_payloads/commands.rs b/rust/tests/execution/external_payloads/commands.rs index bb962993b..e93b63183 100644 --- a/rust/tests/execution/external_payloads/commands.rs +++ b/rust/tests/execution/external_payloads/commands.rs @@ -152,11 +152,11 @@ async fn signal_update_and_query_input_resolve_runtime_references_like_php() { database.remove().await; } -/// The described signal command, which must show only metadata. -async fn described_signal(app: &Router, workflow_id: &str) -> Value { +/// One described command of `kind`, which must show only metadata. +async fn described(app: &Router, workflow_id: &str, kind: &str) -> Value { let (status, body) = json_request(app, "GET", &format!("/api/workflows/{workflow_id}"), "operator-token", Value::Null).await; assert_eq!(status, StatusCode::OK, "{body}"); - let command = body["commands"].as_array().unwrap().iter().find(|command| command["type"] == "signal").unwrap().clone(); + let command = body["commands"].as_array().unwrap().iter().find(|command| command["type"] == kind).unwrap().clone(); for field in ["payload", "external_payload", "external_storage", "reference_id", "sha256", "size_bytes", "arguments"] { assert!(command.get(field).is_none(), "{field}: {command}"); } @@ -166,7 +166,9 @@ async fn described_signal(app: &Router, workflow_id: &str) -> Value { // `WorkflowController::oversizedPayloadPreview` over each command's stored // envelope: only a stored external reference larger than the payload limit -// (2 MiB by default) omits the preview, and then the target name too. +// (2 MiB by default) omits the preview, and then the target name too. Native +// keeps a signal's stored reference in its signal record and a start's in the +// command row. #[tokio::test] async fn described_commands_omit_only_an_oversized_stored_reference() { let temp = tempfile::tempdir().unwrap(); @@ -177,9 +179,15 @@ async fn described_commands_omit_only_an_oversized_stored_reference() { let (status, body) = json_request(&app, "POST", "/api/workflows/described-reference/signal/s", "operator-token", json!({"input":{"codec":"avro","external_payload":reference}})).await; assert_eq!(status, StatusCode::ACCEPTED, "{body}"); - let command = described_signal(&app, "described-reference").await; + let command = described(&app, "described-reference", "signal").await; assert_eq!((command["payload_preview_omitted"].clone(), command["target_name"].clone()), (json!(false), json!("s")), "{command}"); - let command_id = command["id"].as_str().unwrap().to_owned(); + let signal = command["id"].as_str().unwrap().to_owned(); + // The real stored reference is in the signal record, not the command row. + let identity = reference["reference_id"].as_str().unwrap(); + assert_eq!(count(&database, sqlx::AssertSqlSafe(format!( + "SELECT COUNT(*) FROM workflow_signal_records WHERE workflow_command_id='{signal}' AND arguments LIKE 'dw-native-external-payload-v1:%{identity}%'"))).await, 1); + assert_eq!(count(&database, sqlx::AssertSqlSafe(format!( + "SELECT COUNT(*) FROM workflow_commands WHERE id='{signal}' AND payload LIKE 'dw-native%'"))).await, 0); // The same stored reference, as native writes it, at sizes around the // limit, and counterfactuals that are not native stored references. @@ -189,7 +197,7 @@ async fn described_commands_omit_only_an_oversized_stored_reference() { reference["size_bytes"] = json!(size); format!("dw-native-external-payload-v1:{reference}") }; - for (payload, omitted) in [ + let cases = [ (stored(limit), false), (stored(limit + 1), true), // Not native's exact prefix, so never read. @@ -199,11 +207,29 @@ async fn described_commands_omit_only_an_oversized_stored_reference() { (format!("dw-native-external-payload-v1:{{\"size_bytes\":{}}}", limit + 1), false), // A large inline payload. (format!("dw-{}", "A".repeat(limit as usize + 1)), false), - ] { - database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_commands SET payload='{payload}' WHERE id='{command_id}'"))).await; - let command = described_signal(&app, "described-reference").await; - assert_eq!(command["payload_preview_omitted"], json!(omitted), "{}: {command}", &payload[..payload.len().min(48)]); - assert_eq!(command.get("target_name").is_none(), omitted, "{command}"); + ]; + for (arguments, omitted) in &cases { + database.execute(sqlx::AssertSqlSafe(format!( + "UPDATE workflow_signal_records SET arguments='{arguments}' WHERE workflow_command_id='{signal}'"))).await; + let command = described(&app, "described-reference", "signal").await; + assert_eq!(command["payload_preview_omitted"], json!(omitted), "{}: {command}", &arguments[..arguments.len().min(48)]); + assert_eq!(command.get("target_name").is_none(), *omitted, "{command}"); + } + // An oversized reference in the signal's command row is not its input. + database.execute(sqlx::AssertSqlSafe(format!( + "UPDATE workflow_signal_records SET arguments='{}' WHERE workflow_command_id='{signal}'", stored(limit)))).await; + database.execute(sqlx::AssertSqlSafe(format!( + "UPDATE workflow_commands SET payload='{}' WHERE id='{signal}'", stored(limit + 1)))).await; + let command = described(&app, "described-reference", "signal").await; + assert_eq!((command["payload_preview_omitted"].clone(), command["target_name"].clone()), (json!(false), json!("s")), "{command}"); + + // A start keeps its stored input reference in the command row. + let start = described(&app, "described-reference", "start").await["id"].as_str().unwrap().to_owned(); + for (payload, omitted) in &cases { + database.execute(sqlx::AssertSqlSafe(format!( + "UPDATE workflow_commands SET payload='{payload}' WHERE id='{start}'"))).await; + let command = described(&app, "described-reference", "start").await; + assert_eq!(command["payload_preview_omitted"], json!(omitted), "start {}: {command}", &payload[..payload.len().min(48)]); } runtime.close().await; database.remove().await;