From 8ab5cc6be094e4476e3e5dc8364c1e4979a6df0a Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 19:30:22 +0000 Subject: [PATCH 01/11] Show workers a consumed signal's sequence and keep full task snapshots, like Server 2.11.1 Worker polls and history pages add workflow_sequence to a consumed signal's SignalReceived from its record, as WorkflowTaskPoller does; the run history route still shows the event as recorded. Task snapshots keep compatibility, the dispatch and claim stamps and errors when they are set, as WorkflowHistoryEvent::taskSnapshot does. --- .../2026-10-23-worker-history-presentation.md | 28 ++++++++ rust/src/runtime/store.rs | 33 +++++++++- rust/src/runtime/worker_incarnations.rs | 10 ++- rust/tests/execution.rs | 2 + .../execution/worker_history_presentation.rs | 64 +++++++++++++++++++ 5 files changed, 133 insertions(+), 4 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md create mode 100644 rust/tests/execution/worker_history_presentation.rs diff --git a/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md new file mode 100644 index 000000000..2109fcec2 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md @@ -0,0 +1,28 @@ +# October 23, what a worker reads in history + +- **Found by.** Comparing the message-stream fixture's worker polls against + published Server 2.11.1. +- **The gaps.** + - PHP's poll and history-page responses show a consumed signal's + `workflow_sequence` on its `SignalReceived` event, read from the signal's + record (`WorkflowTaskPoller::historyEventsWithSignalArgumentEnvelopes`). + Native showed the event as recorded. + - PHP's task snapshots in history keep every field that is set + (`WorkflowHistoryEvent::taskSnapshot`). Native's left out + `compatibility`, `last_dispatch_attempt_at`, `last_dispatched_at`, + `last_dispatch_error`, `last_claim_failed_at` and `last_claim_error`. +- **What native does now.** + - Worker polls and history pages add `workflow_sequence` to a consumed + signal's `SignalReceived`, when the payload does not have one. The + run's own history route still shows the events as recorded, as PHP's + does. + - Task snapshots keep those fields when they are set. +- **How it was checked.** A test runs on SQLite, PostgreSQL, MySQL and + MariaDB: + - a condition wait consumes an undeclared signal; + - the next poll and history page show that wait's sequence, and the + run's history does not; + - the completing task's snapshot has its dispatch stamps. +- **Not changed yet.** Some events PHP records with a task snapshot have + none in native, such as `ConditionWaitOpened` and `SignalWaitOpened`. + That audit is a #325 follow-up. diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 93c299d12..03dfa699a 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -1798,10 +1798,11 @@ where } } } - let (events, next) = match page { + let (mut events, next) = match page { Some(size) => Self::history_page_connection(&mut tx, &run_id, 0, size).await?, None => (Self::whole_history(&mut tx, &run_id).await?, None), }; + Self::with_signal_sequences(&mut tx, &run_id, &mut events).await?; let budget = Self::workflow_history_budget( &mut tx, &run_id, @@ -1865,7 +1866,8 @@ where } Self::workflow_task_guard(&mut tx, &task, &body).await?; let run_id = DB::string(&task, "workflow_run_id")?; - let (events, next) = Self::history_page_connection(&mut tx, &run_id, after, size).await?; + let (mut events, next) = Self::history_page_connection(&mut tx, &run_id, after, size).await?; + Self::with_signal_sequences(&mut tx, &run_id, &mut events).await?; let run = Self::query("SELECT last_history_sequence FROM workflow_runs WHERE id=$1") .bind(&run_id) .fetch_one(&mut *tx) @@ -3565,6 +3567,33 @@ where /// A poll's whole history, read in bounded pages: published PHP sends /// every event unless the worker asks for pages. + /// `WorkflowTaskPoller::historyEventsWithSignalArgumentEnvelopes`: the + /// history a worker reads shows a consumed signal's workflow sequence on + /// its `SignalReceived`, from the signal's record. The run's own history + /// route shows the events as recorded. + async fn with_signal_sequences(connection: &mut DB::Connection, run_id: &str, events: &mut [Value]) -> Result<()> { + if !events.iter().any(|event| event["event_type"] == "SignalReceived") { + return Ok(()); + } + let rows = Self::query("SELECT id,workflow_sequence FROM workflow_signal_records WHERE workflow_run_id=$1 AND workflow_sequence IS NOT NULL") + .bind(run_id) + .fetch_all(&mut *connection) + .await?; + let mut sequences = std::collections::HashMap::new(); + for row in rows { + sequences.insert(DB::string(&row, "id")?, DB::number(&row, "workflow_sequence")?); + } + for event in events.iter_mut().filter(|event| event["event_type"] == "SignalReceived") { + let sequence = event["payload"]["signal_id"].as_str().and_then(|id| sequences.get(id)).copied(); + if let (Some(sequence), Some(payload)) = (sequence, event["payload"].as_object_mut()) { + if payload.get("workflow_sequence").is_none_or(Value::is_null) { + payload.insert("workflow_sequence".into(), json!(sequence)); + } + } + } + Ok(()) + } + async fn whole_history(connection: &mut DB::Connection, run_id: &str) -> Result> { let mut events = Vec::new(); let mut after = 0; diff --git a/rust/src/runtime/worker_incarnations.rs b/rust/src/runtime/worker_incarnations.rs index 2a83a3117..5685a92be 100644 --- a/rust/src/runtime/worker_incarnations.rs +++ b/rust/src/runtime/worker_incarnations.rs @@ -339,18 +339,24 @@ where "status":DB::string(task,"status")?,"attempt_count":DB::number(task,"attempt_count")?, "repair_count":DB::number(task,"repair_count")?,"queue":DB::string(task,"queue")?, "connection":DB::PHP_CONNECTION}); + // `WorkflowHistoryEvent::taskSnapshot` keeps every field that is set. for field in [ "available_at", "leased_at", "lease_expires_at", + "last_dispatch_attempt_at", + "last_dispatched_at", + "last_claim_failed_at", "repair_available_at", ] { if let Some(value) = DB::optional_instant(task, field)? { snapshot[field] = json!(value); } } - if let Some(owner) = DB::optional_string(task, "lease_owner")? { - snapshot["lease_owner"] = json!(owner); + for field in ["lease_owner", "compatibility", "last_dispatch_error", "last_claim_error"] { + if let Some(value) = DB::optional_string(task, field)? { + snapshot[field] = json!(value); + } } Ok(snapshot) } diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index c3ce551e4..eabeeade2 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -109,6 +109,8 @@ mod parent_close_policies; mod history_pages; #[path = "execution/declared_signals.rs"] mod declared_signals; +#[path = "execution/worker_history_presentation.rs"] +mod worker_history_presentation; #[path = "execution/poll_waits.rs"] mod poll_waits; #[path = "execution/child_cascade.rs"] diff --git a/rust/tests/execution/worker_history_presentation.rs b/rust/tests/execution/worker_history_presentation.rs new file mode 100644 index 000000000..db76a3d75 --- /dev/null +++ b/rust/tests/execution/worker_history_presentation.rs @@ -0,0 +1,64 @@ +use super::*; + +// Server 2.11.1 `WorkflowTaskPoller::historyEventsWithSignalArgumentEnvelopes` +// and `WorkflowHistoryEvent::taskSnapshot`: the history a worker reads shows a +// consumed signal's workflow sequence, and task snapshots keep their dispatch +// stamps. + +fn wait(key: &str) -> Value { + json!([{"type":"open_condition_wait","condition_key":key, + "condition_definition_fingerprint":format!("sha256:{}", "b".repeat(64))}]) +} + +fn received(history: &Value) -> Vec { + history.as_array().unwrap().iter().filter(|event| event["event_type"] == "SignalReceived") + .map(|event| event["payload"].clone()).collect() +} + +#[tokio::test] +async fn a_worker_reads_the_sequence_that_consumed_a_signal() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + // A run started before any worker declared a contract admits any signal. + let started = start(&app, "presented-signals").await; + let worker = json!({"worker_id":"presenter","task_queue":"test","runtime":"php","supported_workflow_types":["echo"], + "supported_activity_types":[]}); + assert_eq!(request(&app, "POST", "/api/worker/register", worker).await.0, StatusCode::CREATED); + let task = poll(&app, "presenter", "workflow").await; + assert_eq!(finish_task(&app, &task, wait("first")).await.0, StatusCode::OK); + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let opened = history.as_array().unwrap().iter().find(|event| event["event_type"] == "ConditionWaitOpened").unwrap(); + let sequence = opened["payload"]["sequence"].clone(); + + let (status, body) = request(&app, "POST", "/api/workflows/presented-signals/signal/go", json!([])).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let task = poll(&app, "presenter", "workflow").await; + assert!(received(&task["history_events"])[0].get("workflow_sequence").is_none(), "not consumed yet: {task}"); + assert_eq!(finish_task(&app, &task, wait("second")).await.0, StatusCode::OK); + + let (status, body) = request(&app, "POST", "/api/workflows/presented-signals/signal/again", json!([])).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let task = poll(&app, "presenter", "workflow").await; + let signals = received(&task["history_events"]); + assert_eq!(signals[0]["workflow_sequence"], sequence, "the consuming wait's sequence: {signals:?}"); + assert!(signals[1].get("workflow_sequence").is_none(), "{signals:?}"); + // A history page reads the same. + let (status, page) = request(&app, "POST", &format!("/api/worker/workflow-tasks/{}/history", task["task_id"].as_str().unwrap()), + json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"]})).await; + assert_eq!(status, StatusCode::OK, "{page}"); + assert_eq!(received(&page["history_events"])[0]["workflow_sequence"], sequence, "{page}"); + // The run's own history shows the event as recorded. + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + assert!(received(&history)[0].get("workflow_sequence").is_none(), "{history}"); + + // The completing task was dispatched when the signal created it, and its + // snapshot keeps the dispatch stamps. + let done = json!([{"type":"complete_workflow","result":envelope(Payload::Null)}]); + assert_eq!(finish_task(&app, &task, done).await.0, StatusCode::OK); + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let completed = history.as_array().unwrap().iter().find(|event| event["event_type"] == "WorkflowCompleted").unwrap(); + for stamp in ["last_dispatch_attempt_at", "last_dispatched_at"] { + assert!(completed["payload"]["task"][stamp].is_string(), "{stamp}: {completed}"); + } + assert!(completed["payload"]["task"].get("last_claim_error").is_none(), "unset fields stay out: {completed}"); +} From 75c775edf1e5ab0e2b1af699014003e53f236e70 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 19:31:45 +0000 Subject: [PATCH 02/11] Record the dispatch-stamp difference the fuller task snapshots show --- .../2026-10-23-worker-history-presentation.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md index 2109fcec2..79271b917 100644 --- a/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md +++ b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md @@ -26,3 +26,9 @@ - **Not changed yet.** Some events PHP records with a task snapshot have none in native, such as `ConditionWaitOpened` and `SignalWaitOpened`. That audit is a #325 follow-up. +- **Known difference now visible in snapshots.** PHP stamps a dispatch only + on tasks a control-plane call dispatches, such as a signal's task. Tasks a + workflow task completion creates, such as the next message task, carry + none. Native stamps every new task, so those snapshots show stamps where + PHP's do not. This is the cause of the known `tasks_dispatched_last_minute` + difference too; fixing it is a #325 follow-up. From fe8269b2556541a41510e0647a485abb0ef7e9fa Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 21:18:31 +0000 Subject: [PATCH 03/11] Refuse policy maps that are not JSON objects, like PHP #705 The six admission, durable-limit and external-payload policy maps are read like PHP's EnvAuditor::jsonObject: unset, or raw text that decodes to a JSON object. Malformed JSON, null, lists and scalars now stop startup instead of silently becoming an empty policy. A set DW_ name wins over its legacy name even when invalid, and the refusal names the variable without its value. The external payload quota reads its overrides with the same decoder. --- .../2026-10-23-policy-map-json.md | 33 +++++ rust/src/config_audit.rs | 131 ++++++++++++++++-- rust/src/runtime/external_config.rs | 14 +- 3 files changed, 163 insertions(+), 15 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-23-policy-map-json.md diff --git a/docs/rust-server-port-checkpoints/2026-10-23-policy-map-json.md b/docs/rust-server-port-checkpoints/2026-10-23-policy-map-json.md new file mode 100644 index 000000000..80db765d9 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-policy-map-json.md @@ -0,0 +1,33 @@ +# October 23, policy maps must be JSON objects + +- **Found by.** PHP Server #705 (merged to main at `f67f2816e`, not yet in a + published image). Published PHP turned malformed admission-policy JSON + into an empty policy, so a configured cap could disappear without notice. + Native had the same defect. + - `DW_TASK_QUEUE_ADMISSION_OVERRIDES` and + `DW_NAMESPACE_ADMISSION_OVERRIDES` accepted malformed JSON. + - All five maps native refuses accepted `null`, `[]`, booleans and + numbers. + - The external-payload namespace overrides, which native honors, became + an empty policy when malformed. +- **Change.** The six maps PHP now reads with `EnvAuditor::jsonObject` are + read by the same rule in `config_audit::json_object`: + - the value is either unset or raw text that decodes to a JSON object; + - Laravel's `env()` conversions do not apply, so `null`, `(null)`, + `empty`, quoted text and an empty value are refused; + - a set `DW_*` name wins over its legacy name even when the `DW_*` value + is invalid; + - the startup refusal names the variable it read, never the value. + + Native still refuses a non-empty object in the five maps it has no policy + for. It still honors `DW_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES`, and the + external payload quota now reads that value with the same decoder. +- **Reference.** This follows unreleased PHP source, not the published + 2.11.1 image. For any valid configuration the two behave the same. The + only values refused are ones published PHP silently discarded and the + next PHP image refuses at boot. +- **Test.** `policy_maps_must_be_json_objects_like_php_705` covers all six + maps under both names, 17 invalid values, padded whitespace around `{}`, + DW-over-legacy precedence in both directions, and that the refusal never + repeats the value. The shipped Compose and Helm values (`{}` for the + durable maps) still start. diff --git a/rust/src/config_audit.rs b/rust/src/config_audit.rs index 3eb4572f0..82cf036c0 100644 --- a/rust/src/config_audit.rs +++ b/rust/src/config_audit.rs @@ -238,6 +238,44 @@ where ) } +/// The raw value PHP's `EnvAuditor::jsonObject` reads: the `DW_*` name +/// whenever it is set, even to an invalid value, else the legacy name. +/// Laravel's `env()` conversions do not apply. +fn selected<'a, F>(lookup: &F, name: &'a str, legacy: Option<&'a str>) -> Option<(String, &'a str)> +where + F: Fn(&str) -> Option, +{ + lookup(name) + .map(|raw| (raw, name)) + .or_else(|| legacy.and_then(|legacy| lookup(legacy).map(|raw| (raw, legacy)))) +} + +/// How a refusal names the variable it read. +fn read_from(name: &str, source: &str) -> String { + if source == name { name.to_owned() } else { format!("{name} (read from {source})") } +} + +/// `EnvAuditor::jsonObject` (Server #705): a policy map is unset, or raw text +/// that decodes to a JSON object. Malformed JSON, `null`, booleans, numbers, +/// strings and lists are refused instead of becoming an empty policy, so a +/// mistyped cap cannot silently disappear. The refusal never repeats the value. +pub fn json_object(lookup: F, name: &str, legacy: Option<&str>) -> Result, String> +where + F: Fn(&str) -> Option, +{ + let Some((raw, source)) = selected(&lookup, name, legacy) else { + return Ok(serde_json::Map::new()); + }; + match serde_json::from_str(&raw) { + Ok(serde_json::Value::Object(policy)) => Ok(policy), + _ => Err(format!("{} must contain a valid JSON object", read_from(name, source))), + } +} + +/// Policy maps native honors. Each must still be unset or a JSON object. +const HONORED_JSON_OBJECTS: &[(&str, &str)] = + &[("DW_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES", "WORKFLOW_SERVER_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES")]; + /// What a refused setting must be for native to start. #[derive(Clone, Copy, Debug)] pub enum Rule { @@ -256,6 +294,9 @@ pub enum Rule { /// `json_decode` gives nothing. Strict settings also refuse invalid JSON, /// which PHP either rejects or cannot have meant. EmptyJson { strict: bool }, + /// A policy map read by `EnvAuditor::jsonObject`: unset or a JSON object + /// (see [`json_object`]), and empty, since native has no such policy. + EmptyObject, /// The `(int)` cast equals PHP's default. Int(i64), /// `filter_var(..., FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE) ?? default` @@ -287,6 +328,9 @@ impl Rule { Ok(decoded) => !php_truthy(&decoded), Err(_) => !strict, }, + Self::EmptyObject => { + matches!(serde_json::from_str(text), Ok(serde_json::Value::Object(policy)) if policy.is_empty()) + } Self::Int(expected) => value.int() == expected, Self::FilterBool(default) => value.filter_bool().unwrap_or(default) == default, Self::CastBool(default) => value.cast_bool() == default, @@ -308,7 +352,7 @@ impl Rule { Self::AnyCase(value) => format!("unset or `{value}`"), Self::Empty => "unset or empty".into(), Self::NoPositiveInt => "unset, empty or not a positive number".into(), - Self::EmptyJson { .. } => "unset or `{}`".into(), + Self::EmptyJson { .. } | Self::EmptyObject => "unset or `{}`".into(), Self::Int(value) => format!("unset or `{value}`"), Self::FilterBool(value) | Self::CastBool(value) => format!("unset or {value}"), Self::DatabaseConnection => "unset, empty or the DB_CONNECTION name".into(), @@ -380,6 +424,17 @@ pub fn audit(environment: &BTreeMap) -> Report { let Disposition::Refused(rule, reason) = setting.disposition else { continue; }; + if let Rule::EmptyObject = rule { + match json_object(lookup, setting.name, setting.legacy) { + Err(message) => report.refusals.push(message), + Ok(policy) if !policy.is_empty() => { + let source = selected(&lookup, setting.name, setting.legacy).map_or(setting.name, |(_, source)| source); + report.refusals.push(format!("{} must be {}: {reason}", read_from(setting.name, source), rule.expectation())); + } + Ok(_) => {} + } + continue; + } let read = |name: &'static str| lookup(name).map(|raw| (PhpValue::from_env(&raw), name)); let primary = read(setting.name).filter(|(value, _)| *value != PhpValue::Null); let Some((value, source)) = primary.or_else(|| setting.legacy.and_then(read)) else { @@ -395,6 +450,11 @@ pub fn audit(environment: &BTreeMap) -> Report { if let Err(message) = activity_task_lease_seconds(lookup) { report.refusals.push(message.into()); } + for (name, legacy) in HONORED_JSON_OBJECTS { + if let Err(message) = json_object(lookup, name, Some(legacy)) { + report.refusals.push(message); + } + } let unknown: Vec<&str> = environment .keys() .map(String::as_str) @@ -458,7 +518,7 @@ const STORAGE_CONNECTION: &str = "native keeps workflow tables on its one databa const LEGACY_SCHEDULE_TIMEZONE: &str = "native does not convert pre-UTC schedule timestamps"; use Disposition::{Honored, Ignored, Refused}; -use Rule::{AnyCase, CastBool, DatabaseConnection, Empty, EmptyJson, Exact, FilterBool, Int, NoPositiveInt, NotAnyCase, NotIn}; +use Rule::{AnyCase, CastBool, DatabaseConnection, Empty, EmptyJson, EmptyObject, Exact, FilterBool, Int, NoPositiveInt, NotAnyCase, NotIn}; /// Settings PHP reads that its contract does not list. PHP's audit warns about /// them as unknown names; native also refuses values it cannot honor. @@ -530,10 +590,10 @@ pub const SETTINGS: &[Setting] = &[ Setting { name: "DW_METRICS_PROMETHEUS_WORKFLOW_SERIES_LIMIT", legacy: Some("WORKFLOW_SERVER_METRICS_PROMETHEUS_WORKFLOW_SERIES_LIMIT"), disposition: Refused(Int(100), METRICS) }, Setting { name: "DW_METRICS_WORKFLOW_TASK_FAILURE_TYPE_LIMIT", legacy: Some("WORKFLOW_SERVER_METRICS_WORKFLOW_TASK_FAILURE_TYPE_LIMIT"), disposition: Refused(Int(20), METRICS) }, Setting { name: "DW_MODE", legacy: Some("WORKFLOW_SERVER_MODE"), disposition: Refused(Exact(&["service"]), TOPOLOGY) }, - Setting { name: "DW_NAMESPACE_ADMISSION_OVERRIDES", legacy: Some("WORKFLOW_SERVER_NAMESPACE_ADMISSION_OVERRIDES"), disposition: Refused(EmptyJson { strict: false }, NAMESPACE_ADMISSION) }, - Setting { name: "DW_NAMESPACE_DURABLE_HARD_LIMITS", legacy: Some("WORKFLOW_SERVER_NAMESPACE_DURABLE_HARD_LIMITS"), disposition: Refused(EmptyJson { strict: true }, DURABLE_LIMITS) }, - Setting { name: "DW_NAMESPACE_DURABLE_LIMITS", legacy: Some("WORKFLOW_SERVER_NAMESPACE_DURABLE_LIMITS"), disposition: Refused(EmptyJson { strict: true }, DURABLE_LIMITS) }, - Setting { name: "DW_NAMESPACE_DURABLE_OVERRIDES", legacy: Some("WORKFLOW_SERVER_NAMESPACE_DURABLE_OVERRIDES"), disposition: Refused(EmptyJson { strict: true }, DURABLE_LIMITS) }, + Setting { name: "DW_NAMESPACE_ADMISSION_OVERRIDES", legacy: Some("WORKFLOW_SERVER_NAMESPACE_ADMISSION_OVERRIDES"), disposition: Refused(EmptyObject, NAMESPACE_ADMISSION) }, + Setting { name: "DW_NAMESPACE_DURABLE_HARD_LIMITS", legacy: Some("WORKFLOW_SERVER_NAMESPACE_DURABLE_HARD_LIMITS"), disposition: Refused(EmptyObject, DURABLE_LIMITS) }, + Setting { name: "DW_NAMESPACE_DURABLE_LIMITS", legacy: Some("WORKFLOW_SERVER_NAMESPACE_DURABLE_LIMITS"), disposition: Refused(EmptyObject, DURABLE_LIMITS) }, + Setting { name: "DW_NAMESPACE_DURABLE_OVERRIDES", legacy: Some("WORKFLOW_SERVER_NAMESPACE_DURABLE_OVERRIDES"), disposition: Refused(EmptyObject, DURABLE_LIMITS) }, Setting { name: "DW_NAMESPACE_HARD_MAX_CONCURRENT_REQUESTS", legacy: Some("WORKFLOW_SERVER_NAMESPACE_HARD_MAX_CONCURRENT_REQUESTS"), disposition: Refused(Empty, NAMESPACE_ADMISSION) }, Setting { name: "DW_NAMESPACE_HARD_MAX_REQUESTS_PER_MINUTE", legacy: Some("WORKFLOW_SERVER_NAMESPACE_HARD_MAX_REQUESTS_PER_MINUTE"), disposition: Refused(Empty, NAMESPACE_ADMISSION) }, Setting { name: "DW_NAMESPACE_MAX_CONCURRENT_REQUESTS", legacy: Some("WORKFLOW_SERVER_NAMESPACE_MAX_CONCURRENT_REQUESTS"), disposition: Refused(Empty, NAMESPACE_ADMISSION) }, @@ -573,7 +633,7 @@ pub const SETTINGS: &[Setting] = &[ Setting { name: "DW_STORAGE_ADMISSION_MAX_AGE_SECONDS", legacy: None, disposition: Ignored }, Setting { name: "DW_STORAGE_ADMISSION_SOURCE", legacy: None, disposition: Refused(Empty, STORAGE_ADMISSION) }, Setting { name: "DW_TASK_DISPATCH_MODE", legacy: Some("WORKFLOW_V2_TASK_DISPATCH_MODE"), disposition: Refused(Exact(&["poll"]), POLL_DISPATCH) }, - Setting { name: "DW_TASK_QUEUE_ADMISSION_OVERRIDES", legacy: Some("WORKFLOW_SERVER_TASK_QUEUE_ADMISSION_OVERRIDES"), disposition: Refused(EmptyJson { strict: false }, TASK_ADMISSION) }, + Setting { name: "DW_TASK_QUEUE_ADMISSION_OVERRIDES", legacy: Some("WORKFLOW_SERVER_TASK_QUEUE_ADMISSION_OVERRIDES"), disposition: Refused(EmptyObject, TASK_ADMISSION) }, Setting { name: "DW_TRUST_FORWARDED_ATTRIBUTION_HEADERS", legacy: Some("WORKFLOW_SERVER_TRUST_FORWARDED_ATTRIBUTION_HEADERS"), disposition: Refused(FilterBool(false), FORWARDED) }, Setting { name: "DW_UPDATE_VALIDATION_LEASE_TIMEOUT", legacy: Some("WORKFLOW_SERVER_UPDATE_VALIDATION_LEASE_TIMEOUT"), disposition: Ignored }, Setting { name: "DW_UPDATE_VALIDATION_TIMEOUT", legacy: Some("WORKFLOW_SERVER_UPDATE_VALIDATION_TIMEOUT"), disposition: Ignored }, @@ -875,8 +935,10 @@ mod tests { &[("DW_WORKFLOW_TASK_MAX_ACTIVE_LEASES_PER_QUEUE", "0")], &[("DW_WORKFLOW_TASK_MAX_ACTIVE_LEASES_PER_QUEUE", "true")], &[("DW_WORKFLOW_TASK_MAX_ACTIVE_LEASES_PER_QUEUE", "none")], - &[("DW_TASK_QUEUE_ADMISSION_OVERRIDES", "not json")], - &[("DW_TASK_QUEUE_ADMISSION_OVERRIDES", "[]")], + &[("DW_TASK_QUEUE_ADMISSION_OVERRIDES", "{}")], + &[("DW_TASK_QUEUE_ADMISSION_OVERRIDES", " \n{ } ")], + &[("WORKFLOW_SERVER_NAMESPACE_ADMISSION_OVERRIDES", "{}")], + &[("DW_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES", "{\"billing\":{\"max_bytes\":5}}")], &[("DW_V2_FLEET_VALIDATION_MODE", "warn")], &[("DW_V2_PIN_TO_RECORDED_FINGERPRINT", "no")], &[("DW_SERIALIZER", "")], @@ -926,6 +988,57 @@ mod tests { } } + #[test] + fn policy_maps_must_be_json_objects_like_php_705() { + // The six maps PHP reads with `EnvAuditor::jsonObject`. + let maps = [ + ("DW_TASK_QUEUE_ADMISSION_OVERRIDES", "WORKFLOW_SERVER_TASK_QUEUE_ADMISSION_OVERRIDES"), + ("DW_NAMESPACE_ADMISSION_OVERRIDES", "WORKFLOW_SERVER_NAMESPACE_ADMISSION_OVERRIDES"), + ("DW_NAMESPACE_DURABLE_LIMITS", "WORKFLOW_SERVER_NAMESPACE_DURABLE_LIMITS"), + ("DW_NAMESPACE_DURABLE_HARD_LIMITS", "WORKFLOW_SERVER_NAMESPACE_DURABLE_HARD_LIMITS"), + ("DW_NAMESPACE_DURABLE_OVERRIDES", "WORKFLOW_SERVER_NAMESPACE_DURABLE_OVERRIDES"), + ("DW_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES", "WORKFLOW_SERVER_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES"), + ]; + let invalid = [ + "", " ", "not json", "{", "{\"cap\":1", "null", "(null)", "empty", "true", "false", "0", "1", "[]", + "[{\"cap\":1}]", "\"{}\"", "'{}'", "\"text\"", + ]; + for (name, legacy) in maps { + for value in ["{}", " \n\t{}\r\n"] { + assert_eq!(refusals(&[(name, value)]), Vec::::new(), "{name}={value:?}"); + assert_eq!(refusals(&[(legacy, value)]), Vec::::new(), "{legacy}={value:?}"); + } + for value in invalid { + assert_eq!(refusals(&[(name, value)]), [format!("{name} must contain a valid JSON object")], "{name}={value:?}"); + assert_eq!( + refusals(&[(legacy, value)]), + [format!("{name} (read from {legacy}) must contain a valid JSON object")], + "{legacy}={value:?}" + ); + // A set DW name wins even when it is invalid and the legacy + // value is valid. + assert_eq!(refusals(&[(name, value), (legacy, "{}")]).len(), 1, "{name}={value:?}"); + } + // A valid DW name hides an invalid legacy value. + assert_eq!(refusals(&[(name, "{}"), (legacy, "not json")]), Vec::::new(), "{name}"); + } + // Native honors the external payload overrides; it refuses the other + // five unless they are empty, as before. + let policy = "{\"billing\":{\"max_bytes\":5}}"; + for (name, legacy) in &maps[..5] { + assert_eq!(refusals(&[(name, policy)]).len(), 1, "{name}"); + assert!(refusals(&[(legacy, policy)])[0].starts_with(&format!("{name} (read from {legacy}) must be unset or `{{}}`"))); + } + assert_eq!(refusals(&[(maps[5].0, policy)]), Vec::::new()); + assert_eq!( + json_object(|_| Some(policy.to_owned()), maps[5].0, Some(maps[5].1)).unwrap()["billing"]["max_bytes"], + 5 + ); + // The refusal never repeats the value. + let secret = "{\"secret-tenant-value\":"; + assert!(refusals(&[(maps[5].0, secret)]).iter().all(|refusal| !refusal.contains("secret-tenant-value"))); + } + #[test] fn refusals_never_repeat_the_value() { let secret = "principal-secret-value"; diff --git a/rust/src/runtime/external_config.rs b/rust/src/runtime/external_config.rs index e4992b199..070638fc7 100644 --- a/rust/src/runtime/external_config.rs +++ b/rust/src/runtime/external_config.rs @@ -122,11 +122,13 @@ fn member<'a>(value: &'a Value, key: &str) -> Option<&'a Value> { impl ExternalPayloadConfig { pub(super) fn from_env() -> Result { let completion = process_value("COMPLETION_MAX_BYTES", Value::Null)?; - let overrides = process_value("NAMESPACE_OVERRIDES", Value::String("{}".into()))?; - let overrides = overrides - .as_str() - .and_then(|value| serde_json::from_str(value).ok()) - .unwrap_or(Value::Null); + // Startup refuses a value that is not a JSON object (Server #705). + let overrides = crate::config_audit::json_object( + |name| env::var_os(name).map(|value| value.to_string_lossy().into_owned()), + "DW_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES", + Some("WORKFLOW_SERVER_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES"), + ) + .map_err(|_| configuration_error())?; Ok(Self { completion_max_bytes: (!completion.is_null()) .then(|| integer(&completion).max(1) as u64), @@ -150,7 +152,7 @@ impl ExternalPayloadConfig { "HARD_MAX_OBJECTS_PER_NAMESPACE", Value::Null, )?, - namespace_overrides: overrides, + namespace_overrides: Value::Object(overrides), ..Self::default() }) } From f31c4c36130c77c46ca6cc4f7cab5307f1480714 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 22:40:17 +0000 Subject: [PATCH 04/11] Send MySQL statement closes with the next command COM_STMT_CLOSE has no reply. The published sqlx-mysql flushed it on its own write after a cache eviction or an uncached execute, and flushed it again before draining results. The next command then waited behind Nagle's algorithm for the server's delayed acknowledgement, about 40 ms. Across the 30-48 statements of a transition that held the global lock for 0.3-1 s under load, and a due timer behind it fired more than the 1 s bound late (#714). The vendored driver now queues each close with sequence 0 and leaves a buffer of replyless closes unflushed until the next command, so both go out in one write. A proxy-recorded qualification test fails with the published driver and passes with the change on MySQL 8.0 and MariaDB. --- .../2026-10-23-mysql-statement-close.md | 57 +++++++++ rust/src/runtime/mysql_pool_tests.rs | 119 ++++++++++++++++++ rust/vendor/sqlx-mysql/DURABLE-PATCH.md | 38 +++++- .../sqlx-mysql/src/connection/executor.rs | 11 +- rust/vendor/sqlx-mysql/src/connection/mod.rs | 5 +- .../sqlx-mysql/src/connection/stream.rs | 25 +++- rust/vendor/sqlx-mysql/src/connection/tls.rs | 1 + 7 files changed, 239 insertions(+), 17 deletions(-) create mode 100644 docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md diff --git a/docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md b/docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md new file mode 100644 index 000000000..d5377cc5d --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md @@ -0,0 +1,57 @@ +# October 23, MySQL statement closes and late timers + +- **Found by.** Server #714. Twice on October 23, the hosted MariaDB 10.11 + `codec-deadline-health` cell recorded a native timer more than the 1 s + maximum late. In one, `TimerFired` came 1.098 s after `fire_at`. MySQL 8.0 + and 8.4 cells fail the same way, less often. +- **Reproduction.** A disposable 4-vCPU VM ran the unchanged cell on main + `f67f2816e`, with temporary timing in a local-only build: + - MariaDB failed 2 of the first 8 cells, with timers 1.228 s and 1.213 s + late; + - PostgreSQL 17 and SQLite passed all 8 cells, at most 171 ms late. +- **Attribution.** Every late timer waited behind the global transition lock: + - In the worst MariaDB cell, the tick's activity-deadline sweep waited + 543 ms behind a workflow completion that held the lock 690 ms. The sweep + then held it 407 ms itself, and the timer fired 1.228 s late. + - Ordinary transitions held the lock 300 ms to 1 s. With every statement + logged by MariaDB, those transactions spent a median of 5 ms in the + database (p90 9 ms) across their 30 to 48 statements. + - Native was not CPU-starved. Wake-up probes on an async task and an OS + thread rarely saw 50 ms delays, and giving native 2 CPUs left the holds + unchanged. + - Timestamps on each statement inside the transactions showed gaps of + 40 to 100 ms spread across ordinary statements, many near 58 ms. That is + Nagle's algorithm waiting for a delayed acknowledgement. +- **Cause.** `COM_STMT_CLOSE` has no reply. The published `sqlx-mysql` 0.9.0 + flushes it in its own write: when its 100-entry statement cache evicts a + statement, and after executing a statement it does not cache. Before the + next command it also flushes anything still buffered. The next command is + then a second small write while the close is still unacknowledged, so the + kernel holds it about 40 ms. SQLx sets no `TCP_NODELAY`; PHP's mysqlnd does. + PostgreSQL's driver sends its closes together with the next request. +- **Change.** In the vendored driver, `MySqlStream::queue_packet` buffers each + statement close with sequence number 0. `wait_until_ready` no longer flushes + a buffer that holds only replyless closes, so they leave in the same write as + the next command. Packets, order and server behavior are unchanged. The + patch record is `rust/vendor/sqlx-mysql/DURABLE-PATCH.md`. +- **Test.** `qualification_statement_close_leaves_with_the_next_command` + records the client's writes through a proxy. It uses a one-entry statement + cache and one uncached statement, so both close paths are exercised. + - With the fix it passes on MySQL 8.0.46 and MariaDB 10.11 in 0.01 s. + - With the published driver it fails on both: every close ends its own + write, and the same 14 statements take 0.6 s. + - The existing cancelled-boundary qualification still passes. +- **Hosted-cell A/B on the VM.** Interleaved cells with the spec and limits + unchanged. Both builds were instrumented: one without the driver change, one + with its first revision. That revision already buffered the closes, but + `wait_until_ready` still flushed them before draining. + + | | Cells passed | Timer lateness | Holds of 300 ms or more | + | --- | --- | --- | --- | + | MariaDB without the fix | 3 of 6 | 190 to 1045 ms | 54 | + | MariaDB with the fix | 6 of 6 | 49 to 289 ms | 0 | + | MySQL 8.0 and 8.4 without the fix | 5 of 8 | 138 to 1099 ms | 77 | + | MySQL 8.0 and 8.4 with the fix | 8 of 8 | 116 to 299 ms | 3 | + + Clean release builds of main and of this branch are compared across all five + families in the PR's evidence comment. diff --git a/rust/src/runtime/mysql_pool_tests.rs b/rust/src/runtime/mysql_pool_tests.rs index a4780a930..e6f40c497 100644 --- a/rust/src/runtime/mysql_pool_tests.rs +++ b/rust/src/runtime/mysql_pool_tests.rs @@ -221,3 +221,122 @@ async fn qualification_cancelled_pool_boundaries_preserve_fresh_reads_and_commit .await .expect("bounded real transaction-boundary qualification"); } + +/// Forward one client to `upstream`, recording each client write as it +/// arrives. +async fn record_client_writes( + listener: TcpListener, + upstream: String, + writes: Arc>>>, +) { + loop { + let (client, _) = listener.accept().await.unwrap(); + let server = TcpStream::connect(&upstream).await.unwrap(); + let writes = writes.clone(); + tokio::spawn(async move { + let (mut client_read, mut client_write) = client.into_split(); + let (mut server_read, mut server_write) = server.into_split(); + let upstream = async { + let mut bytes = vec![0; 65536]; + loop { + let count = client_read.read(&mut bytes).await?; + if count == 0 { + break; + } + writes.lock().unwrap().push(bytes[..count].to_vec()); + server_write.write_all(&bytes[..count]).await?; + } + server_write.shutdown().await + }; + let downstream = async { + tokio::io::copy(&mut server_read, &mut client_write).await?; + client_write.shutdown().await + }; + let _: (std::io::Result<()>, std::io::Result<()>) = tokio::join!(upstream, downstream); + }); + } +} + +/// Each client command packet (sequence 0): its command byte, and whether a +/// recorded write ended with it. +fn recorded_commands(writes: &[Vec]) -> Vec<(u8, bool)> { + let mut stream = Vec::new(); + let mut write_ends = std::collections::BTreeSet::new(); + for write in writes { + stream.extend_from_slice(write); + write_ends.insert(stream.len()); + } + let mut commands = Vec::new(); + let mut at = 0; + while at + 4 <= stream.len() { + let length = usize::from(stream[at]) + | usize::from(stream[at + 1]) << 8 + | usize::from(stream[at + 2]) << 16; + let end = at + 4 + length; + if end > stream.len() { + break; + } + if stream[at + 3] == 0 && length > 0 { + commands.push((stream[at + 4], write_ends.contains(&end))); + } + at = end; + } + commands +} + +/// A statement close has no reply. Flushed on its own, it leaves +/// unacknowledged data in flight, and Nagle's algorithm then holds the next +/// command until the server's delayed acknowledgement, about 40 ms per close. +/// Under the transition lock those stalls added up to a late timer (#714). +#[tokio::test] +#[ignore = "requires disposable MySQL/MariaDB; run qualification_ explicitly"] +async fn qualification_statement_close_leaves_with_the_next_command() { + const COM_STMT_CLOSE: u8 = 0x19; + tokio::time::timeout(Duration::from_secs(15), async { + let url = std::env::var("DW_TEST_MYSQL_URL").unwrap(); + let direct = MySqlConnectOptions::from_str(&url).unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let writes = Arc::new(std::sync::Mutex::new(Vec::new())); + let proxy = tokio::spawn(record_client_writes( + listener, + format!("{}:{}", direct.get_host(), direct.get_port()), + writes.clone(), + )); + // Plain text so the proxy can read packets; one cached statement, so + // each new statement evicts and closes the previous one. + let options = direct + .host("127.0.0.1") + .port(port) + .ssl_mode(sqlx::mysql::MySqlSslMode::Disabled) + .statement_cache_capacity(1); + let mut connection = MySqlConnection::connect_with(&options).await.unwrap(); + for round in 0..4_i64 { + for (sql, step) in [("SELECT ? + 1", 1), ("SELECT ? + 2", 2), ("SELECT ? + 3", 3)] { + let value: i64 = sqlx::query_scalar(sql).bind(round).fetch_one(&mut connection).await.unwrap(); + assert_eq!(value, round + step); + } + } + // An uncached statement is closed right after it executes. + let value: i64 = sqlx::query_scalar("SELECT ? + 4") + .bind(1_i64) + .persistent(false) + .fetch_one(&mut connection) + .await + .unwrap(); + assert_eq!(value, 5); + let value: i64 = sqlx::query_scalar("SELECT ? + 5").bind(1_i64).fetch_one(&mut connection).await.unwrap(); + assert_eq!(value, 6); + connection.close().await.unwrap(); + proxy.abort(); + let commands = recorded_commands(&writes.lock().unwrap()); + let closes = commands.iter().filter(|(command, _)| *command == COM_STMT_CLOSE).count(); + assert!(closes >= 12, "eviction and uncached closes were exercised: {closes}"); + assert!( + commands.iter().all(|(command, ended)| *command != COM_STMT_CLOSE || !ended), + "a statement close ended its own write: {commands:?}" + ); + }) + .await + .expect("bounded statement-close write qualification"); +} diff --git a/rust/vendor/sqlx-mysql/DURABLE-PATCH.md b/rust/vendor/sqlx-mysql/DURABLE-PATCH.md index f08664b46..117531a0c 100644 --- a/rust/vendor/sqlx-mysql/DURABLE-PATCH.md +++ b/rust/vendor/sqlx-mysql/DURABLE-PATCH.md @@ -1,7 +1,7 @@ -# SQLx MySQL hard-close patch +# SQLx MySQL driver patches -This directory contains the official `sqlx-mysql` 0.9.0 crate with one scoped -driver correction. Its published archive has SHA-256 +This directory contains the official `sqlx-mysql` 0.9.0 crate with two scoped +driver corrections. Its published archive has SHA-256 `90b8020fe17c5f2c245bfa2505d7ef59c5604839527c740266ad2214acebea27` and VCS commit `003b698e99e024f3621b8043a2426fde5b741171`. @@ -19,10 +19,36 @@ The original and flush-before-gate TLS controls fail with the published driver; both pass with this correction. Qualification and original evidence are in [Server #429](https://github.com/durable-workflow/server/pull/429#issuecomment-6135933370). -All upstream source is retained. The only source-code change is in -`src/connection/mod.rs`. The packaged license-reference files are replaced with +## Statement closes leave with the next command + +`COM_STMT_CLOSE` has no server reply. The published driver flushes it on its +own write: when its statement cache evicts a statement, and right after +executing a statement it does not cache. The next command is then a second +small write while the close is still unacknowledged, so Nagle's algorithm +holds it until the server's delayed acknowledgement, about 40 ms. SQLx does not +set `TCP_NODELAY`; PHP's mysqlnd and libmysqlclient do. + +Inside the native transition lock, a transaction issues 30 to 48 statements, +so these stalls added up to holds of 0.3 to 1 s under load. A due timer +waiting behind them fired more than the 1 s maximum late (Server #714). + +`MySqlStream::queue_packet` now buffers each close with the reset sequence +number that `send_packet` would use, so it leaves in the same write as the next +command. The server receives the same packets in the same order. A connection +dropped before its next command leaves the close unsent; the server releases +that connection's prepared statements when it closes. + +`qualification_statement_close_leaves_with_the_next_command` in +`rust/src/runtime/mysql_pool_tests.rs` records every client write through a +proxy. It fails with the published driver, where each close ends its write, and +passes with this correction. + +All upstream source is retained. The source-code changes are in +`src/connection/mod.rs`, `src/connection/stream.rs` and +`src/connection/executor.rs`. The packaged license-reference files are replaced with the full MIT and Apache license texts from the same upstream commit. This file records the patch; no upstream release is represented as containing it. Remove this override when a reviewed published driver provides equivalent -hard-close behavior and passes the same database and TLS qualification. +hard-close behavior and statement-close batching (or `TCP_NODELAY`), and passes +the same database and TLS qualification. diff --git a/rust/vendor/sqlx-mysql/src/connection/executor.rs b/rust/vendor/sqlx-mysql/src/connection/executor.rs index ee59d03d0..6c12d29fc 100755 --- a/rust/vendor/sqlx-mysql/src/connection/executor.rs +++ b/rust/vendor/sqlx-mysql/src/connection/executor.rs @@ -92,8 +92,7 @@ impl MySqlConnection { { self.inner .stream - .send_packet(StmtClose { statement: id }) - .await?; + .queue_packet(StmtClose { statement: id })?; } Ok((id, metadata)) @@ -168,7 +167,7 @@ impl MySqlConnection { }) .await?; - self.inner.stream.send_packet(StmtClose { statement: id }).await?; + self.inner.stream.queue_packet(StmtClose { statement: id })?; MySqlValueFormat::Binary } @@ -337,8 +336,7 @@ impl<'c> Executor<'c> for &'c mut MySqlConnection { self.inner .stream - .send_packet(StmtClose { statement: id }) - .await?; + .queue_packet(StmtClose { statement: id })?; metadata }; @@ -367,8 +365,7 @@ impl<'c> Executor<'c> for &'c mut MySqlConnection { self.inner .stream - .send_packet(StmtClose { statement: id }) - .await?; + .queue_packet(StmtClose { statement: id })?; let columns = (*metadata.columns).clone(); diff --git a/rust/vendor/sqlx-mysql/src/connection/mod.rs b/rust/vendor/sqlx-mysql/src/connection/mod.rs index 110ae4f1e..c20bfc026 100755 --- a/rust/vendor/sqlx-mysql/src/connection/mod.rs +++ b/rust/vendor/sqlx-mysql/src/connection/mod.rs @@ -105,10 +105,9 @@ impl Connection for MySqlConnection { while let Some((statement_id, _)) = self.inner.cache_statement.remove_lru() { self.inner .stream - .send_packet(StmtClose { + .queue_packet(StmtClose { statement: statement_id, - }) - .await?; + })?; } Ok(()) diff --git a/rust/vendor/sqlx-mysql/src/connection/stream.rs b/rust/vendor/sqlx-mysql/src/connection/stream.rs index e6aa8b48c..590ce1ecd 100755 --- a/rust/vendor/sqlx-mysql/src/connection/stream.rs +++ b/rust/vendor/sqlx-mysql/src/connection/stream.rs @@ -19,6 +19,9 @@ pub struct MySqlStream> { pub(crate) sequence_id: u8, pub(crate) waiting: VecDeque, pub(crate) is_tls: bool, + // The write buffer holds only commands without a reply, queued to leave + // with the next command. + pub(crate) queued_unanswered: bool, } #[derive(Debug, PartialEq, Eq)] @@ -56,11 +59,14 @@ impl MySqlStream { sequence_id: 0, socket: BufferedSocket::new(socket), is_tls: false, + queued_unanswered: false, } } pub(crate) async fn wait_until_ready(&mut self) -> Result<(), Error> { - if !self.socket.write_buffer().is_empty() { + // Queued commands without a reply wait for the next command; nothing + // pending needs them to be sent first. + if !self.socket.write_buffer().is_empty() && !self.queued_unanswered { self.socket.flush().await?; } @@ -108,10 +114,26 @@ impl MySqlStream { Ok(()) } + /// Buffer a command that has no server reply (`COM_STMT_CLOSE`) so it + /// leaves in the same write as the next command. Flushing it alone leaves + /// unacknowledged data in flight, and Nagle's algorithm then holds the next + /// command until the server's delayed acknowledgement (about 40 ms). + pub(crate) fn queue_packet<'en, T>(&mut self, payload: T) -> Result<(), Error> + where + T: ProtocolEncode<'en, Capabilities>, + { + let unanswered = self.socket.write_buffer().is_empty() || self.queued_unanswered; + self.sequence_id = 0; + self.write_packet(payload)?; + self.queued_unanswered = unanswered; + Ok(()) + } + pub(crate) fn write_packet<'en, T>(&mut self, payload: T) -> Result<(), Error> where T: ProtocolEncode<'en, Capabilities>, { + self.queued_unanswered = false; self.socket .write_with(Packet(payload), (self.capabilities, &mut self.sequence_id)) } @@ -213,6 +235,7 @@ impl MySqlStream { sequence_id: self.sequence_id, waiting: self.waiting, is_tls: self.is_tls, + queued_unanswered: self.queued_unanswered, } } } diff --git a/rust/vendor/sqlx-mysql/src/connection/tls.rs b/rust/vendor/sqlx-mysql/src/connection/tls.rs index 9034fbd63..9ac71f676 100755 --- a/rust/vendor/sqlx-mysql/src/connection/tls.rs +++ b/rust/vendor/sqlx-mysql/src/connection/tls.rs @@ -97,6 +97,7 @@ impl WithSocket for MapStream { sequence_id: self.sequence_id, waiting: self.waiting, is_tls: true, + queued_unanswered: false, } } } From dc58f9345c05b3185426baee2d72249887605470 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Fri, 23 Oct 2026 23:48:10 +0000 Subject: [PATCH 05/11] Keep explicit MySQL flushes and cache clears sending statement closes Security's review of f31c4c361: Connection::flush delegated to wait_until_ready, which now leaves a buffer of queued closes unsent, and clear_cached_statements only queued its closes. Both report success with the bytes still buffered. Explicit flushes and cache clears now send every buffered byte before returning (MySqlStream::flush_queued); only the path before the next command keeps closes to share that command's write. New MySQL and MariaDB controls prove both calls without a later command; they fail on f31c4c361. The close test now describes what it observes: TCP read chunks at a proxy, not write calls. --- .../2026-10-23-mysql-statement-close.md | 36 +++-- rust/src/runtime/mysql_pool_tests.rs | 144 ++++++++++++++++-- rust/vendor/sqlx-mysql/DURABLE-PATCH.md | 31 ++-- rust/vendor/sqlx-mysql/src/connection/mod.rs | 9 +- .../sqlx-mysql/src/connection/stream.rs | 11 ++ 5 files changed, 199 insertions(+), 32 deletions(-) diff --git a/docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md b/docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md index d5377cc5d..7d4c22a89 100644 --- a/docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md +++ b/docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md @@ -30,17 +30,31 @@ kernel holds it about 40 ms. SQLx sets no `TCP_NODELAY`; PHP's mysqlnd does. PostgreSQL's driver sends its closes together with the next request. - **Change.** In the vendored driver, `MySqlStream::queue_packet` buffers each - statement close with sequence number 0. `wait_until_ready` no longer flushes - a buffer that holds only replyless closes, so they leave in the same write as - the next command. Packets, order and server behavior are unchanged. The - patch record is `rust/vendor/sqlx-mysql/DURABLE-PATCH.md`. -- **Test.** `qualification_statement_close_leaves_with_the_next_command` - records the client's writes through a proxy. It uses a one-entry statement - cache and one uncached statement, so both close paths are exercised. - - With the fix it passes on MySQL 8.0.46 and MariaDB 10.11 in 0.01 s. - - With the published driver it fails on both: every close ends its own - write, and the same 14 statements take 0.6 s. - - The existing cancelled-boundary qualification still passes. + statement close with sequence number 0. On the path before the next command, + `wait_until_ready` no longer flushes a buffer that holds only replyless + closes, so they leave in the same write as that command. `Connection::flush` + and `clear_cached_statements` still send queued closes before returning; + Security's review of the first revision found that they did not. Packets, + order and server behavior are unchanged. The patch record is + `rust/vendor/sqlx-mysql/DURABLE-PATCH.md`. +- **Tests.** A proxy records the TCP read chunks it receives from the client. + TCP is a byte stream, so a chunk boundary shows bytes that arrived + separately. + - `qualification_statement_close_leaves_with_the_next_command` uses a + one-entry statement cache and one uncached statement, so both close paths + are exercised. With the fix it passes on MySQL 8.0.46 and MariaDB 10.11 in + 0.01 s. With the published driver it fails on both: every close arrives + as a chunk of its own, and the same 14 statements take 0.6 s. + - `qualification_explicit_flush_sends_queued_statement_closes`: a queued + close stays buffered until `flush`, which sends it with no later command + and leaves `should_flush` false. + - `qualification_clearing_the_statement_cache_closes_them_now`: + `clear_cached_statements` sends all three closes before it returns. On + MySQL 8.0 with the performance schema on, the server's statements for the + connection drop from three to none. + - Both lifecycle tests fail with the first revision (`f31c4c361`) on MySQL + 8.0 and MariaDB, and pass with this one. The existing cancelled-boundary + qualification still passes. - **Hosted-cell A/B on the VM.** Interleaved cells with the spec and limits unchanged. Both builds were instrumented: one without the driver change, one with its first revision. That revision already buffered the closes, but diff --git a/rust/src/runtime/mysql_pool_tests.rs b/rust/src/runtime/mysql_pool_tests.rs index e6f40c497..20b524bf7 100644 --- a/rust/src/runtime/mysql_pool_tests.rs +++ b/rust/src/runtime/mysql_pool_tests.rs @@ -222,9 +222,12 @@ async fn qualification_cancelled_pool_boundaries_preserve_fresh_reads_and_commit .expect("bounded real transaction-boundary qualification"); } -/// Forward one client to `upstream`, recording each client write as it -/// arrives. -async fn record_client_writes( +const COM_STMT_CLOSE: u8 = 0x19; + +/// Forward clients to `upstream`, recording each TCP read chunk received from +/// a client. TCP is a byte stream: a chunk boundary shows bytes that arrived +/// separately, not a write call. +async fn record_client_chunks( listener: TcpListener, upstream: String, writes: Arc>>>, @@ -258,7 +261,7 @@ async fn record_client_writes( } /// Each client command packet (sequence 0): its command byte, and whether a -/// recorded write ended with it. +/// received chunk ended with it. fn recorded_commands(writes: &[Vec]) -> Vec<(u8, bool)> { let mut stream = Vec::new(); let mut write_ends = std::collections::BTreeSet::new(); @@ -284,21 +287,21 @@ fn recorded_commands(writes: &[Vec]) -> Vec<(u8, bool)> { commands } -/// A statement close has no reply. Flushed on its own, it leaves +/// A statement close has no reply. Sent on its own, it leaves /// unacknowledged data in flight, and Nagle's algorithm then holds the next /// command until the server's delayed acknowledgement, about 40 ms per close. -/// Under the transition lock those stalls added up to a late timer (#714). +/// The proxy then receives the close as a chunk of its own. Under the +/// transition lock those stalls added up to a late timer (#714). #[tokio::test] #[ignore = "requires disposable MySQL/MariaDB; run qualification_ explicitly"] async fn qualification_statement_close_leaves_with_the_next_command() { - const COM_STMT_CLOSE: u8 = 0x19; tokio::time::timeout(Duration::from_secs(15), async { let url = std::env::var("DW_TEST_MYSQL_URL").unwrap(); let direct = MySqlConnectOptions::from_str(&url).unwrap(); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let port = listener.local_addr().unwrap().port(); let writes = Arc::new(std::sync::Mutex::new(Vec::new())); - let proxy = tokio::spawn(record_client_writes( + let proxy = tokio::spawn(record_client_chunks( listener, format!("{}:{}", direct.get_host(), direct.get_port()), writes.clone(), @@ -334,9 +337,130 @@ async fn qualification_statement_close_leaves_with_the_next_command() { assert!(closes >= 12, "eviction and uncached closes were exercised: {closes}"); assert!( commands.iter().all(|(command, ended)| *command != COM_STMT_CLOSE || !ended), - "a statement close ended its own write: {commands:?}" + "a statement close arrived as a chunk of its own: {commands:?}" + ); + }) + .await + .expect("bounded statement-close chunk qualification"); +} + +/// A plaintext connection through a recording proxy. +async fn proxied_connection( + capacity: usize, +) -> (MySqlConnection, Arc>>>, tokio::task::JoinHandle<()>, MySqlConnectOptions) { + let url = std::env::var("DW_TEST_MYSQL_URL").unwrap(); + let direct = MySqlConnectOptions::from_str(&url).unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let writes = Arc::new(std::sync::Mutex::new(Vec::new())); + let proxy = tokio::spawn(record_client_chunks( + listener, + format!("{}:{}", direct.get_host(), direct.get_port()), + writes.clone(), + )); + let options = direct + .clone() + .host("127.0.0.1") + .port(port) + .ssl_mode(sqlx::mysql::MySqlSslMode::Disabled) + .statement_cache_capacity(capacity); + (MySqlConnection::connect_with(&options).await.unwrap(), writes, proxy, direct) +} + +fn closes(writes: &std::sync::Mutex>>) -> usize { + recorded_commands(&writes.lock().unwrap()) + .iter() + .filter(|(command, _)| *command == COM_STMT_CLOSE) + .count() +} + +/// Wait, bounded, until the proxy has received `count` statement closes. +async fn closes_arrive(writes: &std::sync::Mutex>>, count: usize) { + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while closes(writes) < count { + assert!( + tokio::time::Instant::now() < deadline, + "{count} statement closes did not leave the client: {:?}", + recorded_commands(&writes.lock().unwrap()) ); + tokio::time::sleep(Duration::from_millis(10)).await; + } +} + +/// `Connection::flush` sends queued closes by itself: no later command is +/// needed, and the connection reports nothing left to flush. +#[tokio::test] +#[ignore = "requires disposable MySQL/MariaDB; run qualification_ explicitly"] +async fn qualification_explicit_flush_sends_queued_statement_closes() { + use sqlx::Connection as _; + tokio::time::timeout(Duration::from_secs(15), async { + let (mut connection, writes, proxy, _) = proxied_connection(1).await; + // An uncached statement queues its close after it executes. + let value: i64 = sqlx::query_scalar("SELECT ? + 4") + .bind(1_i64) + .persistent(false) + .fetch_one(&mut connection) + .await + .unwrap(); + assert_eq!(value, 5); + assert!(connection.should_flush(), "the close is queued"); + tokio::time::sleep(Duration::from_millis(200)).await; + assert_eq!(closes(&writes), 0, "the queued close waits for a command or flush"); + connection.flush().await.unwrap(); + assert!(!connection.should_flush(), "an explicit flush leaves nothing buffered"); + closes_arrive(&writes, 1).await; + connection.close().await.unwrap(); + proxy.abort(); + }) + .await + .expect("bounded explicit-flush qualification"); +} + +/// `clear_cached_statements` closes every cached statement on the server +/// before it returns, without a later command. +#[tokio::test] +#[ignore = "requires disposable MySQL/MariaDB; run qualification_ explicitly"] +async fn qualification_clearing_the_statement_cache_closes_them_now() { + use sqlx::Connection as _; + tokio::time::timeout(Duration::from_secs(15), async { + let (mut connection, writes, proxy, direct) = proxied_connection(100).await; + // A separate session can see this connection's server-side + // statements where the performance schema is on (MySQL 8 default). + let thread: u64 = sqlx::query_scalar("SELECT CONNECTION_ID()").fetch_one(&mut connection).await.unwrap(); + connection.clear_cached_statements().await.unwrap(); + let before = closes(&writes); + for sql in ["SELECT ? + 1", "SELECT ? + 2", "SELECT ? + 3"] { + let _: i64 = sqlx::query_scalar(sql).bind(1_i64).fetch_one(&mut connection).await.unwrap(); + } + assert_eq!(connection.cached_statements_size(), 3); + let mut observer = MySqlConnection::connect_with(&direct).await.unwrap(); + let schema: i64 = sqlx::query_scalar("SELECT @@performance_schema").fetch_one(&mut observer).await.unwrap(); + let server_statements = async |observer: &mut MySqlConnection| -> i64 { + sqlx::query_scalar("SELECT COUNT(*) FROM performance_schema.prepared_statements_instances p JOIN performance_schema.threads t ON t.THREAD_ID=p.OWNER_THREAD_ID WHERE t.PROCESSLIST_ID=?") + .bind(thread) + .fetch_one(observer) + .await + .unwrap() + }; + if schema == 1 { + let held = server_statements(&mut observer).await; + assert_eq!(held, 3, "the server holds this connection's three statements"); + } + connection.clear_cached_statements().await.unwrap(); + assert_eq!(connection.cached_statements_size(), 0); + assert!(!connection.should_flush(), "clearing leaves nothing buffered"); + closes_arrive(&writes, before + 3).await; + if schema == 1 { + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while server_statements(&mut observer).await != 0 { + assert!(tokio::time::Instant::now() < deadline, "the server still holds the statements"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + observer.close().await.unwrap(); + connection.close().await.unwrap(); + proxy.abort(); }) .await - .expect("bounded statement-close write qualification"); + .expect("bounded statement-cache clear qualification"); } diff --git a/rust/vendor/sqlx-mysql/DURABLE-PATCH.md b/rust/vendor/sqlx-mysql/DURABLE-PATCH.md index 117531a0c..f4792828c 100644 --- a/rust/vendor/sqlx-mysql/DURABLE-PATCH.md +++ b/rust/vendor/sqlx-mysql/DURABLE-PATCH.md @@ -33,15 +33,28 @@ so these stalls added up to holds of 0.3 to 1 s under load. A due timer waiting behind them fired more than the 1 s maximum late (Server #714). `MySqlStream::queue_packet` now buffers each close with the reset sequence -number that `send_packet` would use, so it leaves in the same write as the next -command. The server receives the same packets in the same order. A connection -dropped before its next command leaves the close unsent; the server releases -that connection's prepared statements when it closes. - -`qualification_statement_close_leaves_with_the_next_command` in -`rust/src/runtime/mysql_pool_tests.rs` records every client write through a -proxy. It fails with the published driver, where each close ends its write, and -passes with this correction. +number that `send_packet` would use. On the path before the next command +(`wait_until_ready`), a buffer that holds only such closes is not flushed, so +they leave in the same write as that command. The server receives the same +packets in the same order. A connection dropped before its next command leaves +the close unsent; the server releases that connection's prepared statements +when it closes. + +The public lifecycle is unchanged. `Connection::flush` sends every buffered +byte, queued closes included, and leaves `should_flush` false. +`clear_cached_statements` sends its closes before it returns. Neither needs a +later command. + +Three tests in `rust/src/runtime/mysql_pool_tests.rs` record the TCP read +chunks a proxy receives from the client: +- `qualification_statement_close_leaves_with_the_next_command` fails with the + published driver, where each close arrives as a chunk of its own, and passes + with this correction; +- `qualification_explicit_flush_sends_queued_statement_closes` and + `qualification_clearing_the_statement_cache_closes_them_now` prove the two + lifecycle calls without a later command. Where the performance schema is on, + the second also sees the server's statements for the connection drop from + three to none. All upstream source is retained. The source-code changes are in `src/connection/mod.rs`, `src/connection/stream.rs` and diff --git a/rust/vendor/sqlx-mysql/src/connection/mod.rs b/rust/vendor/sqlx-mysql/src/connection/mod.rs index c20bfc026..2513cf3cd 100755 --- a/rust/vendor/sqlx-mysql/src/connection/mod.rs +++ b/rust/vendor/sqlx-mysql/src/connection/mod.rs @@ -94,7 +94,11 @@ impl Connection for MySqlConnection { #[doc(hidden)] fn flush(&mut self) -> impl Future> + Send + '_ { - self.inner.stream.wait_until_ready() + async move { + // `Connection::flush` sends everything, queued closes included. + self.inner.stream.flush_queued().await?; + self.inner.stream.wait_until_ready().await + } } fn cached_statements_size(&self) -> usize { @@ -110,7 +114,8 @@ impl Connection for MySqlConnection { })?; } - Ok(()) + // The server closes them now, not with some later command. + self.inner.stream.flush_queued().await } #[doc(hidden)] diff --git a/rust/vendor/sqlx-mysql/src/connection/stream.rs b/rust/vendor/sqlx-mysql/src/connection/stream.rs index 590ce1ecd..c16a21eb0 100755 --- a/rust/vendor/sqlx-mysql/src/connection/stream.rs +++ b/rust/vendor/sqlx-mysql/src/connection/stream.rs @@ -63,6 +63,17 @@ impl MySqlStream { } } + /// Send every buffered byte now, including queued commands without a + /// reply. Explicit flushes and statement cache clears use this; only the + /// path before the next command leaves queued closes to share its write. + pub(crate) async fn flush_queued(&mut self) -> Result<(), Error> { + if !self.socket.write_buffer().is_empty() { + self.socket.flush().await?; + } + self.queued_unanswered = false; + Ok(()) + } + pub(crate) async fn wait_until_ready(&mut self) -> Result<(), Error> { // Queued commands without a reply wait for the next command; nothing // pending needs them to be sent first. From faded3552843755a810b62abee09f6098ded29f0 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 24 Oct 2026 00:12:45 +0000 Subject: [PATCH 06/11] Keep declared signals' record sequence empty under a condition wait Workflow 2.10.2's applySignal takes a workflow sequence only from the signal's own wait, so a declared signal applied while a condition wait is open keeps no sequence; only an undeclared signal takes the condition wait's. Native recorded the condition wait's sequence for both, which became visible once worker polls showed the record: the published-SDK signal and query cohorts require worker history to equal the run's history prefix (all nine SDK pairings on every database). A new test covers the declared case and fails without the change. --- .../2026-10-23-worker-history-presentation.md | 14 ++++++++ rust/src/runtime/store.rs | 16 +++++---- .../execution/worker_history_presentation.rs | 33 +++++++++++++++++++ 3 files changed, 57 insertions(+), 6 deletions(-) diff --git a/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md index 79271b917..63f110abb 100644 --- a/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md +++ b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md @@ -17,12 +17,26 @@ run's own history route still shows the events as recorded, as PHP's does. - Task snapshots keep those fields when they are set. +- **Which signals carry a sequence.** PHP's `applySignal` takes a sequence + only from the signal's own wait. A signal the run's contract declares is + applied without one even while a condition wait is open, and + `markConditionWaitSignalConsumed` then finds it applied, so its record keeps + no sequence. Only an undeclared signal takes the condition wait's sequence. + Native had recorded the condition wait's sequence for declared signals too. + That difference was invisible until worker polls showed the record, and the + published-SDK signal and query cohorts then caught it: their worker history + must equal the run's history prefix. Native now records a sequence only from + the signal's own wait, or from a condition wait for an undeclared signal. - **How it was checked.** A test runs on SQLite, PostgreSQL, MySQL and MariaDB: - a condition wait consumes an undeclared signal; - the next poll and history page show that wait's sequence, and the run's history does not; - the completing task's snapshot has its dispatch stamps. + + A second test applies a declared signal under a condition wait twice. The + worker's history shows no sequence on either `SignalReceived`, and it fails + without the record change. - **Not changed yet.** Some events PHP records with a task snapshot have none in native, such as `ConditionWaitOpened` and `SignalWaitOpened`. That audit is a #325 follow-up. diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 03dfa699a..d02b6ee5c 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -1124,11 +1124,10 @@ where let wait = open.clone().filter(|wait| { wait.get("signal_name").is_none() || (wait["signal_name"] == name && wait["signal_wait_id"] == wait_id) }); - if wait.is_none() - && !Self::declared_signals(tx, run_id) - .await? - .is_some_and(|names| names.contains(&name)) - { + let declared = Self::declared_signals(tx, run_id) + .await? + .is_some_and(|names| names.contains(&name)); + if wait.is_none() && !declared { let reason = if open.is_some() { "signal_wait_mismatch" } else { "signal_wait_not_open" }; return Err(refuse(StatusCode::CONFLICT, reason)); } @@ -1156,8 +1155,13 @@ where .bind(&command_id) .execute(&mut **tx) .await?; + // PHP's `applySignal` takes a sequence only from the signal's own + // wait, so a declared signal applied under a condition wait keeps no + // workflow sequence. Only an undeclared signal takes the condition + // wait's (`markConditionWaitSignalConsumed`). + let record_sequence = if direct || !declared { sequence } else { None }; Self::query("UPDATE workflow_signal_records SET status='applied',workflow_sequence=$1,applied_at=$2,closed_at=$3 WHERE id=$4") - .bind(DB::optional_integer(sequence)).bind(DB::bind_time(applied_at)).bind(DB::bind_time(applied_at)).bind(signal_id).execute(&mut **tx).await?; + .bind(DB::optional_integer(record_sequence)).bind(DB::bind_time(applied_at)).bind(DB::bind_time(applied_at)).bind(signal_id).execute(&mut **tx).await?; let run = Self::query( "SELECT workflow_instance_id,message_cursor_position FROM workflow_runs WHERE id=$1", ) diff --git a/rust/tests/execution/worker_history_presentation.rs b/rust/tests/execution/worker_history_presentation.rs index db76a3d75..a5cafa602 100644 --- a/rust/tests/execution/worker_history_presentation.rs +++ b/rust/tests/execution/worker_history_presentation.rs @@ -62,3 +62,36 @@ 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}"); } + +// Workflow 2.10.2 `DefaultWorkflowTaskBridge::applySignal` takes a workflow +// sequence only from the signal's own wait. A signal the run's contract +// declares is applied without one even while a condition wait is open, so +// `markConditionWaitSignalConsumed` finds it applied and its record keeps no +// sequence: the history a worker reads matches the run's history. +#[tokio::test] +async fn a_declared_signal_consumed_by_a_condition_wait_keeps_no_sequence() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + update_worker(&app, "declared-presenter").await; + let started = start(&app, "declared-signals").await; + let task = poll(&app, "declared-presenter", "workflow").await; + assert_eq!(finish_task(&app, &task, wait("first")).await.0, StatusCode::OK); + + let finish = json!({"input":envelope(Payload::Array(vec![Payload::Long(7)]))}); + let (status, body) = request(&app, "POST", "/api/workflows/declared-signals/signal/finish", finish.clone()).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let task = poll(&app, "declared-presenter", "workflow").await; + assert_eq!(finish_task(&app, &task, wait("second")).await.0, StatusCode::OK); + let (status, body) = request(&app, "POST", "/api/workflows/declared-signals/signal/finish", finish).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let task = poll(&app, "declared-presenter", "workflow").await; + + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let polled = received(&task["history_events"]); + assert_eq!(polled.len(), 2, "{task}"); + assert!(polled.iter().all(|signal| signal.get("workflow_sequence").is_none()), "{polled:?}"); + let recorded = received(&history); + assert_eq!(polled[0]["signal_id"], recorded[0]["signal_id"]); + let (_, described) = request(&app, "GET", "/api/workflows/declared-signals", Value::Null).await; + assert_eq!(described["status"], "waiting", "{described}"); +} From 31467e575a494850f014a4dd8891f062941b2355 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Sat, 24 Oct 2026 04:26:13 +0000 Subject: [PATCH 07/11] CI: cancel a superseded pull request run of the fixture and PHPUnit workflows A newer push to the same pull request now cancels the run in progress for Shared Server fixtures (13 jobs, about 415 job-minutes) and PHPUnit Feature, as Sonar already does. The organization's 20 shared runners no longer spend hours on replaced heads. Runs on main and manual dispatches always finish. --- .github/workflows/phpunit-feature.yml | 6 ++++++ .github/workflows/server-parity.yml | 7 +++++++ 2 files changed, 13 insertions(+) diff --git a/.github/workflows/phpunit-feature.yml b/.github/workflows/phpunit-feature.yml index 9cecaf1bf..177f18693 100644 --- a/.github/workflows/phpunit-feature.yml +++ b/.github/workflows/phpunit-feature.yml @@ -16,6 +16,12 @@ on: required: false type: string +# A newer push to the same pull request supersedes the run in progress. Runs on +# main and manual dispatches always finish. +concurrency: + group: phpunit-feature-${{ github.event_name }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + permissions: contents: read diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml index 81878ea9f..9960470e7 100644 --- a/.github/workflows/server-parity.yml +++ b/.github/workflows/server-parity.yml @@ -114,6 +114,13 @@ on: - rust/** workflow_dispatch: +# A newer push to the same pull request supersedes the run in progress, so a +# replaced head does not keep holding the organization's shared runners. +# Manual dispatches always finish. +concurrency: + group: shared-server-fixtures-${{ github.event_name }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: ${{ github.event_name == 'pull_request' }} + permissions: contents: read From 49a5099ba955e08d43e7975b4e639e795cd57c10 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 24 Oct 2026 08:00:29 +0000 Subject: [PATCH 08/11] Keep PHPUnit Feature's runs uncancelled; Shared Server fixtures keeps #724's cancellation The central action policy requires always() on the PHPUnit feature suite gate, and GitHub never cancels an always() job, so cancel-in-progress on PHPUnit Feature would only queue a newer run behind it. The workflow returns to main's definition, as train #723 decided at 1fd9e549a. --- .github/workflows/phpunit-feature.yml | 6 ------ 1 file changed, 6 deletions(-) diff --git a/.github/workflows/phpunit-feature.yml b/.github/workflows/phpunit-feature.yml index 177f18693..9cecaf1bf 100644 --- a/.github/workflows/phpunit-feature.yml +++ b/.github/workflows/phpunit-feature.yml @@ -16,12 +16,6 @@ on: required: false type: string -# A newer push to the same pull request supersedes the run in progress. Runs on -# main and manual dispatches always finish. -concurrency: - group: phpunit-feature-${{ github.event_name }}-${{ github.event.pull_request.number || github.ref }} - cancel-in-progress: ${{ github.event_name == 'pull_request' }} - permissions: contents: read From 9eb3bb838f1b68d0d9a09a60ba27f3ecdb2acedd Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 24 Oct 2026 06:53:46 +0000 Subject: [PATCH 09/11] Retry documented lock-pressure poll refusals in the Prometheus and lease probes The published PHP SQLite reference shares its file with a one-second maintenance loop. #723's run 40069368271 recorded the Prometheus fixture's activity claim as 503 (then 422 for the completions that had no task). An activity poll under lock pressure answers the documented 503 backend_lock_pressure refusal, which hands out no task and tells the worker to retry. The system-metrics and operator-dashboard probes already retry exactly that refusal through lock-pressure-middleware.php. The Prometheus and activity lease renewal probes build their own Guzzle clients for raw worker polls and did not. They now push the same middleware. Expected results are unchanged: the claim must still end at 200, and every other status or body still reaches the contract. ServerParityRunnerTest now requires the middleware in every corpus probe that polls over its own HTTP client; it fails against the previous Prometheus probe. --- .../activity-lease-renewal-probe.php | 7 +++++- .../prometheus-metrics-probe.php | 7 +++++- tests/Unit/ServerParityRunnerTest.mjs | 23 +++++++++++++++++++ 3 files changed, 35 insertions(+), 2 deletions(-) diff --git a/scripts/conformance/server-parity/activity-lease-renewal-probe.php b/scripts/conformance/server-parity/activity-lease-renewal-probe.php index 08a9bf8b0..57350cf9e 100644 --- a/scripts/conformance/server-parity/activity-lease-renewal-probe.php +++ b/scripts/conformance/server-parity/activity-lease-renewal-probe.php @@ -4,6 +4,7 @@ use DurableWorkflow\Client; use GuzzleHttp\Client as LeaseRenewalHttp; +use GuzzleHttp\HandlerStack; // Runtime activity lease renewal (POST /api/worker/activity-tasks/{task}/renew-lease) // over raw worker HTTP in a fresh namespace. Three runs each claim one @@ -17,8 +18,12 @@ function finishHttpActivityLeaseRenewal(Client $admin, array $fixture, string $w $namespace = strtolower($workflowId.'-lease'); $admin->createNamespace($namespace, 'activity lease renewal λ', 30); $token = getenv('DW_PARITY_TOKEN') ?: null; + // The SQLite reference shares its database with a one-second maintenance + // loop; retry its retryable lock-pressure refusals, as the other probes do. + $stack = HandlerStack::create(); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $http = new LeaseRenewalHttp(['base_uri' => rtrim($url, '/'), 'http_errors' => false, - 'connect_timeout' => 1, 'timeout' => 30]); + 'connect_timeout' => 1, 'timeout' => 30, 'handler' => $stack]); $raw = static function (string $method, string $path, ?array $body = [], string $protocol = '1.20') use ($http, $token, $namespace): array { $options = ['headers' => array_filter([ 'Authorization' => $token === null ? null : 'Bearer '.$token, diff --git a/scripts/conformance/server-parity/prometheus-metrics-probe.php b/scripts/conformance/server-parity/prometheus-metrics-probe.php index 3fbbe285c..2192b7dc5 100644 --- a/scripts/conformance/server-parity/prometheus-metrics-probe.php +++ b/scripts/conformance/server-parity/prometheus-metrics-probe.php @@ -4,6 +4,7 @@ use DurableWorkflow\Client; use GuzzleHttp\Client as PrometheusMetricsHttp; +use GuzzleHttp\HandlerStack; // The Prometheus summary of a fresh namespace before and after a known // sequence over raw HTTP: a worker with declared slots on the first queue; a @@ -19,8 +20,12 @@ function finishHttpPrometheusMetrics(Client $admin, array $fixture, string $work $worker = $namespace.'-worker'; $admin->createNamespace($namespace, 'prometheus metrics λ', 30); $token = getenv('DW_PARITY_TOKEN') ?: null; + // The SQLite reference shares its database with a one-second maintenance + // loop; retry its retryable lock-pressure refusals, as the other probes do. + $stack = HandlerStack::create(); + $stack->push(LockPressurePollRetries::middleware(), 'lock_pressure_poll_retry'); $http = new PrometheusMetricsHttp(['base_uri' => rtrim($url, '/').'/api/', 'http_errors' => false, - 'connect_timeout' => 1, 'timeout' => 30]); + 'connect_timeout' => 1, 'timeout' => 30, 'handler' => $stack]); $raw = static function (string $method, string $path, ?array $body = null, bool $worker = false) use ($http, $token, $namespace): array { $response = $http->request($method, $path, ['headers' => array_filter([ 'Authorization' => $token === null ? null : 'Bearer '.$token, diff --git a/tests/Unit/ServerParityRunnerTest.mjs b/tests/Unit/ServerParityRunnerTest.mjs index 081b0372b..adf2a32fe 100644 --- a/tests/Unit/ServerParityRunnerTest.mjs +++ b/tests/Unit/ServerParityRunnerTest.mjs @@ -888,3 +888,26 @@ for (const [name, corrupt] of [ assert.throws(() => checkObservation(updateFixture, raw, 'test-one-activity')); }); } + +// Raw-HTTP probes in the corpus process poll the published PHP SQLite +// reference while its maintenance loop runs. A poll it refuses with the +// documented 503 backend_lock_pressure reply hands out no task and asks the +// worker to retry, so each probe that polls must retry exactly that refusal +// (lock-pressure-middleware.php) rather than record it as the case's answer. +test('corpus probes that poll over their own HTTP client retry documented lock-pressure refusals', () => { + const directory = new URL('../../scripts/conformance/server-parity/', import.meta.url); + const probe = readFileSync(new URL('probe.php', directory), 'utf8'); + const required = [...probe.matchAll(/require __DIR__\.'\/([a-z0-9-]+\.php)';/g)].map(match => match[1]); + const pollers = required.filter(name => { + const source = readFileSync(new URL(name, directory), 'utf8'); + return /new \w+Http\(\[/.test(source) && source.includes('tasks/poll'); + }); + assert.ok(pollers.includes('prometheus-metrics-probe.php') && pollers.includes('activity-lease-renewal-probe.php'), pollers.join(', ')); + for (const name of pollers) { + const source = readFileSync(new URL(name, directory), 'utf8'); + assert.match(source, /\$stack->push\(LockPressurePollRetries::middleware\(\), 'lock_pressure_poll_retry'\);/, name); + for (const construction of source.matchAll(/new \w+Http\(\[[^;]*\]\);/gs)) { + assert.match(construction[0], /'handler' => \$stack/, `${name}: ${construction[0]}`); + } + } +}); From 71c0eff781afc3a73fad46928b8490dfc18d7085 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 24 Oct 2026 07:46:47 +0000 Subject: [PATCH 10/11] Report the measured shutdown time when the SDK pressure window is missed Train #723's MySQL 8.0 pressure profile failed 'one real SDK budget expires while pressure stays held' on the published PHP target, and the log could not say whether shutdown took under 8 s or over 10 s. The assertion now prints the measured time and backend. The window itself is unchanged. --- .../conformance/server-parity/worker-persistent-contract.mjs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/scripts/conformance/server-parity/worker-persistent-contract.mjs b/scripts/conformance/server-parity/worker-persistent-contract.mjs index a2c85df6a..36102eeb9 100644 --- a/scripts/conformance/server-parity/worker-persistent-contract.mjs +++ b/scripts/conformance/server-parity/worker-persistent-contract.mjs @@ -10,7 +10,8 @@ export function checkSdkPersistentPressure(fixture,state,workflowId,token) { assert.equal(p.bounded_transport,true); assert.ok(['sqlite','mysql','pgsql'].includes(p.backend)); assert.ok(Number.isFinite(p.shutdown_elapsed_seconds) && p.shutdown_elapsed_seconds>=8 - && p.shutdown_elapsed_seconds<=10,'one real SDK budget expires while pressure stays held'); + && p.shutdown_elapsed_seconds<=10, + `one real SDK budget expires while pressure stays held: shutdown took ${p.shutdown_elapsed_seconds} s on ${p.backend}, outside 8-10 s`); for(const sample of [p.reference_php_session_before,p.reference_php_session_after]) { assert.equal(sample.origin,'php_application_cli_connection'); assert.equal(sample.driver,p.backend); From 284aaba8f89d8b8bee4ca96a14444b5d2c783103 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 24 Oct 2026 09:52:09 +0000 Subject: [PATCH 11/11] Refuse non-UTF-8 quota maps and hydrate only a history page's own signals Security's R2 review of #729 found two defects. The namespace quota map was read with a lossy decode, so a value that is not UTF-8 could become a different namespace key and drop the intended cap. Both the DW and legacy names now go through a fallible read: the DW name wins when set, the legacy name is read only in its absence, and a non-UTF-8 value under either name is a configuration error that never echoes the value. Worker history signal hydration fetched every sequenced signal record of the run on each page. It now reads only the page's distinct non-empty SignalReceived signal IDs, each fenced by workflow_run_id, as PHP's whereIn('id', $signalIds) does, so the work is bounded by the page. --- rust/src/runtime/external_config.rs | 65 ++++++++++++++-- rust/src/runtime/store.rs | 74 +++++++++++++++++-- .../execution/worker_history_presentation.rs | 66 +++++++++++++++++ 3 files changed, 192 insertions(+), 13 deletions(-) diff --git a/rust/src/runtime/external_config.rs b/rust/src/runtime/external_config.rs index 070638fc7..1bd3fa84c 100644 --- a/rust/src/runtime/external_config.rs +++ b/rust/src/runtime/external_config.rs @@ -119,16 +119,36 @@ fn member<'a>(value: &'a Value, key: &str) -> Option<&'a Value> { } } +const OVERRIDES: &str = "DW_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES"; +const LEGACY_OVERRIDES: &str = "WORKFLOW_SERVER_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES"; + +/// The honored namespace quota map: unset, or a JSON object (Server #705). +/// The DW name wins whenever it is set; the legacy name is read only in its +/// absence. A value that is not UTF-8 is refused, as PHP's +/// `JSON_THROW_ON_ERROR` refuses invalid UTF-8, rather than decoded lossily +/// into a different namespace key. Refusals never echo the value. +fn namespace_overrides( + read: impl Fn(&str) -> std::result::Result, +) -> Result> { + let fetch = |name: &str| match read(name) { + Ok(value) => Ok(Some(value)), + Err(env::VarError::NotPresent) => Ok(None), + Err(env::VarError::NotUnicode(_)) => Err(configuration_error()), + }; + let primary = fetch(OVERRIDES)?; + let legacy = if primary.is_none() { fetch(LEGACY_OVERRIDES)? } else { None }; + crate::config_audit::json_object( + |name| if name == OVERRIDES { primary.clone() } else { legacy.clone() }, + OVERRIDES, + Some(LEGACY_OVERRIDES), + ) + .map_err(|_| configuration_error()) +} + impl ExternalPayloadConfig { pub(super) fn from_env() -> Result { let completion = process_value("COMPLETION_MAX_BYTES", Value::Null)?; - // Startup refuses a value that is not a JSON object (Server #705). - let overrides = crate::config_audit::json_object( - |name| env::var_os(name).map(|value| value.to_string_lossy().into_owned()), - "DW_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES", - Some("WORKFLOW_SERVER_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES"), - ) - .map_err(|_| configuration_error())?; + let overrides = namespace_overrides(|name| env::var(name))?; Ok(Self { completion_max_bytes: (!completion.is_null()) .then(|| integer(&completion).max(1) as u64), @@ -245,4 +265,35 @@ mod tests { assert!(config.quota("other").is_err()); assert!(config.quota("default").is_ok()); } + + /// An environment holding exactly these variables. + fn environment(values: Vec<(&'static str, std::ffi::OsString)>) -> impl Fn(&str) -> std::result::Result { + move |name| match values.iter().find(|(key, _)| *key == name) { + None => Err(env::VarError::NotPresent), + Some((_, value)) => value.clone().into_string().map_err(env::VarError::NotUnicode), + } + } + + #[cfg(unix)] + #[test] + fn namespace_overrides_refuse_non_utf8_under_either_name() { + use std::os::unix::ffi::OsStringExt; + // Valid JSON once lossily decoded: "tenant\u{FFFD}" would become a + // different namespace key and drop the intended cap. + let invalid = || std::ffi::OsString::from_vec(b"{\"tenant\xff\":{\"max_bytes\":5}}".to_vec()); + assert!(namespace_overrides(environment(vec![(OVERRIDES, invalid())])).is_err()); + assert!(namespace_overrides(environment(vec![(LEGACY_OVERRIDES, invalid())])).is_err()); + // The DW value wins over the legacy one, as before, even when only the + // DW value is invalid. + assert!(namespace_overrides(environment(vec![(OVERRIDES, invalid()), (LEGACY_OVERRIDES, r#"{"a":{}}"#.into())])).is_err()); + // Valid objects under either name still load. + let valid = namespace_overrides(environment(vec![(OVERRIDES, r#"{"tenant":{"max_bytes":5}}"#.into())])).unwrap(); + assert_eq!(valid["tenant"], json!({"max_bytes": 5})); + let legacy = namespace_overrides(environment(vec![(LEGACY_OVERRIDES, r#"{"other":{"max_objects":3}}"#.into())])).unwrap(); + assert_eq!(legacy["other"], json!({"max_objects": 3})); + assert!(namespace_overrides(environment(vec![])).unwrap().is_empty()); + // A refusal does not echo the value. + let refused = format!("{:?}", namespace_overrides(environment(vec![(OVERRIDES, invalid())])).unwrap_err()); + assert!(!refused.contains("tenant"), "{refused}"); + } } diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 0b36704f1..edeba7358 100644 --- a/rust/src/runtime/store.rs +++ b/rust/src/runtime/store.rs @@ -80,6 +80,61 @@ pub(super) fn id() -> String { .to_string() } +/// The distinct, non-empty signal IDs a history page's `SignalReceived` +/// events name: the only signal records its hydration reads. +fn page_signal_ids(events: &[Value]) -> std::collections::BTreeSet { + events + .iter() + .filter(|event| event["event_type"] == "SignalReceived") + .filter_map(|event| event["payload"]["signal_id"].as_str()) + .filter(|id| !id.is_empty()) + .map(str::to_owned) + .collect() +} + +#[cfg(test)] +#[test] +fn a_history_page_reads_only_its_own_signal_records() { + let signal = |id: &str| json!({"event_type":"SignalReceived","payload":{"signal_id":id}}); + let page = [signal("b"), signal("a"), signal("a"), signal(""), json!({"event_type":"TimerFired","payload":{"signal_id":"c"}}), + json!({"event_type":"SignalReceived","payload":{}})]; + assert_eq!(page_signal_ids(&page).into_iter().collect::>(), ["a", "b"]); + // One event, one record, however many signals the run holds. + assert_eq!(page_signal_ids(&[signal("only")]).len(), 1); + assert!(page_signal_ids(&[]).is_empty()); +} + +// Hydration never reads a record the page does not name. The run's other +// record holds a sequence that cannot be decoded (SQLite keeps the text in +// an integer column), so a run-wide read fails where a page-scoped one +// succeeds; a same-ID record of another run stays fenced out. +#[cfg(test)] +#[tokio::test] +async fn signal_hydration_reads_no_record_outside_the_page() { + use sqlx::Connection; + let mut connection = sqlx::SqliteConnection::connect("sqlite::memory:").await.unwrap(); + sqlx::query("CREATE TABLE workflow_signal_records (id TEXT, workflow_run_id TEXT, workflow_sequence INTEGER)") + .execute(&mut connection) + .await + .unwrap(); + for (id, run, sequence) in [("page", "run", "7"), ("unrelated", "run", "not-a-sequence"), ("fenced", "other", "9")] { + sqlx::query("INSERT INTO workflow_signal_records VALUES ($1, $2, $3)") + .bind(id) + .bind(run) + .bind(sequence) + .execute(&mut connection) + .await + .unwrap(); + } + let mut events = vec![ + json!({"event_type":"SignalReceived","payload":{"signal_id":"page"}}), + json!({"event_type":"SignalReceived","payload":{"signal_id":"fenced"}}), + ]; + Store::::with_signal_sequences(&mut connection, "run", &mut events).await.unwrap(); + assert_eq!(events[0]["payload"]["workflow_sequence"], 7, "{events:?}"); + assert!(events[1]["payload"].get("workflow_sequence").is_none(), "{events:?}"); +} + #[cfg(test)] #[test] fn identifiers_keep_creation_order_within_a_millisecond() { @@ -3579,13 +3634,20 @@ where if !events.iter().any(|event| event["event_type"] == "SignalReceived") { return Ok(()); } - let rows = Self::query("SELECT id,workflow_sequence FROM workflow_signal_records WHERE workflow_run_id=$1 AND workflow_sequence IS NOT NULL") - .bind(run_id) - .fetch_all(&mut *connection) - .await?; + // Only this page's signals, as PHP's `whereIn('id', $signalIds)` does: + // the work stays bounded by the page, not by the run's signal count, + // and the run fence keeps another run's record out. + let ids = page_signal_ids(events); let mut sequences = std::collections::HashMap::new(); - for row in rows { - sequences.insert(DB::string(&row, "id")?, DB::number(&row, "workflow_sequence")?); + for id in ids { + if let Some(row) = Self::query("SELECT workflow_sequence FROM workflow_signal_records WHERE id=$1 AND workflow_run_id=$2 AND workflow_sequence IS NOT NULL") + .bind(id.as_str()) + .bind(run_id) + .fetch_optional(&mut *connection) + .await? + { + sequences.insert(id, DB::number(&row, "workflow_sequence")?); + } } for event in events.iter_mut().filter(|event| event["event_type"] == "SignalReceived") { let sequence = event["payload"]["signal_id"].as_str().and_then(|id| sequences.get(id)).copied(); diff --git a/rust/tests/execution/worker_history_presentation.rs b/rust/tests/execution/worker_history_presentation.rs index a5cafa602..c1f0de28d 100644 --- a/rust/tests/execution/worker_history_presentation.rs +++ b/rust/tests/execution/worker_history_presentation.rs @@ -95,3 +95,69 @@ async fn a_declared_signal_consumed_by_a_condition_wait_keeps_no_sequence() { let (_, described) = request(&app, "GET", "/api/workflows/declared-signals", Value::Null).await; assert_eq!(described["status"], "waiting", "{described}"); } + +// `historyEventsWithSignalArgumentEnvelopes` looks up only the current +// page's signal IDs (`whereIn('id', $signalIds)`), fenced to the run. A run +// with many consumed signals, walked one event per page, hydrates exactly +// each page's own signal, and a record moved to another run is not used. +#[tokio::test] +async fn small_history_pages_hydrate_only_their_own_signals() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + let started = start(&app, "paged-signals").await; + let other = start(&app, "paged-signals-other").await; + let worker = json!({"worker_id":"pager","task_queue":"test","runtime":"php","supported_workflow_types":["echo"], + "supported_activity_types":[]}); + assert_eq!(request(&app, "POST", "/api/worker/register", worker).await.0, StatusCode::CREATED); + // Both runs' first tasks open a wait, so only the signals leave work. + for _ in 0..2 { + let task = poll(&app, "pager", "workflow").await; + assert_eq!(finish_task(&app, &task, wait("many")).await.0, StatusCode::OK); + } + for index in 0..40 { + let (status, body) = request(&app, "POST", &format!("/api/workflows/paged-signals/signal/s{index}"), json!([])).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + } + // Give every recorded signal its own consuming sequence. + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let ids: Vec = received(&history).iter().map(|payload| payload["signal_id"].as_str().unwrap().to_owned()).collect(); + assert_eq!(ids.len(), 40); + for (index, id) in ids.iter().enumerate() { + database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_signal_records SET workflow_sequence={} WHERE id='{id}'", 1000 + index))).await; + } + // One record moves to another run; the page fence must not use it. + let moved = &ids[7]; + database.execute(sqlx::AssertSqlSafe(format!("UPDATE workflow_signal_records SET workflow_run_id='{}' WHERE id='{moved}'", + other["run_id"].as_str().unwrap()))).await; + + let (status, first) = request(&app, "POST", "/api/worker/workflow-tasks/poll", + json!({"worker_id":"pager","task_queue":"test","timeout_seconds":0,"history_page_size":1})).await; + assert_eq!(status, StatusCode::OK, "{first}"); + let task = first["task"].clone(); + assert_eq!(task["run_id"], started["run_id"], "{task}"); + let mut pages = vec![task["history_events"].clone()]; + let mut token = task["next_history_page_token"].clone(); + while let Some(next) = token.as_str().map(str::to_owned) { + let (status, page) = request(&app, "POST", &format!("/api/worker/workflow-tasks/{}/history", task["task_id"].as_str().unwrap()), + json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"], + "next_history_page_token":next,"history_page_size":1})).await; + assert_eq!(status, StatusCode::OK, "{page}"); + assert!(page["history_events"].as_array().unwrap().len() <= 1, "{page}"); + pages.push(page["history_events"].clone()); + token = page["next_history_page_token"].clone(); + } + let mut seen = 0; + for page in &pages { + for payload in received(page) { + let id = payload["signal_id"].as_str().unwrap(); + let index = ids.iter().position(|known| known == id).unwrap(); + if id == moved { + assert!(payload.get("workflow_sequence").is_none(), "another run's record: {payload}"); + } else { + assert_eq!(payload["workflow_sequence"], json!(1000 + index), "{payload}"); + } + seen += 1; + } + } + assert_eq!(seen, 40, "every signal appears on its own page"); +}