diff --git a/rust/src/runtime/schedule_execution.rs b/rust/src/runtime/schedule_execution.rs index 2c5ef80e5..8b7e7f39c 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,159 @@ 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 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(); - assert_eq!(history["events"].as_array().unwrap().len(), 3); + // 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(); + 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(); + 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; } } 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] {