Rust #325: task snapshots on every event PHP records with a task, and describe command class/preview

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).

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,

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.

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"));