diff --git a/docs/README.md b/docs/README.md index 35648c31..c4baa156 100644 --- a/docs/README.md +++ b/docs/README.md @@ -125,6 +125,7 @@ open-item lists. - [Changelog](../CHANGELOG.md) — release history and unreleased changes. - [Rust review](RUST_REVIEW.md) — July 2026 best-practice review record. - [Security audit](SECURITY_AUDIT.md) — second Rust security audit record. +- [Long-running engine review](reviews/2026-10-18-long-running-engine.md) — October 2026 lease, lifecycle, and recovery audit. ## Accuracy contract diff --git a/docs/reviews/2026-10-18-long-running-engine.md b/docs/reviews/2026-10-18-long-running-engine.md new file mode 100644 index 00000000..e5d3512a --- /dev/null +++ b/docs/reviews/2026-10-18-long-running-engine.md @@ -0,0 +1,34 @@ +# Long-running engine review — 2026-10-18 + +## Scope and method + +Reviewed the scheduler's claim, heartbeat, retry, preemption, shutdown, and terminal paths; the Postgres, SQLite, and encrypted storage implementations; the shared lifecycle transition API; targeted storage contracts; ignored regressions; and CI/dependency checks. The repository has roughly 488 Rust source files and 174 SQL migrations, so this is a focused reliability review of the long-running execution path, not a line-by-line proof of the whole codebase. + +## Fixed in this review + +| Severity | Finding | Resolution | +| --- | --- | --- | +| Critical | A stale worker could use a state-only compare-and-swap after its lease was reclaimed, changing a replacement owner's `Running` instance to `Failed` or `Scheduled`. The heartbeat-then-update handback also had a check/write race. | Added a required atomic state-and-claim-token transition in both databases and the encrypted wrapper; used it in scheduler error, retry, handback, setup rollback, concurrency deferral, and preemption paths. | +| Critical | Terminal transitions from a claimed run could complete or fail a replacement owner's row and enqueue webhooks for the wrong run. | Added transactional claim-fenced terminal transitions with outbox rows and routed claimed scheduler and step execution paths through them. | +| High | Tree handlers used `updated_at` as a lease token after context writes had changed that field, so delayed work could park in `Waiting` instead of `Scheduled`. | Carried the original claim token through evaluator dispatch to step and loop handlers; the far-future cancellation regression now exercises the delayed tree path. | +| High | A fixed/virtual clock could issue the same `claimed_at` token to consecutive owners, defeating fencing. A node clock ahead of Postgres could also delay stale-lease recovery. | Claim timestamps now advance at least one microsecond beyond the prior token. Postgres uses its own clock; SQLite uses wall time. Added a fixed-clock replacement-owner test for all four storage modes. | +| High | Rows explicitly created as `Running` or `Waiting` had no claim token, so claim-fenced transitions could not process them. | Canonical instance inserts now initialize the token from the row timestamp for active states. Added a backend contract for explicitly created `Running` rows. | +| Medium | `InstanceStore::conditional_update_instance_state` had an unconditional default that let new backends silently violate its compare-and-swap contract. | Made the method required for every backend implementation. | +| High | A governed memory record could shed `schema_version` and `content_sha256`, then be accepted as legacy shared memory after its content changed. | Governed records now require both fields. Truly ungoverned legacy records remain readable only from instance-local KV. | +| Medium | Six regression tests stayed ignored; five had already been fixed, while the memory test still exposed a bug. | Re-enabled tests for API capability grants, cross-tenant push acknowledgements, federation digest normalization, executor RPC exposure, native field collisions, and memory content commitment. | + +Existing records that have a governance envelope but lack its version or content hash will now be unreadable. Recover them only from a trusted source; accepting the partial envelope would reopen the tampering bypass. + +## Residual risks and recommended work + +1. **Hook delivery under overload (high).** `HookTasks` drops cancel/escalation hooks when its 1,024-item in-memory queue is full and aborts remaining hooks when the shutdown grace expires (`orch8-engine/src/scheduler.rs`). A durable hook queue with retry/recovery is needed if these hooks are contractual. Until then, alert on `orch8_hooks_dropped_total` and exercise overload and restart cases in an integration test. +2. **In-flight writes after lease loss (high).** State and webhook transitions are fenced, but a handler already in progress can still save a block output, execution node, or context update after another worker reclaims the instance. Add a claim or run epoch to those write contracts and test a paused handler against reaper recovery. +3. **External effects during lease loss (high).** An already-running handler can also finish a remote effect after its heartbeat fails. The engine should pass a stable idempotency key to every external worker/connector and test recovery during a slow effect. At-least-once execution remains the contract until that is enforced end to end. +4. **Cross-backend failure testing (medium).** Add a fault-injection test that pauses a handler, reclaims its instance, and proves the original task cannot publish outputs or terminal events after the new claim begins. The storage contract now tests atomic state and outbox fencing, but does not simulate every in-flight handler side effect. +5. **Dependency advisory (medium).** The current audit reports [RUSTSEC-2026-0367](https://rustsec.org/advisories/RUSTSEC-2026-0367.html), [0368](https://rustsec.org/advisories/RUSTSEC-2026-0368.html), and [0369](https://rustsec.org/advisories/RUSTSEC-2026-0369.html) for `zip 8.6.0`, a build dependency of Swagger UI generation. `cargo deny check` passes with the repository's configured exceptions. Reassess when the upstream build dependency has a usable patched version; do not classify the advisory as an engine runtime exploit without an affected call path. + +## Verification + +The new `claim_fence` contract exercises a replacement owner with an unchanged scheduler clock, stale failure/reschedule attempts, a stale terminal webhook, and a valid current-owner terminal webhook on SQLite, Postgres, and both encrypted wrappers. CI should continue to run these legs with `DATABASE_URL` set. The six restored tests should remain enabled in the workspace suite. + +Re-review triggers: a new storage backend; changes to claim or reaper SQL; lifecycle/step terminal paths; hook dispatch persistence; or the next `zip` dependency update. diff --git a/orch8-engine/src/dataflow.rs b/orch8-engine/src/dataflow.rs index bca9f6e5..91547398 100644 --- a/orch8-engine/src/dataflow.rs +++ b/orch8-engine/src/dataflow.rs @@ -1937,7 +1937,6 @@ mod tests { } #[test] - #[ignore = "BUG: `order-id` and `order_id` both sanitize to `order_id`, emitting duplicate native fields — fixed on test/bugs"] fn colliding_sanitized_field_names_get_distinct_native_identifiers() { let schema = json!({ "type": "object", diff --git a/orch8-engine/src/evaluator.rs b/orch8-engine/src/evaluator.rs index c19a45dc..f84434e3 100644 --- a/orch8-engine/src/evaluator.rs +++ b/orch8-engine/src/evaluator.rs @@ -73,6 +73,8 @@ pub enum EvalOutcome { struct EvalContext { tree: Vec, instance: TaskInstance, + /// Stable lease token; ordinary context writes can change `updated_at`. + claim_token: chrono::DateTime, clock: orch8_types::clock::SharedClock, /// When true, the tree must be re-read from storage before the next use. tree_stale: bool, @@ -556,7 +558,6 @@ pub async fn evaluate_with_clock( /// set (the scheduler's lease-lost flag), so no further node is dispatched /// after the instance's lease is presumed lost. Stopping early reports /// [`EvalOutcome::MoreWork`]. -#[allow(clippy::too_many_lines)] pub async fn evaluate_fenced( storage: &Arc, handlers: &HandlerRegistry, @@ -564,6 +565,29 @@ pub async fn evaluate_fenced( sequence: &SequenceDefinition, clock: &orch8_types::clock::SharedClock, stop: &std::sync::atomic::AtomicBool, +) -> Result { + evaluate_with_claim_token( + storage, + handlers, + instance, + sequence, + clock, + instance.updated_at, + stop, + ) + .await +} + +/// Evaluate work claimed by the scheduler using its immutable lease token. +#[allow(clippy::too_many_lines)] +pub(crate) async fn evaluate_with_claim_token( + storage: &Arc, + handlers: &HandlerRegistry, + instance: &TaskInstance, + sequence: &SequenceDefinition, + clock: &orch8_types::clock::SharedClock, + claim_token: chrono::DateTime, + stop: &std::sync::atomic::AtomicBool, ) -> Result { // Merge sequence blocks with any dynamically injected blocks. let blocks = merged_blocks(storage.as_ref(), instance.id, sequence).await?; @@ -587,6 +611,7 @@ pub async fn evaluate_fenced( let mut ctx = EvalContext { tree: initial_tree, instance: initial_instance, + claim_token, clock: clock.clone(), tree_stale: false, instance_stale: false, @@ -907,6 +932,7 @@ async fn phase_root_activation( sequence.interceptors.as_ref(), outputs_snapshot, &ctx.clock, + ctx.claim_token, ) .await?; ctx.tree_stale = true; @@ -969,6 +995,7 @@ async fn phase_running_steps( sequence.interceptors.as_ref(), outputs_snapshot, &ctx.clock, + ctx.claim_token, ), dispatch_block( storage, @@ -980,6 +1007,7 @@ async fn phase_running_steps( sequence.interceptors.as_ref(), outputs_snapshot, &ctx.clock, + ctx.claim_token, ), ) }) @@ -1015,6 +1043,7 @@ async fn phase_running_steps( sequence.interceptors.as_ref(), outputs_snapshot, &ctx.clock, + ctx.claim_token, ) .await?; ctx.tree_stale = true; @@ -1137,6 +1166,7 @@ async fn phase_composite_reevaluation( sequence.interceptors.as_ref(), outputs_snapshot, &ctx.clock, + ctx.claim_token, ) .await?; if may_mutate_instance(block) { diff --git a/orch8-engine/src/evaluator/dispatch.rs b/orch8-engine/src/evaluator/dispatch.rs index a256174a..58d3e1fb 100644 --- a/orch8-engine/src/evaluator/dispatch.rs +++ b/orch8-engine/src/evaluator/dispatch.rs @@ -218,6 +218,7 @@ pub(super) async fn dispatch_block( interceptors: Option<&orch8_types::interceptor::InterceptorDef>, outputs: &OutputsSnapshot, clock: &orch8_types::clock::SharedClock, + claim_token: chrono::DateTime, ) -> Result { // Mark node as running. if node.state == NodeState::Pending { @@ -241,7 +242,14 @@ pub(super) async fn dispatch_block( // Boxed: the step-dispatch future (templating, placement, // effect guard, worker enqueue) is large; keep `evaluate` small. let result = Box::pin(crate::handlers::step_block::execute_step_node_with_clock( - storage, handlers, instance, node, step_def, outputs, clock, + storage, + handlers, + instance, + node, + step_def, + outputs, + clock, + claim_token, )) .await; // Interceptor: after_step @@ -297,6 +305,7 @@ pub(super) async fn dispatch_block( loop_def, tree, clock, + claim_token, ) .await } diff --git a/orch8-engine/src/handlers/loop_block.rs b/orch8-engine/src/handlers/loop_block.rs index 5de3d040..13b4fd67 100644 --- a/orch8-engine/src/handlers/loop_block.rs +++ b/orch8-engine/src/handlers/loop_block.rs @@ -73,7 +73,16 @@ pub async fn execute_loop( tree: &[ExecutionNode], ) -> Result { let clock = orch8_types::clock::SharedClock::default(); - execute_loop_with_clock(storage, instance, node, loop_def, tree, &clock).await + execute_loop_with_clock( + storage, + instance, + node, + loop_def, + tree, + &clock, + instance.updated_at, + ) + .await } #[allow(clippy::too_many_lines, clippy::too_many_arguments)] @@ -84,6 +93,7 @@ pub(crate) async fn execute_loop_with_clock( loop_def: &LoopDef, tree: &[ExecutionNode], clock: &orch8_types::clock::SharedClock, + claim_token: chrono::DateTime, ) -> Result { // Empty body: nothing to run. Completing immediately prevents a // condition-truthy loop from spinning on a no-op body forever. @@ -267,8 +277,9 @@ pub(crate) async fn execute_loop_with_clock( if let Some(interval_secs) = loop_def.poll_interval { let next_fire = poll_fire_at(clock.now(), interval_secs); let transitioned = storage - .conditional_update_instance_state( + .conditional_update_claimed_instance_state( instance.id, + claim_token, InstanceState::Running, InstanceState::Scheduled, Some(next_fire), diff --git a/orch8-engine/src/handlers/memory.rs b/orch8-engine/src/handlers/memory.rs index 015eb3c6..328360fc 100644 --- a/orch8-engine/src/handlers/memory.rs +++ b/orch8-engine/src/handlers/memory.rs @@ -935,15 +935,15 @@ fn memory_content_sha256(content: &Value) -> Result { } fn matches_content_commitment(record: &Value, governance: &Value) -> bool { - let legacy = match governance.get("schema_version") { - None => true, - Some(version) if version.as_u64() == Some(1) => false, - // Unknown governed schemas must not be interpreted as this version. - Some(_) => return false, - }; + // An ungoverned record is handled separately by `governed_records` for + // instance-local compatibility. Once a governance envelope exists, its + // version and commitment are both mandatory: treating a missing version + // as legacy lets a tampered record remove both fields and bypass hashing. + if governance.get("schema_version").and_then(Value::as_u64) != Some(1) { + return false; + } let Some(expected) = governance.get("content_sha256") else { - // Only unversioned legacy records may omit the commitment. - return legacy; + return false; }; let Some(expected) = expected.as_str() else { return false; @@ -1827,7 +1827,6 @@ mod tests { } #[test] - #[ignore = "BUG: governed record with schema_version and content_sha256 stripped is accepted as legacy (T4 #6) — fixed on test/bugs"] fn stripping_the_schema_version_does_not_bypass_the_content_commitment() { let mut record = governed_record(&authorization(), now() + Duration::seconds(10)); let governance = record["governance"].as_object_mut().unwrap(); diff --git a/orch8-engine/src/handlers/step_block.rs b/orch8-engine/src/handlers/step_block.rs index b5ba3ade..4693ab16 100644 --- a/orch8-engine/src/handlers/step_block.rs +++ b/orch8-engine/src/handlers/step_block.rs @@ -337,7 +337,17 @@ pub async fn execute_step_node( outputs: &OutputsSnapshot, ) -> Result { let clock = orch8_types::clock::SharedClock::default(); - execute_step_node_with_clock(storage, handlers, instance, node, step_def, outputs, &clock).await + execute_step_node_with_clock( + storage, + handlers, + instance, + node, + step_def, + outputs, + &clock, + instance.updated_at, + ) + .await } #[allow(clippy::too_many_lines, clippy::too_many_arguments)] @@ -349,6 +359,7 @@ pub(crate) async fn execute_step_node_with_clock( step_def: &StepDef, outputs: &OutputsSnapshot, clock: &orch8_types::clock::SharedClock, + claim_token: chrono::DateTime, ) -> Result { // Conditional guard: if `when` is set, evaluate the expression against // the current context + outputs. Skip the step entirely if falsy. Strict @@ -394,7 +405,13 @@ pub(crate) async fn execute_step_node_with_clock( crate::scheduler::step_preamble_deferral(storage.as_ref(), instance, step_def, clock) .await? { - crate::scheduler::park_tree_instance_until(storage.as_ref(), instance.id, fire_at).await?; + crate::scheduler::park_tree_instance_until( + storage.as_ref(), + instance.id, + claim_token, + fire_at, + ) + .await?; return Ok(false); } @@ -431,12 +448,13 @@ pub(crate) async fn execute_step_node_with_clock( patched.context.runtime.current_step_started_at = Some(node_started); patched }); - let deferred = crate::scheduler::check_human_input_at( + let deferred = crate::scheduler::check_human_input_at_with_claim( storage.as_ref(), gate_instance.as_ref().unwrap_or(instance), step_def, human_def, clock.now(), + claim_token, ) .await?; if deferred { @@ -554,8 +572,9 @@ pub(crate) async fn execute_step_node_with_clock( BreakerDecision::Proceed(cow) => cow, BreakerDecision::Defer { fire_at } => { storage - .conditional_update_instance_state( + .conditional_update_claimed_instance_state( instance.id, + claim_token, orch8_types::instance::InstanceState::Running, orch8_types::instance::InstanceState::Scheduled, Some(fire_at), @@ -882,10 +901,11 @@ pub(crate) async fn execute_step_node_with_clock( message = %message, "tree path: retryable failure, re-scheduling instance with backoff" ); - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( storage.as_ref(), instance.id, Some(&instance.tenant_id), + claim_token, orch8_types::instance::InstanceState::Running, orch8_types::instance::InstanceState::Scheduled, Some(fire_at), diff --git a/orch8-engine/src/lifecycle.rs b/orch8-engine/src/lifecycle.rs index eb0fea81..6a02b3ed 100644 --- a/orch8-engine/src/lifecycle.rs +++ b/orch8-engine/src/lifecycle.rs @@ -23,7 +23,41 @@ pub async fn transition_instance( to: InstanceState, next_fire_at: Option>, ) -> Result<(), EngineError> { - transition_instance_inner(storage, instance_id, tenant_id, from, to, next_fire_at, &[]).await + transition_instance_inner( + storage, + instance_id, + tenant_id, + None, + from, + to, + next_fire_at, + &[], + ) + .await +} + +/// Transition a Running instance only while the caller still owns its claim. +#[allow(clippy::too_many_arguments)] +pub async fn transition_claimed_instance( + storage: &dyn StorageBackend, + instance_id: InstanceId, + tenant_id: Option<&TenantId>, + claim_token: DateTime, + from: InstanceState, + to: InstanceState, + next_fire_at: Option>, +) -> Result<(), EngineError> { + transition_instance_inner( + storage, + instance_id, + tenant_id, + Some(claim_token), + from, + to, + next_fire_at, + &[], + ) + .await } /// Transition an instance and durably enqueue one webhook row per configured @@ -45,6 +79,34 @@ pub async fn transition_instance_with_webhook( storage, instance_id, tenant_id, + None, + from, + to, + next_fire_at, + &entries, + ) + .await +} + +/// Claim-fenced transition that commits its webhook rows with the state. +#[allow(clippy::too_many_arguments)] +pub async fn transition_claimed_instance_with_webhook( + storage: &dyn StorageBackend, + instance_id: InstanceId, + tenant_id: Option<&TenantId>, + claim_token: DateTime, + from: InstanceState, + to: InstanceState, + next_fire_at: Option>, + webhook_config: &WebhookConfig, + event: &crate::webhooks::WebhookEvent, +) -> Result<(), EngineError> { + let entries = crate::webhooks::pending_entries(webhook_config, event); + transition_instance_inner( + storage, + instance_id, + tenant_id, + Some(claim_token), from, to, next_fire_at, @@ -58,6 +120,7 @@ async fn transition_instance_inner( storage: &dyn StorageBackend, instance_id: InstanceId, tenant_id: Option<&TenantId>, + claim_token: Option>, from: InstanceState, to: InstanceState, next_fire_at: Option>, @@ -80,20 +143,46 @@ async fn transition_instance_inner( // Atomic CAS: only update if the row is still in `from` state. // Prevents a concurrent cancel/fail from being silently overwritten // by a late scheduler or worker callback. - let updated = if outbox_entries.is_empty() { - storage - .conditional_update_instance_state(instance_id, from, to, next_fire_at) - .await? - } else { - storage - .conditional_update_instance_state_with_outbox( - instance_id, - from, - to, - next_fire_at, - outbox_entries, - ) - .await? + let updated = match (claim_token, outbox_entries.is_empty()) { + (Some(token), true) => { + storage + .conditional_update_claimed_instance_state( + instance_id, + token, + from, + to, + next_fire_at, + ) + .await? + } + (Some(token), false) => { + storage + .conditional_update_claimed_instance_state_with_outbox( + instance_id, + token, + from, + to, + next_fire_at, + outbox_entries, + ) + .await? + } + (None, true) => { + storage + .conditional_update_instance_state(instance_id, from, to, next_fire_at) + .await? + } + (None, false) => { + storage + .conditional_update_instance_state_with_outbox( + instance_id, + from, + to, + next_fire_at, + outbox_entries, + ) + .await? + } }; if !updated { warn!( @@ -493,6 +582,92 @@ mod tests { })); } + #[tokio::test] + async fn claimed_terminal_transition_rejects_replaced_owner_without_webhook() { + let storage = orch8_storage::sqlite::SqliteStorage::in_memory() + .await + .unwrap(); + let inst = mk_scheduled_instance(); + storage.create_instance(&inst).await.unwrap(); + let now = Utc::now() + chrono::Duration::seconds(1); + let first = storage + .claim_due_instances(now, 1, 0) + .await + .unwrap() + .pop() + .unwrap(); + transition_claimed_instance( + &storage, + inst.id, + Some(&inst.tenant_id), + first.updated_at, + InstanceState::Running, + InstanceState::Scheduled, + Some(now), + ) + .await + .unwrap(); + let second = storage + .claim_due_instances(now, 1, 0) + .await + .unwrap() + .pop() + .unwrap(); + assert_ne!(first.updated_at, second.updated_at); + + let config = WebhookConfig { + urls: vec!["https://hooks.example.com/failure".into()], + ..WebhookConfig::default() + }; + let event = crate::webhooks::instance_event("instance.failed", inst.id, json!({})); + let stale = transition_claimed_instance_with_webhook( + &storage, + inst.id, + Some(&inst.tenant_id), + first.updated_at, + InstanceState::Running, + InstanceState::Failed, + None, + &config, + &event, + ) + .await; + assert!(matches!(stale, Err(EngineError::InvalidTransition { .. }))); + assert_eq!( + storage.get_instance(inst.id).await.unwrap().unwrap().state, + InstanceState::Running + ); + assert!( + storage + .claim_due_webhook_outbox(Utc::now(), 10) + .await + .unwrap() + .is_empty() + ); + + transition_claimed_instance_with_webhook( + &storage, + inst.id, + Some(&inst.tenant_id), + second.updated_at, + InstanceState::Running, + InstanceState::Failed, + None, + &config, + &event, + ) + .await + .unwrap(); + assert_eq!( + storage + .claim_due_webhook_outbox(Utc::now(), 10) + .await + .unwrap() + .len(), + 1 + ); + } + #[tokio::test] async fn transition_instance_waiting_to_running_valid() { let storage = orch8_storage::sqlite::SqliteStorage::in_memory() diff --git a/orch8-engine/src/scheduler.rs b/orch8-engine/src/scheduler.rs index 937cb260..40c6e729 100644 --- a/orch8-engine/src/scheduler.rs +++ b/orch8-engine/src/scheduler.rs @@ -29,6 +29,7 @@ mod step_exec; #[cfg(test)] mod tests; +pub(crate) use step_exec::check_human_input_at_with_claim; pub(crate) use step_exec::{ HUMAN_GATE_MARKER, auto_decide_marker_key, clamped_fire_at, delay_marker_key, park_tree_instance_until, step_preamble_deferral, @@ -464,7 +465,10 @@ async fn process_tick(ctx: &TickContext<'_>) -> Result>, Engi // instances share a concurrency_key, more than `max_concurrency` may be // Running simultaneously. We put the excess back to Scheduled here, // synchronously, so the observable Running count never exceeds the limit. - let claimed_ids: Vec<_> = instances.iter().map(|instance| instance.id).collect(); + let claimed_ids: Vec<_> = instances + .iter() + .map(|instance| (instance.id, instance.updated_at)) + .collect(); let mut instances = match enforce_concurrency_limits(ctx.storage, instances, ctx.clock).await { Ok(instances) => instances, Err(error) => { @@ -504,7 +508,10 @@ async fn process_tick(ctx: &TickContext<'_>) -> Result>, Engi let (signals_map, completed_map) = match prefetch { Ok(prefetched) => prefetched, Err(error) => { - let ids: Vec<_> = instances.iter().map(|instance| instance.id).collect(); + let ids: Vec<_> = instances + .iter() + .map(|instance| (instance.id, instance.updated_at)) + .collect(); reschedule_claimed_after_setup_error( ctx.storage, &ids, @@ -540,8 +547,9 @@ async fn process_tick(ctx: &TickContext<'_>) -> Result>, Engi ); if let Err(e) = ctx .storage - .conditional_update_instance_state( + .conditional_update_claimed_instance_state( instance.id, + instance.updated_at, InstanceState::Running, InstanceState::Scheduled, Some(ctx.clock.now()), @@ -684,7 +692,7 @@ async fn process_tick(ctx: &TickContext<'_>) -> Result>, Engi cancel: &cancel, clock: &clock, lease_lost: &lease_lost_inner, - claim_token: Some(claim_token), + claim_token, }; process_instance(&ctx, instance, data).await }) @@ -719,8 +727,9 @@ async fn process_tick(ctx: &TickContext<'_>) -> Result>, Engi + chrono::Duration::from_std(delay) .unwrap_or(chrono::Duration::seconds(60)); if let Err(te) = storage - .conditional_update_instance_state( + .conditional_update_claimed_instance_state( instance_id, + claim_token, InstanceState::Running, InstanceState::Scheduled, Some(fire_at), @@ -736,15 +745,18 @@ async fn process_tick(ctx: &TickContext<'_>) -> Result>, Engi } transient_retries::Decision::Exhausted { attempts } => { crate::metrics::inc(crate::metrics::INSTANCE_TRANSIENT_RETRY_EXHAUSTED); - record_transient_exhaustion(&storage, instance_id, attempts, &e).await; - mark_instance_failed( + if mark_instance_failed( &storage, instance_id, + claim_token, &e, "instance kept failing with a transient-classified error; retries exhausted (transient_retry_exhausted)", "failed to transition instance to Failed after transient retries were exhausted", ) - .await; + .await + { + record_transient_exhaustion(&storage, instance_id, attempts, &e).await; + } } } } @@ -752,6 +764,7 @@ async fn process_tick(ctx: &TickContext<'_>) -> Result>, Engi mark_instance_failed( &storage, instance_id, + claim_token, &e, "instance processing failed", "failed to transition instance to Failed after processing error", @@ -762,6 +775,7 @@ async fn process_tick(ctx: &TickContext<'_>) -> Result>, Engi mark_instance_failed( &storage, instance_id, + claim_token, &join_err, "instance processing panicked", "failed to transition instance to Failed after panic", @@ -925,14 +939,15 @@ async fn record_transient_exhaustion( /// concurrent signal or operator already moved to another state. async fn defer_running_instances( storage: &Arc, - ids: &[InstanceId], + claims: &[(InstanceId, chrono::DateTime)], fire_at: chrono::DateTime, ) -> Result<(), EngineError> { let mut first_error = None; - for &id in ids { + for &(id, claim_token) in claims { if let Err(error) = storage - .conditional_update_instance_state( + .conditional_update_claimed_instance_state( id, + claim_token, InstanceState::Running, InstanceState::Scheduled, Some(fire_at), @@ -950,12 +965,12 @@ async fn defer_running_instances( /// setup error while making every claimed row immediately eligible to retry. async fn reschedule_claimed_after_setup_error( storage: &Arc, - ids: &[InstanceId], + claims: &[(InstanceId, chrono::DateTime)], fire_at: chrono::DateTime, phase: &'static str, ) { - if let Err(error) = defer_running_instances(storage, ids, fire_at).await { - error!(%error, phase, count = ids.len(), "failed to reschedule claimed instances after scheduler setup error"); + if let Err(error) = defer_running_instances(storage, claims, fire_at).await { + error!(%error, phase, count = claims.len(), "failed to reschedule claimed instances after scheduler setup error"); } } @@ -966,30 +981,38 @@ async fn reschedule_claimed_after_setup_error( async fn mark_instance_failed( storage: &Arc, instance_id: InstanceId, + claim_token: chrono::DateTime, cause: &(dyn std::fmt::Display + Send + Sync), cause_label: &'static str, transition_label: &'static str, -) { - crate::metrics::inc(crate::metrics::INSTANCES_FAILED); +) -> bool { error!( instance_id = %instance_id, error = %cause, "{cause_label}" ); - if let Err(te) = storage - .conditional_update_instance_state( + match storage + .conditional_update_claimed_instance_state( instance_id, + claim_token, InstanceState::Running, InstanceState::Failed, None, ) .await { - error!( - instance_id = %instance_id, - error = %te, - "{transition_label}" - ); + Ok(true) => { + crate::metrics::inc(crate::metrics::INSTANCES_FAILED); + true + } + Ok(false) => { + debug!(instance_id = %instance_id, "failure transition skipped: claim no longer owns instance"); + false + } + Err(te) => { + error!(instance_id = %instance_id, error = %te, "{transition_label}"); + false + } } } @@ -1065,7 +1088,7 @@ async fn enforce_concurrency_limits( } let defer_at = clock.now() + chrono::Duration::seconds(1); - let mut deferred_ids: Vec = Vec::with_capacity(deferred_indices.len()); + let mut deferred_claims = Vec::with_capacity(deferred_indices.len()); for &idx in &deferred_indices { let inst = &instances[idx]; debug!( @@ -1073,9 +1096,9 @@ async fn enforce_concurrency_limits( concurrency_key = %inst.concurrency_key.as_deref().unwrap_or(""), "concurrency limit exceeded at claim time, deferring" ); - deferred_ids.push(inst.id); + deferred_claims.push((inst.id, inst.updated_at)); } - defer_running_instances(storage, &deferred_ids, defer_at).await?; + defer_running_instances(storage, &deferred_claims, defer_at).await?; // Order-preserving removal: `swap_remove` would pull tail elements into // the holes, scrambling the priority order established by @@ -2689,9 +2712,8 @@ struct InstanceRunCtx<'a> { /// Set once the instance's lease heartbeat has failed /// [`HEARTBEAT_LEASE_LOST_AFTER`] times in a row (ENG-R-N8). pub lease_lost: &'a std::sync::atomic::AtomicBool, - /// The claim token this run holds (see `heartbeat_instance`); `None` - /// when the instance was not claimed through `claim_due_instances`. - pub claim_token: Option>, + /// The claim token this run holds (see `heartbeat_instance`). + pub claim_token: chrono::DateTime, } /// Process a single claimed instance: execute ALL pending steps in one go. @@ -2717,10 +2739,11 @@ async fn process_instance( crate::ownership::LocalOwnership::Owned => {} crate::ownership::LocalOwnership::Transferring => { info!(%instance_id, "continuity execution is transferring; deferring instance"); - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( storage.as_ref(), instance_id, Some(&instance.tenant_id), + ctx.claim_token, InstanceState::Running, InstanceState::Scheduled, Some(chrono::Utc::now() + chrono::Duration::seconds(30)), @@ -2730,10 +2753,11 @@ async fn process_instance( } crate::ownership::LocalOwnership::Superseded => { warn!(%instance_id, "continuity execution is owned elsewhere; parking local instance"); - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( storage.as_ref(), instance_id, Some(&instance.tenant_id), + ctx.claim_token, InstanceState::Running, InstanceState::Waiting, None, @@ -2818,10 +2842,11 @@ async fn finalize_completed( let completed_event = crate::webhooks::instance_event("instance.completed", instance_id, serde_json::json!({})); - crate::lifecycle::transition_instance_with_webhook( + crate::lifecycle::transition_claimed_instance_with_webhook( ctx.storage.as_ref(), instance_id, Some(&instance.tenant_id), + ctx.claim_token, InstanceState::Running, InstanceState::Completed, None, @@ -2979,8 +3004,9 @@ async fn defer_if_over_concurrency( if position > i64::from(max) { let defer_at = ctx.clock.now() + chrono::Duration::seconds(2); ctx.storage - .conditional_update_instance_state( + .conditional_update_claimed_instance_state( instance_id, + ctx.claim_token, InstanceState::Running, InstanceState::Scheduled, Some(defer_at), @@ -3151,10 +3177,11 @@ async fn execute_step_loop( max = ctx.max_steps_per_instance, "instance exceeded max_steps_per_instance, failing" ); - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( ctx.storage.as_ref(), instance_id, Some(&instance.tenant_id), + ctx.claim_token, InstanceState::Running, InstanceState::Failed, None, @@ -3162,7 +3189,7 @@ async fn execute_step_loop( .await?; return Ok(false); } - if cooperatively_preempt(ctx.storage, instance, ctx.clock).await? { + if cooperatively_preempt(ctx.storage, instance, ctx.clock, ctx.claim_token).await? { return Ok(false); } } @@ -3190,7 +3217,7 @@ async fn execute_step_loop( if let Err(pos) = completed_blocks.binary_search(&step_def.id) { completed_blocks.insert(pos, step_def.id.clone()); } - if cooperatively_preempt(ctx.storage, instance, ctx.clock).await? { + if cooperatively_preempt(ctx.storage, instance, ctx.clock, ctx.claim_token).await? { return Ok(false); } } @@ -3207,6 +3234,7 @@ async fn cooperatively_preempt( storage: &Arc, instance: &orch8_types::instance::TaskInstance, clock: &SharedClock, + claim_token: chrono::DateTime, ) -> Result { let now = clock.now(); if !storage @@ -3216,8 +3244,9 @@ async fn cooperatively_preempt( return Ok(false); } let yielded = storage - .conditional_update_instance_state( + .conditional_update_claimed_instance_state( instance.id, + claim_token, InstanceState::Running, InstanceState::Scheduled, Some(now), @@ -3364,10 +3393,11 @@ async fn handle_tree_more_work( if current == InstanceState::Paused || current.is_terminal() { // Terminal/paused — leave state as-is. } else if has_waiting_nodes { - match crate::lifecycle::transition_instance( + match crate::lifecycle::transition_claimed_instance( ctx.storage.as_ref(), instance_id, Some(&instance.tenant_id), + ctx.claim_token, InstanceState::Running, InstanceState::Waiting, None, @@ -3385,10 +3415,11 @@ async fn handle_tree_more_work( ); } } - } else if let Err(e) = crate::lifecycle::transition_instance( + } else if let Err(e) = crate::lifecycle::transition_claimed_instance( ctx.storage.as_ref(), instance_id, Some(&instance.tenant_id), + ctx.claim_token, current, InstanceState::Scheduled, Some(ctx.clock.now()), @@ -3590,10 +3621,11 @@ async fn fail_if_over_step_budget( max = ctx.max_steps_per_instance, "instance exceeded max_steps_per_instance (tree path), failing" ); - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( ctx.storage.as_ref(), instance_id, Some(&instance.tenant_id), + ctx.claim_token, InstanceState::Running, InstanceState::Failed, None, @@ -3628,16 +3660,11 @@ fn lease_presumed_lost(ctx: &InstanceRunCtx<'_>) -> bool { /// another claim already owns the row: releasing it then would yank the new /// owner's run back to `Scheduled`. async fn hand_back_after_lease_loss(ctx: &InstanceRunCtx<'_>, instance_id: InstanceId) { - if let Some(token) = ctx.claim_token { - let lease = ctx.storage.heartbeat_instance(instance_id, token).await; - if matches!(lease, Ok(orch8_storage::InstanceLease::Lost)) { - return; - } - } if let Err(e) = ctx .storage - .conditional_update_instance_state( + .conditional_update_claimed_instance_state( instance_id, + ctx.claim_token, InstanceState::Running, InstanceState::Scheduled, Some(ctx.clock.now()), @@ -3675,12 +3702,13 @@ async fn process_instance_tree( return Ok(()); } - match crate::evaluator::evaluate_fenced( + match crate::evaluator::evaluate_with_claim_token( storage, handlers, instance, sequence, ctx.clock, + ctx.claim_token, ctx.lease_lost, ) .await @@ -3763,10 +3791,11 @@ async fn handle_tree_done( _ => None, }; let transition = if let Some(event) = terminal_event.as_ref() { - crate::lifecycle::transition_instance_with_webhook( + crate::lifecycle::transition_claimed_instance_with_webhook( ctx.storage.as_ref(), instance_id, Some(&instance.tenant_id), + ctx.claim_token, current, target, None, @@ -3775,10 +3804,11 @@ async fn handle_tree_done( ) .await } else { - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( ctx.storage.as_ref(), instance_id, Some(&instance.tenant_id), + ctx.claim_token, current, target, None, @@ -3891,10 +3921,11 @@ async fn handle_tree_eval_error( ) -> Result<(), EngineError> { let instance_id = instance.id; error!(instance_id = %instance_id, error = %error, "tree evaluation failed"); - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( ctx.storage.as_ref(), instance_id, Some(&instance.tenant_id), + ctx.claim_token, InstanceState::Running, InstanceState::Failed, None, diff --git a/orch8-engine/src/scheduler/step_exec.rs b/orch8-engine/src/scheduler/step_exec.rs index 0a7715d1..10070945 100644 --- a/orch8-engine/src/scheduler/step_exec.rs +++ b/orch8-engine/src/scheduler/step_exec.rs @@ -112,10 +112,11 @@ pub(super) async fn check_budget( ) .await?; - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( storage.as_ref(), instance.id, Some(&instance.tenant_id), + instance.updated_at, InstanceState::Running, InstanceState::Paused, None, @@ -326,11 +327,13 @@ pub(crate) async fn step_preamble_deferral( pub(crate) async fn park_tree_instance_until( storage: &dyn StorageBackend, instance_id: orch8_types::ids::InstanceId, + claim_token: DateTime, fire_at: DateTime, ) -> Result<(), EngineError> { if storage - .conditional_update_instance_state( + .conditional_update_claimed_instance_state( instance_id, + claim_token, InstanceState::Running, InstanceState::Scheduled, Some(fire_at), @@ -344,8 +347,9 @@ pub(crate) async fn park_tree_instance_until( && current.next_fire_at.is_none_or(|t| t > fire_at) { storage - .conditional_update_instance_state( + .conditional_update_claimed_instance_state( instance_id, + claim_token, InstanceState::Scheduled, InstanceState::Scheduled, Some(fire_at), @@ -361,10 +365,11 @@ async fn defer_flat_instance( instance: &orch8_types::instance::TaskInstance, fire_at: DateTime, ) -> Result<(), EngineError> { - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( storage, instance.id, Some(&instance.tenant_id), + instance.updated_at, InstanceState::Running, InstanceState::Scheduled, Some(fire_at), @@ -708,13 +713,33 @@ pub async fn check_human_input( /// [`check_human_input`] with an explicit "now" — the scheduler passes its /// clock's instant here so input timeouts respect virtual time in tests. -#[allow(clippy::too_many_lines)] pub async fn check_human_input_at( storage: &dyn StorageBackend, instance: &orch8_types::instance::TaskInstance, step_def: &orch8_types::sequence::StepDef, human_def: &orch8_types::sequence::HumanInputDef, now: DateTime, +) -> Result { + check_human_input_at_with_claim( + storage, + instance, + step_def, + human_def, + now, + instance.updated_at, + ) + .await +} + +/// Check a tree gate with the lease token captured at claim time. +#[allow(clippy::too_many_lines, clippy::too_many_arguments)] +pub(crate) async fn check_human_input_at_with_claim( + storage: &dyn StorageBackend, + instance: &orch8_types::instance::TaskInstance, + step_def: &orch8_types::sequence::StepDef, + human_def: &orch8_types::sequence::HumanInputDef, + now: DateTime, + claim_token: DateTime, ) -> Result { // Dry-run auto-approve: resolve the gate with the default (first) choice so // the simulation flows past human gates instead of pausing. Behaviorally @@ -900,10 +925,11 @@ pub async fn check_human_input_at( storage.save_block_output(&output).await?; // Re-schedule immediately so the next tick sees the block // output in completed_blocks and skips this step. - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( storage, instance.id, Some(&instance.tenant_id), + claim_token, InstanceState::Running, InstanceState::Scheduled, Some(now), @@ -1014,10 +1040,11 @@ pub(super) async fn execute_step_block( block_id = %step_def.id, "debug breakpoint hit, pausing instance" ); - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( storage.as_ref(), instance_id, Some(&instance.tenant_id), + instance.updated_at, InstanceState::Running, InstanceState::Paused, None, @@ -1184,10 +1211,11 @@ pub(super) async fn execute_step_block( // satisfied, so without this the instance would park with // nothing listening for its events. crate::event_correlation::register_wait_before_park(storage, instance, step_def).await; - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( storage.as_ref(), instance_id, Some(&instance.tenant_id), + instance.updated_at, InstanceState::Running, InstanceState::Waiting, None, @@ -1252,10 +1280,11 @@ pub(super) async fn execute_step_block( ) { crate::handlers::step_block::BreakerDecision::Proceed(cow) => cow, crate::handlers::step_block::BreakerDecision::Defer { fire_at } => { - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( storage.as_ref(), instance_id, Some(&instance.tenant_id), + instance.updated_at, InstanceState::Running, InstanceState::Scheduled, Some(fire_at), @@ -1471,10 +1500,11 @@ pub(super) async fn execute_step_block( .await { crate::handlers::param_resolve::SelfModifyResult::Applied => { - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( storage.as_ref(), instance_id, Some(&instance.tenant_id), + instance.updated_at, InstanceState::Running, InstanceState::Scheduled, Some(clock.now()), @@ -1603,10 +1633,11 @@ pub(super) async fn execute_step_block( instance_id, serde_json::json!({ "error": e.to_string() }), ); - crate::lifecycle::transition_instance_with_webhook( + crate::lifecycle::transition_claimed_instance_with_webhook( storage.as_ref(), instance_id, Some(&instance.tenant_id), + instance.updated_at, InstanceState::Running, InstanceState::Failed, None, @@ -1684,10 +1715,11 @@ async fn fail_retryable_instance( "block_id": step_def.id.as_str(), }), ); - crate::lifecycle::transition_instance_with_webhook( + crate::lifecycle::transition_claimed_instance_with_webhook( storage, instance_id, Some(&instance.tenant_id), + instance.updated_at, InstanceState::Running, InstanceState::Failed, None, @@ -1787,10 +1819,11 @@ pub(super) async fn handle_retryable_failure( "retryable failure, re-scheduling with backoff" ); - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( storage, instance_id, Some(&instance.tenant_id), + instance.updated_at, InstanceState::Running, InstanceState::Scheduled, Some(fire_at), @@ -1799,16 +1832,26 @@ pub(super) async fn handle_retryable_failure( return Ok(()); } - // No retry policy — fail the instance. + fail_without_retry_policy(storage, instance, webhook_config, message).await +} + +async fn fail_without_retry_policy( + storage: &dyn StorageBackend, + instance: &orch8_types::instance::TaskInstance, + webhook_config: &WebhookConfig, + message: &str, +) -> Result<(), EngineError> { + let instance_id = instance.id; let failed_event = crate::webhooks::instance_event( "instance.failed", instance_id, serde_json::json!({ "error": message }), ); - crate::lifecycle::transition_instance_with_webhook( + crate::lifecycle::transition_claimed_instance_with_webhook( storage, instance_id, Some(&instance.tenant_id), + instance.updated_at, InstanceState::Running, InstanceState::Failed, None, @@ -1854,10 +1897,11 @@ pub(super) async fn fail_instance_with_error( instance_id, serde_json::json!({ "error": error }), ); - crate::lifecycle::transition_instance_with_webhook( + crate::lifecycle::transition_claimed_instance_with_webhook( storage, instance_id, Some(&instance.tenant_id), + instance.updated_at, InstanceState::Running, InstanceState::Failed, None, @@ -1908,10 +1952,11 @@ pub(super) async fn dispatch_to_external_worker( } // Transition instance Running -> Waiting so the scheduler doesn't re-claim it. - crate::lifecycle::transition_instance( + crate::lifecycle::transition_claimed_instance( storage, instance.id, Some(&instance.tenant_id), + instance.updated_at, InstanceState::Running, InstanceState::Waiting, None, diff --git a/orch8-engine/src/scheduler/tests.rs b/orch8-engine/src/scheduler/tests.rs index 153ee635..b3f57124 100644 --- a/orch8-engine/src/scheduler/tests.rs +++ b/orch8-engine/src/scheduler/tests.rs @@ -109,17 +109,27 @@ async fn cooperative_preemption_reschedules_lower_priority_instance() { created_at: now, updated_at: now, }; - let low = make(Priority::Low, InstanceState::Running, None); + let low = make( + Priority::Low, + InstanceState::Scheduled, + Some(now - chrono::Duration::seconds(1)), + ); let high = make( Priority::High, InstanceState::Scheduled, Some(now - chrono::Duration::seconds(1)), ); storage.create_instance(&low).await.unwrap(); + let low = storage + .claim_due_instances(Utc::now(), 1, 0) + .await + .unwrap() + .pop() + .unwrap(); storage.create_instance(&high).await.unwrap(); assert!( - cooperatively_preempt(&storage, &low, &SharedClock::default()) + cooperatively_preempt(&storage, &low, &SharedClock::default(), low.updated_at) .await .unwrap() ); @@ -396,20 +406,15 @@ async fn scheduler_fast_path_retryable_failure_honours_max_attempts() { let mut final_state = None; for _ in 0..10 { // Simulate the scheduler claim: a rescheduled instance sits in - // `Scheduled`; move it back to `Running` before the next execution. + // `Scheduled`; claim it again to issue a fresh ownership token. let inst = storage.get_instance(instance_id).await.unwrap().unwrap(); let inst = if inst.state == InstanceState::Scheduled { - crate::lifecycle::transition_instance( - storage.as_ref(), - instance_id, - Some(&inst.tenant_id), - InstanceState::Scheduled, - InstanceState::Running, - None, - ) - .await - .unwrap(); - storage.get_instance(instance_id).await.unwrap().unwrap() + storage + .claim_due_instances(Utc::now() + chrono::Duration::seconds(1), 1, 0) + .await + .unwrap() + .pop() + .expect("rescheduled instance is due") } else { inst }; @@ -999,9 +1004,15 @@ async fn deferral_only_reschedules_instances_that_are_still_running() { running_id, None, None, - InstanceState::Running, + InstanceState::Scheduled, ) .await; + let running_claim = storage + .claim_due_instances(Utc::now(), 1, 0) + .await + .unwrap() + .pop() + .unwrap(); seed_instance_with_concurrency( storage.as_ref(), cancelled_id, @@ -1012,9 +1023,17 @@ async fn deferral_only_reschedules_instances_that_are_still_running() { .await; let fire_at = Utc::now() + chrono::Duration::seconds(5); - defer_running_instances(&storage, &[running_id, cancelled_id], fire_at) - .await - .unwrap(); + let cancelled = storage.get_instance(cancelled_id).await.unwrap().unwrap(); + defer_running_instances( + &storage, + &[ + (running_id, running_claim.updated_at), + (cancelled_id, cancelled.updated_at), + ], + fire_at, + ) + .await + .unwrap(); let running = storage.get_instance(running_id).await.unwrap().unwrap(); assert_eq!(running.state, InstanceState::Scheduled); @@ -1163,13 +1182,18 @@ async fn wait_for_input_baseline_survives_reentry() { .current_step_started_at .expect("first entry stamps current_step_started_at"); - // Simulate a wake: back to Running, then re-enter the same step later. + // Simulate a wake and a fresh claim before re-entering the same step. storage - .update_instance_state(instance_id, InstanceState::Running, None) + .update_instance_state(instance_id, InstanceState::Scheduled, Some(Utc::now())) .await .unwrap(); tokio::time::sleep(Duration::from_millis(15)).await; - let instance = storage.get_instance(instance_id).await.unwrap().unwrap(); + let instance = storage + .claim_due_instances(Utc::now() + chrono::Duration::seconds(1), 1, 0) + .await + .unwrap() + .pop() + .expect("woken instance is due"); let outcome = execute_step_block( &storage, ®istry, @@ -1586,7 +1610,7 @@ async fn lease_lost_stops_flat_loop_before_next_step() { cancel: &cancel, clock: &clock, lease_lost: &lease_lost, - claim_token: None, + claim_token: instance.updated_at, }; let finished = execute_step_loop(&ctx, &instance, &sequence, &blocks, Vec::new()) @@ -1638,7 +1662,7 @@ async fn lease_lost_to_other_claim_does_not_release_instance() { cancel: &cancel, clock: &clock, lease_lost: &lease_lost, - claim_token: Some(chrono::Utc::now() - chrono::Duration::hours(1)), + claim_token: chrono::Utc::now() - chrono::Duration::hours(1), }; let finished = execute_step_loop(&ctx, &instance, &sequence, &blocks, Vec::new()) @@ -1704,7 +1728,7 @@ async fn run_tree_with_lease_flag( cancel: &cancel, clock: &clock, lease_lost, - claim_token, + claim_token: claim_token.unwrap_or(instance.updated_at), }; process_instance_tree(&ctx, instance, &two_step_composite_sequence()) .await @@ -2228,7 +2252,7 @@ async fn run_process_instance(storage: &Arc, instance: TaskI cancel: &cancel, clock: &clock, lease_lost: &lease_lost, - claim_token: None, + claim_token: instance.updated_at, }; process_instance( &ctx, @@ -2473,7 +2497,7 @@ async fn run_more_work_with( cancel: &cancel, clock: &clock, lease_lost: &lease_lost, - claim_token: None, + claim_token: parent.updated_at, }; handle_tree_more_work(&ctx, parent, seq, has_waiting_nodes) .await diff --git a/orch8-engine/tests/common/mod.rs b/orch8-engine/tests/common/mod.rs index c3e259cf..089d2cf9 100644 --- a/orch8-engine/tests/common/mod.rs +++ b/orch8-engine/tests/common/mod.rs @@ -684,10 +684,7 @@ pub async fn drive( | InstanceState::Waiting | InstanceState::Paused => return, InstanceState::Scheduled => { - storage - .update_instance_state(instance_id, InstanceState::Running, None) - .await - .unwrap(); + claim_for_drive(storage, &inst).await; } _ => {} } @@ -727,10 +724,7 @@ pub async fn drive_n( return; } if inst.state == InstanceState::Scheduled { - storage - .update_instance_state(instance_id, InstanceState::Running, None) - .await - .unwrap(); + claim_for_drive(storage, &inst).await; } let inst = storage.get_instance(instance_id).await.unwrap().unwrap(); let outcome = evaluator::evaluate(storage, handlers, &inst, sequence) @@ -742,6 +736,22 @@ pub async fn drive_n( } } +/// Resume a scheduled fixture through the real lease path. A direct state +/// update leaves its previous `claimed_at` token in place and makes fenced +/// tree retries fail even though the test is acting as the new owner. +async fn claim_for_drive(storage: &Arc, instance: &TaskInstance) { + let due_at = instance + .next_fire_at + .unwrap_or_else(Utc::now) + .max(Utc::now()); + let claimed = storage.claim_due_instances(due_at, 1024, 0).await.unwrap(); + assert!( + claimed.iter().any(|candidate| candidate.id == instance.id), + "fixture instance {} was not claimed", + instance.id + ); +} + // ------------------------------------------------------------------ // Tree inspection // ------------------------------------------------------------------ diff --git a/orch8-server/src/main.rs b/orch8-server/src/main.rs index f45db76a..5e84ab28 100644 --- a/orch8-server/src/main.rs +++ b/orch8-server/src/main.rs @@ -2390,7 +2390,6 @@ mod tests { /// `orch8-grpc/src/worker_client.rs` always releases a task through the /// unary `ReleaseTask`, so an executor node must accept it. #[test] - #[ignore = "BUG: EXECUTOR_GRPC_RPCS omits /orch8.Orch8Service/ReleaseTask — fixed on test/bugs"] fn executor_surface_lets_remote_workers_release_tasks() { assert!(EXECUTOR_GRPC_RPCS.contains(&"/orch8.Orch8Service/ReleaseTask")); } diff --git a/orch8-server/src/startup_preflight_tests.rs b/orch8-server/src/startup_preflight_tests.rs index b976588e..050cc799 100644 --- a/orch8-server/src/startup_preflight_tests.rs +++ b/orch8-server/src/startup_preflight_tests.rs @@ -555,7 +555,6 @@ fn federation_peers_beyond_each_limit_are_rejected_with_the_rule_broken() { /// trust root in that form (or be rejected up front); an accepted /// uppercase digest would make every envelope from that peer fail. #[test] -#[ignore = "BUG: uppercase trust_root_sha256 passes preflight but never matches the lowercase digest — fixed on test/bugs"] fn accepted_federation_trust_roots_match_the_verifiers_lowercase_digest() { let peers = peer_with("trust_root_sha256", serde_json::json!("A1".repeat(32))); diff --git a/orch8-storage/src/encrypting.rs b/orch8-storage/src/encrypting.rs index 73515ab5..df072ccb 100644 --- a/orch8-storage/src/encrypting.rs +++ b/orch8-storage/src/encrypting.rs @@ -1521,7 +1521,9 @@ passthrough_impl! { async fn update_instance_state(&self, id: InstanceId, new_state: orch8_types::instance::InstanceState, next_fire_at: Option>) -> Result<(), StorageError>; async fn batch_reschedule_instances(&self, ids: &[InstanceId], fire_at: DateTime) -> Result<(), StorageError>; async fn conditional_update_instance_state(&self, id: InstanceId, expected_state: orch8_types::instance::InstanceState, new_state: orch8_types::instance::InstanceState, next_fire_at: Option>) -> Result; + async fn conditional_update_claimed_instance_state(&self, id: InstanceId, claimed_at: DateTime, expected_state: orch8_types::instance::InstanceState, new_state: orch8_types::instance::InstanceState, next_fire_at: Option>) -> Result; async fn conditional_update_instance_state_with_outbox(&self, id: InstanceId, expected_state: orch8_types::instance::InstanceState, new_state: orch8_types::instance::InstanceState, next_fire_at: Option>, entries: &[orch8_types::webhook_outbox::WebhookOutboxEntry]) -> Result; + async fn conditional_update_claimed_instance_state_with_outbox(&self, id: InstanceId, claimed_at: DateTime, expected_state: orch8_types::instance::InstanceState, new_state: orch8_types::instance::InstanceState, next_fire_at: Option>, entries: &[orch8_types::webhook_outbox::WebhookOutboxEntry]) -> Result; async fn update_instance_sequence(&self, id: InstanceId, new_sequence_id: orch8_types::ids::SequenceId) -> Result<(), StorageError>; async fn count_instances(&self, filter: &orch8_types::filter::InstanceFilter) -> Result; async fn any_instance_in_state(&self, state: orch8_types::instance::InstanceState) -> Result; diff --git a/orch8-storage/src/lib.rs b/orch8-storage/src/lib.rs index 0020dbd0..baa5a3dd 100644 --- a/orch8-storage/src/lib.rs +++ b/orch8-storage/src/lib.rs @@ -594,20 +594,27 @@ pub trait InstanceStore: Send + Sync + 'static { /// `expected_state`. Returns `true` if the row was updated, `false` if /// the state had already moved (concurrent writer won the race). /// - /// Default implementation falls through to `update_instance_state` - /// without the guard -- production backends override with - /// `WHERE id = $1 AND state = $expected`. + /// Implementations must enforce the guard in storage. A fallback to an + /// unconditional update would silently break scheduler race protection. async fn conditional_update_instance_state( &self, id: InstanceId, - _expected_state: InstanceState, + expected_state: InstanceState, new_state: InstanceState, next_fire_at: Option>, - ) -> Result { - self.update_instance_state(id, new_state, next_fire_at) - .await?; - Ok(true) - } + ) -> Result; + + /// Change a claimed instance only while this exact claim still owns it. + /// The state and claim token must be checked in the same database update; + /// a heartbeat followed by a state-only CAS leaves a reclaim race. + async fn conditional_update_claimed_instance_state( + &self, + id: InstanceId, + claimed_at: DateTime, + expected_state: InstanceState, + new_state: InstanceState, + next_fire_at: Option>, + ) -> Result; /// CAS the instance state AND enqueue webhook outbox rows in one /// transaction, so a terminal transition can never commit without its @@ -626,6 +633,17 @@ pub trait InstanceStore: Send + Sync + 'static { entries: &[orch8_types::webhook_outbox::WebhookOutboxEntry], ) -> Result; + /// Claim-fenced terminal transition with webhook rows in one transaction. + async fn conditional_update_claimed_instance_state_with_outbox( + &self, + id: InstanceId, + claimed_at: DateTime, + expected_state: InstanceState, + new_state: InstanceState, + next_fire_at: Option>, + entries: &[orch8_types::webhook_outbox::WebhookOutboxEntry], + ) -> Result; + /// Reset the per-run bookkeeping in `context.runtime` for a NEW run: /// regenerate `run_id`, zero `total_steps_executed`, and clear /// `current_step` / `current_step_started_at` / `started_at` (the start diff --git a/orch8-storage/src/postgres/instances.rs b/orch8-storage/src/postgres/instances.rs index 6af48da3..c1b24177 100644 --- a/orch8-storage/src/postgres/instances.rs +++ b/orch8-storage/src/postgres/instances.rs @@ -41,8 +41,8 @@ pub(super) const INSTANCE_INSERT_SQL: &str = r" priority, timezone, metadata, context, concurrency_key, max_concurrency, idempotency_key, session_id, parent_instance_id, budget, - created_at, updated_at, sub_tenant) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19) + created_at, updated_at, sub_tenant, claimed_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20) "; /// Bind a `TaskInstance` to an already-prepared query in the canonical column @@ -79,6 +79,10 @@ pub(super) fn bind_instance_insert<'q>( .bind(inst.created_at) .bind(inst.updated_at) .bind(inst.sub_tenant.as_deref()) + .bind( + matches!(inst.state, InstanceState::Running | InstanceState::Waiting) + .then_some(inst.updated_at), + ) } pub(super) async fn create( @@ -202,7 +206,7 @@ async fn insert_batch_tx( priority, timezone, metadata, context, concurrency_key, max_concurrency, idempotency_key, session_id, parent_instance_id, budget, - created_at, updated_at, sub_tenant) ", + created_at, updated_at, sub_tenant, claimed_at) ", ); qb.push_values( chunk.iter().zip(rows.iter()), @@ -228,7 +232,11 @@ async fn insert_batch_tx( .push_bind(budget.as_ref()) .push_bind(inst.created_at) .push_bind(inst.updated_at) - .push_bind(inst.sub_tenant.as_deref()); + .push_bind(inst.sub_tenant.as_deref()) + .push_bind( + matches!(inst.state, InstanceState::Running | InstanceState::Waiting) + .then_some(inst.updated_at), + ); }, ); let result = qb.build().execute(&mut **transaction).await?; @@ -353,6 +361,8 @@ pub(super) async fn claim_due( .await? }; + let previous_claim = candidate_rows.iter().filter_map(|row| row.claimed_at).max(); + let claim_at = next_claim_timestamp(&mut tx, previous_claim).await?; let all_candidates: Vec = candidate_rows .into_iter() .map(InstanceRow::into_instance) @@ -372,7 +382,7 @@ pub(super) async fn claim_due( sqlx::query( r"UPDATE task_instances SET state = 'running', updated_at = $1, claimed_at = $1 WHERE id = ANY($2)", ) - .bind(now) + .bind(claim_at) .bind(&ids) .execute(&mut *tx) .await?; @@ -386,12 +396,37 @@ pub(super) async fn claim_due( .into_iter() .map(|mut i| { i.state = InstanceState::Running; - i.updated_at = now; + i.updated_at = claim_at; i }) .collect()) } +/// Stamp claims from the same clock the stale reaper uses, then advance past +/// the prior token even when a virtual scheduler clock stands still. +async fn next_claim_timestamp( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + previous_claim: Option>, +) -> Result, StorageError> { + let lease_now: DateTime = sqlx::query_scalar("SELECT clock_timestamp()") + .fetch_one(&mut **tx) + .await?; + let claim_at = match previous_claim { + Some(previous) => { + let next = previous + .checked_add_signed(chrono::Duration::microseconds(1)) + .ok_or_else(|| { + StorageError::Constraint("claim token timestamp exhausted".into()) + })?; + lease_now.max(next) + } + None => lease_now, + }; + // Postgres has microsecond precision. Return exactly the stored token. + DateTime::::from_timestamp_micros(claim_at.timestamp_micros()) + .ok_or_else(|| StorageError::Constraint("claim token timestamp exhausted".into())) +} + /// Filter candidates by `concurrency_key` / `max_concurrency` within a transaction. async fn filter_by_concurrency_pg( tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, @@ -560,6 +595,28 @@ pub(super) async fn conditional_update_state( Ok(result.rows_affected() > 0) } +pub(super) async fn conditional_update_claimed_state( + store: &PostgresStorage, + id: InstanceId, + claimed_at: DateTime, + expected_state: InstanceState, + new_state: InstanceState, + next_fire_at: Option>, +) -> Result { + let result = sqlx::query( + "UPDATE task_instances SET state = $2, next_fire_at = $3, updated_at = NOW() \ + WHERE id = $1 AND state = $4 AND claimed_at = $5", + ) + .bind(id.into_uuid()) + .bind(new_state.to_string()) + .bind(next_fire_at) + .bind(expected_state.to_string()) + .bind(claimed_at) + .execute(&store.pool) + .await?; + Ok(result.rows_affected() > 0) +} + /// Transactional variant: CAS the state and enqueue its webhook rows in the /// same commit. A lost CAS queues no events. pub(super) async fn conditional_update_state_with_outbox( @@ -593,6 +650,38 @@ pub(super) async fn conditional_update_state_with_outbox( Ok(true) } +pub(super) async fn conditional_update_claimed_state_with_outbox( + store: &PostgresStorage, + id: InstanceId, + claimed_at: DateTime, + expected_state: InstanceState, + new_state: InstanceState, + next_fire_at: Option>, + entries: &[orch8_types::webhook_outbox::WebhookOutboxEntry], +) -> Result { + let mut tx = store.pool.begin().await?; + let result = sqlx::query( + "UPDATE task_instances SET state = $2, next_fire_at = $3, updated_at = NOW() \ + WHERE id = $1 AND state = $4 AND claimed_at = $5", + ) + .bind(id.into_uuid()) + .bind(new_state.to_string()) + .bind(next_fire_at) + .bind(expected_state.to_string()) + .bind(claimed_at) + .execute(&mut *tx) + .await?; + if result.rows_affected() == 0 { + tx.rollback().await?; + return Ok(false); + } + for entry in entries { + super::webhook_outbox::park_tx(&mut tx, entry).await?; + } + tx.commit().await?; + Ok(true) +} + pub(super) async fn update_context( store: &PostgresStorage, id: InstanceId, @@ -854,7 +943,7 @@ pub(super) async fn create_batch_externalized( priority, timezone, metadata, context, concurrency_key, max_concurrency, idempotency_key, session_id, parent_instance_id, budget, - created_at, updated_at, sub_tenant) ", + created_at, updated_at, sub_tenant, claimed_at) ", ); let mut idx = 0usize; qb.push_values(chunk, |mut b, (inst, _)| { @@ -882,7 +971,11 @@ pub(super) async fn create_batch_externalized( .push_bind(budget) .push_bind(inst.created_at) .push_bind(inst.updated_at) - .push_bind(inst.sub_tenant.as_deref()); + .push_bind(inst.sub_tenant.as_deref()) + .push_bind( + matches!(inst.state, InstanceState::Running | InstanceState::Waiting) + .then_some(inst.updated_at), + ); }); let result = qb.build().execute(&mut *tx).await?; count += result.rows_affected(); diff --git a/orch8-storage/src/postgres/mod.rs b/orch8-storage/src/postgres/mod.rs index 9d8bf5f6..d0468cd2 100644 --- a/orch8-storage/src/postgres/mod.rs +++ b/orch8-storage/src/postgres/mod.rs @@ -595,6 +595,25 @@ impl crate::InstanceStore for PostgresStorage { instances::conditional_update_state(self, id, expected_state, new_state, next_fire_at).await } + async fn conditional_update_claimed_instance_state( + &self, + id: InstanceId, + claimed_at: DateTime, + expected_state: InstanceState, + new_state: InstanceState, + next_fire_at: Option>, + ) -> Result { + instances::conditional_update_claimed_state( + self, + id, + claimed_at, + expected_state, + new_state, + next_fire_at, + ) + .await + } + async fn conditional_update_instance_state_with_outbox( &self, id: InstanceId, @@ -614,6 +633,27 @@ impl crate::InstanceStore for PostgresStorage { .await } + async fn conditional_update_claimed_instance_state_with_outbox( + &self, + id: InstanceId, + claimed_at: DateTime, + expected_state: InstanceState, + new_state: InstanceState, + next_fire_at: Option>, + entries: &[orch8_types::webhook_outbox::WebhookOutboxEntry], + ) -> Result { + instances::conditional_update_claimed_state_with_outbox( + self, + id, + claimed_at, + expected_state, + new_state, + next_fire_at, + entries, + ) + .await + } + async fn update_instance_context( &self, id: InstanceId, diff --git a/orch8-storage/src/postgres/rows.rs b/orch8-storage/src/postgres/rows.rs index 38a096fe..9f0b1f71 100644 --- a/orch8-storage/src/postgres/rows.rs +++ b/orch8-storage/src/postgres/rows.rs @@ -188,6 +188,8 @@ pub(super) struct InstanceRow { pub budget: Option, pub created_at: DateTime, pub updated_at: DateTime, + #[sqlx(default)] + pub claimed_at: Option>, pub sub_tenant: Option, } diff --git a/orch8-storage/src/sqlite/instances.rs b/orch8-storage/src/sqlite/instances.rs index b3349635..1e079778 100644 --- a/orch8-storage/src/sqlite/instances.rs +++ b/orch8-storage/src/sqlite/instances.rs @@ -38,7 +38,7 @@ use super::helpers::{apply_filter_sql, row_to_instance, ts}; /// insert sites (`create`, `create_batch`, `create_externalized`, /// `create_batch_externalized`, `create_instance_with_dedupe`) bind against /// this string via [`bind_instance_insert`]. -pub(super) const INSTANCE_INSERT_SQL: &str = "INSERT INTO task_instances (id,sequence_id,tenant_id,namespace,state,next_fire_at,priority,timezone,metadata,context,concurrency_key,max_concurrency,idempotency_key,session_id,parent_instance_id,budget,created_at,updated_at,sub_tenant) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19)"; +pub(super) const INSTANCE_INSERT_SQL: &str = "INSERT INTO task_instances (id,sequence_id,tenant_id,namespace,state,next_fire_at,priority,timezone,metadata,context,concurrency_key,max_concurrency,idempotency_key,session_id,parent_instance_id,budget,created_at,updated_at,sub_tenant,claimed_at) VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13,?14,?15,?16,?17,?18,?19,?20)"; /// Bind a `TaskInstance` to an already-prepared query in the canonical column /// order used by [`INSTANCE_INSERT_SQL`]. Kept out of `sqlx::query(...)` so @@ -66,7 +66,11 @@ pub(super) fn bind_instance_insert<'q>( .bind(i.budget.as_ref().map(serde_json::to_string).transpose()?) .bind(ts(i.created_at)) .bind(ts(i.updated_at)) - .bind(i.sub_tenant.as_deref())) + .bind(i.sub_tenant.as_deref()) + .bind( + matches!(i.state, InstanceState::Running | InstanceState::Waiting) + .then_some(ts(i.updated_at)), + )) } #[instrument(skip(storage, i), fields(instance_id = %i.id, tenant = %i.tenant_id))] @@ -195,7 +199,7 @@ pub(super) async fn create_batch( .collect::, _>>()?; let mut qb = sqlx::QueryBuilder::new( - "INSERT INTO task_instances (id,sequence_id,tenant_id,namespace,state,next_fire_at,priority,timezone,metadata,context,concurrency_key,max_concurrency,idempotency_key,session_id,parent_instance_id,budget,created_at,updated_at,sub_tenant) ", + "INSERT INTO task_instances (id,sequence_id,tenant_id,namespace,state,next_fire_at,priority,timezone,metadata,context,concurrency_key,max_concurrency,idempotency_key,session_id,parent_instance_id,budget,created_at,updated_at,sub_tenant,claimed_at) ", ); qb.push_values( chunk.iter().zip(rows.iter()), @@ -218,7 +222,11 @@ pub(super) async fn create_batch( .push_bind(budget.as_ref()) .push_bind(ts(i.created_at)) .push_bind(ts(i.updated_at)) - .push_bind(i.sub_tenant.as_deref()); + .push_bind(i.sub_tenant.as_deref()) + .push_bind( + matches!(i.state, InstanceState::Running | InstanceState::Waiting) + .then_some(ts(i.updated_at)), + ); }, ); let result = qb.build().execute(&mut *tx).await?; @@ -264,15 +272,7 @@ pub(super) async fn claim_due( match result { Ok(instances) => { sqlx::query("COMMIT").execute(&mut *conn).await?; - // `updated_at` carries the claim token the lease heartbeat is - // fenced on (parity with Postgres). - Ok(instances - .into_iter() - .map(|mut i| { - i.updated_at = now; - i - }) - .collect()) + Ok(instances) } Err(e) => { // Best-effort rollback — if it fails the connection is returned @@ -341,6 +341,26 @@ async fn claim_due_inner( ? }; + let mut previous_claim: Option> = None; + for row in &rows { + let token: Option = row.try_get("claimed_at")?; + if let Some(token) = token { + let token = super::helpers::parse_ts(&token)?; + previous_claim = Some(previous_claim.map_or(token, |previous| previous.max(token))); + } + } + let claim_at = match previous_claim { + Some(previous) => { + let next = previous + .checked_add_signed(chrono::Duration::microseconds(1)) + .ok_or_else(|| { + StorageError::Constraint("claim token timestamp exhausted".into()) + })?; + Utc::now().max(next) + } + None => Utc::now(), + }; + let claim_at_s = ts(claim_at); let all_candidates: Vec = rows .iter() .map(row_to_instance) @@ -355,10 +375,10 @@ async fn claim_due_inner( if !instances.is_empty() { let mut qb = sqlx::QueryBuilder::new("UPDATE task_instances SET state='running', updated_at="); - qb.push_bind(now_s); + qb.push_bind(&claim_at_s); // Claim token for the lease heartbeat fence (L6-01). qb.push(", claimed_at="); - qb.push_bind(now_s); + qb.push_bind(&claim_at_s); qb.push(" WHERE id IN ("); let mut separated = qb.separated(","); for i in &instances { @@ -368,7 +388,14 @@ async fn claim_due_inner( qb.build().execute(&mut *conn).await?; } - Ok(instances) + Ok(instances + .into_iter() + .map(|mut instance| { + instance.state = InstanceState::Running; + instance.updated_at = claim_at; + instance + }) + .collect()) } /// Within an open transaction on `conn`, filter candidates by @@ -515,6 +542,29 @@ pub(super) async fn conditional_update_state( Ok(result.rows_affected() > 0) } +pub(super) async fn conditional_update_claimed_state( + storage: &SqliteStorage, + id: InstanceId, + claimed_at: DateTime, + expected_state: InstanceState, + new_state: InstanceState, + next_fire_at: Option>, +) -> Result { + let result = sqlx::query( + "UPDATE task_instances SET state=?2, next_fire_at=?3, updated_at=?4 \ + WHERE id=?1 AND state=?5 AND claimed_at=?6", + ) + .bind(id.to_string()) + .bind(new_state.to_string()) + .bind(next_fire_at.map(ts)) + .bind(ts(Utc::now())) + .bind(expected_state.to_string()) + .bind(ts(claimed_at)) + .execute(&storage.pool) + .await?; + Ok(result.rows_affected() > 0) +} + /// Transactional variant: CAS the state AND insert the webhook outbox rows in /// one transaction — a terminal transition can never commit without its /// webhook events being durably queued. @@ -550,6 +600,39 @@ pub(super) async fn conditional_update_state_with_outbox( Ok(true) } +pub(super) async fn conditional_update_claimed_state_with_outbox( + storage: &SqliteStorage, + id: InstanceId, + claimed_at: DateTime, + expected_state: InstanceState, + new_state: InstanceState, + next_fire_at: Option>, + entries: &[orch8_types::webhook_outbox::WebhookOutboxEntry], +) -> Result { + let mut tx = storage.pool.begin().await?; + let result = sqlx::query( + "UPDATE task_instances SET state=?2, next_fire_at=?3, updated_at=?4 \ + WHERE id=?1 AND state=?5 AND claimed_at=?6", + ) + .bind(id.to_string()) + .bind(new_state.to_string()) + .bind(next_fire_at.map(ts)) + .bind(ts(Utc::now())) + .bind(expected_state.to_string()) + .bind(ts(claimed_at)) + .execute(&mut *tx) + .await?; + if result.rows_affected() == 0 { + tx.rollback().await?; + return Ok(false); + } + for entry in entries { + super::webhook_outbox::park_tx(&mut tx, entry).await?; + } + tx.commit().await?; + Ok(true) +} + pub(super) async fn update_context( storage: &SqliteStorage, id: InstanceId, diff --git a/orch8-storage/src/sqlite/mod.rs b/orch8-storage/src/sqlite/mod.rs index a8e78d73..94666a32 100644 --- a/orch8-storage/src/sqlite/mod.rs +++ b/orch8-storage/src/sqlite/mod.rs @@ -809,6 +809,25 @@ impl crate::InstanceStore for SqliteStorage { instances::conditional_update_state(self, id, expected_state, new_state, next_fire_at).await } + async fn conditional_update_claimed_instance_state( + &self, + id: InstanceId, + claimed_at: DateTime, + expected_state: InstanceState, + new_state: InstanceState, + next_fire_at: Option>, + ) -> Result { + instances::conditional_update_claimed_state( + self, + id, + claimed_at, + expected_state, + new_state, + next_fire_at, + ) + .await + } + async fn conditional_update_instance_state_with_outbox( &self, id: InstanceId, @@ -828,6 +847,27 @@ impl crate::InstanceStore for SqliteStorage { .await } + async fn conditional_update_claimed_instance_state_with_outbox( + &self, + id: InstanceId, + claimed_at: DateTime, + expected_state: InstanceState, + new_state: InstanceState, + next_fire_at: Option>, + entries: &[orch8_types::webhook_outbox::WebhookOutboxEntry], + ) -> Result { + instances::conditional_update_claimed_state_with_outbox( + self, + id, + claimed_at, + expected_state, + new_state, + next_fire_at, + entries, + ) + .await + } + async fn update_instance_context_externalized( &self, id: InstanceId, diff --git a/orch8-storage/src/testing/faulty_methods.rs b/orch8-storage/src/testing/faulty_methods.rs index 4a5a023f..0e8c69dc 100644 --- a/orch8-storage/src/testing/faulty_methods.rs +++ b/orch8-storage/src/testing/faulty_methods.rs @@ -42,7 +42,9 @@ fault_impl! { async fn update_instance_state(&self, id: InstanceId, new_state: InstanceState, next_fire_at: Option>) -> Result<(), StorageError>; async fn batch_reschedule_instances(&self, ids: &[InstanceId], fire_at: DateTime) -> Result<(), StorageError>; async fn conditional_update_instance_state(&self, id: InstanceId, _expected_state: InstanceState, new_state: InstanceState, next_fire_at: Option>) -> Result; + async fn conditional_update_claimed_instance_state(&self, id: InstanceId, claimed_at: DateTime, expected_state: InstanceState, new_state: InstanceState, next_fire_at: Option>) -> Result; async fn conditional_update_instance_state_with_outbox(&self, id: InstanceId, expected_state: InstanceState, new_state: InstanceState, next_fire_at: Option>, entries: &[orch8_types::webhook_outbox::WebhookOutboxEntry]) -> Result; + async fn conditional_update_claimed_instance_state_with_outbox(&self, id: InstanceId, claimed_at: DateTime, expected_state: InstanceState, new_state: InstanceState, next_fire_at: Option>, entries: &[orch8_types::webhook_outbox::WebhookOutboxEntry]) -> Result; async fn reset_instance_run(&self, id: InstanceId, run_id: &str) -> Result<(), StorageError>; async fn update_instance_context(&self, id: InstanceId, context: &orch8_types::context::ExecutionContext) -> Result<(), StorageError>; async fn update_instance_context_cas(&self, id: InstanceId, context: &orch8_types::context::ExecutionContext, _expected_updated_at: DateTime) -> Result; diff --git a/orch8-storage/tests/api_key_grant_contract.rs b/orch8-storage/tests/api_key_grant_contract.rs index 25037a60..1c53d6b0 100644 --- a/orch8-storage/tests/api_key_grant_contract.rs +++ b/orch8-storage/tests/api_key_grant_contract.rs @@ -57,6 +57,5 @@ async fn empty_grant_never_reads_back_as_every_capability(s: &dyn StorageBackend backend_contract!( scoped_grant_reads_back_exactly, - #[ignore = "BUG: an empty persisted capability list is read back as ApiCapability::all() (T4 #4) — fixed on test/bugs"] empty_grant_never_reads_back_as_every_capability, ); diff --git a/orch8-storage/tests/claim_fence.rs b/orch8-storage/tests/claim_fence.rs new file mode 100644 index 00000000..014f5df9 --- /dev/null +++ b/orch8-storage/tests/claim_fence.rs @@ -0,0 +1,157 @@ +//! A stale worker must not change a replacement worker's Running instance. + +use chrono::Utc; +use orch8_storage::backend_test; +use orch8_storage::testing::fixtures::{InstanceBuilder, SequenceBuilder}; +use orch8_types::instance::InstanceState; +use orch8_types::webhook_outbox::{WebhookOutboxEntry, WebhookOutboxStatus}; + +fn webhook(instance_id: orch8_types::ids::InstanceId) -> WebhookOutboxEntry { + WebhookOutboxEntry { + id: uuid::Uuid::now_v7(), + url: "https://hooks.example.test/deliver".into(), + event_type: "instance.failed".into(), + instance_id: Some(instance_id.into_uuid()), + payload: serde_json::json!({}), + attempts: 0, + last_error: None, + created_at: Utc::now(), + delivery_id: None, + status: WebhookOutboxStatus::Pending, + next_attempt_at: None, + claimed_at: None, + } +} + +backend_test!( + async fn stale_claim_cannot_change_new_owner(b) { + let seq = SequenceBuilder::new("tenant").step("work", "noop").build(); + b.storage.create_sequence(&seq).await.unwrap(); + let instance = InstanceBuilder::for_sequence(&seq).due_now().build(); + b.storage.create_instance(&instance).await.unwrap(); + // A fixed future scheduling clock must neither reuse the claim token + // nor push lease age into the future. + let claim_time = Utc::now() + chrono::Duration::minutes(1) + chrono::Duration::nanoseconds(123); + + let first = b + .storage + .claim_due_instances(claim_time, 1, 0) + .await + .unwrap() + .pop() + .unwrap(); + let first_token = first.updated_at; + assert!( + first_token <= Utc::now() + chrono::Duration::seconds(5), + "{}: lease timestamp followed the virtual due clock instead of wall/database time", + b.name() + ); + assert!( + b.storage + .conditional_update_claimed_instance_state( + instance.id, + first_token, + InstanceState::Running, + InstanceState::Scheduled, + Some(claim_time), + ) + .await + .unwrap() + ); + + let second = b + .storage + .claim_due_instances(claim_time, 1, 0) + .await + .unwrap() + .pop() + .unwrap(); + assert_ne!(first_token, second.updated_at); + + for next in [InstanceState::Failed, InstanceState::Scheduled] { + assert!( + !b.storage + .conditional_update_claimed_instance_state( + instance.id, + first_token, + InstanceState::Running, + next, + None, + ) + .await + .unwrap(), + "{}: stale claim changed new owner to {next:?}", + b.name() + ); + } + assert_eq!( + b.storage.get_instance(instance.id).await.unwrap().unwrap().state, + InstanceState::Running + ); + let stale_event = webhook(instance.id); + assert!( + !b.storage + .conditional_update_claimed_instance_state_with_outbox( + instance.id, + first_token, + InstanceState::Running, + InstanceState::Failed, + None, + std::slice::from_ref(&stale_event), + ) + .await + .unwrap() + ); + assert!(b.storage.get_webhook_outbox(stale_event.id).await.unwrap().is_none()); + let current_event = webhook(instance.id); + assert!( + b.storage + .conditional_update_claimed_instance_state_with_outbox( + instance.id, + second.updated_at, + InstanceState::Running, + InstanceState::Failed, + None, + std::slice::from_ref(¤t_event), + ) + .await + .unwrap() + ); + assert!(b.storage.get_webhook_outbox(current_event.id).await.unwrap().is_some()); + } +); + +backend_test!( + async fn explicitly_created_running_row_has_a_claim_token(b) { + let seq = SequenceBuilder::new("tenant").step("work", "noop").build(); + b.storage.create_sequence(&seq).await.unwrap(); + for batch in [false, true] { + let instance = InstanceBuilder::for_sequence(&seq) + .state(InstanceState::Running) + .build(); + if batch { + b.storage + .create_instances_batch(std::slice::from_ref(&instance)) + .await + .unwrap(); + } else { + b.storage.create_instance(&instance).await.unwrap(); + } + let stored = b.storage.get_instance(instance.id).await.unwrap().unwrap(); + assert!( + b.storage + .conditional_update_claimed_instance_state( + instance.id, + stored.updated_at, + InstanceState::Running, + InstanceState::Failed, + None, + ) + .await + .unwrap(), + "{}: explicitly created Running instance (batch={batch}) has no owner token", + b.name() + ); + } + } +); diff --git a/orch8-storage/tests/push_outbox_contract.rs b/orch8-storage/tests/push_outbox_contract.rs index 19972d1c..a49b8733 100644 --- a/orch8-storage/tests/push_outbox_contract.rs +++ b/orch8-storage/tests/push_outbox_contract.rs @@ -542,7 +542,6 @@ backend_contract!( device_registration_refuses_cross_tenant_overwrite, empty_ack_list_acknowledges_nothing, ack_marks_only_the_listed_commands_of_that_device, - #[ignore = "BUG: record_command_acks has no tenant predicate, so an ack marks another tenant's wake for the same device id — fixed on test/bugs"] ack_never_marks_another_tenants_wake_for_the_same_device_id, newer_collapsible_wake_supersedes_pending_wake_of_the_same_execution, collapsible_supersede_never_crosses_tenants, diff --git a/orch8-storage/tests/review_fixes_2026_06.rs b/orch8-storage/tests/review_fixes_2026_06.rs index 3adcc648..37aa3931 100644 --- a/orch8-storage/tests/review_fixes_2026_06.rs +++ b/orch8-storage/tests/review_fixes_2026_06.rs @@ -402,6 +402,15 @@ async fn run_start_is_first_writer_wins_and_reset_clears_old_budget_state() { // reclaimed by recover_stale_instances out from under itself. // --------------------------------------------------------------------------- +async fn backdate_lease_heartbeat(s: &SqliteStorage, id: InstanceId) { + sqlx::query("UPDATE task_instances SET updated_at = ? WHERE id = ?") + .bind((Utc::now() - Duration::hours(1)).to_rfc3339()) + .bind(id.to_string()) + .execute(s.pool()) + .await + .unwrap(); +} + #[tokio::test] async fn heartbeat_instance_prevents_stale_recovery() { let s = store().await; @@ -411,12 +420,10 @@ async fn heartbeat_instance_prevents_stale_recovery() { // while (not one that crashed). let inst = make_instance(InstanceState::Scheduled); s.create_instance(&inst).await.unwrap(); - let claimed = s - .claim_due_instances(Utc::now() - Duration::hours(1), 10, 0) - .await - .unwrap(); + let claimed = s.claim_due_instances(Utc::now(), 10, 0).await.unwrap(); assert_eq!(claimed.len(), 1); let token = claimed[0].updated_at; + backdate_lease_heartbeat(&s, inst.id).await; // The scheduler's heartbeat task renews the lease. assert_eq!( @@ -450,12 +457,10 @@ async fn heartbeat_reports_lost_lease_after_reclaim() { s.create_instance(&inst).await.unwrap(); // Node A claims, then stalls (no heartbeat for over an hour). - let a = s - .claim_due_instances(Utc::now() - Duration::hours(1), 10, 0) - .await - .unwrap(); + let a = s.claim_due_instances(Utc::now(), 10, 0).await.unwrap(); assert_eq!(a.len(), 1); let token_a = a[0].updated_at; + backdate_lease_heartbeat(&s, inst.id).await; // The reaper resets it: A's lease is already gone before anyone re-claims. assert_eq!(