From c4e05f65362f6e6bde843abc6075f5d6b64bd0cb Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 18:18:59 -0600 Subject: [PATCH 1/3] Project bounded schedule counters and namespace-bound current run in list --- docs/schedule-list-scalar-summary.md | 66 +++++ docs/server-reference.md | 4 + rust/src/runtime/schedule_list.rs | 369 +++++++++++++++++++++++++-- 3 files changed, 422 insertions(+), 17 deletions(-) create mode 100644 docs/schedule-list-scalar-summary.md diff --git a/docs/schedule-list-scalar-summary.md b/docs/schedule-list-scalar-summary.md new file mode 100644 index 000000000..228acd07f --- /dev/null +++ b/docs/schedule-list-scalar-summary.md @@ -0,0 +1,66 @@ +# Schedule list scalar summary draft + +This is the smaller source slice selected in Server #1014 comment 6644878850. +It is not the full last-dispatch outcome contract or published capability. Waterline +#222 keeps the existing describe fallback. No faster polling is justified by this +partial projection, and describe remains unchanged. + +Each list item adds: + +| Field | Complete value | Unavailable or malformed storage | +| --- | --- | --- | +| `failures_count` | Nonnegative stored signed-64-bit-decodable integer | Decode/negative value fails the request | +| `skipped_trigger_count` | Same counter contract | Same failure | +| `latest_workflow_instance_id` | Stored identity or null, at most 191 Unicode scalars / 764 UTF-8 bytes | Empty/oversized/invalid UTF-8 fails; a pruned instance preserves this identity | +| `current_run` | Nullable `{run_id,status,status_bucket,is_terminal}` from that instance's current run | Missing/pruned/wrong-namespace/wrong-instance target is null, meaning unknown | + +The instance and run joins require exact identities and exact schedule namespace +on both sides. MySQL/MariaDB use binary comparisons to avoid case/accent/trailing- +space collation matches. A current run ID and the instance's complete current-run +pointer have at most 26 Unicode scalars / 104 UTF-8 bytes. Status is a complete +value at most 32 UTF-8 bytes and must be one of the existing statuses: +`pending`, `running`, `waiting` map to running/nonterminal; `completed` maps to +completed/terminal; `failed`, `cancelled`, `terminated` map to failed/terminal. +An unknown or empty status is invalid storage, never a default healthy state. +The dispatch count is not a run status and the current run is not a last-dispatch +outcome. Missing current-run information does not imply completion or no action. + +For newly projected scalars, SQL transfers the **whole** value only within its +byte bound and returns an independent invalid flag otherwise. Rust checks the +complete decoded scalar count and enum before returning a response. No prefix +can become a plausible identity or status. NUL and astral scalars remain exact. +Malformed/oversized scalar metadata refuses the page with the existing typed +HTTP 503 `storage_unavailable` envelope; values are not included in diagnostics. +Backend decode failures retain the existing storage-unavailable response. + +| SQL boundary | SQLite | PostgreSQL | MySQL / MariaDB | +| --- | --- | --- | --- | +| Byte size including NUL | `length(CAST(column AS BLOB))` | `OCTET_LENGTH(column)` | `OCTET_LENGTH(column)` | +| Scalar projection | `CASE WHEN bytes<=bound THEN column ELSE NULL END` | Same | Same | +| Invalid flag | Non-null and empty or over-bound, separately from SQL null | Same | Same | +| Namespace / join identity | Exact text comparison | Exact text comparison | `BINARY left=BINARY right` | + +The page selects explicit existing list columns and bounded scalar expressions. +It excludes `recent_actions`, `buffered_actions`, `last_skip_reason` and +`last_skipped_at` entirely. Recent/last-action, buffer and last-skip fields remain +**deferred**, including their malformed-member, Unicode and timestamp semantics. +No SQL date conversion, substring preview, JSON prefix or invented event ordering +is introduced. The existing legacy action/input/spec/note/search-attribute fields +keep their previous size surface; this does not claim a bounded whole response +or constant database bytes/CPU. + +The one-to-one joins add no row-level calls. Existing request normalization, +closed grammar, bound values, filter fingerprint, signed cursor, stale-anchor +check, order, page size and plus-one lookahead stay intact. Cursor probes select +only schedule ID/created-at and retain the same filters. A plain page uses one +query; either cursor presence or registered-definition lookup adds one; both use +three. Registered attributes still have the existing native unsupported-value +behavior, outside this slice. + +Source vectors cover a 50-row page and its continuation with/without a system +query (1/2/2/3 calls), malformed excluded arrays, independent counters, missing +and wrong-namespace links, all seven statuses, pruned targets, exact NUL/astral +identities and empty/scalar-count/byte-count failures. Normal source checks and +Server review must qualify the exact SQL/decoder on each backend before this +can be called contract-compliant. No new runtime/image/qualification dispatch, +schema/UDF, persisted summary, release or frozen-artifact admission follows. diff --git a/docs/server-reference.md b/docs/server-reference.md index 63696884b..008b90770 100644 --- a/docs/server-reference.md +++ b/docs/server-reference.md @@ -1203,6 +1203,10 @@ heartbeat, complete, and fail responses include `run_closed_reason` and - `POST /api/schedules/{id}/trigger` — Trigger immediately - `POST /api/schedules/{id}/backfill` — Backfill missed runs +The native development [scalar summary draft](schedule-list-scalar-summary.md) adds +counters, latest-instance identity and a nullable current-run summary. Full last- +dispatch/buffer/skip summaries and published capability remain pending Server #1014. + Schedule listing is namespace-scoped and never returns deleted schedules. The optional `status` (`active` or `paused`) and `workflow_type` filters are exact matches. `query` accepts equality predicates joined by `AND` for diff --git a/rust/src/runtime/schedule_list.rs b/rust/src/runtime/schedule_list.rs index ee4f90577..d1ca577cc 100644 --- a/rust/src/runtime/schedule_list.rs +++ b/rust/src/runtime/schedule_list.rs @@ -59,15 +59,15 @@ impl Column { /// The SQL expression Laravel compiles for this column. fn sql(self, backend: &str) -> &'static str { match (self, backend) { - (Self::ScheduleId, _) => "schedule_id", - (Self::Status, _) => "status", - (Self::Note, _) => "note", - (Self::WorkflowType, "SQLite") => "json_extract(action,'$.\"workflow_type\"')", - (Self::TaskQueue, "SQLite") => "json_extract(action,'$.\"task_queue\"')", - (Self::WorkflowType, "PostgreSQL") => "(action->>'workflow_type')", - (Self::TaskQueue, "PostgreSQL") => "(action->>'task_queue')", - (Self::WorkflowType, _) => "JSON_UNQUOTE(JSON_EXTRACT(action,'$.\"workflow_type\"'))", - (Self::TaskQueue, _) => "JSON_UNQUOTE(JSON_EXTRACT(action,'$.\"task_queue\"'))", + (Self::ScheduleId, _) => "s.schedule_id", + (Self::Status, _) => "s.status", + (Self::Note, _) => "s.note", + (Self::WorkflowType, "SQLite") => "json_extract(s.action,'$.\"workflow_type\"')", + (Self::TaskQueue, "SQLite") => "json_extract(s.action,'$.\"task_queue\"')", + (Self::WorkflowType, "PostgreSQL") => "(s.action->>'workflow_type')", + (Self::TaskQueue, "PostgreSQL") => "(s.action->>'task_queue')", + (Self::WorkflowType, _) => "JSON_UNQUOTE(JSON_EXTRACT(s.action,'$.\"workflow_type\"'))", + (Self::TaskQueue, _) => "JSON_UNQUOTE(JSON_EXTRACT(s.action,'$.\"task_queue\"'))", } } } @@ -529,6 +529,101 @@ pub(crate) struct Selection { predicates: Vec, } +// Preserve existing list fields without transferring either stored action array. +const LIST_COLUMNS: &[&str] = &[ + "schedule_id", + "status", + "action", + "next_fire_at", + "last_fired_at", + "overlap_policy", + "note", + "namespace", + "spec", + "search_attributes", + "created_at", + "updated_at", + "fires_count", + "jitter_seconds", + "max_runs", + "remaining_actions", + "failures_count", + "skipped_trigger_count", +]; + +fn byte_length(backend: &str, column: &str) -> String { + if backend == "SQLite" { + format!("length(CAST({column} AS BLOB))") + } else { + format!("OCTET_LENGTH({column})") + } +} + +fn exact_identity(backend: &str, left: &str, right: &str) -> String { + if backend == "MySQL" { + format!("BINARY {left}=BINARY {right}") + } else { + format!("{left}={right}") + } +} + +fn list_selection(backend: &str, cursor_probe: bool) -> String { + if cursor_probe { + return "SELECT s.schedule_id,s.created_at FROM workflow_schedules s".into(); + } + let mut columns = LIST_COLUMNS + .iter() + .map(|name| format!("s.{name}")) + .collect::>(); + let scalars = [ + ( + "s.latest_workflow_instance_id", + "latest_workflow_instance_id", + 764, + ), + ("i.current_run_id", "summary_current_run_pointer", 104), + ("r.id", "summary_run_id", 104), + ("r.status", "summary_run_status", 32), + ]; + let mut invalid = Vec::new(); + for (column, alias, limit) in scalars { + let length = byte_length(backend, column); + columns.push(format!( + "CASE WHEN {length}<={limit} THEN {column} ELSE NULL END AS {alias}" + )); + invalid.push(format!( + "({column} IS NOT NULL AND ({length}=0 OR {length}>{limit}))" + )); + } + columns.push(format!( + "CASE WHEN {} THEN 1 ELSE 0 END AS summary_invalid", + invalid.join(" OR ") + )); + let instance = exact_identity(backend, "i.id", "s.latest_workflow_instance_id"); + let instance_namespace = exact_identity(backend, "i.namespace", "s.namespace"); + let run = exact_identity(backend, "r.id", "i.current_run_id"); + let run_instance = exact_identity(backend, "r.workflow_instance_id", "i.id"); + let run_namespace = exact_identity(backend, "r.namespace", "s.namespace"); + format!( + "SELECT {} FROM workflow_schedules s LEFT JOIN workflow_instances i ON {instance} AND {instance_namespace} LEFT JOIN workflow_runs r ON {run} AND {run_instance} AND {run_namespace}", + columns.join(",") + ) +} + +fn invalid_summary() -> RuntimeError { + super::refuse(StatusCode::SERVICE_UNAVAILABLE, "storage_unavailable") +} + +fn summary_identifier(value: Option, max_characters: usize) -> Result> { + if value + .as_ref() + .is_some_and(|value| value.is_empty() || value.chars().count() > max_characters) + { + return Err(invalid_summary()); + } + Ok(value) +} + enum Bind { Text(String), Time(DateTime), @@ -565,12 +660,21 @@ where limit: i64, ) -> Result> { // A native schedule never holds search attribute values. - if selection.predicates.iter().any(|predicate| predicate.column.is_none()) { + if selection + .predicates + .iter() + .any(|predicate| predicate.column.is_none()) + { return Ok(Vec::new()); } let mysql = DB::NAME == "MySQL"; let mut binds = Vec::new(); - let mut sql = String::from("SELECT * FROM workflow_schedules WHERE namespace="); + let mut sql = list_selection(DB::NAME, only.is_some()); + sql.push_str(if mysql { + " WHERE BINARY s.namespace=BINARY " + } else { + " WHERE s.namespace=" + }); let mut placeholder = |sql: &mut String, bind: Bind| { binds.push(bind); if mysql { @@ -580,7 +684,7 @@ where } }; placeholder(&mut sql, Bind::Text(self.namespace.to_string())); - sql.push_str(" AND status!='deleted'"); + sql.push_str(" AND s.status!='deleted'"); let mut equal = |sql: &mut String, column: Column, value: String| { sql.push_str(" AND "); sql.push_str(column.sql(DB::NAME)); @@ -595,7 +699,11 @@ where } for predicate in &selection.predicates { let (column, _) = predicate.column.expect("system field"); - equal(&mut sql, column, predicate.literal.as_str().unwrap_or_default().to_owned()); + equal( + &mut sql, + column, + predicate.literal.as_str().unwrap_or_default().to_owned(), + ); } if let Some(schedule_id) = only { equal(&mut sql, Column::ScheduleId, schedule_id.to_owned()); @@ -604,15 +712,17 @@ where let time = NaiveDateTime::parse_from_str(created, "%Y-%m-%d %H:%M:%S%.6f") .map(|time| time.and_utc()) .unwrap_or_default(); - sql.push_str(" AND (created_at<"); + sql.push_str(" AND (s.created_at<"); placeholder(&mut sql, Bind::Time(time)); - sql.push_str(" OR (created_at="); + sql.push_str(" OR (s.created_at="); placeholder(&mut sql, Bind::Time(time)); - sql.push_str(" AND schedule_id>"); + sql.push_str(" AND s.schedule_id>"); placeholder(&mut sql, Bind::Text(schedule_id.to_owned())); sql.push_str("))"); } - sql.push_str(&format!(" ORDER BY created_at DESC,schedule_id LIMIT {limit}")); + sql.push_str(&format!( + " ORDER BY s.created_at DESC,s.schedule_id LIMIT {limit}" + )); let mut query = sqlx::query::(AssertSqlSafe(sql)); for bind in binds { query = match bind { @@ -623,6 +733,38 @@ where Ok(query.fetch_all(&self.pool).await?) } + fn list_summary(row: &DB::Row) -> Result { + if DB::number(row, "summary_invalid")? != 0 { + return Err(invalid_summary()); + } + let latest = summary_identifier( + DB::optional_string(row, "latest_workflow_instance_id")?, + 191, + )?; + // Validate the complete bounded pointer even when its target was pruned. + summary_identifier(DB::optional_string(row, "summary_current_run_pointer")?, 26)?; + let run = summary_identifier(DB::optional_string(row, "summary_run_id")?, 26)?; + let status = DB::optional_string(row, "summary_run_status")?; + let current = match (run, status) { + (None, None) => Value::Null, + (Some(run), Some(status)) => { + let bucket = + super::visibility::status_bucket(&status).map_err(|_| invalid_summary())?; + json!({"run_id":run,"status":status,"status_bucket":bucket,"is_terminal":bucket!="running"}) + } + _ => return Err(invalid_summary()), + }; + let failures = DB::number(row, "failures_count")?; + let skipped = DB::number(row, "skipped_trigger_count")?; + if failures < 0 || skipped < 0 { + return Err(invalid_summary()); + } + Ok( + json!({"failures_count":failures,"skipped_trigger_count":skipped, + "latest_workflow_instance_id":latest,"current_run":current}), + ) + } + /// `ScheduleController::formatListItem`. fn list_item(row: &DB::Row) -> Result { let status = DB::string(row, "status")?; @@ -638,6 +780,10 @@ where "search_attributes":DB::optional_document_row(row,"search_attributes")?, "created_at":time("created_at")?,"updated_at":time("updated_at")?, "fires_count":DB::number(row,"fires_count")?}); + let summary = Self::list_summary(row)?; + item.as_object_mut() + .expect("list item") + .extend(summary.as_object().expect("summary").clone()); let jitter = DB::number(row, "jitter_seconds")?; if jitter > 0 { item["jitter_seconds"] = json!(jitter); @@ -814,4 +960,193 @@ mod tests { assert_eq!(query_input(Some("status=a&status=b&x[]=1&x[]=2&y[k]=v")), vec![ ("status".into(), json!("b")), ("x".into(), json!(["1", "2"])), ("y".into(), json!({"k":"v"}))]); } + async fn add_schedule(store: &Store, schedule: &str) { + sqlx::query("INSERT INTO workflow_schedules(id,schedule_id,namespace,action,spec,created_at,updated_at) VALUES (?,?,?,'{\"workflow_type\":\"echo\",\"task_queue\":\"test\"}','{}','2026-10-21 05:00:00.000000','2026-10-21 05:00:00.000000')") + .bind(schedule).bind(schedule).bind("default").execute(&store.pool).await.unwrap(); + } + + async fn summary_store() -> (tempfile::TempDir, Store) { + let directory = tempfile::tempdir().unwrap(); + let file = directory.path().join("schedule-summary.sqlite"); + let pool = super::super::sqlite::open(file.to_str().unwrap()) + .await + .unwrap(); + (directory, Store::new(pool)) + } + + #[tokio::test] + async fn list_summary_has_complete_scalars_and_separate_namespace_bound_current_run() { + let (_directory, store) = summary_store().await; + add_schedule(&store, "summary").await; + sqlx::query("UPDATE workflow_schedules SET failures_count=2,skipped_trigger_count=3,latest_workflow_instance_id='instance',recent_actions='not JSON',buffered_actions='not JSON',last_skip_reason=? WHERE schedule_id='summary'") + .bind("unread deferred reason".repeat(10000)).execute(&store.pool).await.unwrap(); + let missing = store.list_schedules(None, "fixture-secret").await.unwrap(); + let item = &missing["schedules"][0]; + assert_eq!(item["failures_count"], 2); + assert_eq!(item["skipped_trigger_count"], 3); + assert_eq!(item["latest_workflow_instance_id"], "instance"); + assert!( + item["current_run"].is_null(), + "missing is unknown, never completed or healthy" + ); + for field in [ + "recent_actions", + "buffered_actions", + "last_action", + "last_skip_reason", + "last_skipped_at", + ] { + assert!(item.get(field).is_none(), "{field} is deferred"); + } + let rows = store + .select_schedules( + &Selection { + status: None, + workflow_type: None, + predicates: vec![], + }, + None, + None, + 50, + ) + .await + .unwrap(); + use sqlx::Row; + assert!(rows[0].try_column("recent_actions").is_err()); + assert!(rows[0].try_column("buffered_actions").is_err()); + sqlx::query("INSERT INTO workflow_instances(id,workflow_class,workflow_type,namespace,current_run_id) VALUES ('instance','Echo','echo','other','run')") + .execute(&store.pool).await.unwrap(); + sqlx::query("INSERT INTO workflow_runs(id,workflow_instance_id,run_number,workflow_class,workflow_type,namespace,status) VALUES ('run','instance',1,'Echo','echo','other','running')") + .execute(&store.pool).await.unwrap(); + assert!(store.list_schedules(None,"fixture-secret").await.unwrap()["schedules"][0]["current_run"].is_null()); + sqlx::query("UPDATE workflow_instances SET namespace='default'") + .execute(&store.pool) + .await + .unwrap(); + assert!(store.list_schedules(None,"fixture-secret").await.unwrap()["schedules"][0]["current_run"].is_null()); + sqlx::query("UPDATE workflow_runs SET namespace='default'") + .execute(&store.pool) + .await + .unwrap(); + for (status, bucket, terminal) in [ + ("pending", "running", false), + ("running", "running", false), + ("waiting", "running", false), + ("completed", "completed", true), + ("failed", "failed", true), + ("cancelled", "failed", true), + ("terminated", "failed", true), + ] { + sqlx::query("UPDATE workflow_runs SET status=?") + .bind(status) + .execute(&store.pool) + .await + .unwrap(); + assert_eq!( + store.list_schedules(None, "fixture-secret").await.unwrap()["schedules"][0]["current_run"], + json!({"run_id":"run","status":status,"status_bucket":bucket,"is_terminal":terminal}) + ); + } + sqlx::query("UPDATE workflow_runs SET status='unknown-storage-status'").execute(&store.pool).await.unwrap(); + assert!(store.list_schedules(None,"fixture-secret").await.is_err()); + sqlx::query("UPDATE workflow_runs SET status='running'").execute(&store.pool).await.unwrap(); + sqlx::query("DELETE FROM workflow_runs") + .execute(&store.pool) + .await + .unwrap(); + assert!(store.list_schedules(None,"fixture-secret").await.unwrap()["schedules"][0]["current_run"].is_null()); + store.close().await; + } + + #[tokio::test] + async fn list_summary_refuses_complete_malformed_or_oversized_scalars_without_lossy_prefixes() { + let (_directory, store) = summary_store().await; + add_schedule(&store, "scalar").await; + let exact = "ab\0cd🙂"; + sqlx::query("UPDATE workflow_schedules SET latest_workflow_instance_id=?") + .bind(exact) + .execute(&store.pool) + .await + .unwrap(); + assert_eq!( + store.list_schedules(None, "fixture-secret").await.unwrap()["schedules"][0]["latest_workflow_instance_id"], + exact + ); + for invalid in [String::new(), "x".repeat(192), "🙂".repeat(192)] { + sqlx::query("UPDATE workflow_schedules SET latest_workflow_instance_id=?") + .bind(invalid) + .execute(&store.pool) + .await + .unwrap(); + assert!(matches!( + store.list_schedules(None, "fixture-secret").await, + Err(RuntimeError::Refused { + status: StatusCode::SERVICE_UNAVAILABLE, + reason: "storage_unavailable" + }) + )); + } + sqlx::query("UPDATE workflow_schedules SET latest_workflow_instance_id=CAST(X'FF' AS TEXT)").execute(&store.pool).await.unwrap(); + assert!(store.list_schedules(None,"fixture-secret").await.is_err()); + sqlx::query( + "UPDATE workflow_schedules SET latest_workflow_instance_id=NULL,failures_count=-1", + ) + .execute(&store.pool) + .await + .unwrap(); + assert!(store.list_schedules(None, "fixture-secret").await.is_err()); + store.close().await; + } + + #[tokio::test] + async fn list_summary_query_cost_is_constant_per_page_with_existing_cursor_and_filter_guards() { + use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }; + let (directory, store) = summary_store().await; + for index in 0..51 { + add_schedule(&store, &format!("page-{index:03}")).await; + } + store.close().await; + let calls = Arc::new(AtomicUsize::new(0)); + let measured = Arc::clone(&calls); + let pool = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .before_acquire(move |_, _| { + measured.fetch_add(1, Ordering::SeqCst); + Box::pin(async { Ok(true) }) + }) + .connect_with( + sqlx::sqlite::SqliteConnectOptions::new() + .filename(directory.path().join("schedule-summary.sqlite")), + ) + .await + .unwrap(); + let store = Store::::new(pool); + for (query, expected) in [ + ("page_size=50", 1), + ("page_size=50&query=Status%20%3D%20%27active%27", 2), + ] { + calls.store(0, Ordering::SeqCst); + let first = store + .list_schedules(Some(query), "fixture-secret") + .await + .unwrap(); + assert_eq!(first["schedule_count"], 50); + assert_eq!(calls.load(Ordering::SeqCst), expected); + let token = first["next_page_token"].as_str().unwrap(); + calls.store(0, Ordering::SeqCst); + let second = store + .list_schedules( + Some(&format!("{query}&next_page_token={token}")), + "fixture-secret", + ) + .await + .unwrap(); + assert_eq!(second["schedule_count"], 1); + assert_eq!(calls.load(Ordering::SeqCst), expected + 1); + } + store.close().await; + } } From 649ad9cb4759c58fc22dd94d420e018ca62dbc6e Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 4 Nov 2026 18:33:57 -0600 Subject: [PATCH 2/3] Map summary counter overflow and add native backend decoder and join controls --- docs/schedule-list-scalar-summary.md | 21 +++-- rust/src/runtime/mysql.rs | 10 +++ rust/src/runtime/postgres.rs | 10 +++ rust/src/runtime/schedule_list.rs | 120 +++++++++++++++++++++++++-- 4 files changed, 147 insertions(+), 14 deletions(-) diff --git a/docs/schedule-list-scalar-summary.md b/docs/schedule-list-scalar-summary.md index 228acd07f..38b643c48 100644 --- a/docs/schedule-list-scalar-summary.md +++ b/docs/schedule-list-scalar-summary.md @@ -15,8 +15,9 @@ Each list item adds: | `current_run` | Nullable `{run_id,status,status_bucket,is_terminal}` from that instance's current run | Missing/pruned/wrong-namespace/wrong-instance target is null, meaning unknown | The instance and run joins require exact identities and exact schedule namespace -on both sides. MySQL/MariaDB use binary comparisons to avoid case/accent/trailing- -space collation matches. A current run ID and the instance's complete current-run +on both sides. MySQL/MariaDB retain a candidate equality usable by each existing index, then +apply an exact binary residual to avoid case/accent/trailing-space collation +matches. The page namespace predicate uses the same candidate/residual pattern. A current run ID and the instance's complete current-run pointer have at most 26 Unicode scalars / 104 UTF-8 bytes. Status is a complete value at most 32 UTF-8 bytes and must be one of the existing statuses: `pending`, `running`, `waiting` map to running/nonterminal; `completed` maps to @@ -28,7 +29,9 @@ outcome. Missing current-run information does not imply completion or no action. For newly projected scalars, SQL transfers the **whole** value only within its byte bound and returns an independent invalid flag otherwise. Rust checks the complete decoded scalar count and enum before returning a response. No prefix -can become a plausible identity or status. NUL and astral scalars remain exact. +can become a plausible identity or status. NUL and astral scalars remain exact in SQLite/MySQL. PostgreSQL text cannot +store NUL, which is an unsupported storage-domain cell, not a decoder pass. +Invalid UTF-8 setup limitations remain distinct from SQLite decoder refusal. Malformed/oversized scalar metadata refuses the page with the existing typed HTTP 503 `storage_unavailable` envelope; values are not included in diagnostics. Backend decode failures retain the existing storage-unavailable response. @@ -38,7 +41,7 @@ Backend decode failures retain the existing storage-unavailable response. | Byte size including NUL | `length(CAST(column AS BLOB))` | `OCTET_LENGTH(column)` | `OCTET_LENGTH(column)` | | Scalar projection | `CASE WHEN bytes<=bound THEN column ELSE NULL END` | Same | Same | | Invalid flag | Non-null and empty or over-bound, separately from SQL null | Same | Same | -| Namespace / join identity | Exact text comparison | Exact text comparison | `BINARY left=BINARY right` | +| Namespace / join identity | Exact text comparison | Exact text comparison | Candidate `left=right` plus `BINARY left=BINARY right` | The page selects explicit existing list columns and bounded scalar expressions. It excludes `recent_actions`, `buffered_actions`, `last_skip_reason` and @@ -60,7 +63,13 @@ behavior, outside this slice. Source vectors cover a 50-row page and its continuation with/without a system query (1/2/2/3 calls), malformed excluded arrays, independent counters, missing and wrong-namespace links, all seven statuses, pruned targets, exact NUL/astral -identities and empty/scalar-count/byte-count failures. Normal source checks and -Server review must qualify the exact SQL/decoder on each backend before this +identities and empty/scalar-count/byte-count failures. The existing MySQL/MariaDB and PostgreSQL disposable source suites invoke the +same complete decoder/join controls, including maximum legal unsigned counters +and exact HTTP 503/no-value responses on MySQL/MariaDB. The shared unsigned +decoder remains unchanged with its original 409 behavior. MySQL/MariaDB EXPLAIN +controls retain actual primary-key/access-type output for each referenced join +after seeding 256 unrelated instances/runs. Those controls require successful +normal backend qualification; source authorship alone is not plan evidence. +Normal source checks and Server review must qualify the exact SQL/decoder on each backend before this can be called contract-compliant. No new runtime/image/qualification dispatch, schema/UDF, persisted summary, release or frozen-artifact admission follows. diff --git a/rust/src/runtime/mysql.rs b/rust/src/runtime/mysql.rs index 25193b22e..1df6bacc3 100644 --- a/rust/src/runtime/mysql.rs +++ b/rust/src/runtime/mysql.rs @@ -951,6 +951,16 @@ mod tests { } } + #[tokio::test] + #[ignore = "requires the existing disposable mysql source suite"] + async fn qualification_schedule_list_complete_summary_and_join_plans() { + let db=Database::new().await; + let store=super::super::store::Store::new(MySqlStorage::open(db.options.clone()).await.unwrap().into_pool()); + store.qualify_schedule_list_summary().await; + store.close().await; + db.remove().await; + } + #[tokio::test] #[ignore = "requires disposable MySQL/MariaDB; run qualification_ explicitly"] async fn qualification_schedule_creation_keeps_original_audit_and_unsigned_counters() { diff --git a/rust/src/runtime/postgres.rs b/rust/src/runtime/postgres.rs index 55b27aa93..e303eba85 100644 --- a/rust/src/runtime/postgres.rs +++ b/rust/src/runtime/postgres.rs @@ -534,6 +534,16 @@ mod tests { /// 0.9.0 it did (vendor/sqlx-postgres/DURABLE-PATCH.md): the next borrower's /// probe ran inside it, native's read snapshot `BEGIN` then failed with /// 25001, and every later statement on that connection with 25P02. + #[tokio::test] + #[ignore = "requires the existing disposable postgres source suite"] + async fn qualification_schedule_list_complete_summary_and_join_plans() { + let db=Database::new().await; + let store=super::super::store::Store::new(PostgresStorage::open(db.options.clone()).await.unwrap().into_pool()); + store.qualify_schedule_list_summary().await; + store.close().await; + db.remove().await; + } + #[tokio::test] #[ignore = "requires a disposable PostgreSQL; run the qualification_ tests explicitly"] async fn qualification_a_begin_cancelled_in_flight_leaves_no_transaction_behind() { diff --git a/rust/src/runtime/schedule_list.rs b/rust/src/runtime/schedule_list.rs index d1ca577cc..80357d0a5 100644 --- a/rust/src/runtime/schedule_list.rs +++ b/rust/src/runtime/schedule_list.rs @@ -561,7 +561,7 @@ fn byte_length(backend: &str, column: &str) -> String { fn exact_identity(backend: &str, left: &str, right: &str) -> String { if backend == "MySQL" { - format!("BINARY {left}=BINARY {right}") + format!("({left}={right} AND BINARY {left}=BINARY {right})") } else { format!("{left}={right}") } @@ -670,11 +670,7 @@ where let mysql = DB::NAME == "MySQL"; let mut binds = Vec::new(); let mut sql = list_selection(DB::NAME, only.is_some()); - sql.push_str(if mysql { - " WHERE BINARY s.namespace=BINARY " - } else { - " WHERE s.namespace=" - }); + sql.push_str(" WHERE s.namespace="); let mut placeholder = |sql: &mut String, bind: Bind| { binds.push(bind); if mysql { @@ -684,6 +680,10 @@ where } }; placeholder(&mut sql, Bind::Text(self.namespace.to_string())); + if mysql { + sql.push_str(" AND BINARY s.namespace=BINARY "); + placeholder(&mut sql, Bind::Text(self.namespace.to_string())); + } sql.push_str(" AND s.status!='deleted'"); let mut equal = |sql: &mut String, column: Column, value: String| { sql.push_str(" AND "); @@ -754,8 +754,8 @@ where } _ => return Err(invalid_summary()), }; - let failures = DB::number(row, "failures_count")?; - let skipped = DB::number(row, "skipped_trigger_count")?; + let failures = DB::number(row, "failures_count").map_err(|_| invalid_summary())?; + let skipped = DB::number(row, "skipped_trigger_count").map_err(|_| invalid_summary())?; if failures < 0 || skipped < 0 { return Err(invalid_summary()); } @@ -765,6 +765,110 @@ where ) } + /// Shared source controls called by each existing disposable backend suite. + #[cfg(test)] + pub(super) async fn qualify_schedule_list_summary(&self) { + use axum::response::IntoResponse; + async fn refused(error: RuntimeError) { + let response = error.into_response(); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + let body = axum::body::to_bytes(response.into_body(), 1024).await.unwrap(); + assert_eq!(serde_json::from_slice::(&body).unwrap(), + json!({"reason":"storage_unavailable","message":"storage_unavailable"})); + } + Self::query("INSERT INTO workflow_schedules(id,schedule_id,namespace,action,spec,recent_actions,buffered_actions,failures_count,skipped_trigger_count,latest_workflow_instance_id) VALUES ('summary','summary',$1,$2,$3,$4,$5,2,3,'instance')") + .bind(self.namespace.as_ref()).bind(DB::document(&json!({"workflow_type":"echo","task_queue":"test"}))) + .bind(DB::document(&json!({}))).bind(DB::document(&json!({"malformed":"deferred"}))) + .bind(DB::document(&json!({"malformed":"deferred"}))).execute(&self.pool).await.unwrap(); + let rows=self.select_schedules(&Selection {status:None,workflow_type:None,predicates:vec![]},None,None,50).await.unwrap(); + use sqlx::{Row,Column}; + assert!(rows[0].columns().iter().all(|column| !["recent_actions","buffered_actions"].contains(&column.name()))); + let page = self.list_schedules(None,"fixture-secret").await.unwrap(); + assert_eq!(page["schedules"][0]["failures_count"],2); + assert_eq!(page["schedules"][0]["skipped_trigger_count"],3); + assert!(page["schedules"][0]["current_run"].is_null()); + Self::query("INSERT INTO workflow_instances(id,workflow_class,workflow_type,namespace,current_run_id) VALUES ('instance','Echo','echo',$1,'run')") + .bind(self.namespace.as_ref()).execute(&self.pool).await.unwrap(); + Self::query("INSERT INTO workflow_instances(id,workflow_class,workflow_type,namespace) VALUES ('other','Echo','echo',$1)") + .bind(self.namespace.as_ref()).execute(&self.pool).await.unwrap(); + Self::query("INSERT INTO workflow_runs(id,workflow_instance_id,run_number,workflow_class,workflow_type,namespace,status) VALUES ('run','instance',1,'Echo','echo',$1,'running')") + .bind(self.namespace.as_ref()).execute(&self.pool).await.unwrap(); + for (status,bucket,terminal) in [("pending","running",false),("running","running",false),("waiting","running",false), + ("completed","completed",true),("failed","failed",true),("cancelled","failed",true),("terminated","failed",true)] { + Self::query("UPDATE workflow_runs SET status=$1 WHERE id='run'").bind(status).execute(&self.pool).await.unwrap(); + assert_eq!(self.list_schedules(None,"fixture-secret").await.unwrap()["schedules"][0]["current_run"], + json!({"run_id":"run","status":status,"status_bucket":bucket,"is_terminal":terminal})); + } + for invalid in ["Instance","instance ","absent"] { + Self::query("UPDATE workflow_schedules SET latest_workflow_instance_id=$1").bind(invalid).execute(&self.pool).await.unwrap(); + assert!(self.list_schedules(None,"fixture-secret").await.unwrap()["schedules"][0]["current_run"].is_null()); + } + Self::query("UPDATE workflow_schedules SET latest_workflow_instance_id='instance'").execute(&self.pool).await.unwrap(); + for namespace in ["Default","default ","other"] { + Self::query("UPDATE workflow_instances SET namespace=$1 WHERE id='instance'").bind(namespace).execute(&self.pool).await.unwrap(); + assert!(self.list_schedules(None,"fixture-secret").await.unwrap()["schedules"][0]["current_run"].is_null()); + } + Self::query("UPDATE workflow_instances SET namespace=$1 WHERE id='instance'").bind(self.namespace.as_ref()).execute(&self.pool).await.unwrap(); + Self::query("UPDATE workflow_runs SET namespace='other' WHERE id='run'").execute(&self.pool).await.unwrap(); + assert!(self.list_schedules(None,"fixture-secret").await.unwrap()["schedules"][0]["current_run"].is_null()); + Self::query("UPDATE workflow_runs SET namespace=$1,workflow_instance_id='other' WHERE id='run'").bind(self.namespace.as_ref()).execute(&self.pool).await.unwrap(); + assert!(self.list_schedules(None,"fixture-secret").await.unwrap()["schedules"][0]["current_run"].is_null()); + Self::query("UPDATE workflow_runs SET workflow_instance_id='instance' WHERE id='run'").execute(&self.pool).await.unwrap(); + for bad in ["unknown-storage-status".to_owned(),"x".repeat(33),String::new()] { + Self::query("UPDATE workflow_runs SET status=$1 WHERE id='run'").bind(bad).execute(&self.pool).await.unwrap(); + refused(self.list_schedules(None,"fixture-secret").await.unwrap_err()).await; + } + Self::query("UPDATE workflow_runs SET status='running' WHERE id='run'").execute(&self.pool).await.unwrap(); + for bad in ["", "missing"] { + Self::query("UPDATE workflow_instances SET current_run_id=$1 WHERE id='instance'").bind(bad).execute(&self.pool).await.unwrap(); + if bad.is_empty() {refused(self.list_schedules(None,"fixture-secret").await.unwrap_err()).await;} + else {assert!(self.list_schedules(None,"fixture-secret").await.unwrap()["schedules"][0]["current_run"].is_null());} + } + Self::query("UPDATE workflow_instances SET current_run_id='run' WHERE id='instance'").execute(&self.pool).await.unwrap(); + // Complete values that fit every catalog. NUL has no PostgreSQL text cell. + let mut identities=vec!["🙂".repeat(191),"x".repeat(191)]; + if DB::NAME!="PostgreSQL" {identities.push("ab\0cd🙂".into());} + for exact in identities { + Self::query("UPDATE workflow_schedules SET latest_workflow_instance_id=$1").bind(exact.as_str()).execute(&self.pool).await.unwrap(); + assert_eq!(self.list_schedules(None,"fixture-secret").await.unwrap()["schedules"][0]["latest_workflow_instance_id"],exact); + } + println!("schedule summary {}: NUL {}",DB::NAME,if DB::NAME=="PostgreSQL" {"storage-domain unsupported; no decoder pass"} else {"complete scalar decoded"}); + Self::query("UPDATE workflow_schedules SET latest_workflow_instance_id='instance'").execute(&self.pool).await.unwrap(); + let mut run_ids=vec!["🙂".repeat(26),"x".repeat(26)]; + if DB::NAME!="PostgreSQL" {run_ids.push("ab\0cd🙂".into());} + for exact in run_ids { + Self::query("UPDATE workflow_runs SET id=$1 WHERE workflow_instance_id='instance'").bind(exact.as_str()).execute(&self.pool).await.unwrap(); + Self::query("UPDATE workflow_instances SET current_run_id=$1 WHERE id='instance'").bind(exact.as_str()).execute(&self.pool).await.unwrap(); + assert_eq!(self.list_schedules(None,"fixture-secret").await.unwrap()["schedules"][0]["current_run"]["run_id"],exact); + } + Self::query("UPDATE workflow_runs SET id='run' WHERE workflow_instance_id='instance'").execute(&self.pool).await.unwrap(); + Self::query("UPDATE workflow_instances SET current_run_id='run' WHERE id='instance'").execute(&self.pool).await.unwrap(); + if DB::NAME=="MySQL" { + for field in ["failures_count","skipped_trigger_count"] { + sqlx::query::(AssertSqlSafe(format!("UPDATE workflow_schedules SET {field}=18446744073709551615"))).execute(&self.pool).await.unwrap(); + let row=Self::query("SELECT failures_count,skipped_trigger_count FROM workflow_schedules").fetch_one(&self.pool).await.unwrap(); + assert!(matches!(DB::number(&row,field),Err(RuntimeError::Refused {status:StatusCode::CONFLICT,reason:"unsupported_stored_integer"})),"unchanged shared counter decoder exposes old 409 path"); + refused(self.list_schedules(None,"fixture-secret").await.unwrap_err()).await; + sqlx::query::(AssertSqlSafe(format!("UPDATE workflow_schedules SET {field}=0"))).execute(&self.pool).await.unwrap(); + } + // Make a full referenced-table scan materially different from a PK lookup. + for index in 0..256 { + let instance=format!("plan-instance-{index:03}");let run=format!("plan-run-{index:03}"); + Self::query("INSERT INTO workflow_instances(id,workflow_class,workflow_type,namespace) VALUES ($1,'Echo','echo','default')").bind(instance.as_str()).execute(&self.pool).await.unwrap(); + Self::query("INSERT INTO workflow_runs(id,workflow_instance_id,run_number,workflow_class,workflow_type,namespace,status) VALUES ($1,$2,1,'Echo','echo','default','completed')").bind(run.as_str()).bind(instance.as_str()).execute(&self.pool).await.unwrap(); + } + let plan=sqlx::query::(AssertSqlSafe(format!("EXPLAIN {} WHERE s.namespace=? AND BINARY s.namespace=BINARY ? AND s.status!='deleted' ORDER BY s.created_at DESC,s.schedule_id LIMIT 51",list_selection(DB::NAME,false)))) + .bind(self.namespace.as_ref()).bind(self.namespace.as_ref()).fetch_all(&self.pool).await.unwrap(); + for alias in ["i","r"] { + let row=plan.iter().find(|row| DB::optional_string(row,"table").unwrap().as_deref()==Some(alias)).expect("joined plan row"); + let key=DB::optional_string(row,"key").unwrap();let access=DB::string(row,"type").unwrap(); + println!("schedule summary {} EXPLAIN {alias}: key={key:?} access={access}",DB::NAME); + assert_eq!(key.as_deref(),Some("PRIMARY")); + assert!(matches!(access.as_str(),"eq_ref"|"ref"|"const")); + } + } + } + /// `ScheduleController::formatListItem`. fn list_item(row: &DB::Row) -> Result { let status = DB::string(row, "status")?; From 86e3ad85c072a0d51c43d9f10a5e132871ce93aa Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:08:42 +0000 Subject: [PATCH 3/3] Check additive schedule summaries before legacy parity comparison --- docs/schedule-list-scalar-summary.md | 8 +++ scripts/conformance/server-parity.mjs | 2 +- .../conformance/server-parity/contract.mjs | 6 +- .../schedule-catalog-contract.mjs | 51 ++++++++++++++++- ...erverParityScheduleCatalogContractTest.mjs | 57 +++++++++++++++++++ 5 files changed, 118 insertions(+), 6 deletions(-) diff --git a/docs/schedule-list-scalar-summary.md b/docs/schedule-list-scalar-summary.md index 38b643c48..e578b83ed 100644 --- a/docs/schedule-list-scalar-summary.md +++ b/docs/schedule-list-scalar-summary.md @@ -70,6 +70,14 @@ decoder remains unchanged with its original 409 behavior. MySQL/MariaDB EXPLAIN controls retain actual primary-key/access-type output for each referenced join after seeding 256 unrelated instances/runs. Those controls require successful normal backend qualification; source authorship alone is not plan evidence. +The existing HTTP schedule-catalog corpus retains its frozen published-PHP +fixture and expected responses. Native recordings require all four new fields +on every list row. Before comparing the older shared response, the comparator +checks counters and the latest identity against preceding backfill responses, +the unclaimed current-run state, and the retained trigger-audit identity pair. +Only those checked additions are removed from that comparison; raw observations +and every existing field remain intact. Older published PHP recordings can +retain their original list shape. Normal source checks and Server review must qualify the exact SQL/decoder on each backend before this can be called contract-compliant. No new runtime/image/qualification dispatch, schema/UDF, persisted summary, release or frozen-artifact admission follows. diff --git a/scripts/conformance/server-parity.mjs b/scripts/conformance/server-parity.mjs index 617d4a10a..9bdc42bb7 100644 --- a/scripts/conformance/server-parity.mjs +++ b/scripts/conformance/server-parity.mjs @@ -138,7 +138,7 @@ try { checkSdkTimerAuthorPackages(fixture, item.observation, record.artifacts); checkSdkSignalDirectionPackages(fixture, item.observation, record.artifacts); item.outcome = 'product-fail'; - item.projection = checkObservation(fixture, item.observation, workflowId, record.artifacts); + item.projection = checkObservation(fixture, item.observation, workflowId, record.artifacts, record.target); item.outcome = 'pass'; } catch (error) { item.error = error.message; diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs index caf1353ee..b13533b14 100644 --- a/scripts/conformance/server-parity/contract.mjs +++ b/scripts/conformance/server-parity/contract.mjs @@ -104,7 +104,7 @@ export function instantNanoseconds(value) { // Only this fixture's declared semantics are projected. The full observation // remains in the recording, including implementation metadata and timestamps. -export function checkObservation(fixture, observation, workflowId, artifacts) { +export function checkObservation(fixture, observation, workflowId, artifacts, target) { if (fixture.continuation) { assert.equal(observation.workflow_id, workflowId); const record=observation.continuation; @@ -247,7 +247,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) { const messageStreams = fixture.message_streams ? checkMessageStreams(fixture, observation, workflowId) : null; const workflowStreamWrites = fixture.workflow_stream_writes ? checkWorkflowStreamWrites(fixture, observation, workflowId) : null; const schedulePolicies = fixture.schedule_policies ? checkSchedulePolicies(fixture, observation, workflowId) : null; - const scheduleCatalog = fixture.schedule_catalog ? checkScheduleCatalog(fixture, observation, workflowId) : null; + const scheduleCatalog = fixture.schedule_catalog ? checkScheduleCatalog(fixture, observation, workflowId, target) : null; const signalsBeforeWorker = fixture.signals_before_worker ? checkSignalsBeforeWorker(fixture, observation, workflowId) : null; const workerListing = fixture.worker_listing ? checkWorkerListing(fixture, observation, workflowId) : null; const workerReplacement = fixture.worker_replacement ? checkWorkerReplacement(fixture, observation, workflowId) : null; @@ -908,7 +908,7 @@ export function compareRecords(records, expectedFixtureHashes, expectedFixtures, if (record.artifacts.sdk_php_source_commit) assert.equal(item.observation.sdk_php_source, record.artifacts.sdk_php_source_commit, 'installed exact source SDK matches tuple'); if (record.mode === 'embedded') assert.equal(item.observation.workflow_package, record.artifacts.workflow, 'installed Workflow matches tuple'); assert.deepStrictEqual(item.fixture, baseline.fixture, 'same fixture expectations'); - const checked = checkObservation(item.fixture, item.observation, item.projection.workflow_id, record.artifacts); + const checked = checkObservation(item.fixture, item.observation, item.projection.workflow_id, record.artifacts, record.target); assert.deepStrictEqual(item.projection, checked, 'projection matches raw observation'); assert.deepStrictEqual(checked, baseline.projection, `${record.target}: ${item.fixture_id}`); } diff --git a/scripts/conformance/server-parity/schedule-catalog-contract.mjs b/scripts/conformance/server-parity/schedule-catalog-contract.mjs index 1a82da941..c7a1e59ae 100644 --- a/scripts/conformance/server-parity/schedule-catalog-contract.mjs +++ b/scripts/conformance/server-parity/schedule-catalog-contract.mjs @@ -26,7 +26,53 @@ const FIRE_TIMES = [[], [], [], [], const ids = step => step.response.schedules.map(item => item.schedule_id); -export function checkScheduleCatalog(fixture, observation, workflowId) { +// Keep the frozen PHP response contract intact. Native's additive fields are +// checked against the preceding HTTP backfill results before projecting them +// away for comparison with an older published PHP server. +function checkedListSummaries(raw, target) { + const known = new Map(); + return raw.map(step => { + if (step.status === 200 && step.request.endsWith('/backfill')) { + const id = step.response.schedule_id; + const state = known.get(id) ?? {failures: 0, skipped: 0, latest: null}; + for (const result of step.response.results) { + if (result.outcome === 'started') state.latest = result.workflow_id; + if (result.outcome === 'skipped') state.skipped++; + if (result.outcome === 'failed') state.failures++; + } + known.set(id, state); + } + if (!Array.isArray(step.response.schedules)) return step; + const schedules = step.response.schedules.map(item => { + const fields = ['failures_count', 'skipped_trigger_count', 'latest_workflow_instance_id', 'current_run']; + const present = fields.filter(field => Object.hasOwn(item, field)); + if (target !== 'rust' && present.length === 0) return item; + assert.deepStrictEqual(present, fields, 'complete schedule list summary'); + const state = known.get(item.schedule_id) ?? {failures: 0, skipped: 0, latest: null}; + assert.deepStrictEqual([item.failures_count, item.skipped_trigger_count, item.latest_workflow_instance_id], + [state.failures, state.skipped, state.latest], 'summary counters and latest identity match preceding HTTP results'); + if (state.latest === null) assert.equal(item.current_run, null, 'unfired schedule has unknown current run'); + else { + assert.match(state.latest, /^[0-9A-HJKMNP-TV-Z]{26}$/, 'complete generated workflow identity'); + assert.ok(item.current_run && typeof item.current_run === 'object', 'started schedule has current run'); + assert.match(item.current_run.run_id, /^[0-9A-HJKMNP-TV-Z]{26}$/, 'complete current run identity'); + assert.deepStrictEqual(item.current_run, {run_id: item.current_run.run_id, status: 'pending', + status_bucket: 'running', is_terminal: false}, 'unclaimed backfill run remains pending'); + // The skip schedule also exposes both identities in its retained audit. + const audit = raw.find(candidate => candidate.request === `GET schedules/${item.schedule_id}/history?limit=20`); + if (audit) { + const event = audit.response.events.filter(event => event.event_type === 'ScheduleTriggered').at(-1); + assert.deepStrictEqual([state.latest, item.current_run.run_id], + [event.workflow_instance_id, event.workflow_run_id], 'list current run matches retained trigger audit'); + } + } + return Object.fromEntries(Object.entries(item).filter(([field]) => !fields.includes(field))); + }); + return {...step, response: {...step.response, schedules}}; + }); +} + +export function checkScheduleCatalog(fixture, observation, workflowId, target) { const state = observation.schedule_catalog; assert.ok(state && typeof state === 'object', 'actual schedule catalog observation'); if (observation.mode === 'embedded') { @@ -41,7 +87,8 @@ export function checkScheduleCatalog(fixture, observation, workflowId) { assert.ok(Number.isInteger(state.retries) && state.retries >= 0, 'recorded lock retries'); const raw = state.steps; const names = {[namespace]: '@namespace', [state.queue]: '@queue'}; - const steps = raw.map(step => ({request: step.request, status: step.status, response: normalized(step.response, names)})); + const steps = checkedListSummaries(raw, target) + .map(step => ({request: step.request, status: step.status, response: normalized(step.response, names)})); assert.deepStrictEqual(steps.map(step => step.status), STATUSES, 'schedule catalog step statuses'); // Newest first, filtered and paged. assert.deepStrictEqual(ids(steps[4]), ['delta', 'gamma', 'beta', 'alpha'], 'newest schedules first'); diff --git a/tests/Unit/ServerParityScheduleCatalogContractTest.mjs b/tests/Unit/ServerParityScheduleCatalogContractTest.mjs index aae0d771f..92bb0e4b8 100644 --- a/tests/Unit/ServerParityScheduleCatalogContractTest.mjs +++ b/tests/Unit/ServerParityScheduleCatalogContractTest.mjs @@ -32,6 +32,63 @@ function model() { } const projection = {scope: 'schedule_catalog'}; +function nativeModel() { + const observation = model(); + const steps = observation.schedule_catalog.steps; + let sequence = 1; + for (const step of steps) { + for (const result of step.response.results ?? []) { + if (result.outcome === 'started') result.workflow_id = String(sequence++).padStart(26, '0'); + } + for (const item of step.response.schedules ?? []) Object.assign(item, { + failures_count: 0, skipped_trigger_count: 0, latest_workflow_instance_id: null, current_run: null, + }); + } + const latest = {'bf': steps[58].response.results.at(-1).workflow_id, + 'bf-skip': steps[62].response.results.at(-1).workflow_id, + 'bf-buffer': steps[64].response.results.at(-1).workflow_id}; + for (const item of steps[71].response.schedules) { + if (item.schedule_id === 'beta') item.skipped_trigger_count = 2; + if (!Object.hasOwn(latest, item.schedule_id)) continue; + item.latest_workflow_instance_id = latest[item.schedule_id]; + item.skipped_trigger_count = {'bf': 7, 'bf-skip': 2, 'bf-buffer': 0}[item.schedule_id]; + item.current_run = {run_id: String(sequence++).padStart(26, '0'), status: 'pending', status_bucket: 'running', is_terminal: false}; + } + const skipped = steps[71].response.schedules.find(item => item.schedule_id === 'bf-skip'); + const audit = steps[70].response.events.filter(event => event.event_type === 'ScheduleTriggered').at(-1); + audit.workflow_instance_id = skipped.latest_workflow_instance_id; + audit.workflow_run_id = skipped.current_run.run_id; + return observation; +} + +test('native summaries are checked before the unchanged published PHP comparison', () => { + const observation = nativeModel(); + const before = structuredClone(observation); + assert.deepStrictEqual(checkScheduleCatalog(fixture, observation, workflowId, 'rust'), projection); + assert.deepStrictEqual(observation, before, 'raw summary evidence is retained unchanged'); +}); +test('native target cannot omit all new summary fields', () => { + assert.throws(() => checkScheduleCatalog(fixture, model(), workflowId, 'rust'), /complete schedule list summary/); +}); +for (const [label, mutate] of [ + ['partial summary', o => { delete o.schedule_catalog.steps[4].response.schedules[0].failures_count; }], + ['wrong failure count', o => { o.schedule_catalog.steps[4].response.schedules[0].failures_count = 1; }], + ['wrong skip count', o => { o.schedule_catalog.steps[71].response.schedules[1].skipped_trigger_count = 0; }], + ['wrong latest workflow', o => { o.schedule_catalog.steps[71].response.schedules[0].latest_workflow_instance_id = '9'.repeat(26); }], + ['missing current run', o => { o.schedule_catalog.steps[71].response.schedules[0].current_run = null; }], + ['shortened run identity', o => { o.schedule_catalog.steps[71].response.schedules[0].current_run.run_id = 'short'; }], + ['invented terminal state', o => { o.schedule_catalog.steps[71].response.schedules[0].current_run.is_terminal = true; }], + ['wrong audit run', o => { o.schedule_catalog.steps[70].response.events.filter(e => e.event_type === 'ScheduleTriggered').at(-1).workflow_run_id = '9'.repeat(26); }], + ['unreviewed extra field', o => { o.schedule_catalog.steps[4].response.schedules[0].other_summary = 1; }], + ['changed existing field', o => { o.schedule_catalog.steps[4].response.schedules[0].note = 'changed'; }], +]) { + test(`rejects native ${label}`, () => { + const observation = nativeModel(); + mutate(observation); + assert.throws(() => checkScheduleCatalog(fixture, observation, workflowId, 'rust')); + }); +} + test('published PHP responses pass', () => assert.deepStrictEqual(checkScheduleCatalog(fixture, model(), workflowId), projection)); test('a sequence replayed in a new namespace passes', () => { const observation = model();