diff --git a/crates/ghostlight-dungeon/src/runtime.rs b/crates/ghostlight-dungeon/src/runtime.rs index 33620a4..6dd9f82 100644 --- a/crates/ghostlight-dungeon/src/runtime.rs +++ b/crates/ghostlight-dungeon/src/runtime.rs @@ -407,7 +407,7 @@ pub(crate) async fn run(state_root_binding: Option) -> anyhow::Result<( }; tokio::spawn(maintain_mesh_projection(state.clone())); tokio::spawn(drive_cover_tick(state.clone(), configured_tick_interval())); - tokio::spawn(elaborate_world(state.clone())); + tokio::spawn(elaborate_world(state.clone(), ELABORATION_SWEEP_INTERVAL)); let app = app_router(state.clone(), web_root); tracing::info!(address = %bound_address, "Ghostlight Dungeon world owner serving"); let server = axum::serve( @@ -1831,8 +1831,8 @@ const ELABORATION_SWEEP_INTERVAL: Duration = Duration::from_secs(300); /// all, which is the config gate the runtime already has: no mode flag joins it. /// Sweeps never overlap, because this loop awaits one before it ticks the next, /// so one sweep's claim set is the only set of sessions alive. -async fn elaborate_world(state: AppState) { - let mut interval = tokio::time::interval(ELABORATION_SWEEP_INTERVAL); +async fn elaborate_world(state: AppState, interval: Duration) { + let mut interval = tokio::time::interval(interval); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); interval.tick().await; loop { @@ -3594,6 +3594,122 @@ mod tests { assert_eq!(state.elaboration_permits.available_permits(), elaboration); } + /// Panics on its first `infer` call. Every later call announces itself on + /// `again` and answers with text. + struct PanicOnceInferencePort { + first: AtomicBool, + again: Semaphore, + } + + #[async_trait::async_trait] + impl InferencePort for PanicOnceInferencePort { + fn prepare(&self, request: InferenceRequest) -> Result { + PreparedInference::prepare("ghostlight-controller-test", 4_102_444_800_000, request) + } + + async fn infer( + &self, + _request: PreparedInference, + ) -> Result { + if self.first.swap(false, Ordering::SeqCst) { + panic!("the fixture port panics inside the first elaboration session"); + } + self.again.add_permits(1); + Ok(InferenceOutput::new( + vec![InferenceEvent::Text("The driver fixture speaks.".into())], + "sha256:panic-once-port", + )) + } + } + + /// `AlwaysFreshWorkStore`, announcing each in-flight listing: one per sweep. + struct SweepCountingWorkStore { + listed: Semaphore, + } + + #[async_trait::async_trait] + impl ControllerWorkStore for SweepCountingWorkStore { + async fn lookup( + &self, + command_id: CommandId, + ) -> Result { + AlwaysFreshWorkStore.lookup(command_id).await + } + + async fn persist( + &self, + work: &ControllerWork, + ) -> Result { + AlwaysFreshWorkStore.persist(work).await + } + + async fn custody_probe(&self) -> Result { + AlwaysFreshWorkStore.custody_probe().await + } + + async fn elaboration_in_flight( + &self, + ) -> Result, ControllerWorkStoreError> { + self.listed.add_permits(1); + AlwaysFreshWorkStore.elaboration_in_flight().await + } + } + + /// The elaboration driver outlives a sweep whose session panicked. The + /// driver runs at a one-millisecond interval over a world with a deficit, + /// under an elaboration pool of one, so a sweep that stops after a panic + /// has no sibling session. The first sweep's session panics; the driver + /// then reads a second listing, which only a second sweep does, and that + /// sweep's session reaches the provider. If the driver task ends first, + /// the test fails on that instead of waiting. + #[tokio::test] + async fn the_elaboration_driver_survives_a_panicking_sweep() { + let fixture = fixture().await; + two_cell_world( + &fixture.state, + &fixture.cookie, + BTreeMap::from([(SubjectKind::Person, 4)]), + vec![CreateJurisdictionIntent { + handle: "sere".into(), + label: "The Low Sere".into(), + permille: 1000, + }], + true, + ) + .await; + let port = Arc::new(PanicOnceInferencePort { + first: AtomicBool::new(true), + again: Semaphore::new(0), + }); + let store = Arc::new(SweepCountingWorkStore { + listed: Semaphore::new(0), + }); + let mut state = fixture.state.clone(); + state.controllers = Some(Arc::new( + ControllerRunner::open( + state.world.clone(), + port.clone(), + store.clone(), + test_controller_models(), + ) + .expect("the fixture ports open"), + )); + state.elaboration_permits = Arc::new(Semaphore::new(1)); + + let mut driver = tokio::spawn(elaborate_world(state.clone(), Duration::from_millis(1))); + tokio::select! { + listed = store.listed.acquire_many(2) => listed.unwrap().forget(), + ended = &mut driver => panic!("the elaboration driver ended: {ended:?}"), + } + assert!(!port.first.load(Ordering::SeqCst), "the first sweep's session never panicked"); + tokio::select! { + again = port.again.acquire() => again.unwrap().forget(), + ended = &mut driver => panic!("the elaboration driver ended: {ended:?}"), + } + assert!(!state.controller_quarantined.load(Ordering::SeqCst)); + driver.abort(); + } + /// Raises `ControllerError::requires_quarantine` on every call. Used to /// prove the tick driver's quarantine edge rather than any cognition /// outcome. diff --git a/crates/ghostlight/src/controllers.rs b/crates/ghostlight/src/controllers.rs index de19a35..10432f7 100644 --- a/crates/ghostlight/src/controllers.rs +++ b/crates/ghostlight/src/controllers.rs @@ -11489,6 +11489,136 @@ mod tests { task.await.unwrap(); } + /// Holds the first session on each of two dead ends in the provider + /// boundary together, then panics the one on the Unwalked Road while the + /// Far Road's waits for the test. Every later call answers its road with a + /// shed. + struct PanickingPort { + both: tokio::sync::Barrier, + panicking: tokio::sync::Semaphore, + release: tokio::sync::Semaphore, + first_unwalked: AtomicBool, + first_far: AtomicBool, + roads: Box Result + Send + Sync>, + calls: std::sync::atomic::AtomicUsize, + } + + #[async_trait] + impl InferencePort for PanickingPort { + fn prepare(&self, request: InferenceRequest) -> Result { + PreparedInference::prepare("ghostlight-controller-test", 4_102_444_800_000, request) + } + + async fn infer( + &self, + request: PreparedInference, + ) -> Result { + let call = self.calls.fetch_add(1, Ordering::SeqCst); + if prompt_of(&request).contains("The Unwalked Road") { + if self.first_unwalked.swap(false, Ordering::SeqCst) { + self.both.wait().await; + self.panicking.add_permits(1); + panic!("the fixture port panics inside one session"); + } + } else if self.first_far.swap(false, Ordering::SeqCst) { + self.both.wait().await; + self.release.acquire().await.unwrap().forget(); + } + (self.roads)(call, &request) + } + } + + /// A session that panics does not take its sweep down with it. Two + /// sessions are in the provider boundary together; one panics while the + /// other is still held there. The held session finishes its inference and + /// commits in the same sweep, the sweep returns a quarantine-class error + /// naming the panic, and every permit is free. A second sweep then runs + /// normally: it resumes the panicked session's row and commits it. + #[tokio::test] + async fn a_panicking_session_lets_its_sibling_finish() { + let (_directory, mailbox, task, _commons, roads) = dead_end_world( + crate::tests::stock_weights(), + &["The Unwalked Road", "The Far Road"], + Roots::Commons, + 0, + ) + .await; + let port = Arc::new(PanickingPort { + both: tokio::sync::Barrier::new(2), + panicking: tokio::sync::Semaphore::new(0), + release: tokio::sync::Semaphore::new(0), + first_unwalked: AtomicBool::new(true), + first_far: AtomicBool::new(true), + roads: Box::new(shed_on_the_named_road(vec![ + ("The Unwalked Road", roads[0]), + ("The Far Road", roads[1]), + ])), + calls: std::sync::atomic::AtomicUsize::new(0), + }); + let permits = pool(2); + let store = fresh_store(); + let runner = sweep_runner(&mailbox, port.clone(), store.clone()); + let sweep = tokio::spawn({ + let runner = runner.clone(); + let permits = permits.clone(); + async move { runner.sweep(permits).await } + }); + port.panicking.acquire().await.unwrap().forget(); + port.release.add_permits(1); + let Err(error) = sweep.await.expect("a session's panic ended the sweep") else { + panic!("a panicking session did not end the sweep with an error"); + }; + assert!(error.requires_quarantine(), "{error:?}"); + assert!(error.to_string().contains("panic"), "{error}"); + let snapshot = mailbox.snapshot().await.unwrap(); + assert!( + place_labelled(&snapshot, "The Shed on The Far Road").is_some(), + "the held sibling did not commit in the panicking sweep" + ); + assert_eq!(snapshot.boundaries.len(), 1); + assert_eq!(port.calls.load(Ordering::SeqCst), 2); + assert_eq!(permits.available_permits(), 2, "a permit outlived its session"); + + runner.sweep(permits.clone()).await.expect("the sweep after a panic did not run"); + assert_eq!(port.calls.load(Ordering::SeqCst), 3); + let snapshot = mailbox.snapshot().await.unwrap(); + assert!(snapshot.boundaries.is_empty()); + assert!(place_labelled(&snapshot, "The Shed on The Unwalked Road").is_some()); + assert_eq!(store.work.lock().unwrap().len(), 2, "the resumed session opened a row"); + assert_eq!(permits.available_permits(), 2); + drop(runner); + drop(mailbox); + task.await.unwrap(); + } + + /// No session starts after one has panicked. Under a pool of one the + /// second entry waits for the first session's permit; the first session + /// panics, and the second never reaches the provider boundary. + #[tokio::test] + async fn no_session_starts_after_a_panic() { + let (_directory, mailbox, task, _commons, _roads) = dead_end_world( + crate::tests::stock_weights(), + &["The Unwalked Road", "The Far Road"], + Roots::Commons, + 0, + ) + .await; + let port = Arc::new(SweepPort::new(|_, _| { + panic!("the fixture port panics inside every session") + })); + let permits = pool(1); + let runner = sweep_runner(&mailbox, port.clone(), fresh_store()); + let Err(error) = runner.sweep(permits.clone()).await else { + panic!("a panicking session did not end the sweep with an error"); + }; + assert!(error.requires_quarantine(), "{error:?}"); + assert_eq!(port.calls(), 1, "a session started after a panic"); + assert_eq!(permits.available_permits(), 1); + drop(runner); + drop(mailbox); + task.await.unwrap(); + } + /// Probe 3's deficit half, through the sweep. Sweep one's deficit session /// is refused; a clock tick moves the ancestry, and with it the id a fresh /// derivation would give the deficit. Sweep two finds the refused session by diff --git a/crates/ghostlight/src/elaboration.rs b/crates/ghostlight/src/elaboration.rs index 534eef0..9dc48dd 100644 --- a/crates/ghostlight/src/elaboration.rs +++ b/crates/ghostlight/src/elaboration.rs @@ -639,7 +639,9 @@ impl ElaborationRunner { /// /// Every spawned session finishes before this returns. A quarantine-class /// error is returned after that, and no session starts after one has - /// returned such an error; any other error ends only its own session. + /// returned such an error. A session that panics is that same error: its + /// siblings finish, nothing starts after it, and it is returned. Any other + /// error ends only its own session. pub async fn sweep(&self, permits: Arc) -> Result<(), ControllerError> { let snapshot = self.mailbox.snapshot().await.map_err(snapshot_error)?; if snapshot.phase != WorldPhase::Active { @@ -666,14 +668,16 @@ impl ElaborationRunner { let snapshot = snapshot.clone(); let stopped = stopped.clone(); running.spawn(async move { + let mut held = SessionPermit { + _permit: permit, + stopped, + stop: true, + }; let outcome = runner.step_session(command_id, session, &snapshot).await; - if outcome + held.stop = outcome .as_ref() - .is_err_and(ControllerError::requires_quarantine) - { - stopped.store(true, Ordering::SeqCst); - } - drop(permit); + .is_err_and(ControllerError::requires_quarantine); + drop(held); outcome }); } @@ -687,7 +691,11 @@ impl ElaborationRunner { Ok(Err(error)) => { tracing::debug!(%error, "elaboration step did not admit a patch"); } - Err(error) => std::panic::resume_unwind(error.into_panic()), + Err(error) => { + quarantine.get_or_insert(ControllerError::Serialization(format!( + "an elaboration session did not return: {error}" + ))); + } } } quarantine.map_or(Ok(()), Err) @@ -717,6 +725,24 @@ impl ElaborationRunner { } } +/// One session's permit. Raises the sweep's stop before the permit is released +/// on every exit but a clean return: a quarantine-class error, or a session that +/// never returned because it panicked. `drop` runs before the fields drop, so +/// the sweep waiting on this permit sees the stop when it wakes. +struct SessionPermit { + _permit: tokio::sync::OwnedSemaphorePermit, + stopped: Arc, + stop: bool, +} + +impl Drop for SessionPermit { + fn drop(&mut self) { + if self.stop { + self.stopped.store(true, Ordering::SeqCst); + } + } +} + fn snapshot_error(error: MailboxError) -> ControllerError { ControllerError::Snapshot(error) }