From 7ac87a519038889c15c914f8e99c2cbe470384c5 Mon Sep 17 00:00:00 2001 From: Rita Agafonova <36133358+rita-aga@users.noreply.github.com> Date: Tue, 14 Jul 2026 09:05:57 -0700 Subject: [PATCH 1/4] test(server): failing restart/create survival for pending state timeouts (ARN-203) RED: hard-kill gen-A runtime so boot resume alone must fire timers; overdue case; creation into timed Open without dispatch. --- .../tests/state_timeout_restart.rs | 226 ++++++++++++++++++ 1 file changed, 226 insertions(+) create mode 100644 crates/temper-server/tests/state_timeout_restart.rs diff --git a/crates/temper-server/tests/state_timeout_restart.rs b/crates/temper-server/tests/state_timeout_restart.rs new file mode 100644 index 000000000..954b2221c --- /dev/null +++ b/crates/temper-server/tests/state_timeout_restart.rs @@ -0,0 +1,226 @@ +//! ARN-203: pending state-timeouts must survive a server restart. +//! +//! `[[state_timeout]]` declarations (ADR-0049) arm in-memory tokio timers at +//! dispatch time, and ADR-0056 added hydration re-arm — but only on the NEXT +//! dispatch to the entity. An entity sitting in a timed state across a +//! restart receives no dispatch by definition (the timeout exists to fire +//! when nothing happens), so its `on_timeout` never fires. These tests drive +//! the same boot sequence `temper serve` runs (fresh `ServerState` over the +//! persisted store + `populate_index_from_store`) and require the pending +//! timeout to fire without any post-restart traffic. + +use std::time::Duration; + +use temper_runtime::ActorSystem; +use temper_runtime::tenant::TenantId; +use temper_store_turso::TursoEventStore; + +use temper_server::registry::SpecRegistry; +use temper_server::state::ServerState; +use temper_server::storage::StorageStack; +use temper_spec::csdl::parse_csdl; + +const CSDL_XML: &str = include_str!("../../../test-fixtures/specs/model.csdl.xml"); + +/// Ticket spec whose `Open` state times out after 1 second into +/// `AssignAgent` (Open → InProgress). +const TICKET_WITH_TIMEOUT_IOA: &str = r#" +[automaton] +name = "Ticket" +states = ["Open", "InProgress", "WaitingOnCustomer", "Resolved", "Closed"] +initial = "Open" +allow_indefinite_states = ["InProgress", "WaitingOnCustomer", "Resolved", "Closed"] + +[[state]] +name = "replies" +type = "counter" +initial = "0" + +[[action]] +name = "AssignAgent" +kind = "input" +from = ["Open"] +to = "InProgress" + +[[state_timeout]] +state = "Open" +after_seconds = 1 +on_timeout = "AssignAgent" +"#; + +fn build_state(system_name: &str, store: TursoEventStore) -> ServerState { + let mut registry = SpecRegistry::new(); + let csdl = parse_csdl(CSDL_XML).expect("CSDL should parse"); + registry.register_tenant( + "tenant-a", + csdl, + CSDL_XML.to_string(), + &[("Ticket", TICKET_WITH_TIMEOUT_IOA)], + ); + let mut state = ServerState::from_registry(ActorSystem::new(system_name), registry); + state.set_storage_stack(StorageStack::from_turso(store)); + state +} + +async fn open_store(db_url: &str) -> TursoEventStore { + TursoEventStore::new(db_url, None) + .await + .expect("open local turso db") +} + +/// Poll the entity status until it matches `expected` or the deadline passes. +async fn wait_for_status( + state: &ServerState, + tenant: &TenantId, + entity_id: &str, + expected: &str, + deadline: Duration, +) -> String { + let start = std::time::Instant::now(); + loop { + let current = state + .get_tenant_entity_state(tenant, "Ticket", entity_id) + .await + .expect("entity should load") + .state + .status; + if current == expected || start.elapsed() > deadline { + return current; + } + tokio::time::sleep(Duration::from_millis(200)).await; + } +} + +/// Run generation A (create one ticket) on its OWN tokio runtime and then +/// DROP that runtime — aborting every task it spawned, including the timer +/// the creation-arm (ARN-203) legitimately arms in generation A. This makes +/// the "crash" real: nothing from generation A can fire afterwards, so the +/// post-restart transition can only come from generation B's boot resume. +fn run_and_hard_kill_generation_a(system_name: &str, db_url: &str, entity_id: &str) { + let system_name = system_name.to_string(); + let db_url = db_url.to_string(); + let entity_id = entity_id.to_string(); + std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .expect("build generation-A runtime"); + rt.block_on(async move { + let tenant = TenantId::from("tenant-a".to_string()); + let state_a = build_state(&system_name, open_store(&db_url).await); + let created = state_a + .get_or_create_tenant_entity(&tenant, "Ticket", &entity_id, serde_json::json!({})) + .await + .expect("create ticket"); + assert_eq!(created.state.status, "Open"); + }); + // Hard kill: dropping the runtime aborts all spawned tasks. + rt.shutdown_timeout(Duration::from_millis(100)); + }) + .join() + .expect("generation A thread"); +} + +/// An entity that entered a timed state before a restart, with time still +/// left on the budget, must have its timer re-armed at boot and fire on +/// schedule — with NO post-restart dispatch to the entity. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn pending_state_timeout_fires_after_restart() { + let db_path = + std::env::temp_dir().join(format!("temper-arn203-pending-{}.db", uuid::Uuid::new_v4())); + let db_url = format!("file:{}", db_path.display()); + let tenant = TenantId::from("tenant-a".to_string()); + + // Generation A: create the ticket (entering the timed `Open` state), + // then hard-kill its runtime with the 1s budget not yet elapsed. The + // creation-arm DOES arm an in-memory timer in generation A; the hard + // kill aborts it, exactly like a crashed server process. + run_and_hard_kill_generation_a("arn203-gen-a", &db_url, "t-restart-1"); + + // Generation B: the boot sequence `temper serve` runs for a tenant. + let state_b = build_state("arn203-gen-b", open_store(&db_url).await); + state_b.populate_index_from_store(&tenant).await; + + // No dispatch to the entity. The pending timeout alone must fire. + let status = wait_for_status( + &state_b, + &tenant, + "t-restart-1", + "InProgress", + Duration::from_secs(20), + ) + .await; + assert_eq!( + status, "InProgress", + "pending state timeout must fire after restart without any dispatch to the entity" + ); +} + +/// An entity whose timeout budget fully elapsed while the server was down +/// must have `on_timeout` fired promptly at boot (the overdue case). +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn overdue_state_timeout_fires_after_restart() { + let db_path = + std::env::temp_dir().join(format!("temper-arn203-overdue-{}.db", uuid::Uuid::new_v4())); + let db_url = format!("file:{}", db_path.display()); + let tenant = TenantId::from("tenant-a".to_string()); + + // Hard-kill generation A immediately after creation, before its + // creation-armed timer can fire. + run_and_hard_kill_generation_a("arn203-gen-a2", &db_url, "t-overdue-1"); + + // The 1s budget expires entirely while the server is "down". + tokio::time::sleep(Duration::from_millis(1500)).await; + + let state_b = build_state("arn203-gen-b2", open_store(&db_url).await); + state_b.populate_index_from_store(&tenant).await; + + // Overdue at boot: the fire should happen promptly, well within one + // fresh budget (which would indicate the clock was wrongly reset). + let status = wait_for_status( + &state_b, + &tenant, + "t-overdue-1", + "InProgress", + Duration::from_secs(20), + ) + .await; + assert_eq!( + status, "InProgress", + "an overdue state timeout must fire at boot, not wait another full budget or never fire" + ); +} + +/// Sibling of the restart cases (same defect class, found during review of +/// the fix): an entity CREATED into a timed initial state has no dispatch +/// to arm its timer, so without arming at creation its `on_timeout` never +/// fires while the server stays up. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn state_timeout_fires_for_entity_created_into_timed_state() { + let db_path = + std::env::temp_dir().join(format!("temper-arn203-create-{}.db", uuid::Uuid::new_v4())); + let db_url = format!("file:{}", db_path.display()); + let tenant = TenantId::from("tenant-a".to_string()); + + let state = build_state("arn203-create", open_store(&db_url).await); + let created = state + .get_or_create_tenant_entity(&tenant, "Ticket", "t-created-1", serde_json::json!({})) + .await + .expect("create ticket"); + assert_eq!(created.state.status, "Open"); + + // No restart, no dispatch — the creation itself must arm the timer. + let status = wait_for_status( + &state, + &tenant, + "t-created-1", + "InProgress", + Duration::from_secs(20), + ) + .await; + assert_eq!( + status, "InProgress", + "a state timeout must fire for an entity created into the timed state, without any dispatch" + ); +} From 000fb3f513b10e55a0703ebaf1ea7bbb9d0c4fe3 Mon Sep 17 00:00:00 2001 From: Rita Agafonova <36133358+rita-aga@users.noreply.github.com> Date: Tue, 14 Jul 2026 09:06:00 -0700 Subject: [PATCH 2/4] fix(server): re-arm pending state timeouts at boot and on creation (ARN-203) Boot resume sweep after index population, creation-time arm for timed initial states, shared spawn path, bump_if_zero CAS. ADR-0170. --- .../src/state/dispatch/state_timeouts.rs | 546 +++++++++++++----- crates/temper-server/src/state/entity_ops.rs | 40 ++ docs/adrs/0170-state-timeout-boot-resume.md | 46 ++ 3 files changed, 483 insertions(+), 149 deletions(-) create mode 100644 docs/adrs/0170-state-timeout-boot-resume.md diff --git a/crates/temper-server/src/state/dispatch/state_timeouts.rs b/crates/temper-server/src/state/dispatch/state_timeouts.rs index a6d77ef80..3d5867cc0 100644 --- a/crates/temper-server/src/state/dispatch/state_timeouts.rs +++ b/crates/temper-server/src/state/dispatch/state_timeouts.rs @@ -32,11 +32,13 @@ //! This closes the gap where an orphaned entity (actor passivated or //! server restarted while in a timed state) would otherwise never have //! its timer re-armed because no state transition happened on the -//! hydrated actor. Fully event-log-backed scheduling remains the -//! longer-term direction; hydration re-arm is the 80%-value prefix -//! that makes timeouts reliable across the common failure modes. +//! hydrated actor — when traffic arrives. Restart durability without +//! traffic is handled by the boot resume sweep (ARN-203, ADR-0170): +//! [`ServerState::resume_pending_state_timeouts`] runs after boot index +//! population and re-arms every entity sitting in a timed state with its +//! remaining budget, firing overdue entities immediately. -use std::collections::{HashMap, VecDeque}; +use std::collections::{BTreeMap, VecDeque}; use std::sync::Mutex; use std::time::Duration; @@ -50,6 +52,17 @@ use crate::entity_actor::{EntityEvent, EntityResponse}; use super::effects::PostDispatchContext; +/// Spawn a wall-clock timer/side-effect task. Isolated so the determinism +/// guard's `tokio::spawn` pattern has exactly one annotated production +/// occurrence in this module; the actions these tasks fire are DST-covered +/// via `sim_now()`. +pub(crate) fn spawn_timer_task(fut: F) +where + F: std::future::Future + Send + 'static, +{ + tokio::spawn(fut); // determinism-ok: wall-clock timer side-effect task +} + /// Walk the event history backward to find the timestamp of the most recent /// "progress" signal for the given state: either the transition that entered /// `current_state`, or any later event whose action is in `reset_on`. This @@ -87,7 +100,7 @@ fn compute_state_clock_reset_ts( } /// Composite key identifying an entity instance inside the arm-seq tracker. -#[derive(Debug, Clone, Hash, PartialEq, Eq)] +#[derive(Debug, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)] struct EntityKey { tenant: String, entity_type: String, @@ -104,16 +117,27 @@ impl EntityKey { } } +/// Arguments for arming one state-timeout timer with a pre-acquired seq. +struct TimerArm<'a> { + tenant: &'a TenantId, + entity_type: &'a str, + entity_id: &'a str, + st: &'a temper_spec::automaton::StateTimeout, + delay: Duration, + agent_ctx: &'a crate::request_context::AgentContext, + armed_seq: u64, +} + /// In-memory cancellation counter keyed by entity instance. /// /// Each arm increments and captures the new value; firings compare captured /// against current and drop the fire when they diverge. #[derive(Default, Debug)] pub struct StateTimeoutTracker { - seqs: Mutex>, + seqs: Mutex>, /// ADR-0049: per-entity-type count of armed-but-unfired timers. /// Emitted as `temper_scheduler_pending_timers` by the canary loop. - pending_by_type: Mutex>, + pending_by_type: Mutex>, } impl StateTimeoutTracker { @@ -128,6 +152,21 @@ impl StateTimeoutTracker { *entry } + /// CAS arm for the untracked paths (boot sweep, creation): bump ONLY if + /// no arm has ever happened for this entity in this process (seq == 0), + /// atomically under the seqs mutex. Returns the new seq on success and + /// `None` when another arm won the race — closing the check-then-arm + /// window between reading an entity's state and spawning its timer. + fn bump_if_zero(&self, key: &EntityKey) -> Option { + let mut map = self.seqs.lock().expect("state_timeout tracker poisoned"); + let entry = map.entry(key.clone()).or_insert(0); + if *entry != 0 { + return None; + } + *entry = 1; + Some(1) + } + fn current(&self, key: &EntityKey) -> u64 { self.seqs .lock() @@ -194,8 +233,8 @@ impl crate::state::ServerState { /// /// Invoked from `run_post_dispatch_effects`. Walks the spec's /// `state_timeouts`, bumps the per-entity seq appropriately, and spawns - /// a tokio task per armed timer. Non-durable under the MVP — timers are - /// lost across restarts. + /// a tokio task per armed timer. Timers lost to a restart are re-armed + /// by the boot resume sweep (ARN-203, ADR-0170). pub(crate) fn arm_state_timeouts_if_needed( &self, ctx: &PostDispatchContext<'_>, @@ -271,145 +310,342 @@ impl crate::state::ServerState { // entered the current state (or the most recent `reset_on` // event, whichever is later). If elapsed >= budget, delay is 0 // and the on_timeout action fires on the next tokio tick. - let mut delay = Duration::from_secs(st.after_seconds); - if needs_hydration_rearm { - let clock_reset = - compute_state_clock_reset_ts(&response.state.events, &post_state, &st.reset_on); - if let Some(reset_ts) = clock_reset { - let now = sim_now(); - let elapsed = now - .signed_duration_since(reset_ts) - .to_std() - .unwrap_or(Duration::ZERO); - let budget = Duration::from_secs(st.after_seconds); - let overdue = elapsed >= budget; - delay = if overdue { - Duration::ZERO - } else { - budget - elapsed - }; - crate::runtime_metrics::record_state_timeout_armed_on_hydration( - ctx.tenant.as_str(), - ctx.entity_type, - &st.state, - if overdue { "overdue" } else { "budgeted" }, - ); - } else { - // No entry event found — treat as freshly entered. - // Safe default; worst case is one extra budget of wait. - crate::runtime_metrics::record_state_timeout_armed_on_hydration( - ctx.tenant.as_str(), - ctx.entity_type, - &st.state, - "budgeted", - ); + let delay = if needs_hydration_rearm { + hydration_delay(ctx.tenant, ctx.entity_type, &response.state.events, st) + } else { + Duration::from_secs(st.after_seconds) + }; + + self.spawn_state_timeout_timer( + ctx.tenant, + ctx.entity_type, + ctx.entity_id, + st, + delay, + ctx.agent_ctx, + ); + } + } + + /// ARN-203: re-arm pending state timeouts for every entity of `tenant` + /// currently sitting in a timed state, at boot. + /// + /// The dispatch-time arm (above) and the ADR-0056 hydration re-arm both + /// require traffic to the entity — but a state timeout exists precisely + /// to fire when nothing happens, so a restart left pending timeouts + /// silently dead. This sweep runs after the boot entity-index population: + /// for each entity type whose spec declares `[[state_timeout]]`s, it + /// hydrates the type's entities, and every entity in a declared state + /// with no live timer (tracker seq == 0) is armed with its REMAINING + /// budget (overdue entities fire on the next tick), reusing the same + /// clock-reset reconstruction as the hydration re-arm. + /// + /// Returns the number of timers armed. Idempotent per process: armed + /// entities have seq > 0 and are skipped on a repeat sweep. + pub async fn resume_pending_state_timeouts(&self, tenant: &TenantId) -> usize { + let timed_types: Vec<(String, Vec)> = { + let Ok(registry) = self.registry.read() else { + return 0; + }; + registry + .entity_types(tenant) + .into_iter() + .filter_map(|entity_type| { + registry.get_spec(tenant, entity_type).and_then(|spec| { + if spec.automaton.state_timeouts.is_empty() { + None + } else { + Some(( + entity_type.to_string(), + spec.automaton.state_timeouts.clone(), + )) + } + }) + }) + .collect() + }; + + let mut armed = 0usize; + let service_ctx = crate::request_context::AgentContext::for_service("timeout-scheduler"); + for (entity_type, state_timeouts) in timed_types { + let entity_ids: Vec = { + let index = self + .entity_index + .read() + .expect("entity index lock poisoned"); + index + .get(&format!("{tenant}:{entity_type}")) + .map(|ids| ids.iter().cloned().collect()) + .unwrap_or_default() + }; + for entity_id in entity_ids { + let key = EntityKey::new(tenant, &entity_type, &entity_id); + if self.state_timeout_tracker.current(&key) != 0 { + continue; // a live timer already covers this entity } + let Ok(response) = self + .get_tenant_entity_state(tenant, &entity_type, &entity_id) + .await + else { + continue; + }; + armed += self.arm_untracked_state_timeouts( + tenant, + &entity_type, + &entity_id, + &response, + &state_timeouts, + &service_ctx, + ); } + } + if armed > 0 { + tracing::info!( + tenant = %tenant, + armed, + "resumed pending state timeouts at boot" + ); + } + armed + } - let armed_seq = self.state_timeout_tracker.bump(&key); - self.state_timeout_tracker.inc_pending(ctx.entity_type); - let params: serde_json::Value = serde_json::to_value(&st.params) - .unwrap_or_else(|_| serde_json::Value::Object(serde_json::Map::new())); - - let state = self.clone(); - let tracker = self.state_timeout_tracker.clone(); - let tenant = ctx.tenant.clone(); - let entity_type = ctx.entity_type.to_string(); - let entity_id = ctx.entity_id.to_string(); - let target_state = st.state.clone(); - let target_action = st.on_timeout.clone(); - let agent_ctx = ctx.agent_ctx.clone(); - let key_for_task = key.clone(); - let entity_type_for_dec = ctx.entity_type.to_string(); - let workflow_root_entity_type = agent_ctx - .workflow_root_entity_type - .clone() - .unwrap_or_else(|| entity_type.clone()); - let workflow_root_entity_id = agent_ctx - .workflow_root_entity_id - .clone() - .unwrap_or_else(|| entity_id.clone()); - let workflow_run_id = agent_ctx - .workflow_run_id - .clone() - .unwrap_or_else(|| format!("{entity_type}:{entity_id}")); - - tracing::debug!( - tenant = %ctx.tenant, - entity_type = ctx.entity_type, - entity_id = ctx.entity_id, - target_state = st.state.as_str(), - target_action = st.on_timeout.as_str(), - delay_ms = delay.as_millis() as u64, + /// ARN-203: arm timers for one entity in a timed state that has no live + /// timer (tracker seq == 0), with its remaining budget. Used by the boot + /// resume sweep and by entity creation — a newly created entity whose + /// INITIAL state declares a timeout otherwise has no timer until the + /// first dispatch or the next restart, so its `on_timeout` would never + /// fire on an untouched entity. + pub(crate) fn arm_untracked_state_timeouts( + &self, + tenant: &TenantId, + entity_type: &str, + entity_id: &str, + response: &EntityResponse, + state_timeouts: &[temper_spec::automaton::StateTimeout], + agent_ctx: &crate::request_context::AgentContext, + ) -> usize { + let key = EntityKey::new(tenant, entity_type, entity_id); + let mut armed = 0usize; + for st in state_timeouts { + if st.state != response.state.status { + continue; + } + // CAS: arm only if no other path armed since we read the entity's + // state. A concurrent dispatch-time arm wins and this is a no-op. + let Some(armed_seq) = self.state_timeout_tracker.bump_if_zero(&key) else { + break; + }; + let delay = hydration_delay(tenant, entity_type, &response.state.events, st); + self.spawn_state_timeout_timer_with_seq(TimerArm { + tenant, + entity_type, + entity_id, + st, + delay, + agent_ctx, + armed_seq, + }); + armed += 1; + } + armed + } + + /// Declared `[[state_timeout]]`s for a `(tenant, entity_type)`, or an + /// empty vec when none. Cheap early-out accessor for arm call sites. + pub(crate) fn state_timeout_declarations( + &self, + tenant: &TenantId, + entity_type: &str, + ) -> Vec { + self.registry + .read() + .ok() + .and_then(|registry| { + registry + .get_spec(tenant, entity_type) + .map(|spec| spec.automaton.state_timeouts.clone()) + }) + .unwrap_or_default() + } + + /// Bump the cancellation seq and spawn the timer task for one + /// `[[state_timeout]]` declaration. Shared by the dispatch-time arm, + /// the ADR-0056 hydration re-arm, and the ARN-203 boot resume sweep. + fn spawn_state_timeout_timer( + &self, + tenant: &TenantId, + entity_type: &str, + entity_id: &str, + st: &temper_spec::automaton::StateTimeout, + delay: Duration, + agent_ctx: &crate::request_context::AgentContext, + ) { + let key = EntityKey::new(tenant, entity_type, entity_id); + let armed_seq = self.state_timeout_tracker.bump(&key); + self.spawn_state_timeout_timer_with_seq(TimerArm { + tenant, + entity_type, + entity_id, + st, + delay, + agent_ctx, + armed_seq, + }); + } + + /// Timer-task core, taking a pre-acquired arm seq (the CAS paths acquire + /// theirs via [`StateTimeoutTracker::bump_if_zero`]). + fn spawn_state_timeout_timer_with_seq(&self, arm: TimerArm<'_>) { + let TimerArm { + tenant, + entity_type, + entity_id, + st, + delay, + agent_ctx, + armed_seq, + } = arm; + let key = EntityKey::new(tenant, entity_type, entity_id); + self.state_timeout_tracker.inc_pending(entity_type); + let params: serde_json::Value = serde_json::to_value(&st.params) + .unwrap_or_else(|_| serde_json::Value::Object(serde_json::Map::new())); + + let state = self.clone(); + let tracker = self.state_timeout_tracker.clone(); + let tenant = tenant.clone(); + let entity_type = entity_type.to_string(); + let entity_id = entity_id.to_string(); + let target_state = st.state.clone(); + let target_action = st.on_timeout.clone(); + let agent_ctx = agent_ctx.clone(); + let key_for_task = key; + let entity_type_for_dec = entity_type.clone(); + let workflow_root_entity_type = agent_ctx + .workflow_root_entity_type + .clone() + .unwrap_or_else(|| entity_type.clone()); + let workflow_root_entity_id = agent_ctx + .workflow_root_entity_id + .clone() + .unwrap_or_else(|| entity_id.clone()); + let workflow_run_id = agent_ctx + .workflow_run_id + .clone() + .unwrap_or_else(|| format!("{entity_type}:{entity_id}")); + + tracing::debug!( + tenant = %tenant, + entity_type = %entity_type, + entity_id = %entity_id, + target_state = st.state.as_str(), + target_action = st.on_timeout.as_str(), + delay_ms = delay.as_millis() as u64, + workflow.root_entity_type = %workflow_root_entity_type, + workflow.root_entity_id = %workflow_root_entity_id, + workflow.run_id = %workflow_run_id, + "armed state timeout" + ); + + spawn_timer_task(async move { + // determinism-ok: production timer task; the fired action is deterministic under DST via sim_now() + tokio::time::sleep(delay).await; // determinism-ok: scheduled delay + + let span = tracing::info_span!( + "dispatch.state_timeout.fire", + tenant = %tenant, + entity_type = %entity_type, + entity_id = %entity_id, + target_state = %target_state, + target_action = %target_action, workflow.root_entity_type = %workflow_root_entity_type, workflow.root_entity_id = %workflow_root_entity_id, workflow.run_id = %workflow_run_id, - "armed state timeout" ); - tokio::spawn(async move { - // determinism-ok: wall-clock timer fires a side-effect action; - // the action itself is deterministic under DST via sim_now(). - tokio::time::sleep(delay).await; // determinism-ok: scheduled delay - - let span = tracing::info_span!( - "dispatch.state_timeout.fire", - tenant = %tenant, - entity_type = %entity_type, - entity_id = %entity_id, - target_state = %target_state, - target_action = %target_action, - workflow.root_entity_type = %workflow_root_entity_type, - workflow.root_entity_id = %workflow_root_entity_id, - workflow.run_id = %workflow_run_id, - ); - - async move { - // Sequence-based cancellation check. A newer arm (or a - // state change that bumped the seq on exit) renders - // this timer a no-op. - if tracker.current(&key_for_task) != armed_seq { - tracker.dec_pending(&entity_type_for_dec); - return; - } + async move { + // Sequence-based cancellation check. A newer arm (or a + // state change that bumped the seq on exit) renders + // this timer a no-op. + if tracker.current(&key_for_task) != armed_seq { + tracker.dec_pending(&entity_type_for_dec); + return; + } - // Secondary check: confirm the entity is still in the - // target state. Covers races where the entity left and - // re-entered the same state with a new seq greater than - // this one (in which case the tracker seq would differ, - // already caught above). This is defense in depth. - match state - .get_tenant_entity_state(&tenant, &entity_type, &entity_id) - .await - { - Ok(current) if current.state.status == target_state => { - crate::runtime_metrics::record_state_timeout_fired( - tenant.as_str(), + // Secondary check: confirm the entity is still in the + // target state. Covers races where the entity left and + // re-entered the same state with a new seq greater than + // this one (in which case the tracker seq would differ, + // already caught above). This is defense in depth. + match state + .get_tenant_entity_state(&tenant, &entity_type, &entity_id) + .await + { + Ok(current) if current.state.status == target_state => { + crate::runtime_metrics::record_state_timeout_fired( + tenant.as_str(), + &entity_type, + &target_state, + &target_action, + ); + let _ = state + .dispatch_tenant_action( + &tenant, &entity_type, - &target_state, + &entity_id, &target_action, - ); - let _ = state - .dispatch_tenant_action( - &tenant, - &entity_type, - &entity_id, - &target_action, - params, - &agent_ctx, - ) - .await; - } - _ => { - // State changed or fetch failed — nothing to do. - } + params, + &agent_ctx, + ) + .await; + } + _ => { + // State changed or fetch failed — nothing to do. } - tracker.dec_pending(&entity_type_for_dec); } - .instrument(span) - .await; - }); - } + tracker.dec_pending(&entity_type_for_dec); + } + .instrument(span) + .await; + }); + } +} + +/// Remaining timeout budget for a hydrated entity in a timed state +/// (ADR-0056): the declared budget minus the time already spent in the +/// state per the retained event log. Overdue entities get `Duration::ZERO` +/// (fire on the next tick). When no entry event is retained, the full +/// budget is armed — safe; worst case is one extra budget of wait. +fn hydration_delay( + tenant: &TenantId, + entity_type: &str, + events: &VecDeque, + st: &temper_spec::automaton::StateTimeout, +) -> Duration { + let budget = Duration::from_secs(st.after_seconds); + let Some(reset_ts) = compute_state_clock_reset_ts(events, &st.state, &st.reset_on) else { + crate::runtime_metrics::record_state_timeout_armed_on_hydration( + tenant.as_str(), + entity_type, + &st.state, + "budgeted", + ); + return budget; + }; + let elapsed = sim_now() + .signed_duration_since(reset_ts) + .to_std() + .unwrap_or(Duration::ZERO); + let overdue = elapsed >= budget; + crate::runtime_metrics::record_state_timeout_armed_on_hydration( + tenant.as_str(), + entity_type, + &st.state, + if overdue { "overdue" } else { "budgeted" }, + ); + if overdue { + Duration::ZERO + } else { + budget - elapsed } } @@ -418,6 +654,15 @@ mod tests { use super::*; use temper_runtime::tenant::TenantId; + /// Test-only spawn wrapper so the determinism guard's `tokio::spawn` + /// pattern is annotated once for the load-generation tasks. + fn spawn_load_task(fut: F) -> tokio::task::JoinHandle<()> + where + F: std::future::Future + Send + 'static, + { + tokio::spawn(fut) // determinism-ok: test-only load-generation task + } + fn key() -> EntityKey { EntityKey::new(&TenantId::from("t".to_string()), "E", "e-1") } @@ -683,7 +928,7 @@ queue_timeout_seconds = 10 let barrier = Arc::new(tokio::sync::Barrier::new(N)); let mut handles = Vec::with_capacity(N); - let wall_start = Instant::now(); + let wall_start = Instant::now(); // determinism-ok: test-only wall-clock latency measurement for i in 0..N { let state = state.clone(); let tenant = tenant.clone(); @@ -695,9 +940,10 @@ queue_timeout_seconds = 10 let in_flight = in_flight.clone(); let latencies_ns = latencies_ns.clone(); let barrier = barrier.clone(); - handles.push(tokio::spawn(async move { + handles.push(spawn_load_task(async move { + // determinism-ok: test-only load-generation task barrier.wait().await; // fire all at once - let call_start = Instant::now(); + let call_start = Instant::now(); // determinism-ok: test-only wall-clock latency measurement in_flight.fetch_add(1, Ordering::AcqRel); // Record peak in-flight count. let cur = in_flight.load(Ordering::Acquire); @@ -756,7 +1002,7 @@ queue_timeout_seconds = 10 let o = other.load(Ordering::Acquire); let peak = in_flight_peak.load(Ordering::Acquire); let mut lats = latencies_ns.lock().unwrap().clone(); - lats.sort_unstable(); + lats.sort(); let p = |q: f64| -> u128 { let idx = ((lats.len() as f64 - 1.0) * q).round() as usize; lats[idx.min(lats.len().saturating_sub(1))] @@ -877,7 +1123,7 @@ queue_timeout_seconds = 0 // Prime a synchronization barrier so ALL 300 fire at the same instant. let barrier = Arc::new(tokio::sync::Barrier::new(N)); let mut handles = Vec::with_capacity(N); - let wall_start = std::time::Instant::now(); + let wall_start = std::time::Instant::now(); // determinism-ok: test-only wall-clock latency measurement for _i in 0..N { let state = state.clone(); let tenant = tenant.clone(); @@ -888,9 +1134,10 @@ queue_timeout_seconds = 0 let lat_granted_ns = lat_granted_ns.clone(); let lat_deferred_ns = lat_deferred_ns.clone(); let barrier = barrier.clone(); - handles.push(tokio::spawn(async move { + handles.push(spawn_load_task(async move { + // determinism-ok: test-only load-generation task barrier.wait().await; - let call_start = std::time::Instant::now(); + let call_start = std::time::Instant::now(); // determinism-ok: test-only wall-clock latency measurement let res = state .dispatch_tenant_action_ext_typed( &tenant, @@ -932,8 +1179,8 @@ queue_timeout_seconds = 0 let throughput = N as f64 / wall.as_secs_f64(); let mut gl = lat_granted_ns.lock().unwrap().clone(); let mut dl = lat_deferred_ns.lock().unwrap().clone(); - gl.sort_unstable(); - dl.sort_unstable(); + gl.sort(); + dl.sort(); let p = |v: &[u128], q: f64| -> u128 { if v.is_empty() { return 0; @@ -1045,7 +1292,7 @@ queue_timeout_seconds = 30 let barrier = Arc::new(tokio::sync::Barrier::new(N)); let mut handles = Vec::with_capacity(N); - let wall_start = Instant::now(); + let wall_start = Instant::now(); // determinism-ok: test-only wall-clock latency measurement for i in 0..N { let state = state.clone(); let tenant = tenant.clone(); @@ -1054,9 +1301,10 @@ queue_timeout_seconds = 30 let errored = errored.clone(); let lat_ns = lat_ns.clone(); let barrier = barrier.clone(); - handles.push(tokio::spawn(async move { + handles.push(spawn_load_task(async move { + // determinism-ok: test-only load-generation task barrier.wait().await; - let start = Instant::now(); + let start = Instant::now(); // determinism-ok: test-only wall-clock latency measurement let res = state .dispatch_tenant_action_ext_typed( &tenant, @@ -1091,7 +1339,7 @@ queue_timeout_seconds = 30 let g = granted.load(Ordering::Acquire); let e = errored.load(Ordering::Acquire); let mut lats = lat_ns.lock().unwrap().clone(); - lats.sort_unstable(); + lats.sort(); let p = |q: f64| -> u128 { let idx = ((lats.len() as f64 - 1.0) * q).round() as usize; lats[idx.min(lats.len() - 1)] @@ -1164,7 +1412,7 @@ queue_timeout_seconds = 30 state.arm_state_timeouts_if_needed(&ctx, &response); // Timer is 1s; give it 2s to fire + dispatch + apply. - tokio::time::sleep(std::time::Duration::from_secs(2)).await; + tokio::time::sleep(std::time::Duration::from_secs(2)).await; // determinism-ok: test-only wait for real timer fire let after = state .get_tenant_entity_state(&tenant, "Ticket", "t-1") diff --git a/crates/temper-server/src/state/entity_ops.rs b/crates/temper-server/src/state/entity_ops.rs index 633df47b1..9d5d72843 100644 --- a/crates/temper-server/src/state/entity_ops.rs +++ b/crates/temper-server/src/state/entity_ops.rs @@ -385,6 +385,18 @@ impl ServerState { "populated entity index from event store" ); runtime_metrics::record_server_state_metrics(self); + + // ARN-203: pending state timeouts must survive a restart. + // Now that the index knows every entity, re-arm timers for + // entities sitting in timed states, in the background so + // boot is not blocked on hydrating them. + let state = self.clone(); + let tenant_for_resume = tenant.clone(); + super::dispatch::state_timeouts::spawn_timer_task(async move { + state + .resume_pending_state_timeouts(&tenant_for_resume) + .await; + }); } Err(e) => { tracing::error!( @@ -644,6 +656,17 @@ impl ServerState { "hydrated entities from event store" ); runtime_metrics::record_server_state_metrics(self); + + // ARN-203: same boot-time timeout resume as + // populate_index_from_store — eager hydration is the + // other tenant boot path. + let state = self.clone(); + let tenant_for_resume = tenant.clone(); + super::dispatch::state_timeouts::spawn_timer_task(async move { + state + .resume_pending_state_timeouts(&tenant_for_resume) + .await; + }); } Err(e) => { tracing::error!( @@ -995,6 +1018,23 @@ impl ServerState { ); let _ = self.event_tx.send(change); + // ARN-203: an entity created into a timed INITIAL state has no + // dispatch to arm its timer — without this, on_timeout would never + // fire for an untouched entity. Idempotent: entities with a live + // timer (tracker seq > 0) are skipped, so repeated get-or-create + // calls do not stack timers. + let declarations = self.state_timeout_declarations(tenant, entity_type); + if !declarations.is_empty() { + self.arm_untracked_state_timeouts( + tenant, + entity_type, + entity_id, + &response, + &declarations, + &crate::request_context::AgentContext::for_service("timeout-scheduler"), + ); + } + if let Some(query_plane) = self.query_plane_store() { let status = response.state.status.clone(); let fields = self.query_projection_fields(tenant, entity_type, &response.state.fields); diff --git a/docs/adrs/0170-state-timeout-boot-resume.md b/docs/adrs/0170-state-timeout-boot-resume.md new file mode 100644 index 000000000..5026ca4de --- /dev/null +++ b/docs/adrs/0170-state-timeout-boot-resume.md @@ -0,0 +1,46 @@ +# ADR-0170: State-Timeout Resume at Boot + +- Status: Accepted +- Date: 2026-07-14 +- Deciders: Temper core maintainers +- Related: + - ADR-0049: state_timeout declarations + - ADR-0056: hydration re-arm on dispatch + - ARN-203 + - `crates/temper-server/src/state/dispatch/state_timeouts.rs` + +## Context + +`[[state_timeout]]` timers (ADR-0049) are in-memory `tokio` tasks. ADR-0056 re-arms on +dispatch when tracker seq is 0, but an entity sitting in a timed state across a **restart** +receives no dispatch by definition — the timeout exists to fire when nothing happens. The +timer dies with the process and never re-arms (ARN-203). Creation into a timed initial +state has the same gap: create is not a dispatch. + +## Decision + +1. **Boot resume sweep.** `ServerState::resume_pending_state_timeouts(tenant)` runs after + entity-index population on both tenant boot paths, in a background task. For each type + with state timeouts, hydrate entities and arm any timed state with no live timer. +2. **Remaining budget.** Reuse `compute_state_clock_reset_ts`; overdue entities fire with + zero delay. Missing entry event → full budget (safe default). +3. **Single spawn path.** Dispatch arm, hydration re-arm, boot sweep, and creation share + `spawn_state_timeout_timer`. Untracked arms use `bump_if_zero` for CAS idempotency. +4. **Creation arms.** `get_or_create_tenant_entity` calls `arm_untracked_state_timeouts`. +5. **BTreeMap trackers.** Deterministic iteration for sim-visible crate rules. + +## Consequences + +- Pending timeouts survive restarts; overdue ones fire at boot. +- Sweep cost scales with entities of timed types (background). +- Spec `schedule` effects remain in-memory only (separate residual). + +## Alternatives Considered + +- Durable schedule table: second source of truth; event log already reconstructs clocks. +- Sweep only in CLI bootstrap: miss test/embed boot paths. + +## DST Compliance + +- Wall-clock arm uses `tokio::spawn` with `// determinism-ok` (timer side-effect). +- Tracker maps are `BTreeMap`. Fire path uses existing `__scheduled` dispatch. From cebb022cb1a5eb08ff6aa9d274fe6a81c858b838 Mon Sep 17 00:00:00 2001 From: Rita Agafonova <36133358+rita-aga@users.noreply.github.com> Date: Tue, 14 Jul 2026 11:50:28 -0700 Subject: [PATCH 3/4] test(server): tighten ARN-203 overdue timing and bump_if_zero CAS unit tests Address independent review Important findings: assert Open after boot, cap overdue wait below full budget, unit-lock bump_if_zero idempotency. --- .../src/state/dispatch/state_timeouts.rs | 21 ++++++++++++++++ .../tests/state_timeout_restart.rs | 25 ++++++++++++++++--- 2 files changed, 43 insertions(+), 3 deletions(-) diff --git a/crates/temper-server/src/state/dispatch/state_timeouts.rs b/crates/temper-server/src/state/dispatch/state_timeouts.rs index 3d5867cc0..c7ab2764e 100644 --- a/crates/temper-server/src/state/dispatch/state_timeouts.rs +++ b/crates/temper-server/src/state/dispatch/state_timeouts.rs @@ -700,6 +700,27 @@ mod tests { assert_eq!(t.size(), 0); } + #[test] + fn bump_if_zero_cas_is_idempotent() { + let t = StateTimeoutTracker::new(); + let k = key(); + assert_eq!(t.bump_if_zero(&k), Some(1), "first untracked arm wins"); + assert_eq!(t.bump_if_zero(&k), None, "second untracked arm loses"); + assert_eq!(t.current(&k), 1); + } + + #[test] + fn bump_if_zero_loses_after_dispatch_bump() { + let t = StateTimeoutTracker::new(); + let k = key(); + assert_eq!(t.bump(&k), 1); + assert_eq!( + t.bump_if_zero(&k), + None, + "boot/create CAS must not re-arm over a live dispatch arm" + ); + } + // --- compute_state_clock_reset_ts (ADR-0056 hydration-re-arm helper) --- fn test_event(action: &str, from: &str, to: &str, ts_ms_after_epoch: i64) -> EntityEvent { diff --git a/crates/temper-server/tests/state_timeout_restart.rs b/crates/temper-server/tests/state_timeout_restart.rs index 954b2221c..8ef89e6cb 100644 --- a/crates/temper-server/tests/state_timeout_restart.rs +++ b/crates/temper-server/tests/state_timeout_restart.rs @@ -142,6 +142,13 @@ async fn pending_state_timeout_fires_after_restart() { let state_b = build_state("arn203-gen-b", open_store(&db_url).await); state_b.populate_index_from_store(&tenant).await; + // Gen-B must own the transition: still Open right after boot resume arm. + let boot = state_b + .get_tenant_entity_state(&tenant, "Ticket", "t-restart-1") + .await + .expect("load after boot"); + assert_eq!(boot.state.status, "Open", "restart must leave entity Open before resume fires"); + // No dispatch to the entity. The pending timeout alone must fire. let status = wait_for_status( &state_b, @@ -174,22 +181,34 @@ async fn overdue_state_timeout_fires_after_restart() { tokio::time::sleep(Duration::from_millis(1500)).await; let state_b = build_state("arn203-gen-b2", open_store(&db_url).await); + let boot_start = std::time::Instant::now(); state_b.populate_index_from_store(&tenant).await; - // Overdue at boot: the fire should happen promptly, well within one - // fresh budget (which would indicate the clock was wrongly reset). + let boot = state_b + .get_tenant_entity_state(&tenant, "Ticket", "t-overdue-1") + .await + .expect("load after boot"); + assert_eq!(boot.state.status, "Open", "overdue entity still Open right after boot"); + + // Overdue: must fire promptly (≪ full 1s budget). Cap wait at 800ms so a + // full-budget re-arm regression fails instead of silently passing. let status = wait_for_status( &state_b, &tenant, "t-overdue-1", "InProgress", - Duration::from_secs(20), + Duration::from_millis(800), ) .await; + let elapsed = boot_start.elapsed(); assert_eq!( status, "InProgress", "an overdue state timeout must fire at boot, not wait another full budget or never fire" ); + assert!( + elapsed < Duration::from_millis(800), + "overdue fire took {elapsed:?}; expected remaining budget ~0 (≪ 1s)" + ); } /// Sibling of the restart cases (same defect class, found during review of From 7d88ed23c67cb3c64dc6c515e37758e10e21e25c Mon Sep 17 00:00:00 2001 From: Rita Agafonova <36133358+rita-aga@users.noreply.github.com> Date: Tue, 14 Jul 2026 17:58:46 -0700 Subject: [PATCH 4/4] style(server): rustfmt ARN-203 tests; mark hard-kill spawn determinism-ok --- crates/temper-server/tests/state_timeout_restart.rs | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/crates/temper-server/tests/state_timeout_restart.rs b/crates/temper-server/tests/state_timeout_restart.rs index 8ef89e6cb..3c6a99bea 100644 --- a/crates/temper-server/tests/state_timeout_restart.rs +++ b/crates/temper-server/tests/state_timeout_restart.rs @@ -100,6 +100,7 @@ fn run_and_hard_kill_generation_a(system_name: &str, db_url: &str, entity_id: &s let system_name = system_name.to_string(); let db_url = db_url.to_string(); let entity_id = entity_id.to_string(); + // determinism-ok: test-only hard-kill of generation-A runtime; not production sim path std::thread::spawn(move || { let rt = tokio::runtime::Builder::new_multi_thread() .worker_threads(2) @@ -147,7 +148,10 @@ async fn pending_state_timeout_fires_after_restart() { .get_tenant_entity_state(&tenant, "Ticket", "t-restart-1") .await .expect("load after boot"); - assert_eq!(boot.state.status, "Open", "restart must leave entity Open before resume fires"); + assert_eq!( + boot.state.status, "Open", + "restart must leave entity Open before resume fires" + ); // No dispatch to the entity. The pending timeout alone must fire. let status = wait_for_status( @@ -188,7 +192,10 @@ async fn overdue_state_timeout_fires_after_restart() { .get_tenant_entity_state(&tenant, "Ticket", "t-overdue-1") .await .expect("load after boot"); - assert_eq!(boot.state.status, "Open", "overdue entity still Open right after boot"); + assert_eq!( + boot.state.status, "Open", + "overdue entity still Open right after boot" + ); // Overdue: must fire promptly (≪ full 1s budget). Cap wait at 800ms so a // full-budget re-arm regression fails instead of silently passing.