diff --git a/docs/rust-server-port-checkpoints/2026-10-23-operator-metrics-redispatch-states.md b/docs/rust-server-port-checkpoints/2026-10-23-operator-metrics-redispatch-states.md new file mode 100644 index 000000000..1a82cab36 --- /dev/null +++ b/docs/rust-server-port-checkpoints/2026-10-23-operator-metrics-redispatch-states.md @@ -0,0 +1,81 @@ +# October 23, operator metrics after PHP redispatches an overdue task + +- **Found by.** On PR #673's parity run 39995620421, the Shared Server + fixtures failed on PostgreSQL 17 and MySQL 8.0. The diff came before + any PHP pass line, so published PHP 2.11.1 failed the + `operator-metrics` contract, not native. The failing stage was + `waiting_after_redispatch`. +- **The race in published PHP.** PHP's repair pass redispatches the + overdue ready task in two steps: + - `TaskWatchdog::recoverExistingTask` raises the task's + `repair_count` and projects the run summary, in one transaction. The + run is marked `repair_needed`. + - `TaskDispatcher::markDispatched` then stamps the dispatch, and + `refreshRunSummary` clears `repair_needed` in a separate step. + + `OperatorMetrics::snapshot` reads runs, then tasks, then activities, + then the backlog, with no shared snapshot. `HealthCheck` takes its own + snapshot after the metrics request. So one response can mix states. +- **The four PHP states.** In order, the overdue task is: + 0. overdue and unrepaired; + 1. overdue, repaired, run marked `repair_needed`; + 2. dispatched, run still marked `repair_needed`; + 3. dispatched, run summary refreshed. + + The old contract accepted states 0, 1 and 3 read consistently. It + refused state 2 (the MySQL 8.0 shape) and a backlog read after the + redispatch while the tasks were read before it (the PostgreSQL 17 + shape). +- **First model, refused in review.** The first correction (head + `3fa5cd5288`) let each section of the response show one state, with + sections never going backwards. Security's review (#708 comment + 6373392835) showed this was not enough. Published Workflow `aca7928484` + reads each value with its own query, even inside one section. So a + forward-only change can split the `tasks` section: `dispatch_overdue` + at state 0, `max_repair_count` at state 1, and `unhealthy` at state 2. + The timestamps are read before the counts, so a timestamp can also + outlive its count. +- **Contract decision.** For published PHP only, the + `waiting_after_redispatch` stage follows PHP's actual query order. The + reads that depend on the repair state are, in order: + - runs: `oldest_repair_needed_at`, then `repair_needed`; + - tasks: `oldest_dispatch_overdue_since` (with `oldest_unhealthy_at`, + which PHP builds from the same timestamps), then `dispatch_overdue`, + then `max_repair_count`, then `unhealthy`, counted again; + - backlog: `unhealthy_tasks`, then `repair_needed_runs`. + + The response passes when some sequence of states that never goes + backwards in that order produces exactly its counts and timestamps. The + health request reads its own snapshot afterwards, starting at or after + the response's last read. Each health warning follows the field it + reads: + - `task_transport` follows `tasks.unhealthy`; + - `routing_health` follows `tasks.dispatch_overdue`; + - `durable_resume_paths` follows `backlog.repair_needed_runs`. + + Native runs no repair loop, so native must still report the overdue + task consistently, with each timestamp present exactly when its count + is. A native response that shows a redispatch must show it as one + settled state. +- **How it was checked.** + - Security's counterexample passes for PHP and fails for native. So + does the timestamp kept from before a redispatch. + - New tests refuse each backward step: a recount that is overdue after + the dispatch count was not; a backlog that is overdue after the + recount was not; an overdue count after its timestamp showed the + redispatch; a run timestamp from a repair the later reads have not + reached; and health whose recount is overdue after its dispatch count + was not. + - An exhaustive test covers all 256 combinations of the eight values + that depend on the state. The 37 combinations that some forward-only + order produces pass; the other 219 fail. + - Two earlier tests assumed one read per section. They now accept + health whose backlog was read after the run was projected again, and + whose dispatch count was read before the redispatch. A native + response still fails both. + - The original CI shapes (MySQL 8.0 state 2, PostgreSQL 17 backlog + after the redispatch) still pass. The server-parity unit tests pass: + 5,502 of 5,504, with one intended skip and one test that needs Docker + inside its container. +- **Not changed.** The fixture, its stages and the other stages' expected + values. Native's behaviour is unchanged. diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 773b6b4e4..582784510 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -14,87 +14,83 @@ qualify a later tree. ## Current state -Updated October 22, 2026. - -- **Merged since October 20.** - - **Backup-first takeover** on SQLite, PostgreSQL and MySQL/MariaDB: - [#523](https://github.com/durable-workflow/server/pull/523), - [#529](https://github.com/durable-workflow/server/pull/529), - [#532](https://github.com/durable-workflow/server/pull/532), - [#533](https://github.com/durable-workflow/server/pull/533), - [#536](https://github.com/durable-workflow/server/pull/536), - [#538](https://github.com/durable-workflow/server/pull/538) and - [#539](https://github.com/durable-workflow/server/pull/539). This covers the - fingerprint, the layouts and WAL preparation. It also covers a PHP-accepted - signal and an expired PHP lease resumed under native, and the refusal of - unsupported features. - - **Multi-node and API parity.** - - [#549](https://github.com/durable-workflow/server/pull/549): the - multi-node kill and restart qualification. - - [#548](https://github.com/durable-workflow/server/pull/548): ownership - guard refusals. - - [#550](https://github.com/durable-workflow/server/pull/550) and - [#551](https://github.com/durable-workflow/server/pull/551): the search - attribute registry and namespace update. - - [#594](https://github.com/durable-workflow/server/pull/594): history - export. - - [#595](https://github.com/durable-workflow/server/pull/595) and - [#596](https://github.com/durable-workflow/server/pull/596): retention and - activity timeouts. - - **The October 21-22 trains.** - [#629](https://github.com/durable-workflow/server/pull/629), - [#638](https://github.com/durable-workflow/server/pull/638), - [#641](https://github.com/durable-workflow/server/pull/641), - [#649](https://github.com/durable-workflow/server/pull/649), - [#659](https://github.com/durable-workflow/server/pull/659) and - [#665](https://github.com/durable-workflow/server/pull/665) covered: - - input envelopes, run repair, legacy settings, activity deadlines, - runtime credentials and principals; - - query and update admission, signals rejected after a run closes, start - fingerprints and all-of barriers; - - cooperative waits, child cancellation and parent-close policies, - continue-as-new messages, history pages, declared signals and build IDs. - - **GHSA-6q48-6c7w-9gw2 trust remediation.** - [#671](https://github.com/durable-workflow/server/pull/671) moves the Rust - locks to the fixed webpki roots. - [#672](https://github.com/durable-workflow/server/pull/672) pins the image's - TLS toolchains and checks every trust store. -- **Open.** - - [#675](https://github.com/durable-workflow/server/pull/675): train of - [#667](https://github.com/durable-workflow/server/pull/667) (definition - routing), [#669](https://github.com/durable-workflow/server/pull/669) (poll - waits), [#670](https://github.com/durable-workflow/server/pull/670) (worker - admission) and [#674](https://github.com/durable-workflow/server/pull/674) - (schedule fixture probe). - - [#676](https://github.com/durable-workflow/server/pull/676): accept - published PHP's health read during or after its own task repair. - - [#673](https://github.com/durable-workflow/server/pull/673): pip, Composer - and Cargo trust in CI and conformance, in Security review. - [#677](https://github.com/durable-workflow/server/issues/677) owns the - remaining clients, with an explicit non-completion boundary. -- **Shared Server fixtures on main.** Embedded passes. Native differs on two - fixtures: - - **`signals-before-worker`:** native records `SignalRejected` after a - terminal event. - - **`worker-listing`:** native reports `_worker_protocol_version` in - `process_metrics`. - - In both, native follows published Server 2.10.0 while the fixtures still - expect the frozen 2.5.14 reference. Separately, the published PHP reference - sometimes differs from itself on `operator-metrics` (postgres-17) because of - its own repair window, which #676 handles. Moving the reference needs a - recorded contract decision, not an edited expectation. -- **Parity surface.** Native serves 97 of published PHP's 175 API routes. That - count compares `php artisan route:list --path=api` on this tree with the - native router. The missing routes are listed below. -- **Next action.** - 1. Verify the Rust suites on #675 and #676, then merge them. - 2. Record the reference decision for Server 2.10.0 and re-record the affected - fixtures from that published release, so main's Shared Server fixtures run - green. - 3. Port the missing families below. - - Packaging the native binary into the server image remains an open gate. +Updated October 24, 2026. + +- **Merged October 24.** Train + [#729](https://github.com/durable-workflow/server/pull/729) (merge + `61219609b`): + - MySQL statement closes are sent with the next command + ([#717](https://github.com/durable-workflow/server/pull/717)), ending the + MySQL-family stalls of [#714](https://github.com/durable-workflow/server/issues/714); + - policy maps that are not JSON objects are refused, like PHP #705, and a + namespace quota map that is not UTF-8 is refused under either name + ([#719](https://github.com/durable-workflow/server/pull/719)); + - worker history shows consumed signal sequences, hydrating only each page's + own signals, plus full task snapshots + ([#710](https://github.com/durable-workflow/server/pull/710)); + - a newer push cancels a superseded Shared Server fixtures run + ([#724](https://github.com/durable-workflow/server/pull/724)). +- **Merged October 20-23.** These are summarized in the checkpoint history: + - backup-first takeover on every database; + - multi-node kill and restart; + - ownership guards, search attributes, namespace update, history export, + retention and activity timeouts; + - the October 21-23 trains; + - the move to the published 2.11.1 reference; + - the GHSA-6q48-6c7w-9gw2 fixes. + + Each train's PR holds its evidence. +- **In review or qualifying, in merge order.** + 1. CI speed and the merge queue: + - [#741](https://github.com/durable-workflow/server/pull/741) runs the + SDK direction profiles in parallel lanes, about 14 minutes off every + database job, and fixes a timestamp-comparison test flake; + - [#742](https://github.com/durable-workflow/server/pull/742) adds the + merge-queue prerequisites (`merge_group` triggers, one fixtures + result, and fixture cells skipped when a change cannot affect them). + 2. This train: operator metrics read across PHP's redispatch steps + ([#708](https://github.com/durable-workflow/server/pull/708)), + query-authority clients that wait for the acknowledgement + ([#720](https://github.com/durable-workflow/server/pull/720)), and the + machine-checked route list below. + 3. Condition waits without a key, task snapshots, the SQLite patch cohort and + message streams (#713, #718, #722, #709). + 4. Update validation with PHP's poll request ID contract + ([#712](https://github.com/durable-workflow/server/pull/712)), then the + held-poll registration fence + ([#739](https://github.com/durable-workflow/server/pull/739), #734) and + worker sessions + ([#740](https://github.com/durable-workflow/server/pull/740), #736), + stacked in that order. + 5. Activity cancellation acknowledgement, and pip, Composer and Cargo trust + in CI ([#673](https://github.com/durable-workflow/server/pull/673)). +- **Shared Server fixtures.** Native, embedded and published PHP 2.11.1 are + compared on SQLite, MySQL 8.0, MariaDB 10.11 and PostgreSQL 17. Where the + published reference itself misbehaves, an explicit boundary separates it + from native, and the PHP owner has an issue for each case: + - `operator-metrics` across PHP's redispatch window (#708); + - the published-SDK patch cohort's SQLite poll refusal + ([#721](https://github.com/durable-workflow/server/issues/721), #722); + - update validation under SQLite write locks + ([#726](https://github.com/durable-workflow/server/issues/726), fixed in + PHP by #730); + - `worker-sdk-persistent-pressure`'s shutdown window + ([#728](https://github.com/durable-workflow/server/issues/728)); + - a retried activity's cooperative cancellation + ([Workflow #923](https://github.com/durable-workflow/workflow/issues/923)); + - a reacquired worker session keeping its old failure reason + ([#737](https://github.com/durable-workflow/server/issues/737)). + + No expectation is edited to make native pass. +- **Parity surface.** Native serves 126 of the 175 API route methods in + published 2.11.1's route table. An execution test sends each one through + the native router, and the ones still answering + `501 rust_capability_not_implemented` must match + [`unserved-php-routes.v2.11.1.txt`](../rust/tests/fixtures/unserved-php-routes.v2.11.1.txt). + The work in review serves 12 more: message streams (3), update validation + (3), worker sessions (5) and activity cancellation acknowledgement (1). +- **Next action.** Land the items above in order, keep porting the unserved + routes fixture first, and package the native binary into the server image. ## API surface still missing @@ -601,7 +597,9 @@ October 21, operator metrics and health: - PHP's maintenance loop redispatches an overdue ready task, after which PHP reports it repaired, not overdue. Native runs no repair loop, so the task stays overdue until a worker claims it. The fixture accepts either - PHP state. + PHP state. Since October 23 it also accepts PHP's reads that straddle + the redispatch, provided they never go backwards; see + [the redispatch-states checkpoint](rust-server-port-checkpoints/2026-10-23-operator-metrics-redispatch-states.md). - The operator dashboards, capacity evidence, and the repair, retention, activity-timeout and payload-cleanup pass endpoints remain. diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs index 96f2db7e4..2915050d2 100644 --- a/rust/tests/execution.rs +++ b/rust/tests/execution.rs @@ -287,6 +287,9 @@ mod external_payloads; #[path = "execution/trace_carriers.rs"] mod trace_carriers; +#[path = "execution/route_coverage.rs"] +mod route_coverage; + enum TestDatabase { Sqlite(tempfile::TempDir), Postgres { diff --git a/rust/tests/execution/activity_deadlines.rs b/rust/tests/execution/activity_deadlines.rs index ea2d4d3a9..c67b5f724 100644 --- a/rust/tests/execution/activity_deadlines.rs +++ b/rust/tests/execution/activity_deadlines.rs @@ -483,9 +483,14 @@ async fn heartbeat_renews_only_progress_deadline_and_cannot_resurrect_elapsed_au assert_eq!(after.0, initial.0, "Start-to-close deadline remains fixed"); assert!(after.1 > initial.1, "Heartbeat deadline advances"); assert_eq!(after.2, initial.2, "Lifecycle deadline remains fixed"); + // Compare instants, not text: a lease on a whole second is written + // without a fraction ("...:26Z"), which sorts after "...:26.045Z". + let instant = |value: &Value| chrono::DateTime::parse_from_rfc3339(value.as_str().unwrap()).unwrap(); assert!( - renewed.1["lease_expires_at"].as_str().unwrap() - > activity["lease_expires_at"].as_str().unwrap() + instant(&renewed.1["lease_expires_at"]) > instant(&activity["lease_expires_at"]), + "{} renews {}", + renewed.1["lease_expires_at"], + activity["lease_expires_at"] ); let prefix = run_history(&app, &started["workflow_id"], &started["run_id"]).await; database diff --git a/rust/tests/execution/route_coverage.rs b/rust/tests/execution/route_coverage.rs new file mode 100644 index 000000000..e49e1ff40 --- /dev/null +++ b/rust/tests/execution/route_coverage.rs @@ -0,0 +1,76 @@ +use super::*; + +// The published Server 2.11.1 API routes native does not serve yet. A port +// that serves a route removes it from this list in the same change, so the +// list in docs/rust-server-port.md cannot drift from the code. +const UNSERVED: &str = include_str!("../fixtures/unserved-php-routes.v2.11.1.txt"); + +/// Sends every method of every published PHP API route through the native +/// router, with each path parameter filled in, and lists those that answer +/// 501 rust_capability_not_implemented. +#[tokio::test] +async fn unserved_published_routes_match_the_recorded_list() { + let database = TestDatabase::new().await; + let runtime = database.open().await.unwrap(); + let app = router(runtime.clone()); + let table: Value = serde_json::from_str(include_str!("../../src/runtime/php-routes.v2.11.1.json")).unwrap(); + let mut unserved = Vec::new(); + let mut checked = 0; + for route in table["routes"].as_array().unwrap() { + let (methods, uri) = (route[0].as_str().unwrap(), route[1].as_str().unwrap()); + if !uri.starts_with("api/") { + continue; + } + for method in methods.split('|').filter(|method| *method != "HEAD") { + checked += 1; + let path = format!("/{}", fill_parameters(uri)); + let worker = path.starts_with("/api/worker/"); + let body = if matches!(method, "GET" | "DELETE") { Body::empty() } else { Body::from("{}") }; + let request = Request::builder() + .method(method) + .uri(&path) + .header("authorization", "Bearer test-token") + .header("accept", "application/json") + .header("content-type", "application/json") + .header( + if worker { "x-durable-workflow-protocol-version" } else { "x-durable-workflow-control-plane-version" }, + if worker { "1.20" } else { "2" }, + ) + .body(body) + .unwrap(); + let response = app.clone().oneshot(request).await.unwrap(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), 8 * 1024 * 1024).await.unwrap(); + if status == StatusCode::NOT_IMPLEMENTED + && String::from_utf8_lossy(&bytes).contains("rust_capability_not_implemented") + { + unserved.push(format!("{method} /{uri}")); + } + } + } + assert!(checked > 170, "the published route table lists {checked} API route methods"); + let recorded: Vec<&str> = UNSERVED.lines().filter(|line| !line.is_empty() && !line.starts_with('#')).collect(); + let now_served: Vec<&&str> = recorded.iter().filter(|route| !unserved.iter().any(|found| found == **route)).collect(); + let now_unserved: Vec<&String> = unserved.iter().filter(|found| !recorded.contains(&found.as_str())).collect(); + assert!( + now_served.is_empty() && now_unserved.is_empty(), + "update rust/tests/fixtures/unserved-php-routes.v2.11.1.txt and the port log. Now served: {now_served:?}. Now unserved: {now_unserved:?}" + ); + assert_eq!(unserved, recorded, "keep the recorded list in route-table order"); + runtime.close().await; + database.remove().await; +} + +/// Replaces each `{parameter}` in a route with a placeholder value. +fn fill_parameters(uri: &str) -> String { + let mut path = String::new(); + let mut rest = uri; + while let Some(open) = rest.find('{') { + let close = open + rest[open..].find('}').expect("closed route parameter"); + path.push_str(&rest[..open]); + path.push_str("probe-x"); + rest = &rest[close + 1..]; + } + path.push_str(rest); + path +} diff --git a/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt b/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt new file mode 100644 index 000000000..886219a91 --- /dev/null +++ b/rust/tests/fixtures/unserved-php-routes.v2.11.1.txt @@ -0,0 +1,52 @@ +# Published Server 2.11.1 API routes that native answers with 501 +# rust_capability_not_implemented, in route-table order. Checked by +# rust/tests/execution/route_coverage.rs. +POST /api/bridge-adapters/webhook/{adapter} +GET /api/service-endpoints +POST /api/service-endpoints +GET /api/service-endpoints/{endpointName} +PUT /api/service-endpoints/{endpointName} +DELETE /api/service-endpoints/{endpointName} +GET /api/service-endpoints/{endpointName}/services +POST /api/service-endpoints/{endpointName}/services +GET /api/service-endpoints/{endpointName}/services/{serviceName} +PUT /api/service-endpoints/{endpointName}/services/{serviceName} +DELETE /api/service-endpoints/{endpointName}/services/{serviceName} +GET /api/service-endpoints/{endpointName}/services/{serviceName}/operations +POST /api/service-endpoints/{endpointName}/services/{serviceName}/operations +GET /api/service-endpoints/{endpointName}/services/{serviceName}/operations/{operationName} +PUT /api/service-endpoints/{endpointName}/services/{serviceName}/operations/{operationName} +DELETE /api/service-endpoints/{endpointName}/services/{serviceName}/operations/{operationName} +POST /api/service-endpoints/{endpointName}/services/{serviceName}/operations/{operationName}/execute +GET /api/service-endpoints/{endpointName}/services/{serviceName}/operations/{operationName}/service-calls/{serviceCallId} +POST /api/service-endpoints/{endpointName}/services/{serviceName}/operations/{operationName}/service-calls/{serviceCallId}/cancel +GET /api/worker-sessions +GET /api/worker-sessions/{sessionId} +POST /api/worker/activity-tasks/{taskId}/acknowledge-cancellation +POST /api/worker/sessions +DELETE /api/worker/sessions/{sessionId} +POST /api/worker/sessions/{sessionId}/heartbeat +POST /api/worker/update-validation-tasks/poll +POST /api/worker/update-validation-tasks/{taskId}/approve +POST /api/worker/update-validation-tasks/{taskId}/reject +POST /api/worker/workflow-tasks/{taskId}/cancellation-scopes/checkpoint +POST /api/worker/workflow-tasks/{taskId}/cancellation-scopes/deliver +POST /api/worker/workflow-tasks/{taskId}/cancellation-scopes/open +POST /api/worker/workflow-tasks/{taskId}/cancellation-scopes/prepare +POST /api/worker/workflow-tasks/{taskId}/local-activities/checkpoint +POST /api/worker/workflow-tasks/{taskId}/local-activities/checkpoint-group +POST /api/worker/workflow-tasks/{taskId}/local-activities/prepare +POST /api/worker/workflow-tasks/{taskId}/local-activities/recover +POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/acknowledge-cancellation +POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/control +POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/heartbeat +POST /api/worker/workflow-tasks/{taskId}/local-activities/{attemptId}/outcome +POST /api/workflows/import/embedded-v2 +POST /api/workflows/import/waterline-v1 +GET /api/workflows/{workflowId}/message-streams +GET /api/workflows/{workflowId}/message-streams/{streamName} +POST /api/workflows/{workflowId}/message-streams/{streamName}/messages +GET /api/workflows/{workflowId}/nexus-operations +GET /api/workflows/{workflowId}/runs/{runId}/nexus-operations +POST /api/workflows/{workflowId}/runs/{runId}/streams/{streamName}/close +POST /api/workflows/{workflowId}/runs/{runId}/streams/{streamName}/items diff --git a/scripts/conformance/server-parity/operator-metrics-contract.mjs b/scripts/conformance/server-parity/operator-metrics-contract.mjs index 69df4275e..b3b0d1d69 100644 --- a/scripts/conformance/server-parity/operator-metrics-contract.mjs +++ b/scripts/conformance/server-parity/operator-metrics-contract.mjs @@ -1,8 +1,8 @@ import assert from 'node:assert/strict'; import {isDeepStrictEqual} from 'node:util'; -// Reviewed operator metrics and health of one namespace, as published PHP -// 2.5.14 reports them through a run's stages. Counts must match exactly, with +// Reviewed operator metrics and health of one namespace, as published Server +// 2.11.1 reports them through a run's stages. Counts must match exactly, with // two normalizations: // - Laravel compares "now" in whole seconds, so a task made available within // the current second counts as delayed. Ready-due and delayed tasks are @@ -70,18 +70,99 @@ const EXPECTED = { backlog: {tasks_added_last_minute: 4}, history: {events: 8, max_event_count: 8}, command_contracts: {backfill_needed_runs: 1, backfill_unavailable_runs: 1, closed_backfill_needed_runs: 1}}, }; -// PHP's maintenance loop redispatches an overdue ready task (TaskRepair), so -// past the redispatch window published PHP may already show the task repaired -// instead of overdue. Native runs no repair loop and keeps reporting it overdue. -const REPAIRED = {...EXPECTED.waiting_after_redispatch, tasks: {open: 1, ready: 1, ready_due_or_delayed: 1}, - backlog: {runnable_or_delayed_tasks: 1, tasks_added_last_minute: 1}}; -const REPAIRED_WARNINGS = ['command_contract_snapshots']; -// PHP's TaskWatchdog repairs the task and projects the run before -// TaskDispatcher redispatches it and projects the run again. Until then the -// task is still overdue and the run reads as needing repair. -const REPAIRING = {...EXPECTED.waiting_after_redispatch, runs: {...waitingRun, repair_needed: 1}, - backlog: {...EXPECTED.waiting_after_redispatch.backlog, repair_needed_runs: 1}}; -const REPAIRING_WARNINGS = ['command_contract_snapshots', 'task_transport', 'routing_health', 'durable_resume_paths']; +// Past the redispatch window, published PHP's maintenance loop moves the +// overdue ready task through four states: +// 0 overdue; +// 1 repairing: TaskWatchdog::recoverExistingTask sets the repair count and +// projects the run as needing repair, in one transaction; +// 2 dispatched: TaskDispatcher::markDispatched saves the task as dispatched, +// and its run is not yet projected again; +// 3 redispatched: the run is projected again. +// The task is overdue in states 0 and 1, carries a repair count from state 1, +// and its run reads as needing repair in states 1 and 2. +// Workflow aca7928484's OperatorMetrics::snapshot reads each value with its +// own query, so one document can see several states, never going backwards +// in query order. Its reads that depend on the state, in that order: +// runMetrics: oldest_repair_needed_at, then repair_needed; +// taskMetrics: oldest_dispatch_overdue_since (oldest_unhealthy_at is built +// from the same timestamps), then dispatch_overdue, then +// max_repair_count, then unhealthy, counted again; +// backlogMetrics: unhealthy_tasks, then repair_needed_runs. +// The health request reads its own snapshot afterwards, in the same order: +// task_transport warns on tasks.unhealthy, routing_health on +// tasks.dispatch_overdue, and durable_resume_paths on +// backlog.repair_needed_runs. Native runs no repair loop: it keeps the task +// overdue, or shows it redispatched once repaired. +const PHP_REPAIR_STATES = [0, 1, 2, 3]; +const PHP_READS = 8; +const overdueAt = state => state <= 1; +const runNeedsRepairAt = state => state === 1 || state === 2; +const runsAt = state => (runNeedsRepairAt(state) ? {...waitingRun, repair_needed: 1} : waitingRun); +const backlogAt = (unhealthy, repairNeeded) => ({runnable_or_delayed_tasks: 1, tasks_added_last_minute: 1, + ...(overdueAt(unhealthy) ? {unhealthy_tasks: 1} : {}), ...(runNeedsRepairAt(repairNeeded) ? {repair_needed_runs: 1} : {})}); +const warningsAt = (task, run) => ['command_contract_snapshots', + ...(overdueAt(task) ? ['task_transport', 'routing_health'] : []), ...(runNeedsRepairAt(run) ? ['durable_resume_paths'] : [])]; +const REDISPATCHED = {...EXPECTED.waiting_after_redispatch, runs: runsAt(3), + tasks: {open: 1, ready: 1, ready_due_or_delayed: 1}, backlog: backlogAt(3, 3)}; +// The timestamps a PHP read may take apart from their counts. +const PHP_STAMPED = {runs: ['oldest_repair_needed_at'], tasks: ['oldest_dispatch_overdue_since', 'oldest_unhealthy_at']}; + +// Every way `count` reads can see the states without going backwards. +function forwardReads(count, from = 0) { + if (count === 0) return [[]]; + return PHP_REPAIR_STATES.filter(state => state >= from) + .flatMap(state => forwardReads(count - 1, state).map(rest => [state, ...rest])); +} +const PHP_DOCUMENT_READS = forwardReads(PHP_READS); + +// What a PHP document whose reads saw these states reports. +function phpDocument([runStamp, runCount, taskStamp, overdue, repairs, unhealthy, backlogUnhealthy, backlogRepair]) { + return { + counts: { + runs: runsAt(runCount), + tasks: {open: 1, ready: 1, ready_due_or_delayed: 1, ...(overdueAt(overdue) ? {dispatch_overdue: 1} : {}), + ...(repairs >= 1 ? {max_repair_count: 1} : {}), ...(overdueAt(unhealthy) ? {unhealthy: 1} : {})}, + backlog: backlogAt(backlogUnhealthy, backlogRepair), + }, + timestamps: { + runs: {oldest_repair_needed_at: runNeedsRepairAt(runStamp)}, + tasks: {oldest_dispatch_overdue_since: overdueAt(taskStamp), oldest_unhealthy_at: overdueAt(taskStamp)}, + }, + }; +} + +// The warnings of a PHP health request whose reads saw these states. +const phpWarnings = reads => ['command_contract_snapshots', ...(overdueAt(reads[5]) ? ['task_transport'] : []), + ...(overdueAt(reads[3]) ? ['routing_health'] : []), ...(runNeedsRepairAt(reads[7]) ? ['durable_resume_paths'] : [])]; + +const nonzero = values => Object.fromEntries(Object.entries(values).filter(([, value]) => value !== 0)); + +// Published PHP past the redispatch window: the expected counts, timestamps +// and the warnings its health may report. `timestamps` holds the presence of +// the PHP_STAMPED fields. The document's reads take the states that match +// and end earliest; the health request's reads start from there. With no +// match, the reads that match the counts alone are expected, so a timestamp +// difference is reported; with none either, the overdue state is expected. +export function phpRedispatchExpectation(counts, timestamps) { + const base = EXPECTED.waiting_after_redispatch; + const seen = {runs: nonzero(counts.runs), tasks: nonzero(counts.tasks), backlog: nonzero(counts.backlog)}; + const earliest = matches => matches.reduce((best, reads) => (best === null || reads[PHP_READS - 1] < best[PHP_READS - 1] ? reads : best), null); + const countMatches = PHP_DOCUMENT_READS.filter(reads => isDeepStrictEqual(phpDocument(reads).counts, seen)); + const best = earliest(countMatches.filter(reads => isDeepStrictEqual(phpDocument(reads).timestamps, timestamps))) + ?? earliest(countMatches); + if (best === null) { + const overdue = phpDocument(Array(PHP_READS).fill(0)); + return {counts: {...base, ...overdue.counts}, timestamps: overdue.timestamps, warnings: [WARNINGS.waiting_after_redispatch]}; + } + const document = phpDocument(best); + const warnings = []; + for (const reads of forwardReads(PHP_READS, best[PHP_READS - 1])) { + const list = phpWarnings(reads); + if (!warnings.some(listed => isDeepStrictEqual(listed, list))) warnings.push(list); + } + return {counts: {...base, ...document.counts}, timestamps: document.timestamps, warnings}; +} + // Checks other than ok at each stage (the uncompared check aside). const WARNINGS = { empty: [], @@ -93,16 +174,6 @@ const WARNINGS = { completed: [], }; -// Past the redispatch window published PHP is in one of these states, in -// order: overdue, repairing, or redispatched. Metrics and health are separate -// requests, so its health may be at the state its metrics show or a later one. -// Native runs no repair loop and is held to the state its metrics show. -const REPAIR_STATES = [ - {counts: EXPECTED.waiting_after_redispatch, warnings: WARNINGS.waiting_after_redispatch}, - {counts: REPAIRING, warnings: REPAIRING_WARNINGS}, - {counts: REPAIRED, warnings: REPAIRED_WARNINGS}, -]; - function normalized(sections) { const counts = {}; for (const [name, section] of Object.entries(sections)) { @@ -135,8 +206,12 @@ function checkStage(name, stage, namespace, family) { assert.deepStrictEqual([stage.metrics_status, stage.metrics_namespace, defect ? namespace : stage.health_namespace, stage.expected_namespace], [200, namespace, namespace, namespace], `${label} namespace`); assert.deepStrictEqual(stage.metrics_keys, ['namespace', 'operator_metrics'], `${label} document`); + const phpRedispatch = name === 'waiting_after_redispatch' && stage.runtime === 'php'; for (const [section, pairs] of Object.entries(TIMESTAMPS)) { for (const [field, count] of Object.entries(pairs)) { + // PHP's repair loop may move between a timestamp's query and its + // count's; the redispatch model checks those timestamps. + if (phpRedispatch && PHP_STAMPED[section]?.includes(field)) continue; assert.equal(stage.sections[section].timestamps[field], stage.sections[section].counts[count] > 0, `${label} ${section}.${field} present exactly when ${count} is not zero`); } @@ -152,19 +227,20 @@ function checkStage(name, stage, namespace, family) { } const counts = normalized(raw); const redispatch = name === 'waiting_after_redispatch'; - const php = redispatch && stage.runtime === 'php'; - const repaired = redispatch && counts.tasks.max_repair_count > 0; - if (repaired) delete counts.tasks.max_repair_count; - const states = redispatch ? REPAIR_STATES : [{counts: EXPECTED[name], warnings: WARNINGS[name]}]; - // A repair count means PHP repaired the task: mid-repair (PHP only) or redispatched. - const possible = !repaired ? [0] : php ? [1, 2] : [2]; - const nonzero = values => Object.fromEntries(Object.entries(values).filter(([, value]) => value !== 0)); - const matches = index => Object.entries(counts) - .every(([section, values]) => isDeepStrictEqual(nonzero(values), states[index].counts[section] ?? {})); - const state = possible.find(matches) ?? possible[possible.length - 1]; + let expected = {counts: EXPECTED[name], warnings: [WARNINGS[name]]}; + let stamps = null; + if (phpRedispatch) { + stamps = Object.fromEntries(Object.entries(PHP_STAMPED).map(([section, fields]) => + [section, Object.fromEntries(fields.map(field => [field, raw[section].timestamps[field]]))])); + expected = phpRedispatchExpectation(counts, stamps); + } else if (redispatch && counts.tasks.max_repair_count > 0) { + delete counts.tasks.max_repair_count; + expected = {counts: REDISPATCHED, warnings: [warningsAt(3, 3)]}; + } for (const [section, values] of Object.entries(counts)) { - assert.deepStrictEqual(nonzero(values), states[state].counts[section] ?? {}, `${label} ${section} counts`); + assert.deepStrictEqual(nonzero(values), expected.counts[section] ?? {}, `${label} ${section} counts`); } + if (stamps !== null) assert.deepStrictEqual(stamps, expected.timestamps, `${label} repair timestamps`); assert.equal(stage.active_workers, ['empty', 'waiting', 'waiting_after_redispatch'].includes(name) ? 0 : 1, `${label} active workers`); if (defect) return counts; @@ -172,8 +248,7 @@ function checkStage(name, stage, namespace, family) { const statuses = stage.checks.filter(check => check.name !== UNCOMPARED_CHECK); assert.ok(statuses.every(check => ['ok', 'warning'].includes(check.status)), `${label} no check errors`); const warnings = statuses.filter(check => check.status !== 'ok').map(check => check.name); - const allowed = (php ? states.slice(state) : [states[state]]).map(candidate => candidate.warnings); - const settled = allowed.find(list => isDeepStrictEqual(list, warnings)) ?? states[state].warnings; + const settled = expected.warnings.find(list => isDeepStrictEqual(list, warnings)) ?? expected.warnings[0]; assert.deepStrictEqual(warnings, settled, `${label} warnings`); const warning = settled.length > 0 || stage.checks.some(check => check.status === 'warning'); assert.deepStrictEqual([stage.health_status, stage.health.status, stage.health.healthy], [200, warning ? 'warning' : 'ok', true], diff --git a/scripts/conformance/server-parity/query-authority/python.py b/scripts/conformance/server-parity/query-authority/python.py index 023aa84cb..e3f0d8b1c 100644 --- a/scripts/conformance/server-parity/query-authority/python.py +++ b/scripts/conformance/server-parity/query-authority/python.py @@ -117,6 +117,17 @@ async def main(): if MODE == 'timely': result = await client.query_workflow(namespace, 'value', args=[]) assert result['result'] == EXPECTED and type(result['result']['long']) is int + # The independent caller can observe the commit before the + # Worker receives its acknowledgement. Wait inside the + # original budget before stopping it, as worker.php does. + PHASE = 'acknowledgement' + while not any(x['event'] == 'acknowledgement' for x in events()): + if runner.done(): + await runner + raise RuntimeError('Worker stopped before its acknowledgement') + if time.monotonic() - started >= 30: + raise AssertionError('No acknowledgement inside the original budget') + await asyncio.sleep(.02) else: try: await client.query_workflow(namespace, 'value', args=[]) @@ -125,12 +136,19 @@ async def main(): assert error.status == 504 and error.reason == 'query_worker_execution_timeout' # Delivery is recorded before the SDK receives its response. # Wait for its actual refusal before stopping the worker. + # A late handler's callback counter moves before the SDK + # logs the lost authority, so every mode also waits for + # that evidence, as worker.php waits for its refusal. PHASE = 'expiry-observation' - while (MODE == 'late-poll' and not any( - expired_losses(event['claim']['query_task_id']) - for event in events() if event['event'] == 'claim') + while (not any(expired_losses(event['claim']['query_task_id']) + for event in events() if event['event'] == 'claim') or MODE == 'late-handler-async' and COUNTERS['cancelled'] != 1 or MODE == 'late-handler-sync' and COUNTERS['returned'] != 1): + if runner.done(): + await runner + raise RuntimeError('Worker stopped before its expiry evidence') + if time.monotonic() - started >= 30: + raise AssertionError('No expiry evidence inside the original budget') await asyncio.sleep(.02) finally: PHASE = 'worker-stop' diff --git a/scripts/conformance/server-parity/query-authority/rust/src/main.rs b/scripts/conformance/server-parity/query-authority/rust/src/main.rs index 45fae759d..d413a6dbc 100644 --- a/scripts/conformance/server-parity/query-authority/rust/src/main.rs +++ b/scripts/conformance/server-parity/query-authority/rust/src/main.rs @@ -169,6 +169,13 @@ async fn main() -> Result<()> { let query = client.query_workflow(&namespace, "value", json!([])).await; if mode == "timely" { assert_eq!(query?, expected()); + // The independent caller can observe the commit before the Worker + // receives its acknowledgement. Wait inside the original budget + // before stopping it, as worker.php does. + wait(until, || { + !named(&events(Path::new(journal)), "acknowledgement").is_empty() + }) + .await?; } else { match query { Err(Error::QueryFailed(failure)) => { diff --git a/tests/Unit/ServerParityOperatorMetricsContractTest.mjs b/tests/Unit/ServerParityOperatorMetricsContractTest.mjs index e855bf2c3..3fe4475d4 100644 --- a/tests/Unit/ServerParityOperatorMetricsContractTest.mjs +++ b/tests/Unit/ServerParityOperatorMetricsContractTest.mjs @@ -39,7 +39,7 @@ const NAMES = ['backend_capabilities', 'run_summary_projection', 'selected_run_p 'command_contract_snapshots', 'task_transport', 'activity_path', 'routing_health', 'durable_resume_paths', 'worker_compatibility', 'scheduler_role', 'long_poll_wake_acceleration']; -// PHP 2.5.14's recorded stages, with PHP's thresholds and sizes. +// Published PHP's recorded stages, with PHP's thresholds and sizes. const WAITING = {runs: {total: 1, current: 1, running: 1, waiting: 1}, tasks: {open: 1, ready: 1, delayed: 1}, backlog: {delayed_tasks: 1, tasks_added_last_minute: 1, tasks_dispatched_last_minute: 1}, starts: {pending_runs: 1, pending_commands: 1}, history: {events: 2, max_event_count: 2}, @@ -179,9 +179,16 @@ test('published PHP health read after the repair its metrics caught midway passe setChecks(midRepair(observation), {task_transport: 'ok', routing_health: 'ok'}); assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection); }); -test('rejects PHP health behind metrics caught mid-repair', () => { +// Health reads its backlog with its own query, after its task counts: the run +// may be projected again in between. +test('published PHP health whose backlog was read after the run was projected again passes', () => { const observation = model(); midRepair(observation); + assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection); +}); +test('rejects PHP health whose recount is overdue after metrics caught mid-repair and a dispatch count past it', () => { + const observation = model(); + setChecks(midRepair(observation), {routing_health: 'ok', durable_resume_paths: 'warning'}); assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /warnings/); }); test('rejects native metrics showing a repair native never runs', () => { @@ -189,6 +196,47 @@ test('rejects native metrics showing a repair native never runs', () => { setChecks(midRepair(observation, 'rust'), {durable_resume_paths: 'warning'}); assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /counts/); }); +// The task saved as redispatched before its run is projected again +// (TaskDispatcher::markDispatched): no longer overdue, and the run still needs +// repair. Seen on #673 run 39995620421, mysql-8.0. +const dispatched = (observation, runtime = 'php') => { + const stage = midRepair(observation, runtime); + Object.assign(stage.sections.tasks.counts, {dispatch_overdue: 0, unhealthy: 0}); + Object.assign(stage.sections.tasks.timestamps, {oldest_dispatch_overdue_since: false, oldest_unhealthy_at: false}); + stage.sections.backlog.counts.unhealthy_tasks = 0; + setChecks(stage, {task_transport: 'ok', routing_health: 'ok', durable_resume_paths: 'warning'}); + return stage; +}; +test('published PHP read after the redispatch and before the run is projected again passes', () => { + const observation = model(); + dispatched(observation); + assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection); +}); +test('rejects native showing a redispatch with a run still needing repair', () => { + const observation = model(); + dispatched(observation, 'rust'); + assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /counts/); +}); +// The tasks section read while overdue, the backlog after the redispatch. +// Seen on #673 run 39995620421, postgres-17. +test('published PHP metrics whose backlog was read after the redispatch pass', () => { + const observation = model(); + const stage = observation.operator_metrics.stages.waiting_after_redispatch; + stage.sections.backlog.counts.unhealthy_tasks = 0; + setChecks(stage, {task_transport: 'ok', routing_health: 'ok'}); + assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection); +}); +test('rejects PHP health still overdue after the backlog showed the redispatch', () => { + const observation = model(); + observation.operator_metrics.stages.waiting_after_redispatch.sections.backlog.counts.unhealthy_tasks = 0; + assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /warnings/); +}); +test('rejects a PHP backlog overdue again after the tasks showed the redispatch', () => { + const observation = model(); + const stage = dispatched(observation); + stage.sections.backlog.counts.unhealthy_tasks = 1; + assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /counts/); +}); test('rejects a run needing repair without a repaired task', () => { const observation = model(); const stage = midRepair(observation); @@ -201,11 +249,97 @@ test('rejects a repair warning at another stage', () => { setChecks(observation.operator_metrics.stages.waiting, {durable_resume_paths: 'warning'}); assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /warnings/); }); -test('rejects a repaired task still reported overdue', () => { +// Repaired and still overdue in tasks, with no run needing repair: only a +// backlog read after the run was projected again explains it, and the health +// read after that cannot still warn about the task. +test('rejects health still overdue after a backlog read past the repair', () => { const observation = model(); observation.operator_metrics.stages.waiting_after_redispatch.sections.tasks.counts.max_repair_count = 1; + assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /warnings/); +}); + +// Workflow aca7928484's taskMetrics reads each value with its own query, so +// the repair loop can move on inside the tasks section. +const straddle = (observation, {runs = 0, stamp, overdue, repaired, unhealthy, backlogUnhealthy, backlogRepair, + runStamp = false, runtime = 'php'}) => { + const stage = observation.operator_metrics.stages.waiting_after_redispatch; + stage.runtime = runtime; + stage.sections.runs.counts.repair_needed = runs; + stage.sections.runs.timestamps.oldest_repair_needed_at = runStamp; + Object.assign(stage.sections.tasks.counts, {dispatch_overdue: overdue, max_repair_count: repaired, unhealthy}); + Object.assign(stage.sections.tasks.timestamps, {oldest_dispatch_overdue_since: stamp, oldest_unhealthy_at: stamp}); + Object.assign(stage.sections.backlog.counts, {unhealthy_tasks: backlogUnhealthy, repair_needed_runs: backlogRepair}); + return stage; +}; +// Security's counterexample on #708: runs at 0, overdue at 0, the repair +// count at 1, the recount and backlog at 2. +const securityStraddle = {stamp: true, overdue: 1, repaired: 1, unhealthy: 0, backlogUnhealthy: 0, backlogRepair: 1}; +test('published PHP tasks read across a repair inside taskMetrics pass', () => { + for (const health of [{task_transport: 'ok', routing_health: 'ok', durable_resume_paths: 'warning'}, + {task_transport: 'ok', routing_health: 'ok'}]) { + const observation = model(); + setChecks(straddle(observation, securityStraddle), health); + assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection); + } +}); +test('rejects native tasks read across a repair', () => { + const observation = model(); + setChecks(straddle(observation, {...securityStraddle, runtime: 'rust'}), + {task_transport: 'ok', routing_health: 'ok', durable_resume_paths: 'warning'}); assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST')); }); +test('published PHP timestamps kept from before a redispatch inside taskMetrics pass', () => { + for (const backlogRepair of [1, 0]) { + const observation = model(); + setChecks(straddle(observation, {stamp: true, overdue: 0, repaired: 1, unhealthy: 0, backlogUnhealthy: 0, backlogRepair}), + {task_transport: 'ok', routing_health: 'ok', durable_resume_paths: backlogRepair ? 'warning' : 'ok'}); + assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection); + } +}); +test('published PHP run timestamp read before the repair its count shows passes', () => { + const observation = model(); + setChecks(straddle(observation, {runs: 1, stamp: true, overdue: 1, repaired: 1, unhealthy: 1, backlogUnhealthy: 1, + backlogRepair: 1}), {durable_resume_paths: 'warning'}); + assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection); +}); +test('published PHP health whose dispatch count was read before the redispatch passes', () => { + const observation = model(); + setChecks(observation.operator_metrics.stages.waiting_after_redispatch, {task_transport: 'ok'}); + assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection); +}); +test('rejects PHP health whose recount is overdue after its dispatch count was not', () => { + const observation = model(); + setChecks(observation.operator_metrics.stages.waiting_after_redispatch, {routing_health: 'ok'}); + assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /warnings/); +}); +test('rejects a PHP recount overdue after the dispatch count showed the redispatch', () => { + const observation = model(); + straddle(observation, {stamp: true, overdue: 0, repaired: 1, unhealthy: 1, backlogUnhealthy: 1, backlogRepair: 1}); + assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /counts/); +}); +test('rejects a PHP backlog overdue after the tasks recount showed the redispatch', () => { + const observation = model(); + straddle(observation, {stamp: true, overdue: 1, repaired: 1, unhealthy: 0, backlogUnhealthy: 1, backlogRepair: 1}); + assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /counts/); +}); +test('rejects a PHP overdue count after its timestamp showed the redispatch', () => { + const observation = model(); + straddle(observation, {stamp: false, overdue: 1, repaired: 0, unhealthy: 1, backlogUnhealthy: 1, backlogRepair: 0}); + assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /repair timestamps/); +}); +test('rejects a PHP run timestamp from a repair its later reads have not reached', () => { + const observation = model(); + straddle(observation, {runStamp: true, stamp: true, overdue: 1, repaired: 0, unhealthy: 1, backlogUnhealthy: 1, + backlogRepair: 0}); + assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /repair timestamps/); +}); +test('rejects a native timestamp without its count', () => { + const observation = model(); + const stage = observation.operator_metrics.stages.waiting_after_redispatch; + stage.runtime = 'rust'; + stage.sections.tasks.timestamps.oldest_dispatch_overdue_since = false; + assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /present exactly when/); +}); // Published PHP on MariaDB answers health with 500 once an activity exists. const mariaDbHealthDefect = (observation, runtime, stages) => { observation.operator_metrics.database_family = 'mariadb'; @@ -249,7 +383,10 @@ for (const [label, mutate] of [ ['a closed run still actionable', s => s.stages.completed.sections.command_contracts.counts.actionable_backfill_needed_runs = 1], ['history events miscounted', s => s.stages.completed.sections.history.counts.events = 7], ['a worker never counted', s => s.stages.activity_running.active_workers = 0], - ['a transport warning missed', s => s.stages.waiting_after_redispatch.checks[5].status = 'ok'], + ['a transport warning missed by native', s => { + s.stages.waiting_after_redispatch.runtime = 'rust'; + s.stages.waiting_after_redispatch.checks[5].status = 'ok'; + }], ['an unexpected warning', s => s.stages.completed.checks[9].status = 'warning'], ['a check error', s => s.stages.empty.checks[0].status = 'error'], ['a missing check', s => s.stages.empty.checks.pop()], @@ -265,3 +402,40 @@ for (const [label, mutate] of [ assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST')); }); } + +// Every document the redispatch stage could report, from the eight values +// that depend on the repair state: exactly those some forward-only sequence +// of PHP's per-query reads produces pass, with health read after the run was +// projected again; every other one is rejected. +test('PHP redispatch documents pass exactly when a forward-only read order produces them', () => { + const overdue = state => state <= 1; + const needsRepair = state => state === 1 || state === 2; + const document = reads => [needsRepair(reads[0]), needsRepair(reads[1]), overdue(reads[2]), overdue(reads[3]), + reads[4] >= 1, overdue(reads[5]), overdue(reads[6]), needsRepair(reads[7])].map(Number).join(''); + const forward = new Set(); + const walk = (reads, from) => { + if (reads.length === 8) { + forward.add(document(reads)); + return; + } + for (let state = from; state <= 3; state++) walk([...reads, state], state); + }; + walk([], 0); + let passed = 0; + for (let bits = 0; bits < 256; bits++) { + const [runStamp, runs, stamp, overdueTasks, repaired, unhealthy, backlogUnhealthy, backlogRepair] = + [...bits.toString(2).padStart(8, '0')].map(Number); + const observation = model(); + const stage = straddle(observation, {runStamp: runStamp === 1, runs, stamp: stamp === 1, overdue: overdueTasks, + repaired, unhealthy, backlogUnhealthy, backlogRepair}); + setChecks(stage, {task_transport: 'ok', routing_health: 'ok', durable_resume_paths: 'ok'}); + const key = [runStamp, runs, stamp, overdueTasks, repaired, unhealthy, backlogUnhealthy, backlogRepair].join(''); + if (forward.has(key)) { + assert.deepStrictEqual(checkOperatorMetrics(fixture, observation, 'TEST'), projection, key); + passed++; + } else { + assert.throws(() => checkOperatorMetrics(fixture, observation, 'TEST'), /counts|repair timestamps/, key); + } + } + assert.equal(passed, forward.size); +});