From 7835ea7be820698b3fad15e3a0f3625bf9072e7c Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Wed, 21 Oct 2026 03:41:47 +0000 Subject: [PATCH] Refuse takeover of PHP schedules native cannot run yet Native's scheduler fires only default-namespace schedules with one UTC P1D or PT2S interval, the skip policy, no jitter and at most one run. Once due, a taken-over PHP cron schedule would fail every scheduler tick, and a named-namespace one would silently stop firing. Every upgrade inspection now reports schedule_inspection, counting live schedules outside native's subset by reason without naming them, and takeover refuses such a source with upgrade_source_schedules_unsupported before any change. Co-Authored-By: Claude Opus 5.5 --- docs/rust-server-port.md | 25 +++ docs/rust-server-upgrade-takeover.md | 10 ++ rust/src/runtime/mysql.rs | 4 + rust/src/runtime/postgres.rs | 4 + rust/src/runtime/upgrade.rs | 12 ++ rust/src/runtime/upgrade/schedule_scan.rs | 190 ++++++++++++++++++++++ rust/src/runtime/upgrade/takeover.rs | 7 + rust/src/runtime/upgrade_tests.rs | 40 +++++ 8 files changed, 292 insertions(+) create mode 100644 rust/src/runtime/upgrade/schedule_scan.rs diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md index 0600015c9..f0e542a26 100644 --- a/docs/rust-server-port.md +++ b/docs/rust-server-port.md @@ -229,6 +229,31 @@ support is not evidence of multi-node safety. Newest checkpoints first. Each entry is exact to the source, tree and runs it names, including unsuccessful observations. +October 21, takeover refuses schedules native cannot run: + +- **Found by.** A review of takeover against native's schedule support. + Native's scheduler fires only default-namespace schedules with one UTC + `P1D` or `PT2S` interval, the skip policy, no jitter and at most one run. + Once due, a PHP cron schedule taken over would have failed every + scheduler tick (native refuses its next occurrence), holding back the + schedules due after it. A named-namespace one would have silently stopped + firing. +- **Source.** Every upgrade inspection now reports `schedule_inspection`: the + live schedules and, by reason, how many are outside native's subset. The + reasons are native's own schedule-creation refusals plus namespace, status, + memo/search attributes/labels, note, buffered actions and queue routing, + which native would not describe. Any such schedule makes the inspection + `unqualified`, and takeover refuses with + `upgrade_source_schedules_unsupported` before the fingerprint check or any + write. Deleted schedules do not count. +- **Tests.** SQLite: a cron schedule and a named-namespace schedule each refuse + takeover and leave the file byte-identical; a runnable schedule and a + deleted cron schedule convert; inspection counts reasons without schedule + names or memo values. PostgreSQL 17, MySQL 8.0 and MariaDB 10.11 takeover + qualification each refuse the cron schedule with the source unchanged. +- **Next.** Native schedule parity (cron, time zones, overlap policies, + jitter, quotas, metadata, namespaces) narrows this refusal. + October 20, lost workflow task claims: - **Found by.** The [multi-node runs](../benchmarks/server-port/multinode/results-2026-10-20.md). diff --git a/docs/rust-server-upgrade-takeover.md b/docs/rust-server-upgrade-takeover.md index dfa5d7003..6b62ce342 100644 --- a/docs/rust-server-upgrade-takeover.md +++ b/docs/rust-server-upgrade-takeover.md @@ -58,6 +58,16 @@ check exists, takeover refuses when `upgrade-payload-inspect` reports any external reference, and the operator's object manifest from the backup procedure remains the record for those bytes. +Native schedules are still a bounded subset of PHP's: one UTC interval of a +day or two seconds, the skip overlap policy, no jitter, at most one run, the +default namespace, and no memo, search attributes, note, buffered actions or +queue routing. Every inspection reports `schedule_inspection`: the number of +live (not deleted) schedules and, by reason, how many fall outside that +subset, without naming them. Such a schedule would stop firing or lose +fields under native, so its inspection is `unqualified` and takeover refuses +with `upgrade_source_schedules_unsupported` before any change. Deleted +schedules keep only their audit history and do not block takeover. + ## Exclusive transaction per backend The marker, the fingerprint check and the schema changes must not interleave diff --git a/rust/src/runtime/mysql.rs b/rust/src/runtime/mysql.rs index a4a76dbf5..72da0513d 100644 --- a/rust/src/runtime/mysql.rs +++ b/rust/src/runtime/mysql.rs @@ -1352,6 +1352,10 @@ mod tests { "INSERT INTO runtime_external_payloads(id,namespace,storage_uri,storage_uri_sha256,codec,sha256,size_bytes) VALUES ('p','default','s3://b/k','x','avro','y',1)", "upgrade_source_external_payloads_unsupported", ), + ( + "INSERT INTO workflow_schedules(id,schedule_id,namespace,spec,action,status,overlap_policy,jitter_seconds) VALUES ('s','nightly','default','{\"cron_expressions\":[\"0 2 * * *\"],\"timezone\":\"UTC\"}','{\"workflow_type\":\"echo\",\"task_queue\":\"default\"}','active','skip',0)", + "upgrade_source_schedules_unsupported", + ), ] { let db = Database::new().await; php_layout(&db, Some("php-2.9.7")).await; diff --git a/rust/src/runtime/postgres.rs b/rust/src/runtime/postgres.rs index 06af1a9ce..15d9d5082 100644 --- a/rust/src/runtime/postgres.rs +++ b/rust/src/runtime/postgres.rs @@ -921,6 +921,10 @@ mod tests { "INSERT INTO runtime_external_payloads(id,namespace,storage_uri,storage_uri_sha256,codec,sha256,size_bytes) VALUES ('p','default','s3://b/k','x','avro','y',1)", "upgrade_source_external_payloads_unsupported", ), + ( + "INSERT INTO workflow_schedules(id,schedule_id,namespace,spec,action,status,overlap_policy,jitter_seconds) VALUES ('s','nightly','default','{\"cron_expressions\":[\"0 2 * * *\"],\"timezone\":\"UTC\"}','{\"workflow_type\":\"echo\",\"task_queue\":\"default\"}','active','skip',0)", + "upgrade_source_schedules_unsupported", + ), ] { let db = Database::new().await; let mut connection = php_297(&db).await; diff --git a/rust/src/runtime/upgrade.rs b/rust/src/runtime/upgrade.rs index e3f3f7b44..85cfd7913 100644 --- a/rust/src/runtime/upgrade.rs +++ b/rust/src/runtime/upgrade.rs @@ -3,6 +3,7 @@ pub mod external_reference; pub mod fingerprint; pub mod takeover; mod payload_scan; +mod schedule_scan; pub mod queue; pub mod queue_binding; mod queue_scan; @@ -196,6 +197,7 @@ fn report( counts: BTreeMap, queue_inspection: Option, payload_inspection: Option, + schedule_inspection: Value, ) -> Result { let digest = Sha256::digest(serde_json::to_vec(catalog)?) .iter() @@ -220,6 +222,10 @@ fn report( } report["payload_inspection"] = payloads; } + if schedule_inspection["unsupported"] != 0 { + report["status"] = json!("unqualified"); + } + report["schedule_inspection"] = schedule_inspection; Ok(report) } @@ -392,6 +398,7 @@ pub(super) async fn sqlite_inventory( } else { None }; + let schedule_inspection = schedule_scan::inspect::(&mut *tx).await?; report( "sqlite", &catalog, @@ -399,6 +406,7 @@ pub(super) async fn sqlite_inventory( counts, queue_inspection, payload_inspection, + schedule_inspection, ) } @@ -661,6 +669,7 @@ pub(super) async fn mysql_inventory( } else { None }; + let schedule_inspection = schedule_scan::inspect::(&mut *tx).await?; // MySQL's metadata is not an MVCC snapshot. Refuse a changed visible // catalog rather than joining counts to a different schema. if mysql_source_catalog(&mut *tx, fenced).await?.0 != catalog { @@ -676,6 +685,7 @@ pub(super) async fn mysql_inventory( counts, queue_inspection, payload_inspection, + schedule_inspection, ) } @@ -742,6 +752,7 @@ pub(super) async fn postgres_inventory( } else { None }; + let schedule_inspection = schedule_scan::inspect::(&mut *tx).await?; report( "pgsql", &catalog, @@ -749,6 +760,7 @@ pub(super) async fn postgres_inventory( counts, queue_inspection, payload_inspection, + schedule_inspection, ) } diff --git a/rust/src/runtime/upgrade/schedule_scan.rs b/rust/src/runtime/upgrade/schedule_scan.rs new file mode 100644 index 000000000..57894c5e1 --- /dev/null +++ b/rust/src/runtime/upgrade/schedule_scan.rs @@ -0,0 +1,190 @@ +//! Whether every live PHP schedule is one native can run and describe as PHP +//! does. Native schedules are still a bounded subset (one UTC interval, the +//! skip policy, no jitter, at most one run, the default namespace), so a +//! source schedule outside it would stop firing, or lose fields, after +//! takeover. The report counts them by reason and never contains schedule +//! identities or values. +use crate::runtime::{Result, RuntimeError, backend::Backend, schedules, store::now}; +use serde_json::{Value, json}; +use sqlx::{Executor, IntoArguments}; +use std::collections::BTreeMap; + +const SCHEDULES: &str = "SELECT schedule_id,namespace,status,spec,action,overlap_policy,jitter_seconds,max_runs,note,memo,search_attributes,visibility_labels,buffered_actions,connection,queue FROM workflow_schedules WHERE status<>'deleted' ORDER BY id"; + +pub(super) async fn inspect(connection: &mut DB::Connection) -> Result +where + for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>, + DB::Arguments: IntoArguments, +{ + let rows = sqlx::query(sqlx::AssertSqlSafe(DB::statement(SCHEDULES))) + .fetch_all(&mut *connection) + .await?; + let mut reasons = BTreeMap::<&'static str, i64>::new(); + for row in &rows { + let schedule = Schedule { + schedule_id: DB::string(row, "schedule_id")?, + namespace: DB::optional_string(row, "namespace")?, + status: DB::string(row, "status")?, + spec: DB::optional_document_row(row, "spec")?.unwrap_or(Value::Null), + action: DB::optional_document_row(row, "action")?.unwrap_or(Value::Null), + overlap_policy: DB::string(row, "overlap_policy")?, + jitter_seconds: DB::number(row, "jitter_seconds")?, + max_runs: DB::optional_number(row, "max_runs")?, + note: DB::optional_string(row, "note")?, + metadata: [ + DB::optional_document_row(row, "memo")?, + DB::optional_document_row(row, "search_attributes")?, + DB::optional_document_row(row, "visibility_labels")?, + ], + buffered: DB::optional_document_row(row, "buffered_actions")?, + routing: [ + DB::optional_string(row, "connection")?, + DB::optional_string(row, "queue")?, + ], + }; + if let Some(reason) = schedule.unsupported() { + *reasons.entry(reason).or_default() += 1; + } + } + Ok(json!({ + "schedules": rows.len(), + "unsupported": reasons.values().sum::(), + "reasons": reasons, + })) +} + +struct Schedule { + schedule_id: String, + namespace: Option, + status: String, + spec: Value, + action: Value, + overlap_policy: String, + jitter_seconds: i64, + max_runs: Option, + note: Option, + metadata: [Option; 3], + buffered: Option, + routing: [Option; 2], +} + +/// A JSON column PHP leaves null when empty; an empty array or object is +/// treated the same. +fn present(value: &Option) -> bool { + match value { + None | Some(Value::Null) => false, + Some(Value::Array(items)) => !items.is_empty(), + Some(Value::Object(fields)) => !fields.is_empty(), + Some(_) => true, + } +} + +impl Schedule { + /// The first reason native cannot take this schedule over, if any. + fn unsupported(&self) -> Option<&'static str> { + if self.namespace.as_deref() != Some("default") { + return Some("schedule_namespace_unsupported"); + } + if !matches!(self.status.as_str(), "active" | "paused") { + return Some("schedule_status_unsupported"); + } + if self.metadata.iter().any(present) { + return Some("schedule_metadata_unsupported"); + } + if self.note.is_some() { + return Some("schedule_note_unsupported"); + } + if present(&self.buffered) { + return Some("schedule_buffered_actions_unsupported"); + } + if self.routing.iter().any(Option::is_some) { + return Some("schedule_queue_routing_unsupported"); + } + // Exactly the definition native's own schedule creation accepts. + let body = json!({ + "schedule_id": self.schedule_id, "spec": self.spec, "action": self.action, + "overlap_policy": self.overlap_policy, "jitter_seconds": self.jitter_seconds, + "max_runs": self.max_runs, + }); + match schedules::definition(&body, now()) { + Ok(_) => None, + Err(RuntimeError::Refused { reason, .. }) => Some(reason), + Err(_) => Some("schedule_definition_unsupported"), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn runnable() -> Schedule { + Schedule { + schedule_id: "daily".into(), + namespace: Some("default".into()), + status: "active".into(), + spec: json!({"intervals":[{"every":"P1D"}],"timezone":"UTC"}), + action: json!({"workflow_type":"echo","task_queue":"default", + "input":{"codec":"avro","blob":"AA=="}, + "execution_timeout_seconds":3600,"run_timeout_seconds":600}), + overlap_policy: "skip".into(), + jitter_seconds: 0, + max_runs: None, + note: None, + metadata: [None, Some(Value::Null), Some(json!([]))], + buffered: None, + routing: [None, None], + } + } + + #[test] + fn only_schedules_native_runs_and_describes_unchanged_are_qualified() { + assert_eq!(runnable().unsupported(), None); + let cases: [(fn(&mut Schedule), &str); 10] = [ + ( + |s| s.namespace = Some("billing".into()), + "schedule_namespace_unsupported", + ), + (|s| s.namespace = None, "schedule_namespace_unsupported"), + ( + |s| s.spec = json!({"cron_expressions":["0 9 * * *"],"timezone":"UTC"}), + "unsupported_request_field", + ), + ( + |s| s.spec = json!({"intervals":[{"every":"PT1H"}],"timezone":"UTC"}), + "unsupported_schedule_interval", + ), + (|s| s.overlap_policy = "buffer_one".into(), "unsupported_schedule_policy"), + (|s| s.jitter_seconds = 30, "unsupported_schedule_policy"), + (|s| s.max_runs = Some(5), "unsupported_schedule_quota"), + ( + |s| s.metadata[0] = Some(json!({"team":"ops"})), + "schedule_metadata_unsupported", + ), + (|s| s.note = Some("paused for audit".into()), "schedule_note_unsupported"), + ( + |s| { + s.action["execution_timeout_seconds"] = Value::Null; + }, + "unsupported_schedule_action_timeout", + ), + ]; + for (change, reason) in cases { + let mut schedule = runnable(); + change(&mut schedule); + assert_eq!(schedule.unsupported(), Some(reason)); + } + let mut buffered = runnable(); + buffered.buffered = Some(json!([{"buffered_at":"2026-10-21T00:00:00+00:00"}])); + assert_eq!( + buffered.unsupported(), + Some("schedule_buffered_actions_unsupported") + ); + let mut routed = runnable(); + routed.routing[1] = Some("schedules".into()); + assert_eq!( + routed.unsupported(), + Some("schedule_queue_routing_unsupported") + ); + } +} diff --git a/rust/src/runtime/upgrade/takeover.rs b/rust/src/runtime/upgrade/takeover.rs index 0d9188b83..ff2431d29 100644 --- a/rust/src/runtime/upgrade/takeover.rs +++ b/rust/src/runtime/upgrade/takeover.rs @@ -258,6 +258,13 @@ fn qualified(inventory: &Value) -> Result<&str> { return Err(refuse(StatusCode::CONFLICT, reason)); } } + // A schedule native cannot run would stop firing after takeover. + if inventory["schedule_inspection"]["unsupported"] != 0 { + return Err(refuse( + StatusCode::CONFLICT, + "upgrade_source_schedules_unsupported", + )); + } if inventory["queue_inspection"]["status"] != "known_subset_bound" { return Err(refuse( StatusCode::CONFLICT, diff --git a/rust/src/runtime/upgrade_tests.rs b/rust/src/runtime/upgrade_tests.rs index 045caea82..2cbb2c8e7 100644 --- a/rust/src/runtime/upgrade_tests.rs +++ b/rust/src/runtime/upgrade_tests.rs @@ -294,6 +294,10 @@ async fn inspection_preserves_source_bytes_reports_counts_and_never_grants_takeo assert_eq!(inspected["table_rows"].as_object().unwrap().len(), 49); assert_eq!(inspected["table_rows"]["jobs"], 1); assert_eq!(inspected["table_rows"]["workflow_runs"], 0); + assert_eq!( + inspected["schedule_inspection"], + serde_json::json!({"schedules":0,"unsupported":0,"reasons":{}}) + ); assert_eq!(inspected["catalog_sha256"].as_str().unwrap().len(), 64); for flag in [ "source_mutated", @@ -897,6 +901,10 @@ async fn takeover_converts_a_php_297_database_that_native_then_opens() { "INSERT INTO workflow_worker_registration_incarnations VALUES ('0123456789abcdef0123456789abcdef','default','current','active',0,NULL,'2026-10-20 09:00:00.000000','2026-10-20 09:00:00.000000')", "INSERT INTO workflow_worker_registration_incarnations VALUES ('11111111111111111111111111111111','default','current','superseded',0,'2026-10-20 08:00:00.000000','2026-10-20 07:00:00.000000','2026-10-20 08:00:00.000000')", "INSERT INTO workflow_worker_registration_incarnations VALUES ('22222222222222222222222222222222','default','gone','deregistered',3,'2026-10-20 08:30:00.000000','2026-10-20 07:00:00.000000','2026-10-20 08:30:00.000000')", + // A schedule native runs as PHP stored it, and a deleted cron one, + // which only keeps its audit history. + "INSERT INTO workflow_schedules(id,schedule_id,namespace,spec,action,status,overlap_policy,jitter_seconds) VALUES ('s1','daily','default','{\"intervals\":[{\"every\":\"P1D\"}],\"timezone\":\"UTC\"}','{\"workflow_type\":\"echo\",\"task_queue\":\"default\",\"input\":{\"codec\":\"avro\",\"blob\":\"AA==\"},\"execution_timeout_seconds\":3600,\"run_timeout_seconds\":600}','active','skip',0)", + "INSERT INTO workflow_schedules(id,schedule_id,namespace,spec,action,status,overlap_policy,jitter_seconds,deleted_at) VALUES ('s2','nightly','default','{\"cron_expressions\":[\"0 2 * * *\"]}','{}','deleted','buffer_all',30,'2026-10-19 00:00:00')", ] { mutate(&path, statement).await; } @@ -1062,6 +1070,15 @@ async fn takeover_refusals_leave_the_php_database_unchanged() { "INSERT INTO workflow_messages(workflow_instance_id,direction,channel,stream_key,sequence) VALUES ('w','inbound','c','k',1)", "upgrade_source_message_streams_unsupported", ), + // Schedules native would not fire, or would describe without fields. + ( + "INSERT INTO workflow_schedules(id,schedule_id,namespace,spec,action,status,overlap_policy,jitter_seconds) VALUES ('s','nightly','default','{\"cron_expressions\":[\"0 2 * * *\"],\"timezone\":\"UTC\"}','{\"workflow_type\":\"echo\",\"task_queue\":\"default\"}','active','skip',0)", + "upgrade_source_schedules_unsupported", + ), + ( + "INSERT INTO workflow_schedules(id,schedule_id,namespace,spec,action,status,overlap_policy,jitter_seconds) VALUES ('s','daily','billing','{\"intervals\":[{\"every\":\"P1D\"}],\"timezone\":\"UTC\"}','{\"workflow_type\":\"echo\",\"task_queue\":\"default\",\"input\":{\"codec\":\"avro\",\"blob\":\"AA==\"},\"execution_timeout_seconds\":3600,\"run_timeout_seconds\":600}','paused','skip',0)", + "upgrade_source_schedules_unsupported", + ), ] { let (_directory, path) = php297_layout().await; mutate(&path, statement).await; @@ -1165,3 +1182,26 @@ async fn prepare_leaves_a_rollback_journal_source_byte_identical() { assert_eq!(report["source_mutated"], false); assert_eq!(std::fs::read(&path).unwrap(), original); } + +#[tokio::test] +async fn inspection_counts_unrunnable_schedules_by_reason_without_naming_them() { + let (_directory, path) = php297_layout().await; + for statement in [ + "INSERT INTO workflow_schedules(id,schedule_id,namespace,spec,action,status,overlap_policy,jitter_seconds) VALUES ('s1','nightly-billing','default','{\"cron_expressions\":[\"0 2 * * *\"],\"timezone\":\"UTC\"}','{\"workflow_type\":\"echo\",\"task_queue\":\"default\",\"input\":{\"codec\":\"avro\",\"blob\":\"AA==\"},\"execution_timeout_seconds\":3600,\"run_timeout_seconds\":600}','active','skip',0)", + "INSERT INTO workflow_schedules(id,schedule_id,namespace,spec,action,status,overlap_policy,jitter_seconds,memo) VALUES ('s2','tenant-report','default','{}','{}','paused','skip',0,'{\"tenant\":\"private-tenant\"}')", + "INSERT INTO workflow_schedules(id,schedule_id,namespace,spec,action,status,overlap_policy,jitter_seconds) VALUES ('s3','old','default','{}','{}','deleted','skip',0)", + ] { + mutate(&path, statement).await; + } + let before = std::fs::read(&path).unwrap(); + let inspected = inspect_sqlite(&path).await.unwrap(); + assert_eq!(inspected["status"], "unqualified"); + assert_eq!( + inspected["schedule_inspection"], + serde_json::json!({"schedules":2,"unsupported":2, + "reasons":{"unsupported_request_field":1,"schedule_metadata_unsupported":1}}) + ); + let text = inspected.to_string(); + assert!(!text.contains("nightly-billing") && !text.contains("private-tenant")); + assert_eq!(std::fs::read(&path).unwrap(), before); +}