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..f139ee749 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-task-snapshots.md @@ -0,0 +1,79 @@ +# 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 + 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. +- **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. +- **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 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, 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-`. 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. + + 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 + 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/backend.rs b/rust/src/runtime/backend.rs index 722cd8f75..67a823065 100644 --- a/rust/src/runtime/backend.rs +++ b/rust/src/runtime/backend.rs @@ -35,6 +35,12 @@ 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(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; @@ -185,6 +191,11 @@ 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(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 @@ -457,6 +468,12 @@ 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(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/run_describe.rs b/rust/src/runtime/run_describe.rs index 070e4e364..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. -fn oversized(envelope: &Value) -> bool { - envelope["external_storage"]["size_bytes"].as_i64().is_some_and(|size| size > max_payload_bytes()) +/// PHP's stored envelope names its reference `external_storage`; native's +/// names it `external_payload`. +pub(super) fn oversized(envelope: &Value) -> bool { + ["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 d02b6ee5c..8c0133c61 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -15,6 +15,55 @@ 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", +]; +/// 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 { + ($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 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" + ) + }; +} +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"; @@ -522,18 +571,30 @@ 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(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")?, - "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":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"); @@ -1713,7 +1774,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, @@ -2494,8 +2556,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, @@ -2893,7 +2954,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, @@ -3019,6 +3081,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, @@ -3042,7 +3113,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?; @@ -3371,9 +3442,22 @@ 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 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. 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? @@ -4023,3 +4107,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..9529c9473 --- /dev/null +++ b/rust/src/runtime/store_record_tests.rs @@ -0,0 +1,152 @@ +//! 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: 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, 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)="), + ] { + 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}"); + } +} + +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 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(); + let store = Store::::new( + crate::runtime::sqlite::open(directory.path().join("commands.sqlite").to_str().unwrap()) + .await + .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", "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, 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(*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") + .bind("instance") + .fetch_all(&store.pool) + .await + .unwrap(); + assert_eq!(selected.len(), rows.len()); + 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, "{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 82237259c..e93b63183 100644 --- a/rust/tests/execution/external_payloads/commands.rs +++ b/rust/tests/execution/external_payloads/commands.rs @@ -151,3 +151,86 @@ async fn signal_update_and_query_input_resolve_runtime_references_like_php() { runtime.close().await; database.remove().await; } + +/// 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"] == 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}"); + } + 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. 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(); + 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(&app, "described-reference", "signal").await; + assert_eq!((command["payload_preview_omitted"].clone(), command["target_name"].clone()), (json!(false), json!("s")), "{command}"); + 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. + 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}") + }; + let cases = [ + (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), + ]; + 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; +} diff --git a/rust/tests/execution/worker_history_presentation.rs b/rust/tests/execution/worker_history_presentation.rs index a5cafa602..e95689628 100644 --- a/rust/tests/execution/worker_history_presentation.rs +++ b/rust/tests/execution/worker_history_presentation.rs @@ -63,6 +63,65 @@ 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}"); +} + // 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