From 0656ab9399f7686d421a409e8424c7a4c616529c Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 11:08:45 -0600 Subject: [PATCH 1/2] Keep one failing schedule from stalling the maintenance tick (#955) A non-transient database error in one schedule's admission used to end the whole tick. The schedule kept its occurrence, so it failed again on every tick: later schedules never fired, the repair pass never ran, and /api/ready stayed 503. Settle each failed admission as published PHP's recordStartFailure does. A transient storage error (lock pressure, unavailable backend) still keeps the occurrence and fails the tick. Any other error is recorded, moving the schedule to its next occurrence, and the batch goes on. A database error is recorded as storage_unavailable, so driver text stays out of recent_actions. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/schedule_execution.rs | 242 +++++++++++++++++++------ 1 file changed, 184 insertions(+), 58 deletions(-) diff --git a/rust/src/runtime/schedule_execution.rs b/rust/src/runtime/schedule_execution.rs index 2c5ef80e5..9a4ab7708 100644 --- a/rust/src/runtime/schedule_execution.rs +++ b/rust/src/runtime/schedule_execution.rs @@ -555,10 +555,7 @@ where for candidate in buffered { let key = DB::string(&candidate, "id")?; if let Err(error) = self.drain_one(&key).await { - if matches!(error, RuntimeError::Database(_)) { - return Err(error); - } - self.record_failure(&key, &error.to_string()).await?; + self.settle_failed_admission(&key, error).await?; } } let due = Self::query("SELECT id,next_fire_at FROM workflow_schedules WHERE status='active' AND next_fire_at IS NOT NULL AND next_fire_at<=$1 ORDER BY next_fire_at,id LIMIT 100") @@ -571,13 +568,40 @@ where .await { Ok(_) => self.wake.notify_waiters(), - Err(error @ RuntimeError::Database(_)) => return Err(error), - Err(error) => self.record_failure(&key, &error.to_string()).await?, + Err(error) => self.settle_failed_admission(&key, error).await?, } } Ok(()) } + /// One schedule's failed admission in a maintenance tick, as PHP's + /// `recordStartFailure` settles it (#955). A transient storage failure + /// (lock pressure or an unavailable backend) keeps the stored occurrence + /// and fails the tick, which retries it from durable state. Any other + /// failure is recorded, which moves the schedule to its next occurrence, + /// and the batch goes on. One schedule that always fails can then stop + /// neither the other schedules, the repair pass nor readiness. + async fn settle_failed_admission(&self, key: &str, error: RuntimeError) -> Result<()> { + if super::worker_incarnations::transient_reason(&error).is_some() { + return Err(error); + } + let reason = match &error { + // Database text can carry stored payloads. The schedule's visible + // actions get the HTTP surface's bounded reason, and the log gets + // only the bounded driver diagnostic. + RuntimeError::Database(_) | RuntimeError::Migration(_) => { + if let Some(diagnostic) = error.storage_diagnostic() { + eprintln!("{diagnostic}"); + } + "storage_unavailable".to_owned() + } + _ => error.to_string(), + }; + // A failure to record it is a storage failure of its own: the tick + // fails and the occurrence is retried. + self.record_failure(key, &reason).await + } + /// One buffered action of the tick's first phase. async fn drain_one(&self, key: &str) -> Result<()> { let mut tx = self.begin().await?; @@ -668,19 +692,13 @@ mod tests { use super::*; use crate::codec::{Value as Payload, ValueCodec}; - #[tokio::test] - async fn failed_due_admission_rolls_back_start_and_retains_original_occurrence_and_quota() { - let directory = tempfile::tempdir().unwrap(); - let file = directory.path().join("schedule.sqlite"); - let pool = super::super::sqlite::open(file.to_str().unwrap()) - .await - .unwrap(); - let store = Store::::new(pool); + /// A one-run interval schedule whose stored occurrence is already due. + async fn due_schedule(store: &Store, schedule: &str, overdue_seconds: i64) -> DateTime { let input = ValueCodec::new() .unwrap() .encode(&Payload::Array(vec![Payload::Long(9007199254740993)])) .unwrap(); - let definition = schedules::definition(&json!({"schedule_id":"atomic-occurrence", + let definition = schedules::definition(&json!({"schedule_id":schedule, "spec":{"intervals":[{"every":"PT2S"}],"timezone":"UTC"}, "action":{"workflow_type":"echo","task_queue":"test","input":{"codec":"avro","blob":input}, "execution_timeout_seconds":3600,"run_timeout_seconds":600}, @@ -690,16 +708,51 @@ mod tests { .create_schedule(definition, json!({"test":"original creation"}), json!({})) .await .unwrap(); - // Make the stored occurrence due now. - let occurrence = now() - chrono::Duration::seconds(4); - sqlx::query("UPDATE workflow_schedules SET next_fire_at=?") + make_due(store, schedule, overdue_seconds).await + } + + /// Makes the schedule's stored occurrence due and returns it, in whole + /// seconds as the schedule description reports it. + async fn make_due(store: &Store, schedule: &str, overdue_seconds: i64) -> DateTime { + let occurrence = DateTime::from_timestamp(now().timestamp() - overdue_seconds, 0).unwrap(); + sqlx::query("UPDATE workflow_schedules SET next_fire_at=? WHERE schedule_id=?") .bind(occurrence.format("%Y-%m-%d %H:%M:%S%.6f").to_string()) + .bind(schedule) .execute(&store.pool) .await .unwrap(); - sqlx::query("CREATE TRIGGER reject_schedule_audit BEFORE INSERT ON workflow_schedule_history_events WHEN NEW.event_type='ScheduleTriggered' BEGIN SELECT RAISE(ABORT,'injected schedule audit failure'); END") + occurrence + } + + async fn count(store: &Store, table: &str) -> i64 { + // Compiled test-only table names, never request input. + sqlx::query_scalar(sqlx::AssertSqlSafe(format!("SELECT COUNT(*) FROM {table}"))) + .fetch_one(&store.pool) + .await + .unwrap() + } + + fn instant(value: &Value) -> DateTime { + DateTime::parse_from_rfc3339(value.as_str().unwrap()) + .unwrap() + .with_timezone(&Utc) + } + + async fn sqlite_store(file: &std::path::Path) -> Store { + Store::::new(super::super::sqlite::open(file.to_str().unwrap()).await.unwrap()) + } + + #[tokio::test] + async fn failed_due_admission_rolls_back_start_keeps_quota_and_records_the_failure() { + let directory = tempfile::tempdir().unwrap(); + let store = sqlite_store(&directory.path().join("schedule.sqlite")).await; + due_schedule(&store, "atomic-occurrence", 4).await; + // A database failure that is not transient: the trigger aborts the + // admission's audit row (SQLITE_CONSTRAINT_TRIGGER). + sqlx::query("CREATE TRIGGER reject_schedule_audit BEFORE INSERT ON workflow_schedule_history_events WHEN NEW.event_type='ScheduleTriggered' BEGIN SELECT RAISE(ABORT,'injected schedule audit failure: secret payload'); END") .execute(&store.pool).await.unwrap(); - assert!(store.fire_due_schedules().await.is_err()); + let failed_at = now(); + store.fire_due_schedules().await.unwrap(); for table in [ "workflow_instances", "workflow_runs", @@ -707,59 +760,132 @@ mod tests { "workflow_commands", "workflow_history_events", ] { - // Compiled test-only table names, never request input. - let count: i64 = - sqlx::query_scalar(sqlx::AssertSqlSafe(format!("SELECT COUNT(*) FROM {table}"))) - .fetch_one(&store.pool) - .await - .unwrap(); - assert_eq!( - count, 0, - "partial {table} survived failed schedule admission" - ); + assert_eq!(count(&store, table).await, 0, "partial {table} survived failed schedule admission"); } - let before = store.describe_schedule("atomic-occurrence").await.unwrap(); - assert_eq!(before["fires_count"], 0); - assert_eq!(before["remaining_actions"], 1); - assert_eq!(before["status"], "active"); - assert_eq!( - chrono::DateTime::parse_from_rfc3339(before["next_fire_at"].as_str().unwrap()).unwrap(), - chrono::DateTime::parse_from_rfc3339(&iso8601(occurrence)).unwrap() - ); - assert_eq!( - store - .schedule_history("atomic-occurrence", 0, 100) - .await - .unwrap()["events"] - .as_array() - .unwrap() - .len(), - 1 - ); + let after = store.describe_schedule("atomic-occurrence").await.unwrap(); + assert_eq!(after["fires_count"], 0); + assert_eq!(after["remaining_actions"], 1, "the rolled-back start kept its quota"); + assert_eq!(after["status"], "active"); + assert_eq!(after["failures_count"], 1); + let recent = after["info"]["recent_actions"].as_array().unwrap(); + assert_eq!(recent.len(), 1); + assert_eq!(recent[0]["outcome"], "failed"); + // The database's text, which can carry payloads, stays out of view. + assert_eq!(recent[0]["reason"], "storage_unavailable"); + assert!(!after.to_string().contains("secret payload")); + // As PHP's recordFailure: the schedule moves to its next occurrence. + assert!(instant(&after["next_fire_at"]) > failed_at); + let history = store.schedule_history("atomic-occurrence", 0, 100).await.unwrap(); + assert_eq!(history["events"].as_array().unwrap().len(), 1); sqlx::query("DROP TRIGGER reject_schedule_audit") .execute(&store.pool) .await .unwrap(); + // The kept quota starts exactly one run at the next due occurrence. + let next = make_due(&store, "atomic-occurrence", 1).await; store.fire_due_schedules().await.unwrap(); + make_due(&store, "atomic-occurrence", 1).await; store.fire_due_schedules().await.unwrap(); - let history = store - .schedule_history("atomic-occurrence", 0, 100) + let history = store.schedule_history("atomic-occurrence", 0, 100).await.unwrap(); + assert_eq!(history["events"].as_array().unwrap().len(), 3); + assert_eq!(history["events"][1]["payload"]["occurrence_time"], occurrence_text(next)); + assert_eq!(history["events"][2]["payload"]["reason"], "max_runs_exhausted"); + assert_eq!(count(&store, "workflow_runs").await, 1); + store.close().await; + } + + #[tokio::test] + async fn a_schedule_that_always_fails_does_not_stop_the_rest_of_the_batch() { + // #955: the failing schedule is first in due order on every tick. + let directory = tempfile::tempdir().unwrap(); + let store = sqlite_store(&directory.path().join("schedule.sqlite")).await; + due_schedule(&store, "always-fails", 60).await; + due_schedule(&store, "healthy", 30).await; + let broken: String = + sqlx::query_scalar("SELECT id FROM workflow_schedules WHERE schedule_id='always-fails'") + .fetch_one(&store.pool) + .await + .unwrap(); + // The id is a generated ULID, never request input. + sqlx::query(sqlx::AssertSqlSafe(format!("CREATE TRIGGER reject_one_schedule BEFORE INSERT ON workflow_schedule_history_events WHEN NEW.event_type='ScheduleTriggered' AND NEW.workflow_schedule_id='{broken}' BEGIN SELECT RAISE(ABORT,'injected'); END"))) + .execute(&store.pool).await.unwrap(); + for _ in 0..3 { + store.fire_due_schedules().await.unwrap(); + make_due(&store, "always-fails", 60).await; + } + // Its one allowed run ends it, so its row is read directly. + let (fires, failures): (i64, i64) = sqlx::query_as( + "SELECT fires_count,failures_count FROM workflow_schedules WHERE schedule_id='healthy'", + ) + .fetch_one(&store.pool) + .await + .unwrap(); + assert_eq!(fires, 1, "the later schedule fired despite the failing one"); + assert_eq!(failures, 0); + let failing = store.describe_schedule("always-fails").await.unwrap(); + assert_eq!(failing["fires_count"], 0); + assert_eq!(failing["failures_count"], 3, "each tick recorded the failure and went on"); + assert_eq!(failing["remaining_actions"], 1); + assert_eq!(count(&store, "workflow_runs").await, 1); + store.close().await; + } + + #[tokio::test] + async fn lock_pressure_keeps_the_occurrence_and_fails_the_tick_without_recording() { + let directory = tempfile::tempdir().unwrap(); + let file = directory.path().join("schedule.sqlite"); + let store = sqlite_store(&file).await; + let occurrence = due_schedule(&store, "locked", 4).await; + store.close().await; + // The same database with a short busy wait, while another writer + // holds SQLite's single write lock: SQLITE_BUSY. + let options = sqlx::sqlite::SqliteConnectOptions::new() + .filename(&file) + .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal) + .foreign_keys(true) + .busy_timeout(std::time::Duration::from_millis(50)); + let pool = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(2) + .connect_with(options.clone()) .await .unwrap(); - assert_eq!(history["events"].as_array().unwrap().len(), 3); + let store = Store::::new(pool); + let mut holder = ::connect_with(&options) + .await + .unwrap(); + sqlx::query("BEGIN IMMEDIATE").execute(&mut holder).await.unwrap(); + let error = store.fire_due_schedules().await.unwrap_err(); assert_eq!( - history["events"][1]["payload"]["occurrence_time"], - occurrence_text(occurrence) + super::super::worker_incarnations::transient_reason(&error), + Some("backend_lock_pressure") ); - assert_eq!( - history["events"][2]["payload"]["reason"], - "max_runs_exhausted" + sqlx::query("ROLLBACK").execute(&mut holder).await.unwrap(); + let after = store.describe_schedule("locked").await.unwrap(); + assert_eq!(after["failures_count"], 0, "lock pressure is retried, not recorded"); + assert_eq!(after["info"]["recent_actions"], json!([])); + assert_eq!(instant(&after["next_fire_at"]), occurrence, "the original occurrence is kept"); + // The same real BUSY error, settled once the lock is free again: the + // transient rule, not the held lock, is what keeps it unrecorded. + sqlx::query("BEGIN IMMEDIATE").execute(&mut holder).await.unwrap(); + let busy = RuntimeError::from( + sqlx::query("UPDATE workflow_schedules SET updated_at=updated_at") + .execute(&store.pool) + .await + .unwrap_err(), ); - let runs: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM workflow_runs") + sqlx::query("ROLLBACK").execute(&mut holder).await.unwrap(); + let key: String = sqlx::query_scalar("SELECT id FROM workflow_schedules WHERE schedule_id='locked'") .fetch_one(&store.pool) .await .unwrap(); - assert_eq!(runs, 1); + assert!(store.settle_failed_admission(&key, busy).await.is_err()); + let after = store.describe_schedule("locked").await.unwrap(); + assert_eq!(after["failures_count"], 0); + assert_eq!(instant(&after["next_fire_at"]), occurrence); + // The next tick admits that same occurrence. + store.fire_due_schedules().await.unwrap(); + let history = store.schedule_history("locked", 0, 100).await.unwrap(); + assert_eq!(history["events"][1]["payload"]["occurrence_time"], occurrence_text(occurrence)); store.close().await; } } From b0e523d55ff909339a74dee28afccb85d6c70121 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:02:55 -0600 Subject: [PATCH 2/2] Treat backend resource exhaustion as transient in schedule settlement Security hold on #995: settle_failed_admission advances a schedule on any classifier miss, so backend-wide outages left outside transient_reason were recorded as the schedule's own failure. Classify PostgreSQL SQLSTATE class 53 (insufficient resources) and MySQL 1040, 1203, 1041, 1037, 1135 and 1021 as backend_unavailable, and let an unlisted MySQL number fall back to its SQLSTATE class instead of returning early. Add exact classifier counterfactuals and a settlement test that a backend outage leaves failures_count, recent_actions and the stored occurrence unchanged. Co-Authored-By: Claude Opus 5.5 --- rust/src/runtime/schedule_execution.rs | 27 ++++++++++++ rust/src/runtime/worker_incarnations.rs | 56 ++++++++++++++++++------- 2 files changed, 69 insertions(+), 14 deletions(-) diff --git a/rust/src/runtime/schedule_execution.rs b/rust/src/runtime/schedule_execution.rs index 9a4ab7708..8b7e7f39c 100644 --- a/rust/src/runtime/schedule_execution.rs +++ b/rust/src/runtime/schedule_execution.rs @@ -830,6 +830,33 @@ mod tests { store.close().await; } + #[tokio::test] + async fn backend_unavailability_keeps_the_occurrence_without_recording() { + let directory = tempfile::tempdir().unwrap(); + let store = sqlite_store(&directory.path().join("schedule.sqlite")).await; + let occurrence = due_schedule(&store, "outage", 4).await; + let key: String = sqlx::query_scalar("SELECT id FROM workflow_schedules WHERE schedule_id='outage'") + .fetch_one(&store.pool) + .await + .unwrap(); + // A backend-wide outage observed by one admission, settled while the + // database is reachable again: it must still not be recorded as this + // schedule's failure or consume the occurrence. + for error in [sqlx::Error::PoolTimedOut, sqlx::Error::PoolClosed] { + let error = RuntimeError::from(error); + assert_eq!( + super::super::worker_incarnations::transient_reason(&error), + Some("backend_unavailable") + ); + assert!(store.settle_failed_admission(&key, error).await.is_err()); + } + let after = store.describe_schedule("outage").await.unwrap(); + assert_eq!(after["failures_count"], 0, "an outage is retried, not recorded"); + assert_eq!(after["info"]["recent_actions"], json!([])); + assert_eq!(instant(&after["next_fire_at"]), occurrence, "the original occurrence is kept"); + store.close().await; + } + #[tokio::test] async fn lock_pressure_keeps_the_occurrence_and_fails_the_tick_without_recording() { let directory = tempfile::tempdir().unwrap(); diff --git a/rust/src/runtime/worker_incarnations.rs b/rust/src/runtime/worker_incarnations.rs index 2100717f3..5eaddbeb3 100644 --- a/rust/src/runtime/worker_incarnations.rs +++ b/rust/src/runtime/worker_incarnations.rs @@ -23,29 +23,53 @@ pub(super) fn transient_reason(error: &RuntimeError) -> Option<&'static str> { ) => Some("backend_unavailable"), RuntimeError::Database(sqlx::Error::Database(error)) => { if let Some(mysql) = error.try_downcast_ref::() { - return mysql_transient_reason(mysql.number()); + // An unlisted number still falls back to its SQLSTATE class, + // so a connection-class refusal is never schedule-local. + return mysql_transient_reason(mysql.number()) + .or_else(|| mysql.code().and_then(sqlstate_transient_reason)); } let code = error.code()?; - if matches!(code.as_ref(), "55P03" | "40P01" | "40001") - || (error - .try_downcast_ref::() - .is_some() - && code - .parse::() - .is_ok_and(|code| matches!(code & 0xff, 5 | 6))) + if error.try_downcast_ref::().is_some() + && code + .parse::() + .is_ok_and(|code| matches!(code & 0xff, 5 | 6)) { Some("backend_lock_pressure") - } else if code.starts_with("08") || matches!(code.as_ref(), "57P01" | "57P02" | "57P03") - { - Some("backend_unavailable") } else { - None + sqlstate_transient_reason(code.as_ref()) } } _ => None, } } +/// SQLSTATE values that describe a passing backend condition. Class 08 is a +/// connection exception and class 53 is PostgreSQL's insufficient resources +/// (disk full, out of memory, too many connections, configuration limit +/// exceeded): both are backend-wide, never one schedule's fault (#955). +pub(super) fn sqlstate_transient_reason(code: &str) -> Option<&'static str> { + if matches!(code, "55P03" | "40P01" | "40001") { + Some("backend_lock_pressure") + } else if code.starts_with("08") || code.starts_with("53") || matches!(code, "57P01" | "57P02" | "57P03") { + Some("backend_unavailable") + } else { + None + } +} + +#[test] +fn backend_resource_sqlstates_are_transient() { + for code in ["53000", "53100", "53200", "53300", "53400", "08004", "08006", "57P01"] { + assert_eq!(sqlstate_transient_reason(code), Some("backend_unavailable"), "{code}"); + } + for code in ["55P03", "40P01", "40001"] { + assert_eq!(sqlstate_transient_reason(code), Some("backend_lock_pressure"), "{code}"); + } + for code in ["23505", "22001", "42P01", "HY000", "42S02"] { + assert_eq!(sqlstate_transient_reason(code), None, "{code}"); + } +} + /// MySQL server and client error numbers that describe a passing backend /// condition rather than a request fault. Beyond lock waits and lost /// connections, an InnoDB Cluster primary switch behind MySQL Router can @@ -53,18 +77,22 @@ pub(super) fn transient_reason(error: &RuntimeError) -> Option<&'static str> { /// that is (now) `super_read_only`, 1927 connection killed, 2055 connection /// lost, 3100 a Group Replication commit hook refusal while the member has no /// quorum, 4031 a server-side disconnect; and 3101 a transaction rolled back -/// at commit by group certification, which a retry can pass. +/// at commit by group certification, which a retry can pass. Server resource +/// exhaustion is backend-wide too: 1021 disk full, 1037 out of memory, 1040 +/// too many connections, 1041 out of resources, 1135 cannot create a thread, +/// 1203 the account's connection limit. pub(super) fn mysql_transient_reason(number: u16) -> Option<&'static str> { match number { 1205 | 1213 | 3101 => Some("backend_lock_pressure"), 1053 | 1290 | 1927 | 2002 | 2003 | 2006 | 2013 | 2055 | 3100 | 4031 => Some("backend_unavailable"), + 1021 | 1037 | 1040 | 1041 | 1135 | 1203 => Some("backend_unavailable"), _ => None, } } #[test] fn mysql_cluster_failover_errors_are_transient() { - for number in [1053, 1290, 1927, 2002, 2003, 2006, 2013, 2055, 3100, 4031] { + for number in [1053, 1290, 1927, 2002, 2003, 2006, 2013, 2055, 3100, 4031, 1021, 1037, 1040, 1041, 1135, 1203] { assert_eq!(mysql_transient_reason(number), Some("backend_unavailable"), "{number}"); } for number in [1205, 1213, 3101] {