diff --git a/crates/engram-coordinator/src/api/host_http.rs b/crates/engram-coordinator/src/api/host_http.rs index 2a88be957..70004225e 100644 --- a/crates/engram-coordinator/src/api/host_http.rs +++ b/crates/engram-coordinator/src/api/host_http.rs @@ -863,6 +863,34 @@ pub async fn heartbeat( if recoverable && adv.kind == engram_protocol::heartbeat::CheckpointKind::EvictionFinal { + // The settle's lifecycle facts ride ITS transaction: + // the settle is CAS-once (a re-advert after success is + // a clean no-op), so facts emitted here afterwards had + // a crash window in which they were lost forever — + // the same event-loss class as the fenced + // post-transition emits, crash-shaped. See the trait + // doc on `settle_evicted_session_idle`. + let now = state.services.clock.now_utc(); + let events = vec![ + SessionEvent::Evicted { at: now }, + SessionEvent::StatusChanged { + from: SessionState::Evicting, + to: SessionState::Idle, + at: now, + }, + ]; + let wire = match crate::state::wire_events(&events) { + Ok(wire) => wire, + Err(e) => { + tracing::warn!( + session_id = %adv.session_id, + error = %e, + "eviction settle: event serialize failed; \ + the host's re-advert retries it", + ); + continue; + } + }; match state .services .meta @@ -870,11 +898,11 @@ pub async fn heartbeat( adv.session_id, adv.sandbox_id, adv.snapshot_id, + &wire, ) .await { - Ok(true) => { - let now = state.services.clock.now_utc(); + Ok(Some(indices)) => { // Best-effort rung clear (the park stamp is // ascent/ledger metadata; lifecycle already // settled above). @@ -883,19 +911,16 @@ pub async fn heartbeat( .meta .set_session_park_rung(adv.session_id, 0, None) .await; - let _ = state - .emit(adv.session_id, SessionEvent::Evicted { at: now }) - .await; - let _ = state - .emit( + for (idx, event) in indices.into_iter().zip(events) { + state.events.publish( adv.session_id, - SessionEvent::StatusChanged { - from: SessionState::Evicting, - to: SessionState::Idle, - at: now, + crate::state::IndexedEvent { + idx, + event, + ephemeral: false, }, - ) - .await; + ); + } tracing::info!( session_id = %adv.session_id, snapshot_id = %adv.snapshot_id, @@ -903,7 +928,7 @@ pub async fn heartbeat( (ADR 0101 C durability floor)", ); } - Ok(false) => {} + Ok(None) => {} Err(e) => { tracing::warn!( session_id = %adv.session_id, diff --git a/crates/engram-coordinator/src/idle_evictor.rs b/crates/engram-coordinator/src/idle_evictor.rs index 226953c29..8b234d9be 100644 --- a/crates/engram-coordinator/src/idle_evictor.rs +++ b/crates/engram-coordinator/src/idle_evictor.rs @@ -1007,39 +1007,46 @@ pub(crate) async fn run_evict_pipeline( return Ok(EvictOutcome::Fenced); } - // Step 3 (PG, Idle-before-destroy): clear sandbox_id on the - // session row — ADR 0047, this is the authoritative unbind (no - // in-memory registry to drop). Fenced; `host_id` is re-written - // unchanged (the fenced write sets both columns) to preserve the - // resume path's origin-affinity hint. - match state - .services - .meta - .fenced_assign_sandbox(session_id, ctx.epoch, None, session.host_id) - .await - { - Ok(true) => {} - Ok(false) => { - crate::metrics::note_fenced_write(); - return Ok(EvictOutcome::Fenced); - } - Err(e) => { - tracing::warn!( - session_id = %session_id, - error = %e, - "idle eviction: fenced_assign_sandbox(None) failed", - ); - } - } - // Step 3c (PG, Idle-before-destroy): flip to the target. Once this + // Step 3 + 3c fused (PG, Idle-before-destroy): the authoritative + // unbind (ADR 0047 — no in-memory registry to drop; `host_id` is + // untouched so the resume path's origin-affinity hint survives) + // rides the SAME transaction as the flip, via `detach_sandbox`. A + // separate preceding detach left a partial-failure window: the flip + // rolls back, the detach has already committed, and the scanner's + // retry finds an `evicting` row with no bound sandbox — the + // HostLost fallback — instead of finishing to Idle (#844 review + // finding). Once this // commits, the reconciler will no-op on every subsequent // heartbeat for this session because the reconcile pass keys // on Active status only. - let prev = match crate::session_ops::transition_with_fence( + // + // The transition's own facts (`snapshot_taken`, `evicted`, the final + // `status_changed`) ride the SAME store transaction: the flip makes + // the session immediately claimable, so post-commit `emit_fenced` + // calls raced a successor's claim and the facts were silently fenced + // out of the record (the e2e_resume event-loss flake). `from` is + // `entry_status` by construction — our op claim excludes every other + // lifecycle writer, so under an unmoved epoch the state is exactly + // what we observed at entry (the debug assert pins this). + let prev = match crate::session_ops::transition_with_fence_emitting( state, session_id, ctx.fence(), target_state, + /*detach_sandbox=*/ true, + vec![ + crate::state::SessionEvent::SnapshotTaken { + snapshot_id: metadata.id, + size_bytes: metadata.size_bytes, + at: now, + }, + crate::state::SessionEvent::Evicted { at: now }, + crate::state::SessionEvent::StatusChanged { + from: entry_status, + to: target_state, + at: now, + }, + ], ) .await { @@ -1055,6 +1062,10 @@ pub(crate) async fn run_evict_pipeline( return Err(EvictError::Meta(e.to_string())); } }; + debug_assert_eq!( + prev, entry_status, + "state moved under our op claim without an epoch bump" + ); // The parking ladder leaves no trace in the terminal state — a // descent (or a plain eviction of a never-parked session, where this // is a no-op) clears the rung with the Idle flip. @@ -1087,47 +1098,10 @@ pub(crate) async fn run_evict_pipeline( // ADR 0006: host-agent unregisters its local proxy entry as // part of `destroy`. No coordinator-side cleanup needed. - // Review finding #6: fenced emits. Ok(None) = a successor re-claimed - // between our committed transition and here — stop emitting silently - // (never compensate/abort from a fenced predecessor); the transition - // already committed under our epoch, so the eviction is done. - if let Err(e) = state - .emit_fenced( - session_id, - ctx.fence(), - SessionEvent::SnapshotTaken { - snapshot_id: metadata.id, - size_bytes: metadata.size_bytes, - at: now, - }, - ) - .await - { - abort_inflight_snapshot(ctx, session_id, sandbox_id, "emit SnapshotTaken").await; - return Err(EvictError::Emit(e.to_string())); - } - if let Err(e) = state - .emit_fenced(session_id, ctx.fence(), SessionEvent::Evicted { at: now }) - .await - { - abort_inflight_snapshot(ctx, session_id, sandbox_id, "emit Evicted").await; - return Err(EvictError::Emit(e.to_string())); - } - if let Err(e) = state - .emit_fenced( - session_id, - ctx.fence(), - SessionEvent::StatusChanged { - from: prev, - to: target_state, - at: now, - }, - ) - .await - { - abort_inflight_snapshot(ctx, session_id, sandbox_id, "emit StatusChanged").await; - return Err(EvictError::Emit(e.to_string())); - } + // The lifecycle facts (`snapshot_taken`, `evicted`, `status_changed`) + // landed atomically with the Step 3c transition above — the old + // post-commit emit_fenced block (and its abort-after-commit error + // arms) is gone with the race it carried. // ADR 0016 A.1.1: success log. Pairs with the entry log so a // pipeline that flushes (host log) without committing (no PG @@ -1272,7 +1246,6 @@ pub enum EvictError { Io(String), Sandbox(engram_core::SandboxError), Meta(String), - Emit(String), } impl std::fmt::Display for EvictError { @@ -1281,7 +1254,6 @@ impl std::fmt::Display for EvictError { Self::Io(m) => write!(f, "idle evict io: {m}"), Self::Sandbox(e) => write!(f, "idle evict sandbox: {e}"), Self::Meta(m) => write!(f, "idle evict meta: {m}"), - Self::Emit(m) => write!(f, "idle evict event emit: {m}"), } } } @@ -2220,14 +2192,18 @@ mod tests { events_cursor: Some(0), fc_snapshot_version: None, }); - assert!( - state - .services - .meta - .settle_evicted_session_idle(session_id, sandbox_id, snapshot_id) - .await - .unwrap(), - "a recoverable row + matching binding must settle the session Idle", + let settle_events = vec![("evicted".to_string(), serde_json::json!({"at": "test"}))]; + let indices = state + .services + .meta + .settle_evicted_session_idle(session_id, sandbox_id, snapshot_id, &settle_events) + .await + .unwrap() + .expect("a recoverable row + matching binding must settle the session Idle"); + assert_eq!( + indices.len(), + 1, + "the settle lands its lifecycle facts atomically", ); assert_eq!(meta.session.lock().status, SessionState::Idle); assert_eq!( diff --git a/crates/engram-coordinator/src/session_ops.rs b/crates/engram-coordinator/src/session_ops.rs index 7616be1d7..3f32c8fc7 100644 --- a/crates/engram-coordinator/src/session_ops.rs +++ b/crates/engram-coordinator/src/session_ops.rs @@ -221,6 +221,85 @@ pub(crate) async fn transition_with_fence( } } +/// [`transition_with_fence`] that lands `events` ATOMICALLY with the +/// transition (one store transaction). For transitions that make the +/// session immediately claimable (eviction's Idle flip: "the instant the +/// session is Idle it is resumable"), post-commit `emit_fenced` calls +/// race the successor's claim and the transition's own facts +/// (`evicted`, the final `status_changed`) get fenced out of the record +/// — the e2e_resume event-loss flake. Appending them inside the +/// transition makes a successor order strictly after. +/// +/// The events must not be outbox-acking kinds (`outbox_ack_id` = None +/// for lifecycle events); their only post-commit side effect is the +/// in-process publish to live streams, done here with the committed +/// indices. +pub(crate) async fn transition_with_fence_emitting( + state: &SharedState, + session_id: SessionId, + fence: SessionFence, + to: SessionState, + detach_sandbox: bool, + events: Vec, +) -> Result { + debug_assert!( + events + .iter() + .all(|e| crate::state::outbox_ack_id(session_id, e).is_none()), + "transition_with_fence_emitting only handles the publish side effect; \ + outbox-acking events must go through emit_fenced" + ); + if fence.epoch == 0 { + // Unfenced interim path: plain transition, then plain appends — + // no fence exists to race, so the atomicity doesn't apply. (No + // unfenced caller detaches.) + debug_assert!(!detach_sandbox, "detach requires the fenced path"); + let prev = state + .services + .meta + .transition_session(session_id, to) + .await?; + for event in events { + let _ = state.emit(session_id, event).await; + } + return Ok(prev); + } + let wire = crate::state::wire_events(&events)?; + match state + .services + .meta + .fenced_transition_session_with_events( + session_id, + fence.epoch as i64, + to, + detach_sandbox, + &wire, + ) + .await? + { + Some((prev, indices)) => { + for (idx, event) in indices.into_iter().zip(events) { + state.events.publish( + session_id, + crate::state::IndexedEvent { + idx, + event, + ephemeral: false, + }, + ); + } + Ok(prev) + } + None => { + crate::metrics::note_fenced_write(); + Err(engram_core::MetaError::Conflict(format!( + "fenced: session {session_id} was re-claimed by a successor op (epoch moved past {})", + fence.epoch, + ))) + } + } +} + /// ADR 0079 interim: an inline op-log claim for lifecycle pipelines that /// have NOT yet migrated into verb bodies (the manual snapshot, the evac /// resume, the live teleport). Rides the same `session_ops` row + the diff --git a/crates/engram-coordinator/src/state.rs b/crates/engram-coordinator/src/state.rs index 147b78c5d..ff299c697 100644 --- a/crates/engram-coordinator/src/state.rs +++ b/crates/engram-coordinator/src/state.rs @@ -1095,6 +1095,22 @@ impl AppState { } } +/// Serialize lifecycle events to the `(kind, payload)` wire pairs the +/// atomic store methods (`fenced_transition_session_with_events`, +/// `settle_evicted_session_idle`) append in-transaction. +pub(crate) fn wire_events( + events: &[SessionEvent], +) -> Result, engram_core::MetaError> { + events + .iter() + .map(|e| { + serde_json::to_value(e) + .map(|payload| (e.kind().to_string(), payload)) + .map_err(|e| engram_core::MetaError::Serialization(format!("event serialize: {e}"))) + }) + .collect() +} + /// ADR 0073: which outbox row (if any) does this event confirm? /// - `run_started{prompt_id}` / `prompt_queued{prompt_id}` — the /// harness took ownership of the prompt (running it or holding it in @@ -1102,7 +1118,7 @@ impl AppState { /// harness itself does, and an edit/dequeue of a queued prompt keeps /// its own confirmations). /// - `tool_call_completed{tool_call_id}` — a generic tool result landed. -fn outbox_ack_id(session_id: SessionId, event: &SessionEvent) -> Option { +pub(crate) fn outbox_ack_id(session_id: SessionId, event: &SessionEvent) -> Option { match event { SessionEvent::HarnessRunStarted { prompt_id: Some(pid), @@ -2394,24 +2410,34 @@ pub(crate) mod tests { session_id: engram_core::SessionId, sandbox_id: engram_core::SandboxId, snapshot_id: engram_core::types::SnapshotId, - ) -> Result { + events: &[(String, serde_json::Value)], + ) -> Result>, MetaError> { let row_ok = self.snapshots.lock().iter().any(|s| { s.id == snapshot_id && s.session_id == Some(session_id) && s.recoverable }); if !row_ok { - return Ok(false); + return Ok(None); } - let mut session = self.session.lock(); - if session.id != session_id - || session.status != SessionState::Evicting - || session.sandbox_id != Some(sandbox_id) { - return Ok(false); + let mut session = self.session.lock(); + if session.id != session_id + || session.status != SessionState::Evicting + || session.sandbox_id != Some(sandbox_id) + { + return Ok(None); + } + session.status = SessionState::Idle; + session.sandbox_id = None; } - session.status = SessionState::Idle; - session.sandbox_id = None; - Ok(true) + let mut indices = Vec::with_capacity(events.len()); + for (kind, payload) in events { + indices.push( + self.append_session_event(session_id, kind, payload.clone()) + .await?, + ); + } + Ok(Some(indices)) } async fn get_snapshot( &self, @@ -2816,6 +2842,36 @@ pub(crate) mod tests { self.transition_session(session_id, to).await.map(Some) } + async fn fenced_transition_session_with_events( + &self, + session_id: SessionId, + epoch: i64, + to: engram_core::types::SessionState, + detach_sandbox: bool, + events: &[(String, serde_json::Value)], + ) -> Result)>, MetaError> { + // In-memory "transaction": the fence gates once, then the flip + // and the appends run back to back under the test's + // single-threaded driver — good enough for the unit tests' + // ordering assertions (the real atomicity is conformance-tested + // against SimMetadataStore + PostgresStore, ADR 0098 D4). + if self.ops.current_epoch(session_id) != epoch { + return Ok(None); + } + let prev = self.transition_session(session_id, to).await?; + if detach_sandbox { + self.session.lock().sandbox_id = None; + } + let mut indices = Vec::with_capacity(events.len()); + for (kind, payload) in events { + indices.push( + self.append_session_event(session_id, kind, payload.clone()) + .await?, + ); + } + Ok(Some((prev, indices))) + } + async fn fenced_assign_sandbox( &self, session_id: SessionId, diff --git a/crates/engram-core/src/traits/metadata.rs b/crates/engram-core/src/traits/metadata.rs index c7e583f16..c4cdf25c2 100644 --- a/crates/engram-core/src/traits/metadata.rs +++ b/crates/engram-core/src/traits/metadata.rs @@ -1060,6 +1060,46 @@ pub trait MetadataStore: Send + Sync { )) } + /// [`fenced_transition_session`] plus lifecycle-event appends, in ONE + /// store transaction under the same epoch predicate — all-or-nothing. + /// + /// Why it exists (the e2e_resume event-loss race): a transition that + /// makes the session immediately claimable (eviction's Idle flip — + /// "the instant the session is Idle it is resumable", ADR 0079/0101) + /// followed by separate post-commit `append_session_event_fenced` + /// calls leaves a window where a successor claims the session between + /// the commit and the appends, the epoch moves, and the transition's + /// own facts (`evicted`, `status_changed`) are silently fenced out of + /// the record. Appending them in the transition's transaction makes a + /// successor order strictly after — the facts always land iff the + /// transition lands. + /// + /// `events` are `(kind, payload)` pairs appended in order with + /// consecutive indices. Returns `(previous_state, event_indices)`; + /// `Ok(None)` = fenced (nothing committed). An illegal transition is + /// `Err(Conflict)` (nothing committed). Callers own any post-commit + /// side effects (in-process publish); `pg_notify` fires on commit. + /// + /// `detach_sandbox = true` also clears `sandbox_id` in the SAME + /// update (`host_id` untouched — resume affinity survives, as with + /// the settle). The eviction flip detaches; doing it in a separate + /// preceding write left a partial-failure window where the flip's + /// rollback stranded an `evicting` session with no bound sandbox — + /// which the scanner's retry resolves as HostLost instead of Idle. + async fn fenced_transition_session_with_events( + &self, + session_id: SessionId, + epoch: i64, + to: crate::types::SessionState, + detach_sandbox: bool, + events: &[(String, serde_json::Value)], + ) -> Result)>, MetaError> { + let _ = (session_id, epoch, to, detach_sandbox, events); + Err(MetaError::Serialization( + "fenced writes not supported by this store".into(), + )) + } + /// Fenced sandbox (re)bind — subsumes `rebind_session_guarded`'s /// bespoke expected-state list with the one epoch predicate. async fn fenced_assign_sandbox( @@ -3101,7 +3141,7 @@ pub trait MetadataStore: Send + Sync { /// ADR 0101 C: the durability-floor settle — atomically flip an /// `evicting` session to `idle` and detach its sandbox, GUARDED on - /// the recoverable snapshot row already existing. Returns `false` + /// the recoverable snapshot row already existing. Returns `None` /// (no-op) when the session is no longer `evicting`, its bound /// sandbox is not `sandbox_id` (a successor rebound), or the /// snapshot row is absent / not recoverable — the caller (the @@ -3112,6 +3152,18 @@ pub trait MetadataStore: Send + Sync { /// Idle-on-capture flip is retired. `host_id` is preserved for /// resume affinity, mirroring the old detach. /// + /// `events` are `(kind, payload)` lifecycle facts (`evicted`, the + /// final `status_changed`) appended ATOMICALLY with the settle, in + /// order, with consecutive indices (returned on success). The settle + /// is CAS-once — a re-advert after a successful settle is a clean + /// no-op — so facts emitted by the caller AFTER the settle had a + /// crash window in which they were lost forever (the settle already + /// landed; no retry would ever re-emit them). Same event-loss class + /// as the fenced post-transition emits + /// ([`Self::fenced_transition_session_with_events`]), crash-shaped + /// instead of fence-shaped. Landing them in the settle's transaction + /// closes it: the facts exist iff the settle happened. + /// /// No silent default (ADR 0098 D4): a store that can be reached by /// the reconcile must implement the real semantics. async fn settle_evicted_session_idle( @@ -3119,7 +3171,8 @@ pub trait MetadataStore: Send + Sync { _session_id: SessionId, _sandbox_id: SandboxId, _snapshot_id: SnapshotId, - ) -> Result { + _events: &[(String, serde_json::Value)], + ) -> Result>, MetaError> { unimplemented!( "settle_evicted_session_idle: PG-semantic method; mocks on the reconcile path must \ implement it (ADR 0098 D4)" diff --git a/crates/engram-dst-cosim/src/scheduler.rs b/crates/engram-dst-cosim/src/scheduler.rs index 8fd2de8de..5b02a1982 100644 --- a/crates/engram-dst-cosim/src/scheduler.rs +++ b/crates/engram-dst-cosim/src/scheduler.rs @@ -543,12 +543,25 @@ impl Cosim { // flips now (the REAL store semantics; prod's heartbeat // HTTP glue is e2e territory). Before the floor flip the // D5 verb lied the session Idle at capture time. + // The settle's lifecycle facts ride its transaction (the + // crash-window fix) — the cosim drives the real store + // semantics, so it passes them the same way the + // reconcile does. + let settle_events = [ + ("evicted".to_string(), serde_json::json!({})), + ( + "status_changed".to_string(), + serde_json::json!({"to": "idle"}), + ), + ]; let settled = self .world .meta - .settle_evicted_session_idle(session_id, sandbox, snapshot_id) + .settle_evicted_session_idle(session_id, sandbox, snapshot_id, &settle_events) .await - .unwrap_or(false); + .ok() + .flatten() + .is_some(); self.log(format!( "finalize completed {session_id} snapshot={snapshot_id} cursor={cursor} \ settled_idle={settled}" diff --git a/crates/engram-postgres/src/lib.rs b/crates/engram-postgres/src/lib.rs index f1655317d..242ea96f0 100644 --- a/crates/engram-postgres/src/lib.rs +++ b/crates/engram-postgres/src/lib.rs @@ -2441,7 +2441,16 @@ impl MetadataStore for PostgresStore { session_id: SessionId, sandbox_id: SandboxId, snapshot_id: SnapshotId, - ) -> Result { + events: &[(String, serde_json::Value)], + ) -> Result>, MetaError> { + // One transaction: the guarded settle UPDATE (the CAS — it also + // row-locks the session for the appends below) plus the event + // appends. The settle is CAS-once, so events appended by the + // caller afterwards had a crash window of permanent loss; in-tx + // they exist iff the settle happened. `pg_notify` fires on + // commit only. + let mut tx = self.pool.begin().await.map_err(db_err)?; + let now = self.clock.now_utc(); let res = sqlx::query( r#" UPDATE sessions @@ -2458,11 +2467,53 @@ impl MetadataStore for PostgresStore { .bind(session_id.as_uuid()) .bind(sandbox_id.as_uuid()) .bind(snapshot_id.as_uuid()) - .bind(self.clock.now_utc()) - .execute(&self.pool) + .bind(now) + .execute(&mut *tx) .await .map_err(db_err)?; - Ok(res.rows_affected() == 1) + if res.rows_affected() != 1 { + return Ok(None); + } + let mut indices = Vec::with_capacity(events.len()); + for (kind, payload) in events { + let row = sqlx::query( + r#" + WITH next AS ( + UPDATE sessions + SET next_event_idx = next_event_idx + 1, + updated_at = $4, + last_event_at = $4 + WHERE id = $1 + RETURNING next_event_idx - 1 AS allocated_idx, recovery_epoch + ), + inserted AS ( + INSERT INTO session_events (session_id, idx, kind, payload, recovery_epoch, created_at) + SELECT $1, allocated_idx, $2, $3, recovery_epoch, $4 FROM next + RETURNING idx + ) + SELECT i.idx, + pg_notify( + 'session_events', + json_build_object('session_id', $1::text, 'idx', i.idx)::text + ) + FROM inserted i + "#, + ) + .bind(session_id.as_uuid()) + .bind(kind) + .bind(payload) + .bind(now) + .fetch_one(&mut *tx) + .await + .map_err(db_err)?; + indices.push(sqlx::Row::try_get(&row, "idx").map_err(db_err)?); + } + tx.commit().await.map_err(db_err)?; + // The retired D5 Idle flip fired the queue-scanner wake on + // `evicting → idle` (a memory-reserving state freeing its + // budget); the settle inherits it. + self.notify_placement_changed("session_freed").await; + Ok(Some(indices)) } /// ADR 0034: atomic `+= 1 RETURNING`. The eviction scanner calls @@ -6990,6 +7041,124 @@ impl MetadataStore for PostgresStore { Ok(Some(current)) } + async fn fenced_transition_session_with_events( + &self, + session_id: SessionId, + epoch: i64, + to: SessionState, + detach_sandbox: bool, + events: &[(String, serde_json::Value)], + ) -> Result)>, MetaError> { + // `fenced_transition_session` with the event appends folded into + // the SAME transaction (see the trait doc for the event-loss race + // this closes). The FOR UPDATE row lock spans the fence check, the + // status flip, and every idx allocation, so the appended events + // are contiguous and no successor can interleave; the in-SQL + // pg_notify calls fire only on COMMIT, so subscribers never hear + // about a rolled-back transition. + let mut tx = self.pool.begin().await.map_err(db_err)?; + let row = + sqlx::query("SELECT status, current_epoch FROM sessions WHERE id = $1 FOR UPDATE") + .bind(session_id.as_uuid()) + .fetch_optional(&mut *tx) + .await + .map_err(db_err)? + .ok_or(MetaError::NotFound)?; + // Fence check BEFORE legality — same rationale as + // `fenced_transition_session` (ADR 0079 review finding #3). + let stored_epoch: i64 = row.try_get("current_epoch").map_err(|e| { + MetaError::Serialization(format!( + "fenced_transition_session_with_events: read epoch: {e}" + )) + })?; + if stored_epoch != epoch { + return Ok(None); + } + let current_raw: String = row.try_get("status").map_err(|e| { + MetaError::Serialization(format!( + "fenced_transition_session_with_events: read current: {e}" + )) + })?; + let current = row::parse_session_state_for_lib(¤t_raw)?; + current.try_transition_to(to).map_err(|e| { + tracing::warn!( + session_id = %session_id, + from = %current.as_str(), + to = %to.as_str(), + "rejected illegal session state transition (fenced-with-events path)" + ); + MetaError::Conflict(e.to_string()) + })?; + let now = self.clock.now_utc(); + let n = sqlx::query( + r#" + UPDATE sessions + SET status = $2, + last_active_at = $4, + updated_at = $4, + sandbox_id = CASE WHEN $5 THEN NULL ELSE sandbox_id END, + evac_attempts = CASE WHEN $2 = 'evacuating' THEN 0 ELSE evac_attempts END, + evict_attempts = CASE WHEN $2 = 'evicting' THEN 0 ELSE evict_attempts END, + queued_at = CASE WHEN $2 = 'queued' THEN $4 ELSE queued_at END, + queue_origin = CASE WHEN $2 = 'queued' + THEN COALESCE(queue_origin, 'create') + ELSE queue_origin END + WHERE id = $1 AND current_epoch = $3 + "#, + ) + .bind(session_id.as_uuid()) + .bind(to.as_str()) + .bind(epoch) + .bind(now) + .bind(detach_sandbox) + .execute(&mut *tx) + .await + .map_err(db_err)? + .rows_affected(); + if n == 0 { + return Ok(None); + } + let mut indices = Vec::with_capacity(events.len()); + for (kind, payload) in events { + let row = sqlx::query( + r#" + WITH next AS ( + UPDATE sessions + SET next_event_idx = next_event_idx + 1, + updated_at = $4, + last_event_at = $4 + WHERE id = $1 + RETURNING next_event_idx - 1 AS allocated_idx, recovery_epoch + ), + inserted AS ( + INSERT INTO session_events (session_id, idx, kind, payload, recovery_epoch, created_at) + SELECT $1, allocated_idx, $2, $3, recovery_epoch, $4 FROM next + RETURNING idx + ) + SELECT i.idx, + pg_notify( + 'session_events', + json_build_object('session_id', $1::text, 'idx', i.idx)::text + ) + FROM inserted i + "#, + ) + .bind(session_id.as_uuid()) + .bind(kind) + .bind(payload) + .bind(now) + .fetch_one(&mut *tx) + .await + .map_err(db_err)?; + indices.push(sqlx::Row::try_get(&row, "idx").map_err(db_err)?); + } + tx.commit().await.map_err(db_err)?; + if current.reserves_host_memory() && !to.reserves_host_memory() { + self.notify_placement_changed("session_freed").await; + } + Ok(Some((current, indices))) + } + async fn fenced_assign_sandbox( &self, session_id: SessionId, diff --git a/crates/engram-sim/src/meta/store_impl.rs b/crates/engram-sim/src/meta/store_impl.rs index f64ec2b24..ba8cc1541 100644 --- a/crates/engram-sim/src/meta/store_impl.rs +++ b/crates/engram-sim/src/meta/store_impl.rs @@ -1426,6 +1426,95 @@ impl MetadataStore for SimMetadataStore { Ok(Some(current)) } + /// `fenced_transition_session` + event appends under ONE db lock — + /// the sim's transaction. Mirrors PostgresStore: fence check before + /// legality; on success the events land with consecutive indices and + /// per-event `session_events` notifications; on a stale epoch or an + /// illegal transition NOTHING lands (no state change, no events, no + /// notifications — pg_notify only fires on commit). + async fn fenced_transition_session_with_events( + &self, + session_id: SessionId, + epoch: i64, + to: SessionState, + detach_sandbox: bool, + events: &[(String, serde_json::Value)], + ) -> Result)>, MetaError> { + self.gate()?; + let now = self.now(); + let mut db = self.db.lock(); + let row = db + .sessions + .get_mut(&session_id) + .ok_or(MetaError::NotFound)?; + if row.current_epoch != epoch { + return Ok(None); + } + let current = row.session.status; + current + .try_transition_to(to) + .map_err(|e| MetaError::Conflict(e.to_string()))?; + row.session.status = to; + row.session.last_active_at = now; + row.updated_at = now; + if detach_sandbox { + row.session.sandbox_id = None; + } + if to == SessionState::Evacuating { + row.evac_attempts = 0; + } + if to == SessionState::Evicting { + row.evict_attempts = 0; + } + if to == SessionState::Queued { + row.queued_at = Some(now); + row.queue_origin = Some(row.queue_origin.unwrap_or(QueueOrigin::Create)); + } + let recovery_epoch = row.recovery_epoch; + let mut indices = Vec::with_capacity(events.len()); + for (kind, payload) in events { + // Per-iteration re-borrow: the session row and the event log + // are sibling fields of the same locked db, so the row borrow + // can't span the event push. + let row = db + .sessions + .get_mut(&session_id) + .expect("session row present under the same lock"); + let idx = row.next_event_idx; + row.next_event_idx += 1; + row.last_event_at = Some(now); + indices.push(idx); + db.session_events + .entry(session_id) + .or_default() + .push(PersistedEvent { + idx, + kind: kind.clone(), + payload: payload.clone(), + created_at: now, + recovery_epoch, + rewound_at: None, + }); + } + db.transition_log.push(super::TransitionLogEntry { + session: session_id, + from: current, + to, + exempt: false, + }); + drop(db); + for idx in &indices { + self.notify( + "session_events", + format!("{{\"session_id\":\"{session_id}\",\"idx\":{idx}}}"), + ); + } + if reserves(current) && !reserves(to) { + self.notify("placement_changed", "session_freed"); + } + Ok(Some((current, indices))) + } + /// `WHERE id=$1 AND current_epoch=$4`; bool = landed. async fn fenced_assign_sandbox( &self, @@ -2560,8 +2649,10 @@ impl MetadataStore for SimMetadataStore { session_id: SessionId, sandbox_id: engram_core::SandboxId, snapshot_id: engram_core::types::SnapshotId, - ) -> Result { + events: &[(String, serde_json::Value)], + ) -> Result>, MetaError> { self.gate()?; + let now = self.now(); let mut db = self.db.lock(); // (PG stamps `updated_at`, a column the in-memory `Session` // doesn't carry — nothing to mirror here.) @@ -2570,17 +2661,53 @@ impl MetadataStore for SimMetadataStore { .get(&snapshot_id) .is_some_and(|s| s.session_id == Some(session_id) && s.recoverable); if !row_recoverable { - return Ok(false); + return Ok(None); } let Some(r) = db.sessions.get_mut(&session_id) else { - return Ok(false); + return Ok(None); }; if r.session.status != SessionState::Evicting || r.session.sandbox_id != Some(sandbox_id) { - return Ok(false); + return Ok(None); } r.session.status = SessionState::Idle; r.session.sandbox_id = None; - Ok(true) + // The settle's lifecycle facts land under the same db lock (the + // sim's transaction) — the settle is CAS-once, so a caller-side + // append after it had a crash window of permanent loss. + let recovery_epoch = r.recovery_epoch; + let mut indices = Vec::with_capacity(events.len()); + for (kind, payload) in events { + let r = db + .sessions + .get_mut(&session_id) + .expect("session row present under the same lock"); + let idx = r.next_event_idx; + r.next_event_idx += 1; + r.last_event_at = Some(now); + indices.push(idx); + db.session_events + .entry(session_id) + .or_default() + .push(PersistedEvent { + idx, + kind: kind.clone(), + payload: payload.clone(), + created_at: now, + recovery_epoch, + rewound_at: None, + }); + } + drop(db); + for idx in &indices { + self.notify( + "session_events", + format!("{{\"session_id\":\"{session_id}\",\"idx\":{idx}}}"), + ); + } + // Parity with the retired D5 Idle flip: `evicting → idle` frees + // a memory-reserving state's budget — wake the queue scanner. + self.notify("placement_changed", "session_freed"); + Ok(Some(indices)) } async fn bump_evac_attempts(&self, session_id: SessionId) -> Result { diff --git a/crates/engram-sim/tests/meta_conformance.rs b/crates/engram-sim/tests/meta_conformance.rs index b871dec6b..32e752b19 100644 --- a/crates/engram-sim/tests/meta_conformance.rs +++ b/crates/engram-sim/tests/meta_conformance.rs @@ -859,6 +859,100 @@ async fn fenced_transition(ctx: &Ctx) { assert_eq!(prev, Some(SessionState::Pending)); } +/// Atomic fenced transition + lifecycle events (the eviction event-loss +/// fix): a stale epoch or an illegal transition commits NOTHING — no +/// state flip, no events, no index burn; success lands the flip and the +/// events in order with contiguous indices. Both stores must agree. +async fn fenced_transition_with_events(ctx: &Ctx) { + let meta = &ctx.meta; + let sid = meta.create_session(spec("conf:fence-ev")).await.unwrap(); + let EnqueueOutcome::Claimed(op) = meta + .op_enqueue_and_claim(sid, OpKind::Evict, serde_json::json!({}), None, "pod-a") + .await + .unwrap() + else { + panic!("claimed") + }; + let epoch = op.epoch.unwrap(); + + // Anchor the index sequence with a plain append, and bind a sandbox + // so the success arm can prove the atomic detach. + let baseline = meta + .append_session_event(sid, "status_changed", serde_json::json!({"probe": true})) + .await + .unwrap(); + let sb = engram_core::SandboxId::new(); + meta.assign_session_sandbox(sid, Some(sb)).await.unwrap(); + let events = vec![ + ("evicted".to_string(), serde_json::json!({"at": "t0"})), + ( + "status_changed".to_string(), + serde_json::json!({"to": "failed"}), + ), + ]; + + // Stale epoch: silent None, nothing lands. + assert!(meta + .fenced_transition_session_with_events(sid, epoch + 1, SessionState::Failed, true, &events) + .await + .unwrap() + .is_none()); + // Illegal transition (Pending → Active): Conflict, nothing lands. + let err = meta + .fenced_transition_session_with_events(sid, epoch, SessionState::Active, true, &events) + .await + .unwrap_err(); + assert!(matches!(err, MetaError::Conflict(_))); + assert_eq!( + meta.get_session(sid).await.unwrap().status, + SessionState::Pending, + "rejected calls must not flip state", + ); + let leaked = meta + .list_session_events_since(sid, baseline, 100) + .await + .unwrap(); + assert!( + leaked.is_empty(), + "rejected calls must append nothing: {leaked:?}", + ); + assert_eq!( + meta.get_session(sid).await.unwrap().sandbox_id, + Some(sb), + "rejected calls must not detach", + ); + + // Matching epoch: the flip and both events land together, indices + // contiguous with the baseline append. + let (prev, idxs) = meta + .fenced_transition_session_with_events(sid, epoch, SessionState::Failed, true, &events) + .await + .unwrap() + .expect("matching epoch must land"); + assert_eq!(prev, SessionState::Pending); + assert_eq!(idxs, vec![baseline + 1, baseline + 2]); + let settled = meta.get_session(sid).await.unwrap(); + assert_eq!(settled.status, SessionState::Failed); + assert_eq!( + settled.sandbox_id, None, + "detach_sandbox rides the same transaction as the flip", + ); + let landed = meta + .list_session_events_since(sid, baseline, 100) + .await + .unwrap(); + assert_eq!(landed.len(), 2, "exactly the two events: {landed:?}"); + assert_eq!( + (landed[0].idx, landed[0].kind.as_str()), + (baseline + 1, "evicted") + ); + assert_eq!( + (landed[1].idx, landed[1].kind.as_str()), + (baseline + 2, "status_changed"), + ); + assert_eq!(landed[1].payload, serde_json::json!({"to": "failed"})); +} + /// #800: `enqueue_evacuating_session_resume` — the RESERVED evac-placement /// overflow CAS. Fenced `evacuating → queued` (resume-origin): matches only /// on `status='evacuating'` AND the op's epoch; a wrong epoch, a wrong @@ -1592,12 +1686,25 @@ async fn parked_lifecycle_and_eviction_settle(ctx: &Ctx) { assert!(meta.list_parked_sessions().await.unwrap().is_empty()); let snap_id = SnapshotId::new(); + // The atomic settle facts (see the trait doc: caller-side emits + // after a CAS-once settle had a crash window of permanent loss). + let settle_events = vec![ + ("evicted".to_string(), serde_json::json!({"at": "t"})), + ( + "status_changed".to_string(), + serde_json::json!({"to": "idle"}), + ), + ]; + let ev_floor = meta + .append_session_event(sid, "status_changed", serde_json::json!({"probe": true})) + .await + .unwrap(); // 1. No row yet → no settle. assert!( - !meta - .settle_evicted_session_idle(sid, sb, snap_id) + meta.settle_evicted_session_idle(sid, sb, snap_id, &settle_events) .await - .unwrap(), + .unwrap() + .is_none(), "no settle before the snapshot row exists" ); // 2. A NON-recoverable row → no settle. @@ -1606,10 +1713,10 @@ async fn parked_lifecycle_and_eviction_settle(ctx: &Ctx) { .await .unwrap()); assert!( - !meta - .settle_evicted_session_idle(sid, sb, snap_id) + meta.settle_evicted_session_idle(sid, sb, snap_id, &settle_events) .await - .unwrap(), + .unwrap() + .is_none(), "a non-recoverable row must not settle Idle" ); // 3. A recoverable row for a DIFFERENT session → no settle. @@ -1623,10 +1730,10 @@ async fn parked_lifecycle_and_eviction_settle(ctx: &Ctx) { .await .unwrap()); assert!( - !meta - .settle_evicted_session_idle(sid, sb, other_snap) + meta.settle_evicted_session_idle(sid, sb, other_snap, &settle_events) .await - .unwrap(), + .unwrap() + .is_none(), "another session's row must not settle this one" ); // 4. The right row, but the WRONG sandbox (a rebound successor) → no @@ -1639,30 +1746,52 @@ async fn parked_lifecycle_and_eviction_settle(ctx: &Ctx) { .await .unwrap()); assert!( - !meta - .settle_evicted_session_idle(sid, engram_core::SandboxId::new(), good_snap) - .await - .unwrap(), + meta.settle_evicted_session_idle( + sid, + engram_core::SandboxId::new(), + good_snap, + &settle_events + ) + .await + .unwrap() + .is_none(), "a stale advert against a rebound sandbox must not settle" ); // 5. The exact triple → settle: idle + detached (host kept for // resume affinity). - assert!(meta - .settle_evicted_session_idle(sid, sb, good_snap) + let idxs = meta + .settle_evicted_session_idle(sid, sb, good_snap, &settle_events) .await - .unwrap()); + .unwrap() + .expect("the exact triple settles"); + assert_eq!( + idxs, + vec![ev_floor + 1, ev_floor + 2], + "settle facts land atomically with contiguous indices", + ); let s = meta.get_session(sid).await.unwrap(); assert_eq!(s.status, SessionState::Idle); assert_eq!(s.sandbox_id, None, "the settle detaches the sandbox"); assert_eq!(s.host_id, Some(host), "host affinity preserved"); // 6. Idempotent: a re-advert's second settle is a clean no-op. assert!( - !meta - .settle_evicted_session_idle(sid, sb, good_snap) + meta.settle_evicted_session_idle(sid, sb, good_snap, &settle_events) .await - .unwrap(), + .unwrap() + .is_none(), "an already-settled session no-ops" ); + // Every no-op arm above (and the idempotent re-settle) appended + // NOTHING; only the one successful settle's two facts landed. + let landed = meta + .list_session_events_since(sid, ev_floor, 100) + .await + .unwrap(); + assert_eq!( + landed.iter().map(|e| e.kind.as_str()).collect::>(), + vec!["evicted", "status_changed"], + "exactly the settle's facts, exactly once: {landed:?}", + ); // Parked → HostLost is the host-death edge (never Idle: the parked // RAM died with the host). @@ -1723,6 +1852,10 @@ conformance!( super::op_latest_for_kind_reads_terminal_mints ); conformance!(t_fenced_transition, super::fenced_transition); +conformance!( + t_fenced_transition_with_events, + super::fenced_transition_with_events +); conformance!( t_enqueue_evacuating_resume, super::enqueue_evacuating_resume