From 8e3573b72be891b6a301fc8b2ec6a9ef64c5dd21 Mon Sep 17 00:00:00 2001 From: Drewry Pope Date: Mon, 2 Nov 2026 20:33:23 -0600 Subject: [PATCH] fix(id): reject replication feeds after storage failure --- pkgs/id/src/world/actor.rs | 159 ++++++++++++++++++++++++++++++++++--- 1 file changed, 148 insertions(+), 11 deletions(-) diff --git a/pkgs/id/src/world/actor.rs b/pkgs/id/src/world/actor.rs index a04c0573d..ed5d1641a 100644 --- a/pkgs/id/src/world/actor.rs +++ b/pkgs/id/src/world/actor.rs @@ -421,6 +421,9 @@ impl WorldActor { if let Err(error) = journal.append(&entry) { self.storage_failed = true; tracing::error!("world journal append failed: {error:#}"); + if version_before != journal.replication_header_version() { + let _ = replication_sender.send(ReplicationUpdate::ResetFailed); + } return Err(error.context("world storage failed; the world is read-only")); } self.metrics.record_journal_append(started.elapsed()); @@ -1575,17 +1578,22 @@ impl WorldActor { let _ = reply.send(self.metrics.snapshot(sessions)); } WorldCommand::ReplicationFeed { reply } => { - // Subscribe before capturing the image. The actor is the only - // journal owner and processes commands serially, so no durable - // append can fall between these two operations. - let updates = self.replication_sender.subscribe(); - let world_id = self.core.world_id().to_owned(); - let result = self - .journal - .as_mut() - .ok_or_else(|| anyhow::anyhow!("world has no durable journal")) - .and_then(|journal| journal.replication_snapshot()) - .map(|snapshot| (world_id, snapshot, updates)); + let result = if self.storage_failed { + Err(anyhow::anyhow!( + "world storage failed; replication feed is unavailable" + )) + } else { + // Subscribe before capturing the image. The actor is the + // only journal owner and processes commands serially, so + // no durable append can fall between these two operations. + let updates = self.replication_sender.subscribe(); + let world_id = self.core.world_id().to_owned(); + self.journal + .as_mut() + .ok_or_else(|| anyhow::anyhow!("world has no durable journal")) + .and_then(|journal| journal.replication_snapshot()) + .map(|snapshot| (world_id, snapshot, updates)) + }; let _ = reply.send(result); } WorldCommand::CapsInfo { reply } => { @@ -1686,6 +1694,75 @@ mod replication_feed_tests { } } + /// Write a complete journal entry, then fail as though `sync_data` + /// returned an error. The bytes remain visible to readers but are not + /// confirmed durable, matching the failure window in `FileJournal`. + struct SyncDataFailingJournal { + path: std::path::PathBuf, + version: u32, + entries: Vec, + } + + impl SyncDataFailingJournal { + fn new(path: &std::path::Path) -> Self { + let entries = vec![JournalEntry::Created { + world_id: "lobby".to_owned(), + version: 1, + }]; + std::fs::write( + path, + entries + .iter() + .flat_map(journal_entry_line) + .collect::>(), + ) + .unwrap(); + Self { + path: path.to_owned(), + version: 1, + entries, + } + } + } + + impl WorldJournal for SyncDataFailingJournal { + fn append(&mut self, entry: &JournalEntry) -> Result<()> { + let required_version = journal_version_required_by(entry); + if required_version > self.version { + self.version = required_version; + if let Some(JournalEntry::Created { version, .. }) = self.entries.first_mut() { + *version = required_version; + } + } + self.entries.push(entry.clone()); + + // This succeeds like `write_all`, leaving a complete newline- + // terminated entry; the error models `sync_data` failing after + // that write. + std::fs::write( + &self.path, + self.entries + .iter() + .flat_map(journal_entry_line) + .collect::>(), + )?; + Err(anyhow::anyhow!( + "simulated sync_data failure after complete write" + )) + } + + fn replication_snapshot(&mut self) -> Result { + Ok(JournalSnapshot { + header_version: self.version, + bytes: std::fs::read(&self.path)?, + }) + } + + fn replication_header_version(&self) -> Option { + Some(self.version) + } + } + #[test] fn header_bump_publishes_a_reset_instead_of_a_tail() { let core = WorldCore::new("lobby", WorldLimits::default()).unwrap(); @@ -1724,6 +1801,66 @@ mod replication_feed_tests { ); } + #[tokio::test] + async fn new_replication_feed_refuses_complete_unconfirmed_entry_after_sync_failure() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("journal.jsonl"); + let journal = SyncDataFailingJournal::new(&path); + let core = WorldCore::new("lobby", WorldLimits::default()).unwrap(); + let mut actor = + WorldActor::for_test(core, Box::new(SnapshotProgram), Some(Box::new(journal))); + + let unconfirmed = JournalEntry::Isolation { + isolation: Isolation::Unisolated, + }; + assert!(actor.persist(unconfirmed.clone()).is_err()); + assert!(actor.storage_failed); + + let bytes = std::fs::read(path).unwrap(); + assert!( + bytes.ends_with(b"\n"), + "the failed append wrote a complete line" + ); + let written: Vec = bytes + .split(|byte| *byte == b'\n') + .filter(|line| !line.is_empty()) + .map(|line| serde_json::from_slice(line).unwrap()) + .collect(); + assert_eq!(written.last(), Some(&unconfirmed)); + + let (reply, response) = oneshot::channel(); + actor.handle(WorldCommand::ReplicationFeed { reply }).await; + let feed = response.await.unwrap(); + assert!( + feed.is_err(), + "a new feed must not receive a snapshot containing the failed append" + ); + } + + #[test] + fn sync_failure_after_header_bump_publishes_reset_failed() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("journal.jsonl"); + let journal = SyncDataFailingJournal::new(&path); + let core = WorldCore::new("lobby", WorldLimits::default()).unwrap(); + let mut actor = + WorldActor::for_test(core, Box::new(SnapshotProgram), Some(Box::new(journal))); + let mut updates = actor.replication_sender.subscribe(); + let profile = actor.core.execution_profile(); + + let result = actor.persist(JournalEntry::ExecutionProfile { profile }); + + assert!(result.is_err()); + assert_eq!( + actor.journal.as_ref().unwrap().replication_header_version(), + Some(2) + ); + assert!(matches!( + updates.try_recv().unwrap(), + ReplicationUpdate::ResetFailed + )); + } + #[tokio::test] async fn journal_compaction_publishes_a_reset_snapshot() { let dir = TempDir::new().unwrap();