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 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..7d4c22a89 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-mysql-statement-close.md @@ -0,0 +1,71 @@ +# 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. 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 + `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/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/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md new file mode 100644 index 000000000..63f110abb --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-worker-history-presentation.md @@ -0,0 +1,48 @@ +# October 23, what a worker reads in history + +- **Found by.** Comparing the message-stream fixture's worker polls against + published Server 2.11.1. +- **The gaps.** + - PHP's poll and history-page responses show a consumed signal's + `workflow_sequence` on its `SignalReceived` event, read from the signal's + record (`WorkflowTaskPoller::historyEventsWithSignalArgumentEnvelopes`). + Native showed the event as recorded. + - PHP's task snapshots in history keep every field that is set + (`WorkflowHistoryEvent::taskSnapshot`). Native's left out + `compatibility`, `last_dispatch_attempt_at`, `last_dispatched_at`, + `last_dispatch_error`, `last_claim_failed_at` and `last_claim_error`. +- **What native does now.** + - Worker polls and history pages add `workflow_sequence` to a consumed + signal's `SignalReceived`, when the payload does not have one. The + run's own history route still shows the events as recorded, as PHP's + does. + - Task snapshots keep those fields when they are set. +- **Which signals carry a sequence.** PHP's `applySignal` takes a sequence + only from the signal's own wait. A signal the run's contract declares is + applied without one even while a condition wait is open, and + `markConditionWaitSignalConsumed` then finds it applied, so its record keeps + no sequence. Only an undeclared signal takes the condition wait's sequence. + Native had recorded the condition wait's sequence for declared signals too. + That difference was invisible until worker polls showed the record, and the + published-SDK signal and query cohorts then caught it: their worker history + must equal the run's history prefix. Native now records a sequence only from + the signal's own wait, or from a condition wait for an undeclared signal. +- **How it was checked.** A test runs on SQLite, PostgreSQL, MySQL and + MariaDB: + - a condition wait consumes an undeclared signal; + - the next poll and history page show that wait's sequence, and the + run's history does not; + - the completing task's snapshot has its dispatch stamps. + + A second test applies a declared signal under a condition wait twice. The + worker's history shows no sequence on either `SignalReceived`, and it fails + without the record change. +- **Not changed yet.** Some events PHP records with a task snapshot have + none in native, such as `ConditionWaitOpened` and `SignalWaitOpened`. + That audit is a #325 follow-up. +- **Known difference now visible in snapshots.** PHP stamps a dispatch only + on tasks a control-plane call dispatches, such as a signal's task. Tasks a + workflow task completion creates, such as the next message task, carry + none. Native stamps every new task, so those snapshots show stamps where + PHP's do not. This is the cause of the known `tasks_dispatched_last_minute` + difference too; fixing it is a #325 follow-up. diff --git a/rust/src/config_audit.rs b/rust/src/config_audit.rs index fcbed5eb5..15cf0d2c7 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..1bd3fa84c 100644 --- a/rust/src/runtime/external_config.rs +++ b/rust/src/runtime/external_config.rs @@ -119,14 +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)?; - 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); + let overrides = namespace_overrides(|name| env::var(name))?; Ok(Self { completion_max_bytes: (!completion.is_null()) .then(|| integer(&completion).max(1) as u64), @@ -150,7 +172,7 @@ impl ExternalPayloadConfig { "HARD_MAX_OBJECTS_PER_NAMESPACE", Value::Null, )?, - namespace_overrides: overrides, + namespace_overrides: Value::Object(overrides), ..Self::default() }) } @@ -243,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/mysql_pool_tests.rs b/rust/src/runtime/mysql_pool_tests.rs index a4780a930..20b524bf7 100644 --- a/rust/src/runtime/mysql_pool_tests.rs +++ b/rust/src/runtime/mysql_pool_tests.rs @@ -221,3 +221,246 @@ async fn qualification_cancelled_pool_boundaries_preserve_fresh_reads_and_commit .await .expect("bounded real transaction-boundary qualification"); } + +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>>>, +) { + 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 +/// 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(); + 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. 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. +/// 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() { + 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_chunks( + 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 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-cache clear qualification"); +} diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs index 98d803b40..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() { @@ -1124,11 +1179,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 +1210,13 @@ where .bind(&command_id) .execute(&mut **tx) .await?; + // PHP's `applySignal` takes a sequence only from the signal's own + // wait, so a declared signal applied under a condition wait keeps no + // workflow sequence. Only an undeclared signal takes the condition + // wait's (`markConditionWaitSignalConsumed`). + let record_sequence = if direct || !declared { sequence } else { None }; Self::query("UPDATE workflow_signal_records SET status='applied',workflow_sequence=$1,applied_at=$2,closed_at=$3 WHERE id=$4") - .bind(DB::optional_integer(sequence)).bind(DB::bind_time(applied_at)).bind(DB::bind_time(applied_at)).bind(signal_id).execute(&mut **tx).await?; + .bind(DB::optional_integer(record_sequence)).bind(DB::bind_time(applied_at)).bind(DB::bind_time(applied_at)).bind(signal_id).execute(&mut **tx).await?; let run = Self::query( "SELECT workflow_instance_id,message_cursor_position FROM workflow_runs WHERE id=$1", ) @@ -1798,10 +1857,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 +1925,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 +3626,40 @@ 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(()); + } + // 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 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(); + 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 c2e4b4415..96f2db7e4 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -111,6 +111,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..c1f0de28d --- /dev/null +++ b/rust/tests/execution/worker_history_presentation.rs @@ -0,0 +1,163 @@ +use super::*; + +// Server 2.11.1 `WorkflowTaskPoller::historyEventsWithSignalArgumentEnvelopes` +// and `WorkflowHistoryEvent::taskSnapshot`: the history a worker reads shows a +// consumed signal's workflow sequence, and task snapshots keep their dispatch +// stamps. + +fn wait(key: &str) -> Value { + json!([{"type":"open_condition_wait","condition_key":key, + "condition_definition_fingerprint":format!("sha256:{}", "b".repeat(64))}]) +} + +fn received(history: &Value) -> Vec { + history.as_array().unwrap().iter().filter(|event| event["event_type"] == "SignalReceived") + .map(|event| event["payload"].clone()).collect() +} + +#[tokio::test] +async fn a_worker_reads_the_sequence_that_consumed_a_signal() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + // A run started before any worker declared a contract admits any signal. + let started = start(&app, "presented-signals").await; + let worker = json!({"worker_id":"presenter","task_queue":"test","runtime":"php","supported_workflow_types":["echo"], + "supported_activity_types":[]}); + assert_eq!(request(&app, "POST", "/api/worker/register", worker).await.0, StatusCode::CREATED); + let task = poll(&app, "presenter", "workflow").await; + assert_eq!(finish_task(&app, &task, wait("first")).await.0, StatusCode::OK); + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let opened = history.as_array().unwrap().iter().find(|event| event["event_type"] == "ConditionWaitOpened").unwrap(); + let sequence = opened["payload"]["sequence"].clone(); + + let (status, body) = request(&app, "POST", "/api/workflows/presented-signals/signal/go", json!([])).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let task = poll(&app, "presenter", "workflow").await; + assert!(received(&task["history_events"])[0].get("workflow_sequence").is_none(), "not consumed yet: {task}"); + assert_eq!(finish_task(&app, &task, wait("second")).await.0, StatusCode::OK); + + let (status, body) = request(&app, "POST", "/api/workflows/presented-signals/signal/again", json!([])).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let task = poll(&app, "presenter", "workflow").await; + let signals = received(&task["history_events"]); + assert_eq!(signals[0]["workflow_sequence"], sequence, "the consuming wait's sequence: {signals:?}"); + assert!(signals[1].get("workflow_sequence").is_none(), "{signals:?}"); + // A history page reads the same. + let (status, page) = request(&app, "POST", &format!("/api/worker/workflow-tasks/{}/history", task["task_id"].as_str().unwrap()), + json!({"lease_owner":task["lease_owner"],"workflow_task_attempt":task["workflow_task_attempt"]})).await; + assert_eq!(status, StatusCode::OK, "{page}"); + assert_eq!(received(&page["history_events"])[0]["workflow_sequence"], sequence, "{page}"); + // The run's own history shows the event as recorded. + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + assert!(received(&history)[0].get("workflow_sequence").is_none(), "{history}"); + + // The completing task was dispatched when the signal created it, and its + // snapshot keeps the dispatch stamps. + let done = json!([{"type":"complete_workflow","result":envelope(Payload::Null)}]); + assert_eq!(finish_task(&app, &task, done).await.0, StatusCode::OK); + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let completed = history.as_array().unwrap().iter().find(|event| event["event_type"] == "WorkflowCompleted").unwrap(); + for stamp in ["last_dispatch_attempt_at", "last_dispatched_at"] { + assert!(completed["payload"]["task"][stamp].is_string(), "{stamp}: {completed}"); + } + assert!(completed["payload"]["task"].get("last_claim_error").is_none(), "unset fields stay out: {completed}"); +} + +// Workflow 2.10.2 `DefaultWorkflowTaskBridge::applySignal` takes a workflow +// sequence only from the signal's own wait. A signal the run's contract +// declares is applied without one even while a condition wait is open, so +// `markConditionWaitSignalConsumed` finds it applied and its record keeps no +// sequence: the history a worker reads matches the run's history. +#[tokio::test] +async fn a_declared_signal_consumed_by_a_condition_wait_keeps_no_sequence() { + let database = TestDatabase::new().await; + let app = router(database.open().await.unwrap()); + update_worker(&app, "declared-presenter").await; + let started = start(&app, "declared-signals").await; + let task = poll(&app, "declared-presenter", "workflow").await; + assert_eq!(finish_task(&app, &task, wait("first")).await.0, StatusCode::OK); + + let finish = json!({"input":envelope(Payload::Array(vec![Payload::Long(7)]))}); + let (status, body) = request(&app, "POST", "/api/workflows/declared-signals/signal/finish", finish.clone()).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let task = poll(&app, "declared-presenter", "workflow").await; + assert_eq!(finish_task(&app, &task, wait("second")).await.0, StatusCode::OK); + let (status, body) = request(&app, "POST", "/api/workflows/declared-signals/signal/finish", finish).await; + assert_eq!(status, StatusCode::ACCEPTED, "{body}"); + let task = poll(&app, "declared-presenter", "workflow").await; + + let history = run_history(&app, &started["workflow_id"], &started["run_id"]).await; + let polled = received(&task["history_events"]); + assert_eq!(polled.len(), 2, "{task}"); + assert!(polled.iter().all(|signal| signal.get("workflow_sequence").is_none()), "{polled:?}"); + let recorded = received(&history); + assert_eq!(polled[0]["signal_id"], recorded[0]["signal_id"]); + let (_, described) = request(&app, "GET", "/api/workflows/declared-signals", Value::Null).await; + assert_eq!(described["status"], "waiting", "{described}"); +} + +// `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"); +} diff --git a/rust/vendor/sqlx-mysql/DURABLE-PATCH.md b/rust/vendor/sqlx-mysql/DURABLE-PATCH.md index f08664b46..f4792828c 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,49 @@ 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. 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 +`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..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 { @@ -105,13 +109,13 @@ 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(()) + // 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 e6aa8b48c..c16a21eb0 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,13 +59,27 @@ 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> { + /// 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. + if !self.socket.write_buffer().is_empty() && !self.queued_unanswered { + self.socket.flush().await?; + } while !self.waiting.is_empty() { while self.waiting.front() == Some(&Waiting::Row) { @@ -108,10 +125,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 +246,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, } } } 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/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); 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]}`); + } + } +});