diff --git a/CHANGELOG.md b/CHANGELOG.md index 137768c43..f5599590a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,22 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ### Added +- **Durable execution**: added a crash-orphan sweep to the durable retention loop (`Journal::sweep_orphans`, + #6254) that reclaims `status='running'` executions whose owner process died without finalizing — + previously invisible to the TTL prune (which only ever considers `finalized_at IS NOT NULL` rows) and + stuck `running` forever. A `status='running'` row whose `updated_at` is older than the new + `[durable.retention] stale_running_after_secs` (default 3600s, `0` disables the sweep) becomes a + candidate; it is hard-aborted only after a non-blocking try-acquire of its INV-15 advisory + `ExecutionLock` succeeds — a live owner (`ExecutionLocked`) short-circuits to skip, since staleness of + `updated_at` alone never proves the owner is dead. The sweep runs before `prune()` on every retention + tick (same supervised loop, no new spawn site) and is a documented no-op (warn-once) on backends + without an on-disk lock directory (`:memory:`, Postgres, non-Unix). `zeph durable prune` now runs the + sweep before the TTL prune and `--dry-run` reports both counts separately. +- **Durable execution**: `zeph-orchestration`'s `journal_budget` (P2) and `zeph-scheduler`'s + `fire_with_durable` (P3) now open their execution via `open_execution_exclusive` instead of the plain + `open_execution`, making their `DagRun`/`ScheduledJob` rows' liveness observable to the crash-orphan + sweep; on `DurableError::ExecutionLocked` both adapters log and return `Ok(())` — a graceful skip, never + a task failure or retry. - **TUI**: added a read-only settings view (`S` key or the `settings` command-palette entry) listing configured LLM providers, MCP servers, and sub-agent definitions in three tabs, sourced as a live snapshot from `MetricsSnapshot` (never re-parsed from @@ -290,6 +306,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). bug in replan execution — surfaced by the new end-to-end test added for #6287's review pass. Gap-task IDs are now remapped to local 0-based IDs for the partial scheduler run and back to the original global IDs on the way out. +- **Durable execution**: `open_execution`/`open_execution_exclusive`'s reopen path un-finalized only + `completed`/`failed` rows, leaving `aborted` rows untouched on reopen (INV-16). This was safe before + #6254 because `aborted` was a rare, immediately-redriven outcome, but became a hazard once the new + crash-orphan sweep makes `aborted` the common outcome of a resumable crash: a resumed execution whose + row kept `finalized_at` set was prunable out from under the active resume — the exact hazard the + completed/failed un-finalize was built to prevent. Reopening now resets `status='running'` and clears + `finalized_at` for a row in ANY terminal status. - **Worktree**: `--bare` silently skipped the entire worktree subsystem bootstrap (`WorktreeManager` construction, `probe_capabilities`) with no warning when `worktree.enabled = true` in the active config — the 6th confirmed instance of the `--bare` diff --git a/config/default.toml b/config/default.toml index 433b4d4d3..e9a440758 100644 --- a/config/default.toml +++ b/config/default.toml @@ -1765,6 +1765,10 @@ max_journal_bytes = 1073741824 prune_batch_size = 500 # Background prune poll interval (seconds). prune_interval_secs = 3600 +# Crash-orphan threshold (#6254): a running execution whose owner process died without +# finalizing becomes a sweep candidate after this many seconds of inactivity, subject to an +# advisory-lock liveness check before it is aborted. 0 disables the sweep. +stale_running_after_secs = 3600 [caveman] # Start every session in ultra-compressed (telegraphic) output mode. diff --git a/crates/zeph-config/src/durable.rs b/crates/zeph-config/src/durable.rs index ac5eded21..1c962749f 100644 --- a/crates/zeph-config/src/durable.rs +++ b/crates/zeph-config/src/durable.rs @@ -149,6 +149,10 @@ pub struct RetentionPolicy { pub prune_batch_size: u64, /// Background prune poll interval, in seconds. pub prune_interval_secs: u64, + /// Crash-orphan threshold, in seconds (#6254): a `status='running'` row whose `updated_at` + /// is older than this becomes a sweep candidate, subject to an INV-15 flock liveness check + /// before it is aborted. `0` disables the sweep entirely. + pub stale_running_after_secs: u64, } impl Default for RetentionPolicy { @@ -160,6 +164,7 @@ impl Default for RetentionPolicy { max_journal_bytes: 1_073_741_824, prune_batch_size: 500, prune_interval_secs: 3600, + stale_running_after_secs: 3600, } } } @@ -196,6 +201,7 @@ mod tests { assert_eq!(cfg.retention.max_journal_bytes, 1_073_741_824); assert_eq!(cfg.retention.prune_batch_size, 500); assert_eq!(cfg.retention.prune_interval_secs, 3600); + assert_eq!(cfg.retention.stale_running_after_secs, 3600); } #[test] diff --git a/crates/zeph-config/src/migrate/infra.rs b/crates/zeph-config/src/migrate/infra.rs index 753a3a87b..f8f9a45d5 100644 --- a/crates/zeph-config/src/migrate/infra.rs +++ b/crates/zeph-config/src/migrate/infra.rs @@ -668,7 +668,8 @@ pub fn migrate_durable_config(toml_src: &str) -> Result Result Result { + // Anchored multiline pattern: matches `[durable.retention]` with optional inline comment, + // followed by LF or CRLF. + static DURABLE_RETENTION_HEADER_RE: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + Regex::new(r"(?m)^[ \t]*\[durable\.retention\][ \t]*(?:#[^\r\n]*)?\r?\n") + .expect("static pattern") + }); + + if !section_header_present(toml_src, "durable.retention") { + return Ok(MigrationResult { + output: toml_src.to_owned(), + changed_count: 0, + sections_changed: Vec::new(), + }); + } + + let already_present = toml_src.lines().any(|l| { + l.trim() + .trim_start_matches('#') + .trim() + .starts_with("stale_running_after_secs") + }); + if already_present || !DURABLE_RETENTION_HEADER_RE.is_match(toml_src) { + return Ok(MigrationResult { + output: toml_src.to_owned(), + changed_count: 0, + sections_changed: Vec::new(), + }); + } + + let comment = "# stale_running_after_secs = 3600 # crash-orphan threshold (#6254); 0 disables \ + the sweep\n"; + let output = DURABLE_RETENTION_HEADER_RE + .replacen(toml_src, 1, |caps: ®ex::Captures| { + format!("{}{comment}", &caps[0]) + }) + .into_owned(); + + let changed = output != toml_src; + let changed_count = usize::from(changed); + Ok(MigrationResult { + output, + changed_count, + sections_changed: if changed { + vec!["durable.retention.stale_running_after_secs".to_owned()] + } else { + Vec::new() + }, + }) +} + /// Adds a commented-out `[security.content_isolation.nli]` section to configs that predate the /// SONAR NLI entailment check stage (#5438). Idempotent: no-op when the real or commented /// `[security.content_isolation.nli]` header is already present, so running `--migrate-config` diff --git a/crates/zeph-config/src/migrate/mod.rs b/crates/zeph-config/src/migrate/mod.rs index bc27eb059..8ecdc7109 100644 --- a/crates/zeph-config/src/migrate/mod.rs +++ b/crates/zeph-config/src/migrate/mod.rs @@ -601,33 +601,34 @@ use steps::{ MigrateAcpSubagentsConfig, MigrateAgentBudgetHint, MigrateAgentRetryToToolsRetry, MigrateAutodreamConfig, MigrateCavemanConfig, MigrateCocoonProviderNotice, MigrateCocoonShowBalance, MigrateCompressionPredictorConfig, MigrateDatabaseUrl, - MigrateDeepLinkConfig, MigrateDurableConfig, MigrateDurableSharedDb, MigrateEgressConfig, - MigrateEmbedProviderRename, MigrateEvalModelToProvider, MigrateFidelityTimeoutDefaults, - MigrateFiveSignalConfig, MigrateFocusAutoConsolidateMinWindow, MigrateForgettingConfig, - MigrateGoalsConfig, MigrateGonkagateToGonka, MigrateHooksPermissionDeniedConfig, - MigrateHooksTurnComplete, MigrateKnowledgeConfig, MigrateLlmStreamLimits, - MigrateMagicDocsConfig, MigrateMcpElicitationConfig, MigrateMcpMaxConnectAttempts, - MigrateMcpRetryAndToolTimeout, MigrateMcpTrustLevels, MigrateMemoryGraph, - MigrateMemoryGraphRecallIncludeImported, MigrateMemoryHebbian, - MigrateMemoryHebbianConsolidation, MigrateMemoryHebbianSpread, MigrateMemoryPersonaConfig, - MigrateMemoryReasoning, MigrateMemoryReasoningJudge, MigrateMemoryRetrieval, - MigrateMemoryRetrievalQueryBias, MigrateMemoryTypeAwareCompose, MigrateMicrocompactConfig, - MigrateNliConfig, MigrateOrchestrationAssetSensitivity, MigrateOrchestrationEnsemble, - MigrateOrchestrationPersistence, MigrateOrchestratorProvider, MigrateOtelFilter, - MigratePiiFilterNames, MigratePlannerModelToProvider, MigratePolicyProviderAndUtilityWindow, - MigrateProviderMaxConcurrent, MigrateQdrantApiKey, MigrateQdrantTimeoutSecs, - MigrateQualityConfig, MigrateSandboxConfig, MigrateSandboxEgressFilter, MigrateSchedulerDaemon, - MigrateSecretMaskingConfig, MigrateServeConfig, MigrateSessionPersistProviderOverrides, - MigrateSessionPersistenceConfig, MigrateSessionProviderPersistence, MigrateSessionRecapConfig, - MigrateShadowSentinelConfig, MigrateShellCheckpointsConfig, MigrateShellTransactional, - MigrateSkillTrustRequireCheck, MigrateSkillsRegistry, MigrateSttToProvider, - MigrateSupervisorConfig, MigrateTelemetryConfig, MigrateToolsCompressionConfig, - MigrateTraceMetadata, MigrateTuiDelights, MigrateTuiMouse, MigrateTuiThemeConfig, - MigrateTuiThemeDefaults, MigrateUtilityHighGainTools, MigrateVigilConfig, - MigrateWorktreeConfig, MigrateWorktreeGitTimeout, MigrateWorktreeQuotaFields, + MigrateDeepLinkConfig, MigrateDurableConfig, MigrateDurableSharedDb, + MigrateDurableStaleRunningAfterSecs, MigrateEgressConfig, MigrateEmbedProviderRename, + MigrateEvalModelToProvider, MigrateFidelityTimeoutDefaults, MigrateFiveSignalConfig, + MigrateFocusAutoConsolidateMinWindow, MigrateForgettingConfig, MigrateGoalsConfig, + MigrateGonkagateToGonka, MigrateHooksPermissionDeniedConfig, MigrateHooksTurnComplete, + MigrateKnowledgeConfig, MigrateLlmStreamLimits, MigrateMagicDocsConfig, + MigrateMcpElicitationConfig, MigrateMcpMaxConnectAttempts, MigrateMcpRetryAndToolTimeout, + MigrateMcpTrustLevels, MigrateMemoryGraph, MigrateMemoryGraphRecallIncludeImported, + MigrateMemoryHebbian, MigrateMemoryHebbianConsolidation, MigrateMemoryHebbianSpread, + MigrateMemoryPersonaConfig, MigrateMemoryReasoning, MigrateMemoryReasoningJudge, + MigrateMemoryRetrieval, MigrateMemoryRetrievalQueryBias, MigrateMemoryTypeAwareCompose, + MigrateMicrocompactConfig, MigrateNliConfig, MigrateOrchestrationAssetSensitivity, + MigrateOrchestrationEnsemble, MigrateOrchestrationPersistence, MigrateOrchestratorProvider, + MigrateOtelFilter, MigratePiiFilterNames, MigratePlannerModelToProvider, + MigratePolicyProviderAndUtilityWindow, MigrateProviderMaxConcurrent, MigrateQdrantApiKey, + MigrateQdrantTimeoutSecs, MigrateQualityConfig, MigrateSandboxConfig, + MigrateSandboxEgressFilter, MigrateSchedulerDaemon, MigrateSecretMaskingConfig, + MigrateServeConfig, MigrateSessionPersistProviderOverrides, MigrateSessionPersistenceConfig, + MigrateSessionProviderPersistence, MigrateSessionRecapConfig, MigrateShadowSentinelConfig, + MigrateShellCheckpointsConfig, MigrateShellTransactional, MigrateSkillTrustRequireCheck, + MigrateSkillsRegistry, MigrateSttToProvider, MigrateSupervisorConfig, MigrateTelemetryConfig, + MigrateToolsCompressionConfig, MigrateTraceMetadata, MigrateTuiDelights, MigrateTuiMouse, + MigrateTuiThemeConfig, MigrateTuiThemeDefaults, MigrateUtilityHighGainTools, + MigrateVigilConfig, MigrateWorktreeConfig, MigrateWorktreeGitTimeout, + MigrateWorktreeQuotaFields, }; -/// Ordered registry of all sequential migration steps (steps 1–86). +/// Ordered registry of all sequential migration steps (steps 1–87). /// /// Each entry wraps the corresponding free function and is evaluated lazily at first access. /// The ordering is chronological; the dispatch loop in `src/commands/migrate.rs` iterates @@ -789,6 +790,9 @@ pub static MIGRATIONS: std::sync::LazyLock> // Step 86 — add [orchestration.ensemble] advisory block for ORCH-style // deterministic verifier ensemble-merge (spec 073, #6232) Box::new(MigrateOrchestrationEnsemble), + // Step 87 — add stale_running_after_secs advisory to an existing active + // [durable.retention] table for the crash-orphan sweep (spec-064, #6254) + Box::new(MigrateDurableStaleRunningAfterSecs), ] }); diff --git a/crates/zeph-config/src/migrate/steps.rs b/crates/zeph-config/src/migrate/steps.rs index 0095cb60f..672051d70 100644 --- a/crates/zeph-config/src/migrate/steps.rs +++ b/crates/zeph-config/src/migrate/steps.rs @@ -53,7 +53,9 @@ //! step 85 adds a commented `[memory.type_aware_compose]` advisory block for `MemGuard` //! type-aware retrieval composition (spec 064, #6086); //! step 86 adds a commented `[orchestration.ensemble]` advisory block for ORCH-style -//! deterministic verifier ensemble-merge (spec 073, #6232). +//! deterministic verifier ensemble-merge (spec 073, #6232); +//! step 87 adds a commented `stale_running_after_secs = 3600` advisory to an existing active +//! `[durable.retention]` table for the crash-orphan sweep (spec-064, #6254). //! //! Each struct is a zero-size type that delegates to the corresponding free function in //! `super`. They exist solely to satisfy the object-safe [`super::Migration`] trait so the @@ -66,8 +68,8 @@ use super::{ migrate_autodream_config, migrate_caveman_config, migrate_cocoon_provider_notice, migrate_cocoon_show_balance, migrate_compression_predictor_config, migrate_database_url, migrate_deep_link_config, migrate_durable_config, migrate_durable_shared_db, - migrate_egress_config, migrate_embed_provider_rename, migrate_eval_model_to_provider, - migrate_fidelity_timeout_defaults, migrate_five_signal_config, + migrate_durable_stale_running_after_secs, migrate_egress_config, migrate_embed_provider_rename, + migrate_eval_model_to_provider, migrate_fidelity_timeout_defaults, migrate_five_signal_config, migrate_focus_auto_consolidate_min_window, migrate_forgetting_config, migrate_goals_config, migrate_hooks_permission_denied_config, migrate_hooks_turn_complete_config, migrate_knowledge_config, migrate_llm_stream_limits, migrate_magic_docs_config, @@ -1073,3 +1075,16 @@ impl Migration for MigrateOrchestrationEnsemble { migrate_orchestration_ensemble(toml_src) } } + +/// Step 87 — adds a commented `stale_running_after_secs = 3600` advisory to an existing active +/// `[durable.retention]` table that predates the crash-orphan sweep (spec-064, #6254). +pub(super) struct MigrateDurableStaleRunningAfterSecs; +impl Migration for MigrateDurableStaleRunningAfterSecs { + fn name(&self) -> &'static str { + "migrate_durable_stale_running_after_secs" + } + + fn apply(&self, toml_src: &str) -> Result { + migrate_durable_stale_running_after_secs(toml_src) + } +} diff --git a/crates/zeph-config/src/migrate/tests.rs b/crates/zeph-config/src/migrate/tests.rs index bc7b8f52e..01da86fdb 100644 --- a/crates/zeph-config/src/migrate/tests.rs +++ b/crates/zeph-config/src/migrate/tests.rs @@ -9,8 +9,8 @@ use super::*; fn migrations_registry_has_all_steps() { assert_eq!( MIGRATIONS.len(), - 86, - "MIGRATIONS registry must contain all 86 sequential steps" + 87, + "MIGRATIONS registry must contain all 87 sequential steps" ); for m in MIGRATIONS.iter() { assert!( @@ -1817,7 +1817,7 @@ fn migrate_focus_auto_consolidate_noop_when_only_commented_section() { #[test] fn registry_has_fifty_entries() { - assert_eq!(MIGRATIONS.len(), 86); + assert_eq!(MIGRATIONS.len(), 87); } #[test] @@ -1855,7 +1855,7 @@ fn registry_is_idempotent_on_empty_input() { #[test] fn registry_preserves_order_matches_dispatch() { - // Names must follow the documented step order (steps 1–84). + // Names must follow the documented step order (steps 1–87). let expected = [ "migrate_stt_to_provider", "migrate_planner_model_to_provider", @@ -1943,6 +1943,7 @@ fn registry_preserves_order_matches_dispatch() { "migrate_a2a_server_remove_inert_fields", "migrate_memory_type_aware_compose_config", "migrate_orchestration_ensemble", + "migrate_durable_stale_running_after_secs", ]; let actual: Vec<&str> = MIGRATIONS.iter().map(|m| m.name()).collect(); assert_eq!(actual, expected); @@ -4142,6 +4143,67 @@ fn step_79_still_adds_advisory_when_unsafe_topology_detected() { ); } +// ── migrate_durable_stale_running_after_secs tests (step 87, #6254) ────── + +#[test] +fn step_87_adds_commented_advisory_when_retention_active_and_missing_field() { + let src = "[durable.retention]\nttl_completed_secs = 604800\n"; + let result = migrate_durable_stale_running_after_secs(src).expect("migrate"); + assert_eq!(result.changed_count, 1); + assert!(result.output.contains("# stale_running_after_secs = 3600")); + assert!(!result.output.contains("\nstale_running_after_secs =")); + assert!(result.output.contains("ttl_completed_secs = 604800")); + assert_eq!( + result.sections_changed, + vec!["durable.retention.stale_running_after_secs".to_owned()] + ); +} + +#[test] +fn step_87_noop_when_field_already_present() { + let src = "[durable.retention]\nstale_running_after_secs = 7200\n"; + let result = migrate_durable_stale_running_after_secs(src).expect("migrate"); + assert_eq!(result.changed_count, 0); + assert_eq!(result.output, src); +} + +#[test] +fn step_87_noop_when_field_comment_already_present() { + let src = "[durable.retention]\n# stale_running_after_secs = 3600\n"; + let result = migrate_durable_stale_running_after_secs(src).expect("migrate"); + assert_eq!(result.changed_count, 0); + assert_eq!(result.output, src); +} + +#[test] +fn step_87_noop_when_durable_retention_section_absent() { + let src = "[durable]\nenabled = true\n"; + let result = migrate_durable_stale_running_after_secs(src).expect("migrate"); + assert_eq!(result.changed_count, 0); + assert_eq!(result.output, src); +} + +#[test] +fn step_87_noop_when_durable_retention_only_commented_advisory() { + let src = "# [durable.retention]\n# ttl_completed_secs = 604800\n"; + let result = migrate_durable_stale_running_after_secs(src).expect("migrate"); + assert_eq!(result.changed_count, 0); + assert_eq!(result.output, src); +} + +#[test] +fn step_87_idempotent_on_own_output() { + let src = "[durable.retention]\nttl_completed_secs = 604800\n"; + let first = migrate_durable_stale_running_after_secs(src).expect("first migrate"); + assert_eq!(first.changed_count, 1); + let second = migrate_durable_stale_running_after_secs(&first.output).expect("second migrate"); + assert_eq!(second.changed_count, 0, "second run must be a no-op"); + assert_eq!( + second.output, first.output, + "output unchanged on second run" + ); +} + // ── is_unsafe_shared_topology tests (#6042) ─────────────────────────────── #[test] diff --git a/crates/zeph-durable/src/backend.rs b/crates/zeph-durable/src/backend.rs index 655e96ac7..be7dbc210 100644 --- a/crates/zeph-durable/src/backend.rs +++ b/crates/zeph-durable/src/backend.rs @@ -230,6 +230,12 @@ impl Journal for DurableBackendEnum { Self::Local(backend) => backend.prune(policy).await, } } + + async fn sweep_orphans(&self, policy: &RetentionPolicy) -> Result { + match self { + Self::Local(backend) => backend.sweep_orphans(policy).await, + } + } } impl ExecutionBackend for DurableBackendEnum { diff --git a/crates/zeph-durable/src/backend/local.rs b/crates/zeph-durable/src/backend/local.rs index a419b9c8e..4a503263b 100644 --- a/crates/zeph-durable/src/backend/local.rs +++ b/crates/zeph-durable/src/backend/local.rs @@ -120,6 +120,10 @@ pub struct LocalBackend { /// [`LocalBackend::new`] from a caller-supplied pool, or a non-SQLite (Postgres) deployment, /// where a filesystem lock file cannot express cross-process exclusivity anyway. lock_dir: Option, + /// Set once [`sweep_orphans`](Self::sweep_orphans) has emitted its warn-once log for a + /// `lock_dir = None` backend (#6254), so a background retention tick every + /// `prune_interval_secs` does not spam the log for the lifetime of the process. + orphan_sweep_warned: std::sync::atomic::AtomicBool, } impl fmt::Debug for LocalBackend { @@ -149,6 +153,7 @@ impl LocalBackend { promise_waiters: NotifyRegistry::default(), timer_waiters: NotifyRegistry::default(), lock_dir: None, + orphan_sweep_warned: std::sync::atomic::AtomicBool::new(false), } } @@ -351,21 +356,63 @@ impl LocalBackend { Ok(count.max(0).cast_unsigned()) } + /// Count crash-orphaned executions a [`sweep_orphans`](Journal::sweep_orphans) sweep would + /// abort under `policy` (#6254). + /// + /// Read-only: backs `zeph durable prune --dry-run`. Mirrors the real sweep's staleness scan + /// and INV-15 flock liveness check (acquiring and immediately releasing each candidate's + /// `ExecutionLock`, exactly as the real sweep does, so the count reflects genuinely + /// unowned rows rather than staleness alone) — but never mutates `status`. Returns `0` when + /// the sweep is disabled (`stale_running_after_secs == 0`) or this backend has no `lock_dir`. + /// + /// # Errors + /// + /// Returns [`DurableError::Storage`] if the query fails. + pub async fn count_orphans(&self, policy: &RetentionPolicy) -> Result { + if policy.stale_running_after_secs == 0 { + return Ok(0); + } + let Some(lock_dir) = self.lock_dir.clone() else { + return Ok(0); + }; + let cutoff_ms = orphan_cutoff_ms(policy, now_unix_millis()); + let candidates: Vec<(String,)> = zeph_db::query_as(sql!( + "SELECT execution_id FROM durable_executions WHERE status = 'running' AND updated_at <= ?" + )) + .bind(cutoff_ms) + .fetch_all(&self.pool) + .await + .map_err(|e| DurableError::storage("count_orphans", e))?; + let mut count = 0u64; + for (exec_str,) in &candidates { + let Ok(execution_id) = parse_execution_id(exec_str) else { + continue; + }; + if ExecutionLock::acquire(&lock_dir, execution_id).is_ok() { + count += 1; + } + } + Ok(count) + } + /// Ensure a `durable_executions` row exists for `id`, returning whether this is a resume. /// /// Inserts a fresh `running` row for a new execution (returning `false`) or detects an existing /// row for a resumed one (returning `true`). The journal's foreign key requires this row before /// any entry is appended, so callers open the execution first. /// - /// Reopening a row previously [`finalize`](Journal::finalize)d as `completed` or `failed` - /// un-finalizes it: status resets to `running` and `finalized_at` clears. A caller reopening an - /// execution is, by definition, still using it, so the retention sweep (gated on - /// `finalized_at`) must not consider it prunable while it does — without this, a long-lived - /// execution finalized at one process's graceful shutdown and legitimately resumed by a later - /// process (e.g. a per-conversation `AgentTurn` execution) would keep a stale `finalized_at` - /// and could be pruned out from under its still-active journal. `aborted` rows are left - /// untouched: divergence recovery reopens the same row on purpose and starts a fresh replay - /// cursor without needing the status reset. + /// Reopening a row previously [`finalize`](Journal::finalize)d as `completed`, `failed`, or + /// `aborted` un-finalizes it: status resets to `running` and `finalized_at` clears (INV-16, + /// #6254). A caller reopening an execution is, by definition, still using it, so the retention + /// sweep (gated on `finalized_at`) must not consider it prunable while it does — without this, + /// a long-lived execution finalized at one process's graceful shutdown and legitimately resumed + /// by a later process (e.g. a per-conversation `AgentTurn` execution) would keep a stale + /// `finalized_at` and could be pruned out from under its still-active journal. `aborted` rows + /// are included because the crash-orphan sweep (INV-17) makes `aborted` the common outcome of a + /// resumable crash: a resumed execution whose row keeps `finalized_at` set is prunable out from + /// under the active resume — the exact hazard this un-finalize prevents for `completed`/`failed`. + /// This is also strictly safer for the pre-existing divergence-recovery case, which reopens an + /// `aborted` row on purpose: it now also protects that fresh re-drive from prune. /// /// The un-finalize is attempted as a single guarded `UPDATE` (no preceding `SELECT`) so there /// is no read-then-write window against a concurrent prune sweep (#6251 critic S1): if the row @@ -399,7 +446,7 @@ impl LocalBackend { // "observe completed/failed" and "reset to running" for a concurrent prune to act in. let reopened = zeph_db::query(sql!( "UPDATE durable_executions SET status = 'running', updated_at = ?, finalized_at = NULL - WHERE execution_id = ? AND status IN ('completed', 'failed')" + WHERE execution_id = ? AND status IN ('completed', 'failed', 'aborted')" )) .bind(now_unix_millis()) .bind(&exec) @@ -412,9 +459,10 @@ impl LocalBackend { } // Zero rows: either the row doesn't exist, or it exists but wasn't terminal (already - // `running`/`aborted`, no reset needed). Distinguish the two — if a concurrent prune - // deleted a terminal row between any earlier observation and this check, this SELECT - // sees the authoritative post-delete state instead of a stale belief that it's there. + // `running`, no reset needed — every terminal status is covered by the UPDATE above). + // Distinguish the two — if a concurrent prune deleted a terminal row between any + // earlier observation and this check, this SELECT sees the authoritative post-delete + // state instead of a stale belief that it's there. let existing: Option<(String,)> = zeph_db::query_as(sql!( "SELECT status FROM durable_executions WHERE execution_id = ?" )) @@ -1176,6 +1224,96 @@ impl LocalBackend { Ok(removed) } + /// One batch of the crash-orphan sweep (INV-17, #6254). + /// + /// Selects up to `batch` `status='running'` rows whose `updated_at` is at or before + /// `cutoff_ms`, then for each candidate non-blockingly try-acquires its INV-15 + /// `ExecutionLock`: `ExecutionLocked` (a live owner holds it) short-circuits to skip — + /// staleness of `updated_at` alone is never sufficient grounds to abort. Only when the lock is + /// acquired does the guarded `UPDATE` run, still holding the lock, so the abort is race-free + /// against a concurrent `open_execution_exclusive` reopen for the same id (both require the + /// same non-reentrant flock). The lock releases when it drops at the end of each loop + /// iteration. + /// + /// `cursor` is the previous batch's [`SweepCursor`](crate::retention::SweepCursor) (`None` for + /// the first batch); the candidate scan is keyset-paginated strictly past it so a skipped + /// (lock-held) row is never re-selected by a later batch — #6254 C1: without this, a batch + /// consisting entirely of lock-held rows would re-select the identical rows on every + /// iteration and the caller's batch loop would never terminate. Returns the number of rows + /// scanned (for the caller's batch-continuation decision), the number actually aborted, and + /// the cursor to resume from on the next call. + async fn sweep_orphan_batch( + &self, + lock_dir: &std::path::Path, + cutoff_ms: i64, + batch: u64, + cursor: Option, + ) -> Result { + // Sentinel "no lower bound" cursor: every real `updated_at` (Unix ms) is > i64::MIN, so + // this keyset predicate is a no-op on the first batch while still using one static, + // sql!()-cacheable query for both the first and subsequent calls. + let (after_updated_at, after_exec) = cursor.map_or((i64::MIN, String::new()), |c| { + (c.updated_at_ms, c.execution_id) + }); + + let candidates: Vec<(String, i64)> = zeph_db::query_as(sql!( + "SELECT execution_id, updated_at FROM durable_executions + WHERE status = 'running' AND updated_at <= ? + AND (updated_at > ? OR (updated_at = ? AND execution_id > ?)) + ORDER BY updated_at, execution_id LIMIT ?" + )) + .bind(cutoff_ms) + .bind(after_updated_at) + .bind(after_updated_at) + .bind(&after_exec) + .bind(i64::try_from(batch).unwrap_or(i64::MAX)) + .fetch_all(&self.pool) + .await + .map_err(|e| DurableError::storage("sweep_orphans", e))?; + + let scanned = u64::try_from(candidates.len()).unwrap_or(u64::MAX); + let next_cursor = candidates + .last() + .map(|(id, updated_at)| crate::retention::SweepCursor { + updated_at_ms: *updated_at, + execution_id: id.clone(), + }); + + let now = now_unix_millis(); + let abort = sql!( + "UPDATE durable_executions SET status = 'aborted', finalized_at = ?, updated_at = ? + WHERE execution_id = ? AND status = 'running' AND finalized_at IS NULL" + ); + let mut aborted = 0u64; + for (exec_str, _updated_at) in &candidates { + let Ok(execution_id) = parse_execution_id(exec_str) else { + continue; + }; + match ExecutionLock::acquire(lock_dir, execution_id) { + Ok(_lock) => { + let result = zeph_db::query(abort) + .bind(now) + .bind(now) + .bind(exec_str) + .execute(&self.pool) + .await + .map_err(|e| DurableError::storage("sweep_orphans", e))?; + aborted += result.rows_affected(); + // `_lock` drops here, releasing the flock for the next holder. + } + Err(DurableError::ExecutionLocked { .. }) => { + // A live owner holds this execution — never abort on staleness alone (INV-17). + } + Err(e) => return Err(e), + } + } + Ok(crate::retention::SweepBatchOutcome { + scanned, + aborted, + next_cursor, + }) + } + /// Seal a plaintext payload, or pass it through verbatim when no cipher is configured. fn seal_payload(&self, plaintext: &[u8], aad: &PayloadAad) -> Result, DurableError> { match &self.cipher { @@ -1631,6 +1769,32 @@ impl Journal for LocalBackend { }) .await } + + /// Crash-orphan reclamation (INV-17, #6254). See [`Journal::sweep_orphans`] for the contract. + async fn sweep_orphans(&self, policy: &RetentionPolicy) -> Result { + if policy.stale_running_after_secs == 0 { + return Ok(0); + } + let Some(lock_dir) = self.lock_dir.clone() else { + if !self + .orphan_sweep_warned + .swap(true, std::sync::atomic::Ordering::Relaxed) + { + tracing::warn!( + "durable: crash-orphan sweep requires an on-disk advisory-lock dir; orphan \ + reclamation disabled for this backend (Postgres/:memory:/non-Unix)" + ); + } + return Ok(0); + }; + let cutoff_ms = orphan_cutoff_ms(policy, now_unix_millis()); + crate::retention::sweep_orphans_in_batches( + policy.prune_batch_size, + cutoff_ms, + |cutoff, batch, cursor| self.sweep_orphan_batch(&lock_dir, cutoff, batch, cursor), + ) + .await + } } impl crate::sealed::Sealed for LocalBackend {} @@ -1738,6 +1902,14 @@ pub(crate) fn now_unix_millis() -> i64 { .map_or(0, |d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX)) } +/// The absolute `updated_at` cutoff (Unix ms) at or before which a `status='running'` row becomes +/// a crash-orphan sweep candidate (INV-17, #6254). +fn orphan_cutoff_ms(policy: &RetentionPolicy, now_ms: i64) -> i64 { + let threshold = + i64::try_from(policy.stale_running_after_secs.saturating_mul(1000)).unwrap_or(i64::MAX); + now_ms.saturating_sub(threshold) +} + /// Decode a stored blob into a fixed 32-byte array, failing closed on the wrong length. fn slice_to_array32(bytes: &[u8], field: &'static str) -> Result<[u8; 32], DurableError> { <[u8; 32]>::try_from(bytes).map_err(|_| DurableError::Decode { context: field }) @@ -2415,9 +2587,13 @@ mod tests { } #[tokio::test] - async fn reopening_an_aborted_execution_leaves_it_untouched() { - // Divergence recovery reopens the same row on purpose (a fresh replay cursor, not a status - // reset) — only `completed`/`failed` rows are un-finalized on reopen, `aborted` is not. + async fn reopening_an_aborted_execution_un_finalizes_it() { + // INV-16 (#6254): reopening a row in ANY terminal status — including `aborted` — must + // un-finalize it back to `running` with `finalized_at` cleared. This covers both the + // pre-existing divergence-recovery reopen (which starts a fresh replay cursor on + // purpose) and the new crash-orphan sweep (INV-17), which makes `aborted` the common + // outcome of a resumable crash: a resumed execution whose row keeps `finalized_at` set + // would otherwise be prunable out from under the active resume. let backend = mem_backend(1_048_576).await; let exec = ExecutionId::new(); backend @@ -2429,21 +2605,26 @@ mod tests { .await .unwrap(); - backend + let is_resume = backend .open_execution(exec, ExecutionKind::AgentTurn) .await .unwrap(); + assert!(is_resume, "the row already existed, so this is a resume"); - let (status,): (String,) = zeph_db::query_as(sql!( - "SELECT status FROM durable_executions WHERE execution_id = ?" + let (status, finalized): (String, Option) = zeph_db::query_as(sql!( + "SELECT status, finalized_at FROM durable_executions WHERE execution_id = ?" )) .bind(exec.as_uuid().to_string()) .fetch_one(backend.pool()) .await .unwrap(); assert_eq!( - status, "aborted", - "reopening must not un-finalize an aborted execution" + status, "running", + "reopening an aborted execution must un-finalize it (INV-16)" + ); + assert!( + finalized.is_none(), + "reopening must clear the stale finalized_at" ); } @@ -2874,6 +3055,414 @@ mod tests { assert_eq!(backend.read_execution(live).await.unwrap().len(), 1); } + /// Backdate a `durable_executions` row's `updated_at` so it becomes a sweep candidate. + async fn backdate_updated_at(backend: &LocalBackend, id: ExecutionId, updated_at_ms: i64) { + zeph_db::query(sql!( + "UPDATE durable_executions SET updated_at = ? WHERE execution_id = ?" + )) + .bind(updated_at_ms) + .bind(id.as_uuid().to_string()) + .execute(backend.pool()) + .await + .unwrap(); + } + + #[tokio::test] + async fn sweep_orphans_disabled_when_threshold_is_zero() { + // A file-backed backend so the sweep would otherwise have a lock_dir to work with; + // stale_running_after_secs = 0 must short-circuit before any scan. + let dir = tempfile::tempdir().unwrap(); + let backend = + LocalBackend::open(&dir.path().join("durable.db").to_string_lossy(), 1_048_576) + .await + .unwrap(); + backend.init().await.unwrap(); + + let exec = ExecutionId::new(); + backend + .open_execution(exec, ExecutionKind::AgentTurn) + .await + .unwrap(); + backdate_updated_at(&backend, exec, 0).await; + + let policy = RetentionPolicy { + stale_running_after_secs: 0, + ..RetentionPolicy::default() + }; + let aborted = backend.sweep_orphans(&policy).await.unwrap(); + assert_eq!( + aborted, 0, + "stale_running_after_secs = 0 disables the sweep" + ); + + let (status,): (String,) = zeph_db::query_as(sql!( + "SELECT status FROM durable_executions WHERE execution_id = ?" + )) + .bind(exec.as_uuid().to_string()) + .fetch_one(backend.pool()) + .await + .unwrap(); + assert_eq!(status, "running"); + } + + #[tokio::test] + async fn sweep_orphans_is_a_documented_no_op_on_memory_backend() { + // `:memory:` has no on-disk lock_dir (INV-15 degrade), so the sweep must never abort on + // staleness alone — FR-DE-19. + let backend = mem_backend(1_048_576).await; + let exec = ExecutionId::new(); + backend + .open_execution(exec, ExecutionKind::AgentTurn) + .await + .unwrap(); + backdate_updated_at(&backend, exec, 0).await; + + let policy = RetentionPolicy { + stale_running_after_secs: 1, + ..RetentionPolicy::default() + }; + let aborted = backend.sweep_orphans(&policy).await.unwrap(); + assert_eq!( + aborted, 0, + "a lock_dir=None backend must never abort on staleness alone" + ); + + let (status,): (String,) = zeph_db::query_as(sql!( + "SELECT status FROM durable_executions WHERE execution_id = ?" + )) + .bind(exec.as_uuid().to_string()) + .fetch_one(backend.pool()) + .await + .unwrap(); + assert_eq!(status, "running"); + } + + #[tokio::test] + async fn sweep_orphans_aborts_a_stale_running_execution_with_no_live_owner() { + // FR-DE-16/17: a stale `running` row whose lock is free (no live owner) is hard-aborted. + let dir = tempfile::tempdir().unwrap(); + let backend = + LocalBackend::open(&dir.path().join("durable.db").to_string_lossy(), 1_048_576) + .await + .unwrap(); + backend.init().await.unwrap(); + + let exec = ExecutionId::new(); + backend + .open_execution(exec, ExecutionKind::AgentTurn) + .await + .unwrap(); + // Nothing holds this execution's ExecutionLock — `open_execution` (not `_exclusive`) + // never acquires one, simulating a crashed owner whose flock released on process exit. + backdate_updated_at(&backend, exec, 0).await; + + let policy = RetentionPolicy { + stale_running_after_secs: 1, + ..RetentionPolicy::default() + }; + let aborted = backend.sweep_orphans(&policy).await.unwrap(); + assert_eq!(aborted, 1); + + let (status, finalized): (String, Option) = zeph_db::query_as(sql!( + "SELECT status, finalized_at FROM durable_executions WHERE execution_id = ?" + )) + .bind(exec.as_uuid().to_string()) + .fetch_one(backend.pool()) + .await + .unwrap(); + assert_eq!(status, "aborted"); + assert!(finalized.is_some()); + } + + #[tokio::test] + async fn sweep_orphans_skips_an_execution_whose_lock_is_held_by_a_live_owner() { + // INV-17: staleness of `updated_at` alone is never sufficient — a stale-but-alive + // execution (long single step, parked HITL promise, multi-hour job) must survive the + // sweep as long as its owner still holds the INV-15 flock. + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("durable.db"); + let url = db_path.to_string_lossy().into_owned(); + + let owner = LocalBackend::open(&url, 1_048_576).await.unwrap(); + owner.init().await.unwrap(); + let sweeper = LocalBackend::open(&url, 1_048_576).await.unwrap(); + + let exec = ExecutionId::new(); + let (_, _lock) = owner + .open_execution_exclusive(exec, ExecutionKind::AgentTurn) + .await + .unwrap(); + backdate_updated_at(&owner, exec, 0).await; + + let policy = RetentionPolicy { + stale_running_after_secs: 1, + ..RetentionPolicy::default() + }; + let aborted = sweeper.sweep_orphans(&policy).await.unwrap(); + assert_eq!(aborted, 0, "a live-held lock must never be swept"); + + let (status,): (String,) = zeph_db::query_as(sql!( + "SELECT status FROM durable_executions WHERE execution_id = ?" + )) + .bind(exec.as_uuid().to_string()) + .fetch_one(owner.pool()) + .await + .unwrap(); + assert_eq!(status, "running"); + } + + #[tokio::test] + async fn sweep_orphans_leaves_a_fresh_running_execution_untouched() { + // A recently-updated `running` row is not yet a sweep candidate at all. + let dir = tempfile::tempdir().unwrap(); + let backend = + LocalBackend::open(&dir.path().join("durable.db").to_string_lossy(), 1_048_576) + .await + .unwrap(); + backend.init().await.unwrap(); + + let exec = ExecutionId::new(); + backend + .open_execution(exec, ExecutionKind::AgentTurn) + .await + .unwrap(); + + let policy = RetentionPolicy { + stale_running_after_secs: 3600, + ..RetentionPolicy::default() + }; + let aborted = backend.sweep_orphans(&policy).await.unwrap(); + assert_eq!(aborted, 0); + } + + #[tokio::test] + async fn count_orphans_matches_sweep_without_mutating() { + let dir = tempfile::tempdir().unwrap(); + let backend = + LocalBackend::open(&dir.path().join("durable.db").to_string_lossy(), 1_048_576) + .await + .unwrap(); + backend.init().await.unwrap(); + + let exec = ExecutionId::new(); + backend + .open_execution(exec, ExecutionKind::AgentTurn) + .await + .unwrap(); + backdate_updated_at(&backend, exec, 0).await; + + let policy = RetentionPolicy { + stale_running_after_secs: 1, + ..RetentionPolicy::default() + }; + let counted = backend.count_orphans(&policy).await.unwrap(); + assert_eq!(counted, 1); + + // count_orphans must not have mutated the row. + let (status,): (String,) = zeph_db::query_as(sql!( + "SELECT status FROM durable_executions WHERE execution_id = ?" + )) + .bind(exec.as_uuid().to_string()) + .fetch_one(backend.pool()) + .await + .unwrap(); + assert_eq!(status, "running"); + + let aborted = backend.sweep_orphans(&policy).await.unwrap(); + assert_eq!( + aborted, counted, + "sweep must abort exactly what count_orphans counted" + ); + } + + /// Batching-boundary regression: a candidate set straddling `prune_batch_size` (one more row + /// than a single batch) must be fully processed across multiple batches, not just the first + /// one. Exercises the real `sweep_orphan_batch`/`sweep_orphans_in_batches` composition end to + /// end (not the pure-logic unit test in `retention.rs`), so the SQL `LIMIT` and the + /// `scanned`-driven continuation check are both proven against a real DB. + #[tokio::test] + async fn sweep_orphans_processes_every_batch_when_candidates_straddle_the_batch_size() { + let dir = tempfile::tempdir().unwrap(); + let backend = + LocalBackend::open(&dir.path().join("durable.db").to_string_lossy(), 1_048_576) + .await + .unwrap(); + backend.init().await.unwrap(); + + let batch_size = 2u64; + let candidate_count = batch_size + 1; // straddles the batch boundary + let mut execs = Vec::new(); + for _ in 0..candidate_count { + let exec = ExecutionId::new(); + backend + .open_execution(exec, ExecutionKind::AgentTurn) + .await + .unwrap(); + backdate_updated_at(&backend, exec, 0).await; + execs.push(exec); + } + + let policy = RetentionPolicy { + stale_running_after_secs: 1, + prune_batch_size: batch_size, + ..RetentionPolicy::default() + }; + let aborted = backend.sweep_orphans(&policy).await.unwrap(); + assert_eq!( + aborted, candidate_count, + "every candidate must be aborted, including the one past the first batch" + ); + + for exec in execs { + let (status,): (String,) = zeph_db::query_as(sql!( + "SELECT status FROM durable_executions WHERE execution_id = ?" + )) + .bind(exec.as_uuid().to_string()) + .fetch_one(backend.pool()) + .await + .unwrap(); + assert_eq!(status, "aborted"); + } + } + + /// #6254 C1 regression: when the count of stale-but-live (lock-held) candidates is `>= + /// prune_batch_size`, the sweep must still terminate rather than looping forever re-selecting + /// the same lock-held rows. Before the keyset-pagination fix, `sweep_orphan_batch`'s candidate + /// `SELECT` had no offset/cursor, so a batch consisting entirely of lock-held rows (which the + /// sweep never deletes, mutates, or otherwise removes from the `status='running'` candidate + /// set) would re-select the identical rows on every iteration: `scanned` would stay `== + /// batch` and `aborted` would stay `0` forever, so `sweep_orphans_in_batches`'s `scanned < + /// batch` continuation check would never trip. Exercises the real DB-backed + /// `sweep_orphan_batch`/`sweep_orphans_in_batches` composition (not the pure-logic + /// simulation in `retention.rs`) with more lock-held candidates than `prune_batch_size`, so a + /// naive single-batch-worth-of-locks reproduction would not have caught a bug that only + /// manifests once the candidate set spans multiple batches. + #[tokio::test] + async fn sweep_orphans_terminates_when_lock_held_candidates_exceed_batch_size() { + let dir = tempfile::tempdir().unwrap(); + let db_url = dir.path().join("durable.db").to_string_lossy().into_owned(); + + let owner = LocalBackend::open(&db_url, 1_048_576).await.unwrap(); + owner.init().await.unwrap(); + let sweeper = LocalBackend::open(&db_url, 1_048_576).await.unwrap(); + + let batch_size = 2u64; + let candidate_count = batch_size * 2 + 1; // spans at least three batches, all lock-held + let mut locks = Vec::new(); + for _ in 0..candidate_count { + let exec = ExecutionId::new(); + let (_, lock) = owner + .open_execution_exclusive(exec, ExecutionKind::AgentTurn) + .await + .unwrap(); + backdate_updated_at(&owner, exec, 0).await; + locks.push(lock); // held for the whole test — every candidate stays lock-held + } + + let policy = RetentionPolicy { + stale_running_after_secs: 1, + prune_batch_size: batch_size, + ..RetentionPolicy::default() + }; + + let aborted = tokio::time::timeout( + std::time::Duration::from_secs(10), + sweeper.sweep_orphans(&policy), + ) + .await + .expect( + "sweep_orphans must terminate even when lock-held candidates exceed prune_batch_size \ + (#6254 C1) — it hung instead of returning", + ) + .unwrap(); + + assert_eq!(aborted, 0, "every candidate's lock is held by a live owner"); + drop(locks); + } + + /// INV-17: the sweep's guarded abort `UPDATE` runs only while holding the same non-reentrant + /// flock a concurrent `open_execution_exclusive` reopen for the same execution id requires, so + /// the two can never both mutate the row at once. Drives them as genuinely concurrent tasks + /// against a real multi-connection pool (file-backed — `:memory:` forces a single connection, + /// which would serialize the two calls trivially and prove nothing) across many trials so both + /// orderings ("sweep acquires the lock first" and "reopen acquires the lock first") are + /// exercised without artificial delay injection, mirroring the #6251 + /// `concurrent_prune_and_reopen_never_lose_or_corrupt_the_row` pattern above. + #[tokio::test] + async fn concurrent_sweep_and_reopen_race_never_corrupts_the_row() { + let dir = tempfile::tempdir().unwrap(); + let db_url = dir.path().join("durable.db").to_string_lossy().into_owned(); + let backend = Arc::new(LocalBackend::open(&db_url, 1_048_576).await.unwrap()); + backend.init().await.unwrap(); + + let policy = RetentionPolicy { + stale_running_after_secs: 1, + prune_batch_size: 10, + ..RetentionPolicy::default() + }; + + for _ in 0..20 { + let exec = ExecutionId::new(); + backend + .open_execution(exec, ExecutionKind::AgentTurn) + .await + .unwrap(); + backdate_updated_at(&backend, exec, 0).await; + + let sweep_backend = backend.clone(); + let policy_for_task = policy.clone(); + let sweep = + tokio::spawn(async move { sweep_backend.sweep_orphans(&policy_for_task).await }); + + let reopen_backend = backend.clone(); + let reopen = tokio::spawn(async move { + reopen_backend + .open_execution_exclusive(exec, ExecutionKind::AgentTurn) + .await + }); + + let (sweep_result, reopen_result) = tokio::join!(sweep, reopen); + let aborted = sweep_result + .expect("sweep task must not panic") + .expect("sweep must not error under a concurrent reopen"); + assert!(aborted <= 1, "at most one candidate row exists per trial"); + + match reopen_result.expect("reopen task must not panic") { + Ok((_is_resume, _lock)) => { + // reopen won the race for the lock (either before the sweep even tried, or + // after the sweep aborted the row and released) — the row must be `running` + // with `finalized_at` cleared either way (INV-16 un-finalizes `aborted` too). + let (status, finalized): (String, Option) = zeph_db::query_as(sql!( + "SELECT status, finalized_at FROM durable_executions WHERE execution_id = ?" + )) + .bind(exec.as_uuid().to_string()) + .fetch_one(backend.pool()) + .await + .unwrap(); + assert_eq!(status, "running"); + assert!(finalized.is_none()); + // Finalize before the next trial: when reopen wins because the row was + // already `running` (not terminal) at the time it checked, `open_execution`'s + // existing-row branch never bumps `updated_at` — left alone, this row would + // stay a stale `running` candidate forever and pollute a later trial's + // `aborted` count (the `assert!(aborted <= 1, ...)` above would then see more + // than this trial's own row). Each trial must start with a clean slate of + // exactly its own candidate. + backend + .finalize(exec, ExecutionStatus::Completed) + .await + .unwrap(); + } + Err(DurableError::ExecutionLocked { .. }) => { + // The sweep held the lock at the moment reopen tried — expected under the race. + } + Err(e) => panic!( + "reopen must only ever fail with ExecutionLocked under this race, got {e:?}" + ), + } + } + } + #[tokio::test] async fn checkpoint_fold_compacts_idempotent_prefix_and_replays() { let backend = mem_backend(1_048_576) diff --git a/crates/zeph-durable/src/journal.rs b/crates/zeph-durable/src/journal.rs index 89ec041c8..9ae390113 100644 --- a/crates/zeph-durable/src/journal.rs +++ b/crates/zeph-durable/src/journal.rs @@ -298,6 +298,28 @@ pub trait Journal: Send + Sync { &self, policy: &RetentionPolicy, ) -> impl Future> + Send; + + /// Crash-orphan reclamation (#6254): flock-verify and hard-abort stale `running` rows. + /// + /// A `status='running'` row whose `updated_at` is older than `policy.stale_running_after_secs` + /// is a sweep candidate; it is only hard-aborted after a non-blocking try-acquire of its + /// INV-15 `ExecutionLock` succeeds — a live owner (`ExecutionLocked`) short-circuits to skip, + /// since staleness alone never proves the owner is dead (INV-17). Runs exclusively on a + /// background task, before [`Journal::prune`] on the same tick — never on the dispatch hot + /// path. + /// + /// Returns the number of executions aborted. Returns `Ok(0)` without scanning when + /// `policy.stale_running_after_secs == 0` (disabled), and `Ok(0)` with a warn-once log on + /// backends without a `lock_dir` (`:memory:`, Postgres, non-Unix) — a documented no-op, never + /// a staleness-only abort. + /// + /// # Errors + /// + /// Returns [`DurableError::JournalUnavailable`] if the sweep cannot complete. + fn sweep_orphans( + &self, + policy: &RetentionPolicy, + ) -> impl Future> + Send; } #[cfg(test)] diff --git a/crates/zeph-durable/src/retention.rs b/crates/zeph-durable/src/retention.rs index 64a82be7f..7e58257a3 100644 --- a/crates/zeph-durable/src/retention.rs +++ b/crates/zeph-durable/src/retention.rs @@ -216,8 +216,10 @@ impl DurableRetentionService { /// Run the prune loop until the task is aborted. /// - /// Each tick prunes terminal executions older than their TTL; a prune failure is logged and the - /// loop continues (a transient database error must not kill retention). + /// Each tick first runs the crash-orphan sweep (#6254), then prunes terminal executions older + /// than their TTL — in that order, so a just-aborted orphan is visible to the same tick's TTL + /// check (INV-17, M3). A sweep or prune failure is logged and the loop continues (a transient + /// database error must not kill retention). #[tracing::instrument(name = "durable.retention.run", skip_all)] pub async fn run(self) { let mut tick = tokio::time::interval(self.interval); @@ -228,6 +230,14 @@ impl DurableRetentionService { loop { tick.tick().await; async { + match self.backend.sweep_orphans(&self.policy).await { + Ok(aborted) => { + tracing::debug!(aborted, "durable retention crash-orphan sweep completed"); + } + Err(error) => { + tracing::warn!(%error, "durable retention crash-orphan sweep failed; will retry"); + } + } match self.backend.prune(&self.policy).await { Ok(deleted) => { tracing::debug!(deleted, "durable retention prune sweep completed"); @@ -243,6 +253,96 @@ impl DurableRetentionService { } } +/// A keyset-pagination cursor over `durable_executions(updated_at, execution_id)`, used by the +/// crash-orphan sweep to guarantee forward progress across batches (#6254 C1). +/// +/// Ordering by `(updated_at, execution_id)` (not `updated_at` alone) gives a total order even +/// when several rows share the same `updated_at` millisecond, so no candidate is ever skipped or +/// revisited across batch boundaries. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct SweepCursor { + /// `updated_at` of the last row this batch scanned. + pub(crate) updated_at_ms: i64, + /// `execution_id` (as stored, a UUID string) of the last row this batch scanned. + pub(crate) execution_id: String, +} + +/// The outcome of one batch of the crash-orphan sweep: how many `running` rows this batch +/// scanned (drives the caller's batch-continuation decision, mirroring [`prune_in_batches`]'s +/// `deleted < batch` check), how many of those were actually aborted (a candidate whose +/// [`ExecutionLock`](crate::backend::ExecutionLock) is held by a live owner is scanned but not +/// aborted — INV-17), and the keyset cursor to resume from on the next batch. +/// +/// `next_cursor` is the load-bearing fix for #6254 C1: the sweep never deletes or otherwise +/// removes a skipped (lock-held) candidate from `durable_executions`, so a batch that re-issued +/// the *same* unbounded `SELECT ... LIMIT batch` on every iteration would re-select the exact +/// same lock-held rows forever whenever the live-but-stale count reaches or exceeds `batch` — +/// `scanned` would stay `== batch` and `aborted` would stay `0` on every iteration, so the +/// `scanned < batch` continuation check would never trip and the loop would never terminate. +/// Advancing past `next_cursor` on every batch — whether or not any row in it was aborted — +/// guarantees the candidate set strictly shrinks each iteration regardless of lock outcomes. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct SweepBatchOutcome { + /// Number of `status='running'` candidates this batch scanned. + pub(crate) scanned: u64, + /// Number of those candidates this batch actually hard-aborted. + pub(crate) aborted: u64, + /// Keyset cursor positioned at the last-scanned row, `None` if this batch scanned zero rows. + pub(crate) next_cursor: Option, +} + +/// Run one batched crash-orphan sweep pass, hard-aborting stale `running` executions whose +/// INV-15 `ExecutionLock` is free (#6254). +/// +/// This is the shared body behind [`Journal::sweep_orphans`](crate::Journal::sweep_orphans) for +/// the local backend. Mirrors [`prune_in_batches`]'s batch/yield discipline so a large sweep +/// never monopolizes the runtime, but continues based on `scanned` (not `aborted`) — a batch +/// where every candidate's lock is held by a live owner still scanned a full `batch` and must not +/// be mistaken for "sweep exhausted". Termination is guaranteed by keyset pagination +/// ([`SweepCursor`]): each batch's `sweep_batch` call is handed the previous batch's cursor and +/// scans strictly past it, so the same lock-held row is never re-selected across batches (#6254 +/// C1 — without this, an all-lock-held batch >= `batch_size` would loop forever). +pub(crate) async fn sweep_orphans_in_batches( + batch_size: u64, + cutoff_ms: i64, + sweep_batch: F, +) -> Result +where + F: Fn(i64, u64, Option) -> Fut, + Fut: Future>, +{ + let batch = batch_size.max(1); + let mut total_aborted = 0u64; + let mut cursor: Option = None; + let span = tracing::info_span!( + "durable.retention.sweep_orphans", + aborted_count = tracing::field::Empty + ); + async { + loop { + let outcome = sweep_batch(cutoff_ms, batch, cursor.take()).await?; + total_aborted = total_aborted.saturating_add(outcome.aborted); + if outcome.scanned < batch { + break; + } + cursor = outcome.next_cursor; + // A full batch with no cursor to resume from cannot happen (scanned == batch >= 1 + // rows implies a last row to build a cursor from) — but fail closed rather than + // looping forever on a future refactor that breaks this invariant. + if cursor.is_none() { + break; + } + // Release the write lock and let other tasks run before the next batch. + tokio::task::yield_now().await; + } + tracing::Span::current().record("aborted_count", total_aborted); + metrics::counter!("durable.retention.orphans_aborted").increment(total_aborted); + Ok(total_aborted) + } + .instrument(span) + .await +} + /// Run one batched prune pass over the journal, deleting terminal executions past their TTL. /// /// This is the shared body behind [`Journal::prune`](crate::Journal::prune) for the local backend. @@ -427,4 +527,57 @@ mod tests { assert_eq!(total, 1_620); assert_eq!(remaining.get(), 0); } + + /// Mirrors [`prune_in_batches_loops_until_drained_and_yields`] but for the sweep, which + /// continues on `scanned` rather than `aborted` (#6254): a batch where every candidate's + /// `ExecutionLock` is held by a live owner still scans a full batch and must not be mistaken + /// for "sweep exhausted". The closure below draws from a *fixed*, non-shrinking pool of + /// candidate rows addressed purely by the `cursor` it is handed — exactly what + /// `LocalBackend::sweep_orphan_batch`'s keyset-paginated SQL does — rather than an internal + /// counter that shrinks regardless of lock outcome. This is deliberate: a version of this + /// test that shrinks an internal "remaining" counter on every batch (aborted or not) passes + /// even against the pre-fix implementation, which never advanced a cursor and would re-select + /// the identical lock-held rows forever in production — it asserts nothing about the actual + /// #6254 C1 bug. Middle batch (rows `[2, 4)`) aborts nothing, simulating "every candidate in + /// this batch is lock-held"; the sweep must still advance past those rows via `next_cursor` + /// and finish the remaining pool. Wrapped in a timeout so a regression that drops cursor + /// advancement fails this test instead of hanging the whole suite. + #[tokio::test] + async fn sweep_orphans_in_batches_advances_past_an_all_lock_held_batch() { + const POOL_SIZE: u64 = 5; + const BATCH_SIZE: u64 = 2; + + let sweep = sweep_orphans_in_batches(BATCH_SIZE, 0, |_cutoff, batch, cursor| { + let start = cursor.map_or(0, |c| u64::try_from(c.updated_at_ms).unwrap() + 1); + let end = (start + batch).min(POOL_SIZE); + let scanned = end.saturating_sub(start); + // Rows [2, 4) simulate "every candidate in this batch is lock-held": 0 aborted. + let aborted = if start == 2 { 0 } else { scanned }; + let next_cursor = (scanned > 0).then(|| SweepCursor { + updated_at_ms: i64::try_from(end - 1).unwrap(), + execution_id: String::new(), + }); + async move { + Ok(SweepBatchOutcome { + scanned, + aborted, + next_cursor, + }) + } + }); + + let total_aborted = tokio::time::timeout(std::time::Duration::from_secs(5), sweep) + .await + .expect( + "sweep_orphans_in_batches must terminate even when a batch is entirely \ + lock-held (#6254 C1 regression) — it hung instead of returning", + ) + .unwrap(); + + assert_eq!( + total_aborted, + POOL_SIZE - BATCH_SIZE, + "every row except the all-locked middle batch must be aborted" + ); + } } diff --git a/crates/zeph-orchestration/src/durable.rs b/crates/zeph-orchestration/src/durable.rs index bb9ba204d..8837f143e 100644 --- a/crates/zeph-orchestration/src/durable.rs +++ b/crates/zeph-orchestration/src/durable.rs @@ -219,10 +219,32 @@ pub async fn journal_budget( }); }; let local_backend = local_backend.clone(); - // Open a fresh execution (never a resume) for this generation. - local_backend - .open_execution(exec_id, ExecutionKind::DagRun) - .await?; + // Open a fresh execution (never a resume) for this generation, taking the INV-15 exclusive + // lock so this `DagRun` row's liveness is observable to the INV-17 crash-orphan sweep + // (#6254). `_lock` is held for the rest of this function so the sweep never races the write + // below; `is_resume` stays hardcoded `false` below regardless of the lock result — lock + // acquisition and the resume flag are orthogonal (this path always opens a fresh generation). + let _lock = match local_backend + .open_execution_exclusive(exec_id, ExecutionKind::DagRun) + .await + { + Ok((_, lock)) => lock, + Err(DurableError::ExecutionLocked { + execution_id, + holder_pid, + }) => { + tracing::info!( + execution_id = %execution_id, + holder_pid, + graph_id = %graph_id, + generation, + "orch.durable.budget_journal: execution already open in another process; \ + skipping this budget snapshot" + ); + return Ok(()); + } + Err(e) => return Err(e), + }; let ctx = build_budget_ctx(exec_id, false, backend, writer, config); @@ -584,5 +606,65 @@ mod tests { let exec = budget_exec_id(&graph_id, generation); assert_eq!(execution_status(&local, exec).await, "failed"); } + + /// FR-DE-20 (#6254): `journal_budget` opens via `open_execution_exclusive`. When another + /// process/handle already holds the `DagRun` execution's lock, `journal_budget` must + /// return `Ok(())` (a graceful skip), never an error and never a task failure — the + /// existing caller (`plan.rs`) already tolerates a missing snapshot. + #[tokio::test] + async fn journal_budget_skips_gracefully_when_execution_is_locked() { + let dir = tempfile::tempdir().unwrap(); + let url = dir.path().join("durable.db").to_string_lossy().into_owned(); + + let owner = LocalBackend::open(&url, 1_048_576).await.unwrap(); + owner.init().await.unwrap(); + let owner = Arc::new(owner); + + let contender = LocalBackend::open(&url, 1_048_576).await.unwrap(); + let contender = Arc::new(contender); + let backend = Arc::new(DurableBackendEnum::Local(contender.clone())); + let (writer, handle) = JournalWriter::new(contender.clone(), &test_config()); + tokio::spawn(async move { writer.run().await }); // EXEMPT: test-only helper + + let graph_id = GraphId::new(); + let generation: u32 = 0; + let exec_id = budget_exec_id(&graph_id, generation); + + // The "owner" holds the exclusive lock on this exact execution id, simulating + // another live process already journaling this generation's budget. + let (_, _lock) = owner + .open_execution_exclusive(exec_id, ExecutionKind::DagRun) + .await + .unwrap(); + + let config = test_config(); + let result = journal_budget( + &graph_id, + generation, + backend, + handle, + &config, + ReplanBudgetSnapshot::default(), + ) + .await; + assert!( + result.is_ok(), + "ExecutionLocked must degrade to a graceful Ok(()) skip, got {result:?}" + ); + + // The row belongs to `owner` (it opened first and holds the lock); the losing side + // must not have mutated it — it stays `running`, never `completed`/`failed`. + let (status,): (String,) = zeph_db::query_as(zeph_db::sql!( + "SELECT status FROM durable_executions WHERE execution_id = ?" + )) + .bind(exec_id.as_uuid().to_string()) + .fetch_one(owner.pool()) + .await + .unwrap(); + assert_eq!( + status, "running", + "the losing side must not have written a step or finalized the row" + ); + } } } diff --git a/crates/zeph-scheduler/src/durable.rs b/crates/zeph-scheduler/src/durable.rs index 39e09f164..1b0d98275 100644 --- a/crates/zeph-scheduler/src/durable.rs +++ b/crates/zeph-scheduler/src/durable.rs @@ -26,8 +26,8 @@ use std::sync::Arc; use tracing::Instrument as _; use zeph_config::DurableConfig; use zeph_durable::{ - DurableBackendEnum, DurableContext, EffectIntentSubClass, ExecutionId, ExecutionKind, - ExecutionStatus, JournalWriterHandle, LocalBackend, StepDescriptor, StepError, + DurableBackendEnum, DurableContext, DurableError, EffectIntentSubClass, ExecutionId, + ExecutionKind, ExecutionStatus, JournalWriterHandle, LocalBackend, StepDescriptor, StepError, }; use crate::error::SchedulerError; @@ -149,10 +149,35 @@ where } }; - let is_resume = local_backend - .open_execution(exec_id, ExecutionKind::ScheduledJob) + // Exclusive open (INV-15) so this `ScheduledJob` row's liveness is observable to the + // INV-17 crash-orphan sweep (#6254) — it also closes a latent double-drive gap where two + // scheduler daemons could otherwise both fire the same `job_name`+`slot_ms`. `_lock` is + // held for the entire fire body (open → step → finalize) below. + let (is_resume, _lock) = match local_backend + .open_execution_exclusive(exec_id, ExecutionKind::ScheduledJob) .await - .map_err(|e| SchedulerError::TaskFailed(format!("durable open failed: {e}")))?; + { + Ok(result) => result, + Err(DurableError::ExecutionLocked { + execution_id, + holder_pid, + }) => { + tracing::info!( + execution_id = %execution_id, + holder_pid, + job = job_name, + slot_ms, + "sched.durable.fire: execution already open in another process; skipping \ + this fire" + ); + return Ok(()); + } + Err(e) => { + return Err(SchedulerError::TaskFailed(format!( + "durable open failed: {e}" + ))); + } + }; let ctx = DurableContext::new( exec_id, @@ -423,5 +448,54 @@ mod tests { let exec = derive_execution_id("test-job", 1_000); assert_eq!(execution_status(&local, exec).await, "failed"); } + + /// #6254: `fire_with_durable` opens via `open_execution_exclusive`. When a peer daemon + /// already holds the `ScheduledJob` execution's lock, `fire_with_durable` must return + /// `Ok(())` (skip this fire, no retry) — never `SchedulerError::TaskFailed` and never a + /// re-invocation of the fire body. + #[tokio::test] + async fn fire_with_durable_skips_gracefully_when_execution_is_locked() { + let dir = tempfile::tempdir().unwrap(); + let url = dir.path().join("durable.db").to_string_lossy().into_owned(); + + let owner = LocalBackend::open(&url, 1_048_576).await.unwrap(); + owner.init().await.unwrap(); + + let contender = Arc::new(LocalBackend::open(&url, 1_048_576).await.unwrap()); + let backend = Arc::new(DurableBackendEnum::Local(contender.clone())); + let cfg = Arc::new(fast_config()); + let (writer, handle) = JournalWriter::new(contender.clone(), &cfg); + let _task = tokio::spawn(writer.run()); // EXEMPT: test-only helper + let adapter = SchedulerDurableAdapter::new(backend, handle, cfg); + + let exec_id = derive_execution_id("test-job", 1_000); + // The "owner" holds the exclusive lock, simulating a peer scheduler daemon already + // firing this exact job+slot. + let (_, _lock) = owner + .open_execution_exclusive(exec_id, ExecutionKind::ScheduledJob) + .await + .unwrap(); + + let count = Arc::new(AtomicU32::new(0)); + let c = count.clone(); + let result = fire_with_durable(&adapter, "test-job", 1_000, move || async move { + c.fetch_add(1, Ordering::Relaxed); + Ok(()) + }) + .await; + + assert!( + result.is_ok(), + "ExecutionLocked must degrade to a graceful Ok(()) skip, got {result:?}" + ); + assert_eq!( + count.load(Ordering::Relaxed), + 0, + "a locked execution must not re-invoke the fire body" + ); + + // The row belongs to `owner`; the losing side must not have mutated it. + assert_eq!(execution_status(&owner, exec_id).await, "running"); + } } } diff --git a/specs/064-durable-execution/spec.md b/specs/064-durable-execution/spec.md index 6dffc98cb..bf855d37a 100644 --- a/specs/064-durable-execution/spec.md +++ b/specs/064-durable-execution/spec.md @@ -199,6 +199,46 @@ binds `execution_id` (cipher.rs) and `IdempotencyKey::derive` already folds `exe `IdempotencyKey` collision this invariant's introduction considered was only ever possible when `ExecutionId` itself collided, which this lock now prevents structurally. +**INV-16 — Reopening an execution un-finalizes ALL terminal statuses identically (#6254).** +`open_execution`/`open_execution_exclusive`'s reopen path MUST reset `status='running'` and clear +`finalized_at` for a row found in ANY terminal status — `completed`, `failed`, OR `aborted` — not +just `completed`/`failed` as in the pre-#6254 behavior. The deciding fact for prunability is +"is this execution being actively reopened right now", never "which terminal status produced the +row". Leaving `aborted` untouched on reopen (the pre-#6254 behavior, justified at the time because +the only `aborted` producers were rare, immediately-redriven divergence-recovery/`StepCapExceeded` +aborts) becomes unsafe once the crash-orphan sweep (INV-17) makes `aborted` the common outcome of a +resumable crash: a resumed execution whose row keeps `finalized_at` set is prunable out from under +the active resume — the exact hazard the completed/failed un-finalize was built to prevent. Un- +finalizing `aborted` on reopen is strictly safer for the pre-existing divergence-recovery case too +(it now also protects that fresh re-drive from prune) and does not change replay-cursor +fresh-vs-resume selection, which is independent of the `status` column. + +**INV-17 — A crash-orphaned `running` execution is reclaimed only after its liveness is verified +via the INV-15 advisory lock, never by staleness alone (#6254).** +A background sweep (`Journal::sweep_orphans`, folded into the existing retention tick, running +before `prune`) MAY hard-abort a row matching `status='running' AND finalized_at IS NULL AND +updated_at <= now - stale_running_after_secs` (config: `RetentionPolicy::stale_running_after_secs`, +default 3600s; `0` disables the sweep) **only** after a non-blocking try-acquire of that +execution's INV-15 `ExecutionLock` succeeds. `ExecutionLocked` (lock held by a live owner) MUST +short-circuit to skip — staleness of `updated_at` alone is never sufficient grounds to abort, +because `updated_at` is not bumped on every journal append, and a genuinely active-but-idle +execution (a long single step, a parked HITL promise, a multi-hour scheduled-job body) would +otherwise be false-aborted. The abort `UPDATE` runs while the sweep still holds the acquired lock +(guard: `WHERE execution_id=? AND status='running' AND finalized_at IS NULL`), which makes it +race-free against a concurrent `open_execution_exclusive` reopen for the same id — the two can +never both proceed, because both require the same non-reentrant flock. The sweep never retries or +resumes an orphan; it only marks it `aborted` so the pre-existing crash-resume path +(`ensure_session_durable_ctx` → `open_execution` replay) reclaims it on next legitimate open. +Sweep and prune are ordered on the same tick — the sweep MUST run before `prune()` so a +just-aborted orphan is visible to that same tick's TTL check (informational: with +`ttl_failed_secs=0` this makes a freshly-swept orphan prune-eligible on the same tick — intended, +not a bug, since `0` already means "prune failed/aborted immediately"). Correctness rests on every +production execution kind holding its flock while live: `AgentTurn` already opens via +`open_execution_exclusive`; `ScheduledJob` and `DagRun` are converted to it by this change (see +P2/P3 adapter notes) specifically to make flock-liveness universal. On backends where +`lock_dir=None` (`:memory:`, Postgres, non-Unix — the same class INV-15 already degrades), the +sweep is a documented no-op, never a staleness-only abort — see NEVER and Non-Goals. + --- ## NEVER @@ -214,8 +254,9 @@ binds `execution_id` (cipher.rs) and `IdempotencyKey::derive` already folds `exe migration runner (031 §12). Owning a separate migrator would create a divergent source of truth. - **NEVER** journal a domain type or resolved secret in the step payload; journal opaque pre-serialized bytes passed by the consumer adapter. Consumers sanitize before calling `step()`. -- **NEVER** call `journal.prune()` or any other bulk-write on the step dispatch hot path. Pruning - and compaction run exclusively in a background task on a timed interval. +- **NEVER** call `journal.prune()`, `journal.sweep_orphans()`, or any other bulk-write on the step + dispatch hot path. Pruning, orphan reclamation, and compaction run exclusively in a background + task on a timed interval. - **NEVER** consume `Box` on the hot path. Use `DurableBackendEnum` with enum dispatch — consistent with `AnyProvider`/`AnyChannel` precedent. - **NEVER** add a `restate` feature to the `full` bundle. Restate requires an external server; @@ -232,6 +273,21 @@ binds `execution_id` (cipher.rs) and `IdempotencyKey::derive` already folds `exe (INV-15) when two processes could plausibly derive the same id (e.g. any id keyed on externally-observable state like `ConversationId` rather than a runtime-minted `UUIDv7`). Calling the unsynchronized `open_execution` directly from such an adapter reopens the #6122 race. +- **NEVER** convert `zeph-durable/src/writer.rs` or `handle.rs`'s `open_execution` call sites to + `open_execution_exclusive`. These run in the same process that already holds the driving + adapter's `ExecutionLock`; `flock(2)` is associated with the open file description, so a second + `open()`+`flock` on the same lock file from the same process self-errors (`WOULDBLOCK` / + `ExecutionLocked`) against the live execution it is trying to serve, rather than protecting it + (#6254). Only the two named adapter entry points (`zeph-scheduler`'s `fire_with_durable`, + `zeph-orchestration`'s `journal_budget`) convert — see INV-17 and the P2/P3 adapter notes. Every + other `open_execution` call stays plain: it is either covered by the adapter's own lock for + cross-process liveness, or would deadlock if re-locked in-process. +- **NEVER** ship a staleness-only crash-orphan abort (no INV-15 flock liveness check) for + `lock_dir=None` backends (Postgres, `:memory:`, non-Unix). Without a liveness signal, staleness + of `updated_at` alone cannot distinguish a dead execution from a live one — see INV-17. This is + an explicit Non-Goal of #6254 (documented no-op + warn-once); backend-agnostic reclamation for + these backends requires a different mechanism (heartbeat-timeout based) and is tracked as a + separate follow-up, not shipped in this PR. --- @@ -259,6 +315,10 @@ binds `execution_id` (cipher.rs) and `IdempotencyKey::derive` already folds `exe - P3: scheduler exactly-once job fire via `JobStore.record_run()` seam. - P4: subagent durable spawn/await via `DurablePromise`. - Journal retention and compaction (background sweep, per-execution step cap). +- Crash-orphan reclamation (#6254): a flock-verified staleness sweep (`Journal::sweep_orphans`) + that hard-aborts `running` executions whose owner process died without finalizing, folded into + the existing retention tick. See INV-16, INV-17, and the dedicated Retention & Compaction + subsection. - Mandatory integration points: `[durable]` config, `zeph durable` CLI, TUI `DurableView`, `--init` wizard, `--migrate-config`, testing playbook, coverage-status rows. - 10 criterion benchmarks + `bench_step_run_exactly_once_n ≤ 5 ms @ N=5` CI regression gate. @@ -278,6 +338,16 @@ binds `execution_id` (cipher.rs) and `IdempotencyKey::derive` already folds `exe - Replacing existing persistence (`zeph-agent-persistence` messages, orchestration `GraphPersistence`, scheduler `JobStore`, subagent transcripts). The durable layer *complements* them with an execution-flow journal; it does not subsume them. +- **Backend-agnostic crash-orphan reclamation for `lock_dir=None` backends** (Postgres, `:memory:`, + non-Unix) (#6254). The crash-orphan sweep requires the INV-15 advisory flock as its liveness + signal and is a documented no-op there (warn-once at startup: `"durable: crash-orphan sweep + requires an on-disk advisory-lock dir; orphan reclamation disabled for this backend + (Postgres/:memory:/non-Unix)"`). A backend-agnostic mechanism (heartbeat-timeout based, "option + c" in the design debate) is a distinct future mechanism — it is NOT shipped in this PR and is + tracked as a separate follow-up issue filed after this PR merges. Shipping a staleness-only abort + for these backends in the meantime is explicitly forbidden (see NEVER) — it would reintroduce a + false-abort-of-a-live-execution defect on the very backends where liveness cannot be verified. + `// TODO(post-v1): backend-agnostic (Postgres) crash-orphan reclamation via heartbeat timeout` --- @@ -303,7 +373,7 @@ src/ replay.rs # ReplayCursor, ReplayDivergence check, range-read cursor writer.rs # JournalWriter actor, JournalMsg enum, group-commit, ACK protocol cipher.rs # PayloadCipher trait, PayloadAad, CipherError - retention.rs # compaction/prune, in-execution step cap + retention.rs # compaction/prune, crash-orphan sweep (#6254), in-execution step cap config.rs # re-exports DurableConfig/RetentionPolicy/DurableBackend from zeph-config; # owns the EncryptionGate + encryption_gate AEAD policy (free fn) error.rs # DurableError (thiserror) @@ -370,9 +440,17 @@ trait Journal: Send + Sync { ) -> Result, DurableError>; async fn finalize(&self, id: ExecutionId, status: ExecutionStatus) -> Result<(), DurableError>; async fn prune(&self, policy: &RetentionPolicy) -> Result; + /// Crash-orphan reclamation (#6254): flock-verify and hard-abort stale `running` rows. + /// Returns the count aborted. No-op (`Ok(0)`) on backends without a `lock_dir` (INV-17). + async fn sweep_orphans(&self, policy: &RetentionPolicy) -> Result; } ``` +`sweep_orphans` is dispatched through `DurableBackendEnum` exactly like `prune` — the retention +loop already holds `Arc`. `LocalBackend` implements the real flock-probe logic +(INV-17); `RestateBackend` returns `Ok(0)` (Restate has its own crash-recovery semantics, out of +scope here). + `read_execution_range` is the path for long executions (DAG runs, agent sessions). The `ReplayCursor` reads N steps ahead (default 100, configurable), re-queries as replay advances — O(segment) memory. `read_execution` is retained for short executions. @@ -852,11 +930,70 @@ CREATE INDEX idx_durable_timers_due ON durable_timers(fired, due_at); | `max_journal_bytes` | 1073741824 (1 GiB) | Size cap; triggers LRU sweep | | `prune_batch_size` | 500 | Rows deleted per transaction; yield between batches | | `prune_interval_secs` | 3600 (1h) | Background task poll interval | +| `stale_running_after_secs` | 3600 (1h) | Crash-orphan threshold (#6254): a `status='running'` row whose `updated_at` is older than this becomes a sweep candidate. Default 1h is generous enough that no genuinely-active long turn is ever a candidate before the flock check even runs, yet small enough that orphans re-enter retention within the hour. `0` disables the sweep. See Crash-Orphan Sweep below. | Background pruning NEVER runs on the dispatch/append hot path. A background tokio task runs `prune()` every `prune_interval_secs`. Pruning deletes in batches of `prune_batch_size` rows per transaction, releases the lock, yields, and loops — no large-transaction stall. +#### Crash-Orphan Sweep (#6254) + +**Requirement.** An execution whose owner process exits ungracefully (SIGKILL, panic, OOM, +power-loss) leaves its `durable_executions` row `status='running', finalized_at=NULL` forever — no +graceful detach path runs, so the row is invisible to the TTL prune above (which only ever +considers `finalized_at IS NOT NULL` rows) permanently. The sweep gives such crash-orphaned rows a +terminal status so they re-enter the normal retention lifecycle. + +**Mechanism.** The same background task that runs `prune()` (the existing supervised retention +loop — no new `TaskSupervisor` spawn site) calls `Journal::sweep_orphans(&policy)` **before** +`prune()` on every tick: + +``` +tick: + backend.sweep_orphans(&policy).await // NEW — must run before prune() + backend.prune(&policy).await // existing +``` + +`sweep_orphans` (see INV-17 for the full invariant): +1. If `policy.stale_running_after_secs == 0`, return `Ok(0)` (disabled). +2. If the backend has no `lock_dir` (`:memory:`, Postgres, non-Unix), warn-once and return `Ok(0)` + — see Non-Goals and NEVER. This is a documented no-op, not a silent gap. +3. Batch-scan (`prune_batch_size` rows per batch, yielding between batches — same discipline as + `prune`) `status='running' AND updated_at <= now - stale_running_after_secs`. +4. For each candidate, non-blocking try-acquire its INV-15 `ExecutionLock`: + - `ExecutionLocked` (lock held) → live owner → skip. + - Acquired → no live owner → while still holding the lock, run + `UPDATE durable_executions SET status='aborted', finalized_at=now, updated_at=now WHERE + execution_id=? AND status='running' AND finalized_at IS NULL`, then release the lock. +5. Return the count aborted. + +The sweep never deletes rows and never retries/resumes an orphan — it only finalizes. Deletion +stays exclusively with `prune()`; resume stays exclusively with the pre-existing +`ensure_session_durable_ctx` → `open_execution` replay path on the next legitimate open (INV-17). + +**Dispatch & Postgres compile-correctness (M1/M2).** `sweep_orphans` is a `Journal` trait method +(see Journal trait above), dispatched through `DurableBackendEnum` — not called on `LocalBackend` +directly. The new SQL (the candidate `SELECT` and the guarded abort `UPDATE`) MUST go through the +project's `sql!()` macro so both queries compile and rewrite placeholders correctly under +`--features postgres`, even though `sweep_orphans` is a `lock_dir=None` no-op on Postgres at +runtime — the code path still has to compile there. + +**Ordering (M3 tick-safety).** The sweep MUST run before the same tick's `prune()` so a +just-aborted orphan is visible to that tick's TTL check. With `ttl_failed_secs=0`, a freshly-swept +orphan (`finalized_at = now`) becomes prune-eligible on that same tick (cutoff `now - 0 = now`, +`finalized_at <= now` matches) — intended and harmless, since `0` already means "prune +failed/aborted immediately." + +**Zero DB migrations.** `status`, `updated_at`, and `finalized_at` already exist on +`durable_executions`; `'aborted'` is already a legal `CHECK` value. No new column, no new index — +the existing `idx_durable_exec_status_time (status, finalized_at)` index already restricts the scan +to `status='running'`; filtering the (small, bounded-by-live-concurrency) result set on `updated_at` +in memory is cheap. + +**Observability.** Span `durable.retention.sweep_orphans` with an `aborted_count` attribute +(mirrors `durable.journal.prune`), and a metric `durable.retention.orphans_aborted` — see Tracing +Spans. + **In-execution step cap:** `max_steps_per_execution` (default **10000**). On soft exceed (90% of cap): the `JournalWriter` forces a `Checkpoint` fold of the committed-idempotent prefix below the current replay point. On hard exceed: the execution is aborted with @@ -929,6 +1066,20 @@ mechanism. This is not a new contribution — it documents existing behavior. auto-reload-and-resume on startup today; P2 does not add it. A future epic may wire that path; the durable journal provides the substrate. +**Flock-liveness conversion (#6254, m1/m3).** `journal_budget`'s `DagRun` execution-open call +converts from plain `open_execution` to `open_execution_exclusive`, holding the returned +`ExecutionLock` as a local for the function's lifetime — this is what makes the row observable to +the INV-17 crash-orphan sweep (its `budget_exec_id(graph_id, generation)` id is externally-derived +from a monotonic save-generation, so exclusive is the structurally-correct call regardless of the +sweep; see INV-15/#6122). `journal_budget` MUST keep passing `is_resume=false` into +`build_budget_ctx` exactly as before — lock acquisition and the resume flag are orthogonal; +converting the open call does not change the "always a fresh generation, never a resume" semantics +of this path. On `DurableError::ExecutionLocked` (another live instance already owns this +`graph_id`+`generation`), `journal_budget` MUST log and return `Ok(())` — skip journaling this +budget snapshot, do not retry, do not surface an error. This is benign: the existing caller +(`plan.rs:369`) already tolerates a missing snapshot (budget zeroes on next resume). A test MUST +assert `ExecutionLocked` → `Ok(())`, not an error. + ### P3 — Scheduler Exactly-Once (`zeph-scheduler`) Thin adapter in `zeph-scheduler` wrapping `JobStore.record_run()`. Each job fire opens an @@ -943,6 +1094,19 @@ scheduled_fire_time_ms)`. Respects the invariant "fire via `message_queue` injection, never direct agent call." +**Flock-liveness conversion (#6254, m1).** `fire_with_durable`'s `ScheduledJob` execution-open +call converts from plain `open_execution` to `open_execution_exclusive`, holding the returned +`ExecutionLock` as a local for the entire fire-body function (open → step → finalize) — this is +what makes a slow-but-alive job (e.g. a multi-hour nightly job body, held as one `ctx.step`) +correctly skip the INV-17 crash-orphan sweep instead of being false-aborted. Its id +(`derive_execution_id(job_name, slot_ms)`) is externally-derivable, so exclusive is the +structurally-correct call independent of the sweep — it also closes a latent double-drive gap +where two scheduler daemons could otherwise both fire the same `job_name`+`slot_ms` (INV-15/#6122). +On `DurableError::ExecutionLocked` (a peer daemon already owns this slot), the handler MUST log at +info/debug and return `Ok(())` — skip this fire, do not retry (a retry risks a duplicate; the peer +is already firing it), and MUST NOT map to `SchedulerError::TaskFailed`. A test MUST assert +`ExecutionLocked` → `Ok(())`, not a task failure. + ### P4 — Subagent Durable Promise (`zeph-subagent`) Parent opens a `DurablePromise` at spawn time; the subagent resolves it on @@ -1033,6 +1197,7 @@ max_executions = 10000 max_journal_bytes = 1073741824 # 1 GiB prune_batch_size = 500 prune_interval_secs = 3600 +stale_running_after_secs = 3600 # crash-orphan threshold (#6254); 0 disables the sweep # RestateBackend sub-table (only meaningful when backend = "restate" + feature = "restate") [durable.restate] @@ -1063,7 +1228,7 @@ Analogous to `zeph schedule`. Connects directly to `durable.db`; no agent proces | `zeph durable show ` | Show journal entries (metadata only by default; payload redacted) | | `zeph durable show --reveal` | Show with decrypted payload (WARNING printed) | | `zeph durable inspect --step ` | Inspect a single step entry | -| `zeph durable prune [--dry-run]` | Force retention sweep | +| `zeph durable prune [--dry-run]` | Force crash-orphan sweep, then TTL prune (#6254). `--dry-run` reports both counts separately: "N orphaned executions would be aborted" and "M would be pruned" | | `zeph durable resume ` | Manual replay trigger (for supported execution kinds) | **Redaction rule (INV-5):** default output shows only: `entry_kind`, `step_id`, `effect_class`, @@ -1081,12 +1246,20 @@ tokens are never shown without `--reveal`. - `Awaiting external completion…` (promise parked) - `Journal unavailable — non-durable mode` (ACK timeout degradation) +**Crash-orphan sweep (#6254): minimal/no new TUI surface.** The sweep runs on the same background +loop as `prune()`, which has no mandatory palette command today (only the optional `Pruning +journal…` status line above). No new TUI command or palette entry is added for the sweep. If/when +the retention task's status label is extended, it MAY read `Sweeping orphaned executions…` before +`Pruning journal…`; this is optional polish, not a requirement. + ### 3. `--init` Wizard Step in the interactive configuration wizard offering: 1. Enable durable execution? (y/n, default n) 2. Backend: `local` (default) | `restate` -3. Retention defaults (accept defaults or customize TTL/size). +3. Retention defaults (accept defaults or customize TTL/size). `stale_running_after_secs` (#6254) + is not individually surfaced here, consistent with the other `RetentionPolicy` fields — accepting + defaults includes it. 4. (If backend = restate) Vault key configuration for `ZEPH_RESTATE_INGRESS_URL` and `ZEPH_RESTATE_API_KEY`. 5. Generate `ZEPH_DURABLE_KEY` and store in age vault. @@ -1097,6 +1270,16 @@ Migration step adds `[durable]` section with all defaults to existing configs. T purely additive and default-off (`enabled = false`), so no behavior change on upgrade. Migration step is idempotent (skip if `[durable]` already present). +**`stale_running_after_secs` field injection (#6254).** Two migration paths: +1. Fresh migration (no `[durable]` present): the commented `[durable.retention]` block includes + `# stale_running_after_secs = 3600` alongside the other retention defaults. +2. Existing, uncommented `[durable.retention]` table that predates #6254 and lacks the field: a + targeted field-injection step inserts `stale_running_after_secs = 3600` into it (path 1's + early-return on "`[durable]` already present" would otherwise skip this config forever). Because + the field is `#[serde(default)]`, a config that never runs this migration step still loads with + the correct default at runtime — the migration is a self-documentation convenience, not a + correctness requirement. + ### 5. Testing Playbook File: `/Users/rabax/Dev/zeph/.local/testing/playbooks/durable-execution.md` @@ -1120,6 +1303,22 @@ Must cover: 9. **Promise resolution auth** — attempt resolution with wrong resolver token; verify rejection. 10. **Key rotation** — rotate `ZEPH_DURABLE_KEY`; verify in-flight executions complete or drain cleanly. +11. **Crash-orphan sweep (#6254)** — `kill -9` an agent process mid-turn; verify the + `durable_executions` row stays `status='running', finalized_at=NULL` (invisible to plain TTL + prune). Run `zeph durable prune` (or wait a tick with `stale_running_after_secs` lowered for + the test); verify the row flips to `status='aborted', finalized_at=`. Separately, start a + live session and force a sweep tick while it is idle-but-alive (holding its + `ExecutionLock`); verify its row is NOT swept (still `running`, not aborted). Verify + `--dry-run` reports the orphan count without mutating state. +12. **`ExecutionLocked` graceful skip (#6254, m1)** — for both the scheduler `ScheduledJob` and + orchestration `DagRun` adapters: with two processes contending for the same execution id + (same job+slot, or same graph+generation), verify the loser receives `DurableError:: + ExecutionLocked` and the adapter returns `Ok(())` — no `SchedulerError::TaskFailed`, no error + surfaced, no retry. +13. **Reopen unifies `aborted` (#6254, INV-16)** — abort an execution (via the sweep or via + divergence-recovery), then reopen its `ExecutionId` (e.g. resume the same `ConversationId`); + verify the row resets to `status='running', finalized_at=NULL` rather than staying finalized + and prune-eligible. ### 6. Coverage-Status Rows @@ -1141,7 +1340,9 @@ Add to `/Users/rabax/Dev/zeph/.local/testing/coverage-status.md` with status `Un | P3 scheduler exactly-once | 3 | | P4 subagent durable promise | 4 | | Retention sweep | 5 | -| `zeph durable` CLI | 1, 5 | +| Crash-orphan sweep (#6254) | 11, 13 | +| Scheduler/orchestration `ExecutionLocked` graceful skip (#6254) | 12 | +| `zeph durable` CLI | 1, 5, 11 | | TUI `DurableView` + spinners | 1 | --- @@ -1178,6 +1379,7 @@ Span naming convention: `..`. | `durable.journal.read` | `execution_id`, `step_count` | Full read | | `durable.journal.read_segment` | `execution_id`, `from_step_id`, `count` | Range read (replaces full for long sessions) | | `durable.journal.prune` | `deleted_count` | Background sweep | +| `durable.retention.sweep_orphans` | `aborted_count` | Crash-orphan sweep (#6254); runs immediately before `durable.journal.prune` each tick. Metric: `durable.retention.orphans_aborted`. | | `durable.journal.writer.queue_depth` | gauge value | Gauge event per commit cycle | | `durable.step.run` | `step_id`, `effect_class`, `replayed: bool` | Per step; `replayed` drives perf regression gate | | `durable.step.replay` | `step_id`, `effect_class` | Replay path only | @@ -1221,12 +1423,12 @@ wiring; `zeph-durable` provides the key via `StepHandle`, not a config field. | **001** §10 concurrency | Single-threaded async; concurrent tasks, not parallel OS threads | `DurableContext` is `&self` + `AtomicU32`; concurrent `step()` calls are safe; `parallel()` uses `fetch_add(n)` for contiguous reserved blocks. | | **001** §13 DB backend | SQLite/Postgres parity; `zeph_db::DbPool`; all SQL through `sql!()` macro | Dedicated `durable.db` pool using `DatabaseDriver`/`DbPool`/`sql!()`; durable schema files in `zeph-db/migrations/` applied via `zeph_db::run_migrations`; Postgres variant uses same types. | | **001** §15 RuntimeLayer | `&self` hooks, non-fatal, observation-only | RuntimeLayer receives `StepOutcome::Replayed` to suppress double-print. No replay *control* flows through it. | -| **009** orchestration | `GraphPersistence::save()` after every transition; `DagScheduler::resume_from` | P2 adds a parallel journal; `resume_from` restores replan counters from journal instead of zeroing. `pending_permits` use existing lazy re-acquisition. | -| **018** scheduler | `JobStore.record_run()` sole persistence path | P3 wraps `record_run()` as a `DurableStep`; `JobStore` retains sole ownership of `scheduled_jobs`. | +| **009** orchestration | `GraphPersistence::save()` after every transition; `DagScheduler::resume_from` | P2 adds a parallel journal; `resume_from` restores replan counters from journal instead of zeroing. `pending_permits` use existing lazy re-acquisition. #6254 converts `journal_budget`'s execution-open to `open_execution_exclusive`, making `DagRun` liveness observable to the crash-orphan sweep (INV-17); `ExecutionLocked` degrades to a graceful `Ok(())` skip, never a task failure. | +| **018** scheduler | `JobStore.record_run()` sole persistence path | P3 wraps `record_run()` as a `DurableStep`; `JobStore` retains sole ownership of `scheduled_jobs`. #6254 converts `fire_with_durable`'s execution-open to `open_execution_exclusive`, making `ScheduledJob` liveness observable to the crash-orphan sweep (INV-17) and closing a latent cross-daemon double-fire gap; `ExecutionLocked` degrades to a graceful `Ok(())` skip, never `SchedulerError::TaskFailed`. | | **029** feature flags | Flags gate real optional deps; no behavioral markers | `restate` flag gates `dep:restate-sdk`. Core has no flag. | | **031** database abstraction | Single migration runner (`sqlx::migrate!` only in `zeph-db`); `DbPool` from `DatabaseDriver` | Dedicated `durable.db` pool (own `DbConfig::connect()` — valid precedent from `JobStore`). Schema files added to `zeph-db/migrations/{sqlite,postgres}/`; applied via `zeph_db::run_migrations(&durable_pool)`. `zeph-durable` owns NO `.sql` files and NO `sqlx::migrate!` — single source of truth preserved (031 §12). | | **038** vault | All secrets vault-resolved; `ZEPH_*` keys | `ZEPH_DURABLE_KEY`, `ZEPH_RESTATE_*` are vault-resolved; never inline TOML. | -| **039** background-task-supervisor | Tracked via `TaskSupervisor`; supervised restart | `JournalWriter` tokio task is tracked via `zeph-common::TaskSupervisor` (the unified `JoinSet` wrapper, spec-039) under the daemon supervisor; on panic, supervisor restarts the writer, which re-reads the last committed `JournalSeq` and resumes (INV-12, FR-DE-12). | +| **039** background-task-supervisor | Tracked via `TaskSupervisor`; supervised restart | `JournalWriter` tokio task is tracked via `zeph-common::TaskSupervisor` (the unified `JoinSet` wrapper, spec-039) under the daemon supervisor; on panic, supervisor restarts the writer, which re-reads the last committed `JournalSeq` and resumes (INV-12, FR-DE-12). #6254's crash-orphan sweep introduces NO new spawn site: it folds into the already-supervised retention loop (`sweep_orphans()` called before `prune()` each tick), inheriting that loop's existing `TaskSupervisor::spawn` restart policy at both existing spawn call sites. | | **044** subagent lifecycle | Transcript JSONL + `.meta.json` remains the human record | P4 adds a durable promise for control state; transcript unchanged. | | **057** agent persistence | `NEVER double-persist`; `sanitize_tool_pairs` discards orphans | P1 replays journaled steps (no re-insert). `Idempotent` step replay skips `op`. The discard becomes a resume (INV-10). | | **063** worktree subsystem | Subagent spawning, cwd isolation | P4 durable resume reuses the existing respawn path; CwdGuard discipline is unaffected. | @@ -1255,6 +1457,12 @@ wiring; `zeph-durable` provides the key via `StepHandle`, not a config field. | FR-DE-13 | P2: `/plan resume ` MUST restore `task_replan_counts`, `global_replan_count`, `predicate_replans_used`, `predicate_reasons`, and `lineage_chains` from the journal. | | FR-DE-14 | P3: a scheduler job fire whose `EffectIntent` is journaled but `StepResult` is absent on restart MUST apply the job's configured `OnAmbiguous` policy; it MUST NOT unconditionally re-fire. | | FR-DE-15 | Payload encryption MUST use XChaCha20-Poly1305 with a fresh random 24-byte nonce per `seal`. Stored layout: `nonce(24B) \|\| ciphertext \|\| tag(16B)`. | +| FR-DE-16 | (#6254) The crash-orphan sweep MUST abort a `status='running'` row with `updated_at <= now - stale_running_after_secs` ONLY after a non-blocking try-acquire of that execution's `ExecutionLock` succeeds; `ExecutionLocked` MUST short-circuit to skip (no abort). | +| FR-DE-17 | (#6254) The sweep's abort `UPDATE` MUST run while still holding the acquired `ExecutionLock`, guarded by `WHERE execution_id=? AND status='running' AND finalized_at IS NULL`, and MUST run before that tick's `prune()` call. | +| FR-DE-18 | (#6254) `stale_running_after_secs = 0` MUST disable the sweep entirely (`sweep_orphans` returns `Ok(0)` without scanning). | +| FR-DE-19 | (#6254) On a backend with `lock_dir=None` (`:memory:`, Postgres, non-Unix), `sweep_orphans` MUST return `Ok(0)` and emit a warn-once log; it MUST NOT abort any row on staleness alone. | +| FR-DE-20 | (#6254) `zeph-scheduler`'s `fire_with_durable` and `zeph-orchestration`'s `journal_budget` MUST open their execution via `open_execution_exclusive`; on `DurableError::ExecutionLocked` both MUST return `Ok(())` without retry and without surfacing a task-failure error. | +| FR-DE-21 | (#6254) `open_execution`/`open_execution_exclusive`'s reopen path MUST reset `status='running'` and clear `finalized_at` for a row in ANY of `completed`, `failed`, or `aborted` — not only `completed`/`failed`. | ### Non-Functional Requirements (measurable) @@ -1299,6 +1507,11 @@ wiring; `zeph-durable` provides the key via `StepHandle`, not a config field. // TODO(post-v1): auto crash-recovery on process start (no-arg resume of in-flight executions). // v1 fixes only the explicit /plan resume user command path (P2). + +// TODO(post-v1): backend-agnostic (Postgres/:memory:/non-Unix) crash-orphan reclamation via a +// heartbeat-timeout signal ("option c"). #6254's sweep requires the INV-15 advisory flock as its +// liveness signal and is a documented no-op on lock_dir=None backends. Tracked as a separate +// follow-up issue filed after #6254 merges. ``` --- diff --git a/src/cli.rs b/src/cli.rs index 797a66c19..d40c2f274 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -781,9 +781,9 @@ pub(crate) enum DurableCommand { #[arg(long)] json: bool, }, - /// Force a retention sweep over terminal executions past their TTL + /// Force the crash-orphan sweep, then a retention sweep over terminal executions past their TTL Prune { - /// Report how many executions would be pruned without deleting anything + /// Report how many executions would be aborted/pruned without mutating anything #[arg(long)] dry_run: bool, }, diff --git a/src/commands/durable.rs b/src/commands/durable.rs index df5ae9cce..c60afc6ea 100644 --- a/src/commands/durable.rs +++ b/src/commands/durable.rs @@ -329,12 +329,22 @@ pub(crate) async fn handle_durable_command( }; let policy = &config.durable.retention; if dry_run { + let orphans = backend + .count_orphans(policy) + .await + .map_err(|e| anyhow::anyhow!("failed to count orphaned executions: {e}"))?; + println!("Dry run: {orphans} orphaned execution(s) would be aborted."); let n = backend .count_prunable(policy) .await .map_err(|e| anyhow::anyhow!("failed to count prunable executions: {e}"))?; println!("Dry run: {n} terminal execution(s) past TTL would be pruned."); } else { + let aborted = backend + .sweep_orphans(policy) + .await + .map_err(|e| anyhow::anyhow!("failed to sweep orphaned executions: {e}"))?; + println!("Aborted {aborted} orphaned execution(s)."); let n = backend .prune(policy) .await diff --git a/tests/durable_prune_cli.rs b/tests/durable_prune_cli.rs new file mode 100644 index 000000000..e7ee05f95 --- /dev/null +++ b/tests/durable_prune_cli.rs @@ -0,0 +1,247 @@ +// SPDX-FileCopyrightText: 2026 Andrei G +// SPDX-License-Identifier: MIT OR Apache-2.0 + +//! CLI integration coverage for `zeph durable prune` (#6254). +//! +//! `handle_durable_command`'s `Prune` branch only `println!`s its report — there is no return +//! value a unit test could assert on — so this drives the real compiled `zeph` binary as a +//! subprocess (mirroring `tests/daemon_boot.rs`'s `zeph_bin_path` pattern) and asserts on its +//! captured stdout. Seeds a real `durable.db` journal directly through `zeph_durable::LocalBackend` +//! at the exact path `resolve_durable_db_url` resolves for the test config's `memory.sqlite_path`, +//! so both the crash-orphan count and the TTL-prunable count are exercised against genuine rows, +//! not mocks. + +use std::process::Command; + +use zeph_durable::{ExecutionId, ExecutionKind, ExecutionStatus, Journal as _, LocalBackend}; + +/// Resolves the path to the built `zeph` binary at runtime — see `tests/daemon_boot.rs::zeph_bin_path` +/// for why this must be a runtime env var lookup rather than the `env!("CARGO_BIN_EXE_zeph")` macro. +fn zeph_bin_path() -> String { + std::env::var("NEXTEST_BIN_EXE_zeph") + .or_else(|_| std::env::var("CARGO_BIN_EXE_zeph")) + .expect( + "NEXTEST_BIN_EXE_zeph or CARGO_BIN_EXE_zeph must be set by the test runner \ + (cargo test / cargo nextest run / cargo nextest run --archive-file)", + ) +} + +async fn backdate_updated_at(backend: &LocalBackend, id: ExecutionId, updated_at_ms: i64) { + zeph_db::query(zeph_db::sql!( + "UPDATE durable_executions SET updated_at = ? WHERE execution_id = ?" + )) + .bind(updated_at_ms) + .bind(id.as_uuid().to_string()) + .execute(backend.pool()) + .await + .unwrap(); +} + +async fn backdate_finalized_at(backend: &LocalBackend, id: ExecutionId, finalized_at_ms: i64) { + zeph_db::query(zeph_db::sql!( + "UPDATE durable_executions SET finalized_at = ? WHERE execution_id = ?" + )) + .bind(finalized_at_ms) + .bind(id.as_uuid().to_string()) + .execute(backend.pool()) + .await + .unwrap(); +} + +/// `zeph durable prune --dry-run` must report the crash-orphan count and the TTL-prunable count +/// as two separate lines, and must not mutate anything (#6254's CLI wiring: `count_orphans` runs +/// before `count_prunable`, mirroring the non-dry-run `sweep_orphans`-before-`prune` ordering). +#[tokio::test] +async fn durable_prune_dry_run_reports_orphan_and_ttl_counts_separately() { + let tmp = tempfile::tempdir().unwrap(); + let sqlite_path = tmp.path().join("zeph.db"); + + let mut doc: toml_edit::DocumentMut = zeph_core::config::Config::dump_defaults() + .expect("dump default config") + .parse() + .expect("parse default config toml"); + doc["memory"]["sqlite_path"] = toml_edit::value(sqlite_path.display().to_string()); + doc["vault"]["backend"] = toml_edit::value("env"); + doc["durable"]["retention"]["stale_running_after_secs"] = toml_edit::value(1_i64); + doc["durable"]["retention"]["ttl_failed_secs"] = toml_edit::value(1_i64); + let config_path = tmp.path().join("test.toml"); + std::fs::write(&config_path, doc.to_string()).expect("write test config"); + + // Seed the durable journal at the exact URL `resolve_durable_db_url` resolves for this + // `memory.sqlite_path` (no pre-existing legacy `durable.db`, so it's `.durable.db`). + let durable_url = format!("{}.durable.db", sqlite_path.display()); + let backend = LocalBackend::open(&durable_url, 1_048_576) + .await + .expect("open seed backend"); + backend.init().await.expect("init durable schema"); + + // One crash-orphaned `running` execution: stale `updated_at`, no lock held (nothing acquired + // an `ExecutionLock` for it), simulating a crashed owner. + let orphan = ExecutionId::new(); + backend + .open_execution(orphan, ExecutionKind::AgentTurn) + .await + .unwrap(); + backdate_updated_at(&backend, orphan, 0).await; + + // One terminal execution past its TTL. + let stale_failed = ExecutionId::new(); + backend + .open_execution(stale_failed, ExecutionKind::AgentTurn) + .await + .unwrap(); + backend + .finalize(stale_failed, ExecutionStatus::Failed) + .await + .unwrap(); + backdate_finalized_at(&backend, stale_failed, 0).await; + + // Release the pool's connections before the subprocess opens the same sqlite file. + backend.pool().close().await; + drop(backend); + + let bin = zeph_bin_path(); + let output = Command::new(&bin) + .arg("--config") + .arg(&config_path) + .arg("durable") + .arg("prune") + .arg("--dry-run") + .output() + .expect("spawn zeph durable prune --dry-run"); + assert!( + output.status.success(), + "zeph durable prune --dry-run must exit successfully; stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + stdout.contains("Dry run: 1 orphaned execution(s) would be aborted."), + "expected a separate orphan-count line, got stdout:\n{stdout}" + ); + assert!( + stdout.contains("Dry run: 1 terminal execution(s) past TTL would be pruned."), + "expected a separate TTL-prunable-count line, got stdout:\n{stdout}" + ); + + // A dry run must not have mutated anything. + let verify_backend = LocalBackend::open(&durable_url, 1_048_576) + .await + .expect("reopen backend to verify no mutation"); + let (orphan_status,): (String,) = zeph_db::query_as(zeph_db::sql!( + "SELECT status FROM durable_executions WHERE execution_id = ?" + )) + .bind(orphan.as_uuid().to_string()) + .fetch_one(verify_backend.pool()) + .await + .unwrap(); + assert_eq!( + orphan_status, "running", + "dry-run must not abort the orphan" + ); + + let (failed_still_present,): (i64,) = zeph_db::query_as(zeph_db::sql!( + "SELECT COUNT(*) FROM durable_executions WHERE execution_id = ?" + )) + .bind(stale_failed.as_uuid().to_string()) + .fetch_one(verify_backend.pool()) + .await + .unwrap(); + assert_eq!( + failed_still_present, 1, + "dry-run must not delete the prunable row" + ); +} + +/// The non-dry-run `zeph durable prune` must actually sweep the orphan and prune the terminal +/// row, reporting both counts on their own lines. +#[tokio::test] +async fn durable_prune_without_dry_run_sweeps_and_prunes_and_reports_both_counts() { + let tmp = tempfile::tempdir().unwrap(); + let sqlite_path = tmp.path().join("zeph.db"); + + let mut doc: toml_edit::DocumentMut = zeph_core::config::Config::dump_defaults() + .expect("dump default config") + .parse() + .expect("parse default config toml"); + doc["memory"]["sqlite_path"] = toml_edit::value(sqlite_path.display().to_string()); + doc["vault"]["backend"] = toml_edit::value("env"); + doc["durable"]["retention"]["stale_running_after_secs"] = toml_edit::value(1_i64); + doc["durable"]["retention"]["ttl_failed_secs"] = toml_edit::value(1_i64); + let config_path = tmp.path().join("test.toml"); + std::fs::write(&config_path, doc.to_string()).expect("write test config"); + + let durable_url = format!("{}.durable.db", sqlite_path.display()); + let backend = LocalBackend::open(&durable_url, 1_048_576) + .await + .expect("open seed backend"); + backend.init().await.expect("init durable schema"); + + let orphan = ExecutionId::new(); + backend + .open_execution(orphan, ExecutionKind::AgentTurn) + .await + .unwrap(); + backdate_updated_at(&backend, orphan, 0).await; + + let stale_failed = ExecutionId::new(); + backend + .open_execution(stale_failed, ExecutionKind::AgentTurn) + .await + .unwrap(); + backend + .finalize(stale_failed, ExecutionStatus::Failed) + .await + .unwrap(); + backdate_finalized_at(&backend, stale_failed, 0).await; + + backend.pool().close().await; + drop(backend); + + let bin = zeph_bin_path(); + let output = Command::new(&bin) + .arg("--config") + .arg(&config_path) + .arg("durable") + .arg("prune") + .output() + .expect("spawn zeph durable prune"); + assert!( + output.status.success(), + "zeph durable prune must exit successfully; stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!( + stdout.contains("Aborted 1 orphaned execution(s)."), + "expected a separate sweep-count line, got stdout:\n{stdout}" + ); + assert!( + stdout.contains("Pruned 1 execution(s)."), + "expected a separate prune-count line, got stdout:\n{stdout}" + ); + + let verify_backend = LocalBackend::open(&durable_url, 1_048_576) + .await + .expect("reopen backend to verify mutation"); + let (orphan_status,): (String,) = zeph_db::query_as(zeph_db::sql!( + "SELECT status FROM durable_executions WHERE execution_id = ?" + )) + .bind(orphan.as_uuid().to_string()) + .fetch_one(verify_backend.pool()) + .await + .unwrap(); + assert_eq!(orphan_status, "aborted", "the orphan must have been swept"); + + let (failed_still_present,): (i64,) = zeph_db::query_as(zeph_db::sql!( + "SELECT COUNT(*) FROM durable_executions WHERE execution_id = ?" + )) + .bind(stale_failed.as_uuid().to_string()) + .fetch_one(verify_backend.pool()) + .await + .unwrap(); + assert_eq!( + failed_still_present, 0, + "the stale failed execution must have been pruned (deleted)" + ); +}