From 46a04efc91a36298492410c272b3d391666a52f4 Mon Sep 17 00:00:00 2001 From: Nikhil Unni Date: Tue, 6 Oct 2026 10:44:24 -0700 Subject: [PATCH] refactor(metadata): retire assign_session_sandbox (ADR 0123 C4) Rebuilt onto main after the squash merge of #1591; content unchanged. Remove the blind sandbox setter from MetadataStore, both stores, and all mocks. Use the surviving fenced binding writes in test fixtures. Make reconcile compare the exact current binding, including None. Keep assign_session_host for dead-host cleanup. Preserve strike reset and live-manifest cleanup in fenced assignment; extend conformance coverage. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01965DMBwLXzE9baCmj1Wp8Q --- crates/engram-chunk-store/src/gc.rs | 7 - .../src/api/enabled_images.rs | 7 - crates/engram-coordinator/src/api/snapshot.rs | 5 +- crates/engram-coordinator/src/boot_bundle.rs | 7 - .../engram-coordinator/src/host_registry.rs | 7 - crates/engram-coordinator/src/idle_evictor.rs | 60 ++++---- crates/engram-coordinator/src/reconcile.rs | 24 +-- .../engram-coordinator/src/session_verbs.rs | 2 +- crates/engram-coordinator/src/state.rs | 85 ++++++----- .../tests/admin_evac_live_pg.rs | 12 +- crates/engram-coordinator/tests/api.rs | 16 +- .../tests/binding_cas_live_pg.rs | 8 +- .../tests/binding_writer_inventory.rs | 2 - .../tests/chunk_gc_helpers_live_pg.rs | 13 +- .../tests/dead_host_mock.rs | 10 -- .../tests/eviction_live_pg.rs | 4 +- .../tests/outbox_live_pg.rs | 13 +- .../tests/reconcile_integration.rs | 4 +- crates/engram-core/src/traits/metadata.rs | 87 +++-------- crates/engram-core/src/types/session.rs | 2 +- .../src/disk_daemon/flush_scheduler.rs | 2 +- crates/engram-oci-auth/src/lib.rs | 9 +- crates/engram-postgres/src/lib.rs | 138 +++--------------- crates/engram-sim/src/meta/store_impl.rs | 39 ++--- crates/engram-sim/tests/meta_conformance.rs | 114 ++++++++++++++- 25 files changed, 293 insertions(+), 384 deletions(-) diff --git a/crates/engram-chunk-store/src/gc.rs b/crates/engram-chunk-store/src/gc.rs index 50f5f96b2..450cef7ac 100644 --- a/crates/engram-chunk-store/src/gc.rs +++ b/crates/engram-chunk-store/src/gc.rs @@ -472,13 +472,6 @@ mod tests { ) -> Result<(), MetaError> { Ok(()) } - async fn assign_session_sandbox( - &self, - _id: SessionId, - _sandbox_id: Option, - ) -> Result, MetaError> { - Ok(None) - } async fn upsert_host( &self, _host: engram_core::types::HostRecord, diff --git a/crates/engram-coordinator/src/api/enabled_images.rs b/crates/engram-coordinator/src/api/enabled_images.rs index 7a92f42b6..ed16062bd 100644 --- a/crates/engram-coordinator/src/api/enabled_images.rs +++ b/crates/engram-coordinator/src/api/enabled_images.rs @@ -1390,13 +1390,6 @@ mod tests { ) -> Result<(), MetaError> { unreachable!() } - async fn assign_session_sandbox( - &self, - _: engram_core::SessionId, - _: Option, - ) -> Result, MetaError> { - unreachable!() - } async fn upsert_host( &self, _: engram_core::types::host::HostRecord, diff --git a/crates/engram-coordinator/src/api/snapshot.rs b/crates/engram-coordinator/src/api/snapshot.rs index 3ae833ace..85be2de22 100644 --- a/crates/engram-coordinator/src/api/snapshot.rs +++ b/crates/engram-coordinator/src/api/snapshot.rs @@ -1819,8 +1819,7 @@ async fn resume_from_fc_snapshot( // `session.live_disk_manifest` is `None` when: // - The session never went through Phase B (non-NBD host / // never had a publish land). - // - The session is mid-eviction and `assign_session_sandbox(None)` - // cleared the column (commit 3's load-bearing race fix). + // - The session is mid-eviction and the detach cleared the column. // // In both `None` cases the resolver falls back to the // snapshot's manifest, preserving the pre-Phase-B behaviour. @@ -2148,7 +2147,7 @@ async fn bind_resumed_session( /// FlushScheduler's live-manifest publisher uses to attach session_id /// to the publish RPC. The coordinator-side binding is NOT updated /// here — it lives only in `sessions.sandbox_id` (Postgres), written by -/// the caller via `assign_session_sandbox` BEFORE this call, so every +/// the caller through a guarded binding write BEFORE this call, so every /// replica's `/exec` / `/shell` / `/prompt` resolves the new sandbox by /// reading that row ([`AppState::resolve_sandbox`]). /// diff --git a/crates/engram-coordinator/src/boot_bundle.rs b/crates/engram-coordinator/src/boot_bundle.rs index 4bbc65636..e99ff1a39 100644 --- a/crates/engram-coordinator/src/boot_bundle.rs +++ b/crates/engram-coordinator/src/boot_bundle.rs @@ -322,13 +322,6 @@ mod tests { ) -> Result<(), MetaError> { unimplemented!() } - async fn assign_session_sandbox( - &self, - _: engram_core::SessionId, - _: Option, - ) -> Result, MetaError> { - unimplemented!() - } async fn upsert_host(&self, _: HostRecord) -> Result<(), MetaError> { unimplemented!() } diff --git a/crates/engram-coordinator/src/host_registry.rs b/crates/engram-coordinator/src/host_registry.rs index 901b808ce..fff120a9e 100644 --- a/crates/engram-coordinator/src/host_registry.rs +++ b/crates/engram-coordinator/src/host_registry.rs @@ -1006,13 +1006,6 @@ mod tests { ) -> Result<(), engram_core::MetaError> { Ok(()) } - async fn assign_session_sandbox( - &self, - _: engram_core::SessionId, - _: Option, - ) -> Result, engram_core::MetaError> { - Ok(None) - } async fn host_for_sandbox( &self, sandbox_id: SandboxId, diff --git a/crates/engram-coordinator/src/idle_evictor.rs b/crates/engram-coordinator/src/idle_evictor.rs index dc539d227..30a3a87e1 100644 --- a/crates/engram-coordinator/src/idle_evictor.rs +++ b/crates/engram-coordinator/src/idle_evictor.rs @@ -1952,7 +1952,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); assert_eq!( @@ -1965,7 +1965,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, None) + .fenced_assign_sandbox(session_id, 0, None, None) .await .unwrap(); assert_eq!(state.resolve_sandbox(session_id).await, None); @@ -2160,7 +2160,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -2290,7 +2290,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -2551,7 +2551,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -2776,7 +2776,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -2996,7 +2996,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -3061,7 +3061,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -3327,7 +3327,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -3425,7 +3425,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -3469,7 +3469,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -3519,7 +3519,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); *mini.fail_next_record_snapshot.lock() = true; @@ -3674,7 +3674,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -3734,7 +3734,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -3787,7 +3787,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); let host_id = state @@ -3859,7 +3859,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); let host_id = state @@ -3921,7 +3921,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); let host_id = state @@ -3975,7 +3975,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); let host_id = state @@ -4051,7 +4051,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); let host_id = state @@ -4137,7 +4137,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); let host_id = state @@ -4204,7 +4204,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); let host_id = state.host_registry.host_of(sandbox_id).expect("routed"); @@ -4443,7 +4443,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -4697,7 +4697,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); { @@ -4891,7 +4891,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -4949,7 +4949,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -5002,7 +5002,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -5054,7 +5054,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -5127,7 +5127,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -5227,7 +5227,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); @@ -5293,7 +5293,7 @@ mod tests { state .services .meta - .assign_session_sandbox(session_id, Some(sandbox_id)) + .fenced_assign_sandbox(session_id, 0, Some(sandbox_id), None) .await .unwrap(); diff --git a/crates/engram-coordinator/src/reconcile.rs b/crates/engram-coordinator/src/reconcile.rs index d763f6200..1e9ba8d32 100644 --- a/crates/engram-coordinator/src/reconcile.rs +++ b/crates/engram-coordinator/src/reconcile.rs @@ -358,26 +358,10 @@ async fn flip_missing( } } - // Clear sandbox_id so coord routing and a future restart's - // `repopulate_routing` don't try to talk to the dead sandbox. - // - // Issue #211: this MUST be a compare-and-swap on the EXACT sandbox - // that struck out, not a blind `WHERE id = $1` clear. A live - // migration's `rebind_session` can land a FRESH sandbox onto this - // row between our strike-out decision and this clear; a blind null - // would wipe that healthy binding and drive a just-migrated session - // to HostLost→Idle/Dead. With the CAS, if the row no longer points - // at the struck-out sandbox we abort the whole flip (the binding - // moved on — the session is not orphaned). When `sandbox_id` is - // None (the row already had no binding) we fall back to the blind - // clear: there is nothing for a rebind to have replaced. - let clear_result = match sandbox_id { - Some(struck) => { - meta.assign_session_sandbox_guarded(session_id, None, Some(Some(struck)), &[]) - .await - } - None => meta.assign_session_sandbox(session_id, None).await, - }; + // Clear only if the current binding still matches, including an absent binding. + let clear_result = meta + .assign_session_sandbox_guarded(session_id, None, Some(sandbox_id), &[]) + .await; if let Err(e) = clear_result { if matches!(e, engram_core::MetaError::Conflict(_)) { tracing::info!( diff --git a/crates/engram-coordinator/src/session_verbs.rs b/crates/engram-coordinator/src/session_verbs.rs index 2d98de5a8..7392b3e59 100644 --- a/crates/engram-coordinator/src/session_verbs.rs +++ b/crates/engram-coordinator/src/session_verbs.rs @@ -2409,7 +2409,7 @@ mod tests { state .services .meta - .assign_session_sandbox(id, Some(sandbox_id)) + .fenced_assign_sandbox(id, 0, Some(sandbox_id), None) .await .unwrap(); state diff --git a/crates/engram-coordinator/src/state.rs b/crates/engram-coordinator/src/state.rs index b047d02a1..967af3300 100644 --- a/crates/engram-coordinator/src/state.rs +++ b/crates/engram-coordinator/src/state.rs @@ -1067,8 +1067,8 @@ impl AppState { /// /// ADR 0047: the coordinator holds NO in-memory session→sandbox /// authority. `session_boot` persists the binding via - /// `create_session_created`; every rebind/resume path persists it via - /// `assign_session_sandbox`; eviction/teardown clears it to `None`. + /// `transition_session_created`; rebind/resume paths use guarded binding + /// writes; eviction/teardown clears it to `None`. /// Any replica answers `/exec` / `/prompt` / `/shell` / `/snapshot` /// identically by reading this row — one indexed PK select, sub-ms, /// dwarfed by the downstream host exec RPC. A `None` here means the @@ -2287,7 +2287,7 @@ pub(crate) mod tests { /// ADR 0016 Phase C: in-memory mirror of the `chunk_generation` /// counter. Bumped in the same critical section that writes /// `live_disk_manifests` (or clears it via - /// `assign_session_sandbox(None)`) so tests can verify the + /// `fenced_assign_sandbox(None)`) so tests can verify the /// barrier behaviour Phase C will rely on. pub(crate) chunk_generation: PlMutex, /// ADR 0034: in-memory mirror of the eviction retry count. @@ -2645,29 +2645,6 @@ pub(crate) mod tests { Ok(()) } - async fn assign_session_sandbox( - &self, - id: engram_core::SessionId, - sandbox_id: Option, - ) -> Result, MetaError> { - let mut s = self.session.lock(); - if id != s.id { - return Err(MetaError::NotFound); - } - s.sandbox_id = sandbox_id; - // ADR 0016 Phase B: unbind clears the live manifest + - // bumps chunk_generation so Phase C's mid-sweep barrier - // observes the pin-set shrink atomically. Mirrors the - // PG path in engram-postgres::assign_session_sandbox. - if sandbox_id.is_none() && self.live_disk_manifests.lock().remove(&id).is_some() { - *self.chunk_generation.lock() += 1; - } - Ok(sandbox_id.map(|_| { - let mut epoch = self.binding_epoch.lock(); - *epoch += 1; - *epoch - })) - } async fn upsert_host(&self, host: HostRecord) -> Result<(), MetaError> { let mut hosts = self.hosts.lock(); if let Some(existing) = hosts.iter_mut().find(|h| h.id == host.id) { @@ -3418,6 +3395,31 @@ pub(crate) mod tests { Ok(Some((prev, indices))) } + async fn rebind_session_guarded( + &self, + id: SessionId, + host_id: HostId, + sandbox_id: engram_core::SandboxId, + expected_current: Option>, + allowed_states: &[SessionState], + ) -> Result { + let mut session = self.session.lock(); + if session.id != id { + return Err(MetaError::NotFound); + } + if expected_current.is_some_and(|expected| session.sandbox_id != expected) { + return Err(MetaError::Conflict("sandbox binding changed".into())); + } + if !allowed_states.is_empty() && !allowed_states.contains(&session.status) { + return Err(MetaError::Conflict("session state changed".into())); + } + session.host_id = Some(host_id); + session.sandbox_id = Some(sandbox_id); + let mut epoch = self.binding_epoch.lock(); + *epoch += 1; + Ok(*epoch) + } + async fn fenced_assign_sandbox( &self, session_id: SessionId, @@ -3425,12 +3427,25 @@ pub(crate) mod tests { sandbox_id: Option, host_id: Option, ) -> Result, MetaError> { - if self.ops.current_epoch(session_id) != epoch { + let mut s = self.session.lock(); + if session_id != s.id || self.ops.current_epoch(session_id) != epoch { return Err(MetaError::Conflict("stale fence".into())); } - let epoch = self.assign_session_sandbox(session_id, sandbox_id).await?; - self.assign_session_host(session_id, host_id).await?; - Ok(epoch) + s.sandbox_id = sandbox_id; + s.host_id = host_id; + // ADR 0016 Phase B: unbind clears the live manifest + + // bumps chunk_generation so Phase C's mid-sweep barrier + // observes the pin-set shrink atomically. Mirrors the + // PG path in engram-postgres::fenced_assign_sandbox. + if sandbox_id.is_none() { + self.live_disk_manifests.lock().remove(&session_id); + *self.chunk_generation.lock() += 1; + } + Ok(sandbox_id.map(|_| { + let mut epoch = self.binding_epoch.lock(); + *epoch += 1; + *epoch + })) } // ADR 0016 Phase B: in-memory mirror of @@ -3956,7 +3971,7 @@ pub(crate) mod tests { /// commit 6's effective_resume_disk_manifest resolver would /// then prefer it and restore disk-state AHEAD of memory. #[tokio::test] - async fn assign_session_sandbox_none_clears_live_manifest_and_bumps_generation() { + async fn fenced_assign_sandbox_none_clears_live_manifest_and_bumps_generation() { let sandbox_id = engram_core::SandboxId::new(); let (session_id, mini) = build_phase_b_meta(Some(sandbox_id)); let meta: Arc = mini.clone(); @@ -3972,7 +3987,9 @@ pub(crate) mod tests { // Unbind. Live manifest disappears; generation bumps because // the pin set shrunk. - meta.assign_session_sandbox(session_id, None).await.unwrap(); + meta.fenced_assign_sandbox(session_id, 0, None, None) + .await + .unwrap(); assert!(!mini.live_disk_manifests.lock().contains_key(&session_id)); assert_eq!( meta.chunk_generation().await.unwrap(), @@ -3986,7 +4003,7 @@ pub(crate) mod tests { /// as stale if rebinding raced). This is the symmetric case to /// the unbind-clears test. #[tokio::test] - async fn assign_session_sandbox_some_does_not_clear_or_bump() { + async fn fenced_assign_sandbox_some_does_not_clear_or_bump() { let sandbox_id = engram_core::SandboxId::new(); let (session_id, mini) = build_phase_b_meta(Some(sandbox_id)); let meta: Arc = mini.clone(); @@ -3999,7 +4016,7 @@ pub(crate) mod tests { // Rebind to a new sandbox id (the eventual resume path). let new_sandbox = engram_core::SandboxId::new(); - meta.assign_session_sandbox(session_id, Some(new_sandbox)) + meta.fenced_assign_sandbox(session_id, 0, Some(new_sandbox), None) .await .unwrap(); // Generation does NOT bump on a Some(_) rebind — the live diff --git a/crates/engram-coordinator/tests/admin_evac_live_pg.rs b/crates/engram-coordinator/tests/admin_evac_live_pg.rs index f15056538..f974bfefa 100644 --- a/crates/engram-coordinator/tests/admin_evac_live_pg.rs +++ b/crates/engram-coordinator/tests/admin_evac_live_pg.rs @@ -307,7 +307,7 @@ async fn seed_active_session( /// SB1 via `apply_missing_sandbox_strikes`, then rebind to SB2 via the /// production binding writers and assert the column is back at 0 — so a /// single subsequent missing tick does NOT cross the threshold. Before -/// the fix, `assign_session_sandbox` / `rebind_session` left the column +/// the fix, sandbox binding writes left the column /// untouched and the very next missing tick flipped a healthy session. #[tokio::test] #[ignore = "requires live Postgres at ENGRAM_TEST_DATABASE_URL"] @@ -331,9 +331,9 @@ async fn missing_strikes_reset_on_sandbox_rekey() { assert!(flipped.is_empty(), "below threshold → no flip"); } - // --- Path 1: assign_session_sandbox(Some) rebind resets strikes. + // --- Path 1: fenced_assign_sandbox(Some) rebind resets strikes. let sb2 = SandboxId::new(); - meta.assign_session_sandbox(session_id, Some(sb2)) + meta.fenced_assign_sandbox(session_id, 0, Some(sb2), Some(host_a)) .await .expect("rebind to SB2"); let flipped = meta @@ -370,19 +370,19 @@ async fn missing_strikes_reset_on_sandbox_rekey() { "rebind_session must also reset the strike streak" ); - // --- Path 3: unbind (assign_session_sandbox(None)) clears strikes. + // --- Path 3: unbind (fenced_assign_sandbox(None)) clears strikes. // Re-accrue to grace-1, unbind, rebind, then a single miss must not flip. for _ in 0..(grace - 2) { meta.apply_missing_sandbox_strikes(&[], &[session_id], grace) .await .expect("re-accrue before unbind"); } - meta.assign_session_sandbox(session_id, None) + meta.fenced_assign_sandbox(session_id, 0, None, Some(host_b)) .await .expect("unbind"); // Rebind to a fresh sandbox so the row is bound again for the tick. let sb4 = SandboxId::new(); - meta.assign_session_sandbox(session_id, Some(sb4)) + meta.fenced_assign_sandbox(session_id, 0, Some(sb4), Some(host_b)) .await .expect("rebind after unbind"); let flipped = meta diff --git a/crates/engram-coordinator/tests/api.rs b/crates/engram-coordinator/tests/api.rs index 01938d4cd..f334671c2 100644 --- a/crates/engram-coordinator/tests/api.rs +++ b/crates/engram-coordinator/tests/api.rs @@ -686,14 +686,20 @@ async fn live_manifest_publish_unbind_clears_and_bumps_generation() { assert_eq!(resp.status(), StatusCode::OK); let gen_after_publish = meta.chunk_generation().await.unwrap(); - // Direct trait call (no API endpoint for assign_session_sandbox). + // Direct trait call (no API endpoint for fenced_assign_sandbox). // ADR 0016 Phase B: this is the load-bearing eviction-race - // mitigation — assign_session_sandbox(None) clears the live + // mitigation — fenced_assign_sandbox(None) clears the live // manifest + bumps chunk_generation in the same logical step // the sandbox_id NULLs out. - engram_core::traits::MetadataStore::assign_session_sandbox(meta.as_ref(), session_id, None) - .await - .unwrap(); + engram_core::traits::MetadataStore::fenced_assign_sandbox( + meta.as_ref(), + session_id, + 0, + None, + None, + ) + .await + .unwrap(); assert!(meta .get_session(session_id) .await diff --git a/crates/engram-coordinator/tests/binding_cas_live_pg.rs b/crates/engram-coordinator/tests/binding_cas_live_pg.rs index 20d885135..11139dd41 100644 --- a/crates/engram-coordinator/tests/binding_cas_live_pg.rs +++ b/crates/engram-coordinator/tests/binding_cas_live_pg.rs @@ -1,8 +1,8 @@ //! Issue #211: live-Postgres regression tests for the guarded binding //! writers. //! -//! Before this fix, `assign_session_host`, `assign_session_sandbox`, and -//! `rebind_session` were blind `WHERE id = $1` UPDATEs with no state or +//! Before the guarded writers, binding updates used blind +//! `WHERE id = $1` UPDATEs with no state or //! expected-value guard. A racing actor could bind a live sandbox onto a //! row that had concurrently gone terminal (the terminate-races-resume //! interleaving) — the ownership oracle matches `sandbox_id` only and @@ -99,7 +99,7 @@ async fn seed_idle_unbound(meta: &Arc) -> SessionId { meta.transition_session(id, SessionState::Idle, BindingDisposition::Detach) .await .expect("active->idle"); - meta.assign_session_sandbox(id, None) + meta.fenced_assign_sandbox(id, 0, None, None) .await .expect("clear sandbox (evict_local / resume-dispatch shape)"); id @@ -197,7 +197,7 @@ async fn guarded_clear_does_not_null_a_fresh_rebind() { // A migration rebinds A -> B (the fresh, healthy binding). let sandbox_b = SandboxId::new(); - meta.assign_session_sandbox(id, Some(sandbox_b)) + meta.fenced_assign_sandbox(id, 0, Some(sandbox_b), None) .await .expect("rebind to B"); diff --git a/crates/engram-coordinator/tests/binding_writer_inventory.rs b/crates/engram-coordinator/tests/binding_writer_inventory.rs index 904ee2fd5..b3de13884 100644 --- a/crates/engram-coordinator/tests/binding_writer_inventory.rs +++ b/crates/engram-coordinator/tests/binding_writer_inventory.rs @@ -24,7 +24,6 @@ use std::path::Path; /// its call sites are ownership writes too. const METHODS: &[&str] = &[ "transition_session_created", - "assign_session_sandbox", "assign_session_sandbox_guarded", "rebind_session_guarded", "teleport_commit", @@ -62,7 +61,6 @@ const DECLARED: &[(&str, &str, usize, &str)] = &[ ("teleport", "transition_with_fence", 1, "post-attach peer loss enters failed recovery"), ("api/snapshot", "transition_with_fence_emitting", 1, "disk-only cold boot retains the fenced binding"), ("queue_scanner", "transition_session", 4, "Queued→Idle/Failed settles (RequireUnbound: queued rows are unbound)"), - ("reconcile", "assign_session_sandbox", 1, "strike-out unbind fallback"), ("reconcile", "assign_session_sandbox_guarded", 1, "strike-out guarded unbind"), ("reconcile", "transition_session", 1, "→HostLost after the clears (stage-2 now via dead_host::settle_host_lost, ADR 0116 A4)"), ("session_boot", "transition_session", 1, "Created→Active (Retain: the freshly-bound sandbox)"), diff --git a/crates/engram-coordinator/tests/chunk_gc_helpers_live_pg.rs b/crates/engram-coordinator/tests/chunk_gc_helpers_live_pg.rs index 9dbec630a..0bf9d0ded 100644 --- a/crates/engram-coordinator/tests/chunk_gc_helpers_live_pg.rs +++ b/crates/engram-coordinator/tests/chunk_gc_helpers_live_pg.rs @@ -207,11 +207,16 @@ async fn list_live_session_disk_manifest_ids_picks_up_live_writes() { "version must round-trip via the live-set query", ); - // Clear the live manifest via assign_session_sandbox(None) — + // Clear the live manifest via fenced_assign_sandbox(None) — // the trait method already cascades the cleanup per its docs. - meta.assign_session_sandbox(session_id, None) - .await - .expect("unbind sandbox"); + meta.fenced_assign_sandbox( + session_id, + 0, + None, + meta.get_session(session_id).await.unwrap().host_id, + ) + .await + .expect("unbind sandbox"); // Allow PG a tick to fully commit the cascade then re-poll. tokio::time::sleep(Duration::from_millis(50)).await; diff --git a/crates/engram-coordinator/tests/dead_host_mock.rs b/crates/engram-coordinator/tests/dead_host_mock.rs index 0d051daa1..1717052fa 100644 --- a/crates/engram-coordinator/tests/dead_host_mock.rs +++ b/crates/engram-coordinator/tests/dead_host_mock.rs @@ -106,16 +106,6 @@ impl MetadataStore for MiniMeta { s.host_id = host_id; Ok(()) } - async fn assign_session_sandbox( - &self, - id: SessionId, - sandbox_id: Option, - ) -> Result, MetaError> { - let mut g = self.sessions.lock(); - let s = g.get_mut(&id).ok_or(MetaError::NotFound)?; - s.sandbox_id = sandbox_id; - Ok(None) - } async fn upsert_host(&self, _h: HostRecord) -> Result<(), MetaError> { Ok(()) } diff --git a/crates/engram-coordinator/tests/eviction_live_pg.rs b/crates/engram-coordinator/tests/eviction_live_pg.rs index 4bcb547e7..b9ae529e1 100644 --- a/crates/engram-coordinator/tests/eviction_live_pg.rs +++ b/crates/engram-coordinator/tests/eviction_live_pg.rs @@ -183,7 +183,9 @@ async fn backstop_query_filters_on_ttl_and_sandbox() { // Unbinding the sandbox removes it from the backstop's view // (nothing to evict; other lifecycle paths own bare rows). - meta.assign_session_sandbox(id, None).await.expect("unbind"); + meta.fenced_assign_sandbox(id, 0, None, meta.get_session(id).await.unwrap().host_id) + .await + .expect("unbind"); assert!( !meta .list_active_sessions_idle_past(0) diff --git a/crates/engram-coordinator/tests/outbox_live_pg.rs b/crates/engram-coordinator/tests/outbox_live_pg.rs index b7669cdb1..2172ea7cf 100644 --- a/crates/engram-coordinator/tests/outbox_live_pg.rs +++ b/crates/engram-coordinator/tests/outbox_live_pg.rs @@ -335,17 +335,24 @@ async fn binding_epoch_mints_monotonically_per_session() { .await .expect("bind 1"); let e2 = meta - .assign_session_sandbox(sid, Some(engram_core::SandboxId::new())) + .fenced_assign_sandbox(sid, 0, Some(engram_core::SandboxId::new()), None) .await .expect("bind 2"); assert_eq!((e1, e2), (1, Some(2))); // A clear is not a binding write: no mint. assert_eq!( - meta.assign_session_sandbox(sid, None).await.expect("clear"), + meta.fenced_assign_sandbox(sid, 0, None, None) + .await + .expect("clear"), None ); assert!(meta - .assign_session_sandbox(SessionId::new(), Some(engram_core::SandboxId::new())) + .fenced_assign_sandbox( + SessionId::new(), + 0, + Some(engram_core::SandboxId::new()), + None + ) .await .is_err()); } diff --git a/crates/engram-coordinator/tests/reconcile_integration.rs b/crates/engram-coordinator/tests/reconcile_integration.rs index 1c902f219..a0c7b8f2d 100644 --- a/crates/engram-coordinator/tests/reconcile_integration.rs +++ b/crates/engram-coordinator/tests/reconcile_integration.rs @@ -310,7 +310,7 @@ async fn re_appearing_sandbox_within_grace_does_not_flip() { /// (evac/resume/migration). The strike counter encodes "N CONSECUTIVE /// heartbeats THIS binding's sandbox was missing"; rebinding to a fresh /// sandbox breaks consecutiveness, so the new binding is owed a full -/// `grace_ticks` window. Before the fix, `assign_session_sandbox` left +/// `grace_ticks` window. Before the fix, sandbox binding writes left /// the counter untouched, so a freshly-resumed session carrying 2 stale /// strikes was dismantled on its first transient under-report (strike 3 /// of a contract-promised 3-tick grace collapsed to 1). @@ -342,7 +342,7 @@ async fn stale_strikes_do_not_carry_across_sandbox_rekey() { meta.assign_session_host(session, Some(host_b)) .await .unwrap(); - meta.assign_session_sandbox(session, Some(sb2)) + meta.fenced_assign_sandbox(session, 0, Some(sb2), Some(host_b)) .await .unwrap(); diff --git a/crates/engram-core/src/traits/metadata.rs b/crates/engram-core/src/traits/metadata.rs index 1780e8ba3..4599f92da 100644 --- a/crates/engram-core/src/traits/metadata.rs +++ b/crates/engram-core/src/traits/metadata.rs @@ -897,24 +897,13 @@ pub trait MetadataStore: Send + Sync { Ok(Some((prev, target))) } + /// Retained because dead_host.rs still uses it to clear host ownership. async fn assign_session_host( &self, id: SessionId, host_id: Option, ) -> Result<(), MetaError>; - /// Persist the in-memory `SandboxId` of the live sandbox serving - /// this session. Set to `Some` after `host_registry.create_for_session` - /// returns, cleared to `None` on evict/migrate. A set mints and returns - /// the binding epoch; a clear returns None without a mint. The coordinator - /// uses these rows to rebuild its in-memory routing maps after - /// a restart. - async fn assign_session_sandbox( - &self, - id: SessionId, - sandbox_id: Option, - ) -> Result, MetaError>; - /// ADR 0073 phase 4: one row per Active+bound session for the idle /// scan — newest event (kind + time), host, and the shell pin. The /// detector classifies soft/hard client-side so the TTL policy @@ -1328,7 +1317,8 @@ pub trait MetadataStore: Send + Sync { } /// A set returns the new binding epoch; a clear returns None. - /// A stale fence returns Conflict. + /// A stale fence returns Conflict. Both writes reset missing-sandbox strikes; + /// a clear also removes the live manifest and bumps chunk_generation. /// Fenced sandbox (re)bind — subsumes `rebind_session_guarded`'s /// bespoke expected-state list with the one epoch predicate. async fn fenced_assign_sandbox( @@ -1526,14 +1516,7 @@ pub trait MetadataStore: Send + Sync { .await } - /// Issue #211: guarded compare-and-swap variant of - /// [`Self::assign_session_sandbox`]. The three binding writers - /// (`assign_session_{host,sandbox}`, `rebind_session`) are otherwise - /// blind `WHERE id = $1` UPDATEs: a racing actor can bind a live - /// sandbox onto a row that has *concurrently* gone terminal (the - /// terminate-races-resume interleaving), defeating the orphan reap — - /// the ownership oracle matches `sandbox_id` only and never re-checks - /// status, so a live VM pinned to a `Completed` row leaks forever. + /// Compare-and-swap sandbox binding write. /// /// This method conditions the write on: /// * `expected_current` — `Some(prev)` requires the row's current @@ -1548,33 +1531,16 @@ pub trait MetadataStore: Send + Sync { /// row with this id exists at all. Callers that just created a sandbox /// MUST destroy it on `Conflict` rather than leaking it. /// - /// The default impl composes a `get_session` legality check with the - /// blind setter — atomic enough for single-threaded mock stores; the - /// Postgres store overrides it with a true single-statement CAS. async fn assign_session_sandbox_guarded( &self, - id: SessionId, - sandbox_id: Option, - expected_current: Option>, - allowed_states: &[SessionState], + _id: SessionId, + _sandbox_id: Option, + _expected_current: Option>, + _allowed_states: &[SessionState], ) -> Result, MetaError> { - let session = self.get_session(id).await?; - if let Some(expected) = expected_current { - if session.sandbox_id != expected { - return Err(MetaError::Conflict(format!( - "assign_session_sandbox guard: sandbox_id is {:?}, expected {:?}", - session.sandbox_id, expected - ))); - } - } - if !allowed_states.is_empty() && !allowed_states.contains(&session.status) { - return Err(MetaError::Conflict(format!( - "assign_session_sandbox guard: status is {}, not in {:?}", - session.status.as_str(), - allowed_states - ))); - } - self.assign_session_sandbox(id, sandbox_id).await + unimplemented!( + "assign_session_sandbox_guarded: PostgresStore and SimMetadataStore implement it" + ) } /// Issue #211: guarded CAS variant of [`Self::assign_session_host`]. @@ -1616,32 +1582,13 @@ pub trait MetadataStore: Send + Sync { /// landing on a row that went terminal mid-migration. async fn rebind_session_guarded( &self, - id: SessionId, - host_id: HostId, - sandbox_id: SandboxId, - expected_current: Option>, - allowed_states: &[SessionState], + _id: SessionId, + _host_id: HostId, + _sandbox_id: SandboxId, + _expected_current: Option>, + _allowed_states: &[SessionState], ) -> Result { - let session = self.get_session(id).await?; - if let Some(expected) = expected_current { - if session.sandbox_id != expected { - return Err(MetaError::Conflict(format!( - "rebind_session guard: sandbox_id is {:?}, expected {:?}", - session.sandbox_id, expected - ))); - } - } - if !allowed_states.is_empty() && !allowed_states.contains(&session.status) { - return Err(MetaError::Conflict(format!( - "rebind_session guard: status is {}, not in {:?}", - session.status.as_str(), - allowed_states - ))); - } - self.assign_session_host(id, Some(host_id)).await?; - self.assign_session_sandbox(id, Some(sandbox_id)) - .await? - .ok_or_else(|| MetaError::Serialization("binding write returned no epoch".into())) + unimplemented!("rebind_session_guarded: PostgresStore and SimMetadataStore implement it") } /// ADR 0015 M3: PG-authoritative lookup for "which host owns this diff --git a/crates/engram-core/src/types/session.rs b/crates/engram-core/src/types/session.rs index d1aff15db..0a774b53f 100644 --- a/crates/engram-core/src/types/session.rs +++ b/crates/engram-core/src/types/session.rs @@ -619,7 +619,7 @@ pub struct Session { /// ADR 0016 Phase B: the host's last-published live disk /// manifest from the FlushScheduler. Updated by /// `MetadataStore::update_live_disk_manifest`; cleared by - /// `assign_session_sandbox(None)`. Coord's + /// `fenced_assign_sandbox(None)`. Coord's /// `effective_resume_disk_manifest` picks the newer of this /// and `snapshots.disk_manifest` so the first resume after /// continuous flush is enabled doesn't silently roll back to diff --git a/crates/engram-host-agent/src/disk_daemon/flush_scheduler.rs b/crates/engram-host-agent/src/disk_daemon/flush_scheduler.rs index 8e1438f67..8fe0cb5c4 100644 --- a/crates/engram-host-agent/src/disk_daemon/flush_scheduler.rs +++ b/crates/engram-host-agent/src/disk_daemon/flush_scheduler.rs @@ -59,7 +59,7 @@ //! Mitigation lives in commit 3: `transition_session(Idle)` clears //! `live_disk_manifest_*` in the same transaction it flips //! `status = Idle`, so any racing publish that landed between -//! `host.snapshot()` and `assign_session_sandbox(None)` is wiped. +//! `host.snapshot()` and the detach is wiped. //! The `sessions.sandbox_id == publish.sandbox_id` guard ALSO //! drops post-step-4 publishes. Two defences in series; the Idle- //! clear is the load-bearing one for the pre-step-4 window. diff --git a/crates/engram-oci-auth/src/lib.rs b/crates/engram-oci-auth/src/lib.rs index 2f94e2eee..263651da5 100644 --- a/crates/engram-oci-auth/src/lib.rs +++ b/crates/engram-oci-auth/src/lib.rs @@ -204,7 +204,7 @@ mod tests { use engram_core::types::{ HostRecord, PersistedEvent, Session, SessionSpec, SessionState, SnapshotRecord, }; - use engram_core::{HostId, MetaError, SandboxId, SessionId}; + use engram_core::{HostId, MetaError, SessionId}; /// MetadataStore stub that holds at most one registry credential. /// All other methods unreachable / empty — we never call them in @@ -264,13 +264,6 @@ mod tests { ) -> Result<(), MetaError> { Ok(()) } - async fn assign_session_sandbox( - &self, - _: SessionId, - _: Option, - ) -> Result, MetaError> { - Ok(None) - } async fn upsert_host(&self, _: HostRecord) -> Result<(), MetaError> { Ok(()) } diff --git a/crates/engram-postgres/src/lib.rs b/crates/engram-postgres/src/lib.rs index 5179e8aca..1a2c73017 100644 --- a/crates/engram-postgres/src/lib.rs +++ b/crates/engram-postgres/src/lib.rs @@ -3935,113 +3935,8 @@ impl MetadataStore for PostgresStore { Ok(res.rows_affected() > 0) } - async fn assign_session_sandbox( - &self, - id: SessionId, - sandbox_id: Option, - ) -> Result, MetaError> { - // ADR 0016 Phase B: the `live_disk_manifest_*` invariant is - // "set IFF the session is bound to a running sandbox the - // host-side FlushScheduler is publishing for." Unbinding - // (sandbox_id = NULL) means the live manifest is no longer - // authoritative — the snapshot row (if any) is. Clear in the - // same UPDATE so: - // 1. Resume's `effective_resume_disk_manifest` resolver - // (commit 6) sees NULL and falls back to the snapshot's - // `disk_manifest`. Without this, an eviction race - // (scheduler publishes between `host.snapshot()` and - // `assign_session_sandbox(None)`) would leave a live - // manifest AHEAD of the snapshot, producing an - // incoherent (memory at T-from-snapshot, disk at T+delta) - // resume. - // 2. Phase C's pin set (which keys on `live_disk_manifest_id` - // WHERE NOT NULL) drops the post-eviction lineage from - // its live set so its chunks become GC-eligible after - // the snapshot's chunks supersede them. - // - // Rebinding (sandbox_id = Some) does NOT clear — the next - // scheduler flush of the new sandbox populates the columns; - // any leftover value from a prior binding is overwritten by - // that publish (or the sandbox_id guard drops it as stale). - // - // Issue #215: BOTH branches reset `missing_strikes = 0`. The - // reconcile strike counter (`apply_missing_sandbox_strikes`) - // means "N CONSECUTIVE heartbeats in which THIS session's - // bound sandbox was missing from its host's running set". A - // rebind points the row at a brand-new sandbox and an unbind - // detaches it entirely — either way the previous streak is no - // longer consecutive against the current binding, so carrying - // it forward would collapse the 3-tick grace for a freshly - // resumed/migrated session (one transient under-report on the - // new host is strike 3, dismantling a healthy VM). The strike - // column is keyed by session id alone, so re-keying the - // sandbox must explicitly clear it here. - let now = self.clock.now_utc(); - if sandbox_id.is_some() { - let row = sqlx::query( - "UPDATE sessions SET sandbox_id = $2, missing_strikes = 0, updated_at = $3, binding_epoch = binding_epoch + 1 \ - WHERE id = $1 RETURNING binding_epoch", - ) - .bind(id.as_uuid()) - .bind(sandbox_id.map(|s| s.as_uuid())) - .bind(now) - .fetch_optional(&self.pool) - .await - .map_err(db_err)?.ok_or(MetaError::NotFound)?; - return Ok(Some(row.try_get::(0).map_err(db_err)? as u64)); - } else { - // Single TX: clear sandbox + live manifest, AND bump - // chunk_generation in the same step so Phase C's mid- - // sweep barrier observes the pin-set shrink atomically. - let mut tx = self.pool.begin().await.map_err(db_err)?; - let n = sqlx::query( - "UPDATE sessions - SET sandbox_id = NULL, - missing_strikes = 0, - live_disk_manifest_id = NULL, - live_disk_manifest_version = NULL, - live_disk_manifest_at = NULL, - updated_at = $2 - WHERE id = $1", - ) - .bind(id.as_uuid()) - .bind(now) - .execute(&mut *tx) - .await - .map_err(db_err)? - .rows_affected(); - // Only bump generation when we actually cleared a row - // that had a live manifest. A NULL→NULL clear is a no-op - // for the pin set; bumping anyway is harmless (a wasted - // sweep restart) but the conditional keeps generation - // bumps tied to real pin-set deltas. - if n > 0 { - sqlx::query( - "UPDATE chunk_generation SET generation = generation + 1 WHERE id = TRUE", - ) - .execute(&mut *tx) - .await - .map_err(db_err)?; - } - tx.commit().await.map_err(db_err)?; - if n == 0 { - return Err(MetaError::NotFound); - } - } - Ok(None) - } - - // ---- Issue #211: guarded CAS overrides ---- - // - // The three writers above are blind `WHERE id = $1` UPDATEs. These - // overrides condition the same write on the row's current - // `sandbox_id` and `status`, so a racing actor can't bind a live - // sandbox onto a row that concurrently went terminal (defeating the - // orphan reap), and reconcile can't null a freshly-landed rebind. - // - // `0 rows` is disambiguated into `Conflict` (row exists but the guard - // rejected it) vs `NotFound` (no such id) with a cheap follow-up - // existence probe — the same shape `transition_session` uses. + // Guarded binding writes lock the row before checking sandbox and status. + // A concurrent binding change must not be overwritten by a stale writer. async fn assign_session_sandbox_guarded( &self, @@ -4050,9 +3945,8 @@ impl MetadataStore for PostgresStore { expected_current: Option>, allowed_states: &[SessionState], ) -> Result, MetaError> { - // The `None` clear path additionally tears down the live disk - // manifest + bumps chunk_generation; reuse the existing blind - // setter inside a guarded TX rather than duplicating that logic. + // Clear the live manifest and bump chunk_generation on unbind in + // the same transaction, so resume cannot use a post-snapshot publish. let states: Vec = allowed_states .iter() .map(|s| s.as_str().to_string()) @@ -4077,19 +3971,17 @@ impl MetadataStore for PostgresStore { if cur_sandbox != expected { tx.rollback().await.map_err(db_err)?; return Err(MetaError::Conflict(format!( - "assign_session_sandbox CAS: sandbox_id is {cur_sandbox:?}, expected {expected:?}" + "assign_session_sandbox_guarded CAS: sandbox_id is {cur_sandbox:?}, expected {expected:?}" ))); } } if !states.is_empty() && !states.contains(&status) { tx.rollback().await.map_err(db_err)?; return Err(MetaError::Conflict(format!( - "assign_session_sandbox CAS: status is {status}, not in {states:?}" + "assign_session_sandbox_guarded CAS: status is {status}, not in {states:?}" ))); } - // Issue #215: clear `missing_strikes` on both bind and unbind — - // see the comment in `assign_session_sandbox`. A re-key / unbind - // breaks the reconcile strike streak's consecutiveness. + // A bind or unbind breaks the consecutive missing-sandbox streak. let epoch = if sandbox_id.is_some() { let row = sqlx::query( "UPDATE sessions SET sandbox_id = $2, missing_strikes = 0, updated_at = $3, binding_epoch = binding_epoch + 1 \ @@ -4187,7 +4079,7 @@ impl MetadataStore for PostgresStore { .collect(); let expected_uuid = expected_current.map(|o| o.map(|s| s.as_uuid())); // Issue #215: re-keying onto a fresh sandbox clears the stale - // reconcile strike streak (see `assign_session_sandbox`). + // reconcile strike streak. // ADR 0116 A5: a rebind SUPERSEDES the old binding without // host-affirmed absence — the superseded sandbox gets its // tombstone in the SAME statement (the A4 discipline; without it @@ -8700,10 +8592,20 @@ impl MetadataStore for PostgresStore { host_id: Option, ) -> Result, MetaError> { let res = sqlx::query( - "UPDATE sessions + "WITH binding AS ( + UPDATE sessions SET sandbox_id = $2, host_id = $3, last_active_at = $5, updated_at = $5, + missing_strikes = 0, + live_disk_manifest_id = CASE WHEN $2::uuid IS NULL THEN NULL ELSE live_disk_manifest_id END, + live_disk_manifest_version = CASE WHEN $2::uuid IS NULL THEN NULL ELSE live_disk_manifest_version END, + live_disk_manifest_at = CASE WHEN $2::uuid IS NULL THEN NULL ELSE live_disk_manifest_at END, binding_epoch = binding_epoch + CASE WHEN $2::uuid IS NULL THEN 0 ELSE 1 END - WHERE id = $1 AND current_epoch = $4 RETURNING binding_epoch", + WHERE id = $1 AND current_epoch = $4 RETURNING binding_epoch + ), generation AS ( + UPDATE chunk_generation SET generation = generation + 1 + WHERE id = TRUE AND $2::uuid IS NULL AND EXISTS (SELECT 1 FROM binding) + ) + SELECT binding_epoch FROM binding", ) .bind(session_id.as_uuid()) .bind(sandbox_id.map(|s| s.as_uuid())) diff --git a/crates/engram-sim/src/meta/store_impl.rs b/crates/engram-sim/src/meta/store_impl.rs index 0600024e6..bb6bc36d0 100644 --- a/crates/engram-sim/src/meta/store_impl.rs +++ b/crates/engram-sim/src/meta/store_impl.rs @@ -942,31 +942,6 @@ impl MetadataStore for SimMetadataStore { Ok(()) } - /// Bind: set sandbox + reset strikes. Unbind (None): also clears the - /// live disk manifest and bumps chunk_generation (same tx). - async fn assign_session_sandbox( - &self, - id: SessionId, - sandbox_id: Option, - ) -> Result, MetaError> { - self.gate()?; - let now = self.now(); - let mut db = self.db.lock(); - let row = db.sessions.get_mut(&id).ok_or(MetaError::NotFound)?; - row.session.sandbox_id = sandbox_id; - row.missing_strikes = 0; - row.updated_at = now; - let epoch = sandbox_id.map(|_| { - row.binding_epoch += 1; - row.binding_epoch as u64 - }); - if sandbox_id.is_none() { - row.session.live_disk_manifest = None; - db.chunk_generation += 1; - } - Ok(epoch) - } - /// Fenced on status='idle' AND current_epoch=$2; true iff the flip /// landed. Fires the enqueued placement notify on success. async fn enqueue_session_resume(&self, id: SessionId, epoch: i64) -> Result { @@ -2581,13 +2556,19 @@ impl MetadataStore for SimMetadataStore { return Err(MetaError::Conflict("stale session fence".into())); } r.session.sandbox_id = sandbox_id; + r.missing_strikes = 0; r.session.host_id = host_id; r.session.last_active_at = now; r.updated_at = now; - Ok(sandbox_id.map(|_| { + let binding_epoch = sandbox_id.map(|_| { r.binding_epoch += 1; r.binding_epoch as u64 - })) + }); + if sandbox_id.is_none() { + r.session.live_disk_manifest = None; + db.chunk_generation += 1; + } + Ok(binding_epoch) } /// CAS on expected sandbox + allowed states; distinct Conflict @@ -2608,14 +2589,14 @@ impl MetadataStore for SimMetadataStore { if let Some(expected) = expected_current { if r.session.sandbox_id != expected { return Err(MetaError::Conflict(format!( - "assign_session_sandbox CAS: sandbox_id is {:?}, expected {:?}", + "assign_session_sandbox_guarded CAS: sandbox_id is {:?}, expected {:?}", r.session.sandbox_id, expected ))); } } if !allowed_states.is_empty() && !allowed_states.contains(&r.session.status) { return Err(MetaError::Conflict(format!( - "assign_session_sandbox CAS: status is {}, not in {:?}", + "assign_session_sandbox_guarded CAS: status is {}, not in {:?}", r.session.status.as_str(), allowed_states ))); diff --git a/crates/engram-sim/tests/meta_conformance.rs b/crates/engram-sim/tests/meta_conformance.rs index 971402f1c..057f80fe7 100644 --- a/crates/engram-sim/tests/meta_conformance.rs +++ b/crates/engram-sim/tests/meta_conformance.rs @@ -3002,7 +3002,9 @@ async fn resident_sandboxes_rehydrate_list(ctx: &Ctx) { meta.transition_session(s_idle, SessionState::Idle, BindingDisposition::Detach) .await .unwrap(); - meta.assign_session_sandbox(s_idle, None).await.unwrap(); + meta.fenced_assign_sandbox(s_idle, 0, None, Some(host)) + .await + .unwrap(); // Active on ANOTHER host → excluded from this host's list. let s_elsewhere = meta @@ -6008,14 +6010,19 @@ async fn binding_epoch_minted_by_binding_writes(ctx: &Ctx) { .unwrap(), None ); - assert_eq!(meta.assign_session_sandbox(sid, None).await.unwrap(), None); + assert_eq!( + meta.assign_session_sandbox_guarded(sid, None, Some(None), &[]) + .await + .unwrap(), + None + ); assert!(matches!( meta.fenced_assign_sandbox(sid, 99, Some(second), Some(host)) .await, Err(MetaError::Conflict(_)) )); assert_eq!( - meta.assign_session_sandbox(sid, Some(second)) + meta.fenced_assign_sandbox(sid, 0, Some(second), Some(host)) .await .unwrap(), Some(4) @@ -6038,12 +6045,106 @@ async fn binding_epoch_minted_by_binding_writes(ctx: &Ctx) { None ); assert_eq!(meta.rebind_session(sid, host, second).await.unwrap(), 6); + // Reconcile's absent-binding CAS must not clear a newly bound sandbox. + assert!(matches!( + meta.assign_session_sandbox_guarded(sid, None, Some(None), &[]) + .await, + Err(MetaError::Conflict(_)) + )); + assert_eq!( + meta.get_session(sid).await.unwrap().sandbox_id, + Some(second) + ); + assert_eq!(meta.session_binding_generations(sid).await.unwrap().0, 6); } conformance!( t_binding_epoch_minted_by_binding_writes, super::binding_epoch_minted_by_binding_writes ); +async fn fenced_binding_preserves_cleanup_and_strike_reset(ctx: &Ctx) { + let meta = &ctx.meta; + let sid = meta + .create_session(spec("conf:fenced-binding-cleanup")) + .await + .unwrap(); + let first = engram_core::SandboxId::new(); + meta.transition_session_created(sid, first).await.unwrap(); + let manifest = engram_core::types::manifest::ManifestRef::new(); + meta.update_live_disk_manifest(sid, first, manifest) + .await + .unwrap(); + let generation = meta.chunk_generation().await.unwrap(); + assert!(meta + .apply_missing_sandbox_strikes(&[], &[sid], 2) + .await + .unwrap() + .is_empty()); + + // A rejected clear must retain both the live pin and the strike streak. + assert!(matches!( + meta.fenced_assign_sandbox(sid, 1, None, None).await, + Err(MetaError::Conflict(_)) + )); + assert_eq!( + meta.get_session(sid).await.unwrap().live_disk_manifest, + Some(manifest) + ); + assert_eq!(meta.chunk_generation().await.unwrap(), generation); + assert_eq!( + meta.apply_missing_sandbox_strikes(&[], &[sid], 2) + .await + .unwrap(), + vec![sid] + ); + + // A bind resets strikes, retains the manifest, and mints one epoch. + assert!(meta + .apply_missing_sandbox_strikes(&[], &[sid], 2) + .await + .unwrap() + .is_empty()); + let second = engram_core::SandboxId::new(); + assert_eq!( + meta.fenced_assign_sandbox(sid, 0, Some(second), None) + .await + .unwrap(), + Some(2) + ); + assert!(meta + .apply_missing_sandbox_strikes(&[], &[sid], 2) + .await + .unwrap() + .is_empty()); + assert_eq!( + meta.get_session(sid).await.unwrap().live_disk_manifest, + Some(manifest) + ); + assert_eq!(meta.chunk_generation().await.unwrap(), generation); + + // A clear resets strikes and removes the live pin without minting an epoch. + assert_eq!( + meta.fenced_assign_sandbox(sid, 0, None, None) + .await + .unwrap(), + None + ); + let session = meta.get_session(sid).await.unwrap(); + assert_eq!(session.sandbox_id, None); + assert_eq!(session.live_disk_manifest, None); + assert_eq!(meta.chunk_generation().await.unwrap(), generation + 1); + assert_eq!(meta.session_binding_generations(sid).await.unwrap().0, 2); + assert!(meta + .apply_missing_sandbox_strikes(&[], &[sid], 2) + .await + .unwrap() + .is_empty()); +} +conformance!( + t_fenced_binding_preserves_cleanup_and_strike_reset, + super::fenced_binding_preserves_cleanup_and_strike_reset +); + async fn settle_harness_generation_exactly_once(ctx: &Ctx) { let meta = &ctx.meta; let sid = meta.create_session(spec("conf:settlement")).await.unwrap(); @@ -6510,7 +6611,12 @@ async fn teleport_commit_is_one_statement(ctx: &Ctx) { ); let mismatch = restored_teleport(ctx).await; ctx.meta - .assign_session_sandbox(mismatch.session_id, Some(engram_core::SandboxId::new())) + .fenced_assign_sandbox( + mismatch.session_id, + 0, + Some(engram_core::SandboxId::new()), + Some(mismatch.source_host_id), + ) .await .unwrap(); assert_eq!(