feat(id): serve signed world journal replication (#806 phase 2)

pkgs/id/src/commands/serve.rs

@@ -1145,6 +1145,13 @@ pub async fn cmd_serve(opts: ServeOptions) -> Result<()> {
                 crate::world_net::WORLD_ALPN,
                 crate::world_net::WorldProtocol::new(hub.clone()),
             )
+            .accept(
+                crate::world_replication_net::REPLICATION_ALPN,
+                crate::world_replication_net::WorldReplicationProtocol::new(
+                    hub.clone(),
+                    ed25519_dalek::SigningKey::from_bytes(&key.to_bytes()),
+                ),
+            )
             .accept(
                 crate::envelope_net::ENVELOPE_ALPN,
                 crate::envelope_net::EnvelopeProtocol::new(hub.clone()),

pkgs/id/src/lib.rs

@@ -256,6 +256,8 @@ pub mod world_native;
 #[cfg(feature = "world")]
 pub mod world_net;
 #[cfg(feature = "world")]
+pub mod world_replication_net;
+#[cfg(feature = "world")]
 pub use id_core::world_presentation;
 #[cfg(feature = "world")]
 pub mod world_records;

pkgs/id/src/world/actor.rs

@@ -45,6 +45,7 @@ impl WorldCommand {
             Self::SetFileScope { .. } => "SetFileScope",
             Self::PolicyInfo { .. } => "PolicyInfo",
             Self::Metrics { .. } => "Metrics",
+            Self::ReplicationFeed { .. } => "ReplicationFeed",
             Self::Shutdown { .. } => "Shutdown",
         }
     }
@@ -240,6 +241,16 @@ pub(crate) enum WorldCommand {
     Metrics {
         reply: oneshot::Sender<MetricsSnapshot>,
     },
+    /// Atomically capture the journal and subscribe to subsequent durable changes.
+    ReplicationFeed {
+        reply: oneshot::Sender<
+            Result<(
+                String,
+                JournalSnapshot,
+                broadcast::Receiver<ReplicationUpdate>,
+            )>,
+        >,
+    },
     Shutdown {
         reply: oneshot::Sender<()>,
     },
@@ -287,6 +298,7 @@ pub(crate) struct WorldActor {
     pub(crate) event_sender: broadcast::Sender<WorldEvent>,
     pub(crate) revocation_sender: broadcast::Sender<u64>,
     pub(crate) records_sender: watch::Sender<(u64, Arc<Records>)>,
+    pub(crate) replication_sender: broadcast::Sender<ReplicationUpdate>,
     pub(crate) view_changed: watch::Sender<u64>,
     /// Live sessions per participant: presence is refcounted.
     pub(crate) presence: HashMap<u64, usize>,
@@ -399,16 +411,38 @@ impl WorldActor {
             !self.storage_failed,
             "world storage failed; the world is read-only"
         );
+        let world_id = self.core.world_id().to_owned();
+        let replication_sender = self.replication_sender.clone();
         if let Some(journal) = self.journal.as_mut() {
             // The append includes its fsync, so this is the full durable
             // cost of the entry (F11.7 journal append + fsync latency).
             let started = std::time::Instant::now();
+            let version_before = journal.replication_header_version();
             if let Err(error) = journal.append(&entry) {
                 self.storage_failed = true;
                 tracing::error!("world journal append failed: {error:#}");
                 return Err(error.context("world storage failed; the world is read-only"));
             }
             self.metrics.record_journal_append(started.elapsed());
+            if let Some(header_version) = journal.replication_header_version() {
+                if version_before != Some(header_version) {
+                    let update = journal
+                        .replication_snapshot()
+                        .map(ReplicationUpdate::Reset)
+                        .unwrap_or_else(|error| {
+                            tracing::error!(
+                                "world {world_id}: replication reset snapshot failed: {error:#}"
+                            );
+                            ReplicationUpdate::ResetFailed
+                        });
+                    let _ = replication_sender.send(update);
+                } else {
+                    let _ = replication_sender.send(ReplicationUpdate::Append {
+                        header_version,
+                        entry: journal_entry_line(&entry),
+                    });
+                }
+            }
         }
         Ok(())
     }
@@ -1540,6 +1574,20 @@ impl WorldActor {
                 let sessions = self.presence.values().sum::<usize>() as u64;
                 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 _ = reply.send(result);
+            }
             WorldCommand::CapsInfo { reply } => {
                 let _ = reply.send(self.caps_report());
             }
@@ -1557,12 +1605,18 @@ impl WorldActor {
         {
             return;
         }
-        let Some(journal) = self.journal.as_mut() else {
-            return;
-        };
         let started = std::time::Instant::now();
-        match compact_journal(&mut self.core, self.program.as_mut(), journal.as_mut()) {
-            Ok(true) => self.metrics.record_checkpoint(started.elapsed()),
+        let compacted = match self.journal.as_mut() {
+            Some(journal) => {
+                compact_journal(&mut self.core, self.program.as_mut(), journal.as_mut())
+            }
+            None => return,
+        };
+        match compacted {
+            Ok(true) => {
+                self.metrics.record_checkpoint(started.elapsed());
+                self.publish_replication_reset();
+            }
             Ok(false) => self.compaction_off = true,
             Err(error) => {
                 // The journal is still complete; just do not retry until
@@ -1572,6 +1626,126 @@ impl WorldActor {
             }
         }
     }
+
+    fn publish_replication_reset(&mut self) {
+        let Some(journal) = self.journal.as_mut() else {
+            return;
+        };
+        let update = journal
+            .replication_snapshot()
+            .map(ReplicationUpdate::Reset)
+            .unwrap_or_else(|error| {
+                tracing::error!(
+                    "world {}: replication reset snapshot failed: {error:#}",
+                    self.core.world_id()
+                );
+                ReplicationUpdate::ResetFailed
+            });
+        let _ = self.replication_sender.send(update);
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
+mod replication_feed_tests {
+    use super::*;
+    use tempfile::TempDir;
+
+    struct SnapshotProgram;
+
+    impl WorldProgram for SnapshotProgram {
+        fn snapshot(&mut self) -> Result<Option<String>> {
+            Ok(Some("{}".to_owned()))
+        }
+    }
+
+    struct BumpingJournal {
+        version: u32,
+        entries: Vec<JournalEntry>,
+    }
+
+    impl WorldJournal for BumpingJournal {
+        fn append(&mut self, entry: &JournalEntry) -> Result<()> {
+            self.version = 2;
+            if let Some(JournalEntry::Created { version, .. }) = self.entries.first_mut() {
+                *version = self.version;
+            }
+            self.entries.push(entry.clone());
+            Ok(())
+        }
+
+        fn replication_snapshot(&mut self) -> Result<JournalSnapshot> {
+            Ok(JournalSnapshot {
+                header_version: self.version,
+                bytes: self.entries.iter().flat_map(journal_entry_line).collect(),
+            })
+        }
+
+        fn replication_header_version(&self) -> Option<u32> {
+            Some(self.version)
+        }
+    }
+
+    #[test]
+    fn header_bump_publishes_a_reset_instead_of_a_tail() {
+        let core = WorldCore::new("lobby", WorldLimits::default()).unwrap();
+        let journal = BumpingJournal {
+            version: 1,
+            entries: vec![JournalEntry::Created {
+                world_id: "lobby".to_owned(),
+                version: 1,
+            }],
+        };
+        let mut actor =
+            WorldActor::for_test(core, Box::new(SnapshotProgram), Some(Box::new(journal)));
+        let mut updates = actor.replication_sender.subscribe();
+
+        actor
+            .persist(JournalEntry::Isolation {
+                isolation: Isolation::Unisolated,
+            })
+            .unwrap();
+
+        let ReplicationUpdate::Reset(snapshot) = updates.try_recv().unwrap() else {
+            panic!("header bump must publish a reset snapshot");
+        };
+        assert_eq!(snapshot.header_version, 2);
+        let header = snapshot
+            .bytes
+            .split_inclusive(|byte| *byte == b'\n')
+            .next()
+            .unwrap();
+        assert_eq!(
+            serde_json::from_slice::<JournalEntry>(header).unwrap(),
+            JournalEntry::Created {
+                world_id: "lobby".to_owned(),
+                version: 2,
+            }
+        );
+    }
+
+    #[tokio::test]
+    async fn journal_compaction_publishes_a_reset_snapshot() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("journal.jsonl");
+        let mut limits = WorldLimits::default();
+        limits.checkpoint_every = 1;
+        let core = WorldCore::new("lobby", limits).unwrap();
+        let (journal, _) = crate::world_store::FileJournal::open(&path, "lobby").unwrap();
+        let mut actor =
+            WorldActor::for_test(core, Box::new(SnapshotProgram), Some(Box::new(journal)));
+        actor.core.active_module_hash = Some("test-module-hash".to_owned());
+        actor.core.since_checkpoint = 1;
+        let mut updates = actor.replication_sender.subscribe();
+
+        actor.compact_if_due().await;
+
+        let ReplicationUpdate::Reset(snapshot) = updates.recv().await.unwrap() else {
+            panic!("compaction must publish a reset snapshot");
+        };
+        assert_eq!(snapshot.bytes, std::fs::read(path).unwrap());
+        assert!(snapshot.bytes.ends_with(b"\n"));
+    }
 }
 
 #[cfg(test)]
@@ -1586,6 +1760,7 @@ impl WorldActor {
     ) -> Self {
         let (events, _) = broadcast::channel(256);
         let (revocations, _) = broadcast::channel(256);
+        let (replication_sender, _) = broadcast::channel(256);
         let (view_changed, _) = watch::channel(core.sequence);
         let mut program_healthy = true;
         let initial = match program_records(program.as_mut()) {
@@ -1615,6 +1790,7 @@ impl WorldActor {
             event_sender: events,
             revocation_sender: revocations,
             records_sender,
+            replication_sender,
             view_changed,
             presence: HashMap::new(),
             active_subs: BTreeSet::new(),

pkgs/id/src/world/handle.rs

@@ -45,6 +45,7 @@ impl WorldHandle {
         let (commands, receiver) = mpsc::channel(256);
         let (events, _) = broadcast::channel(256);
         let (revocations, _) = broadcast::channel(256);
+        let (replication_sender, _) = broadcast::channel(256);
         let (view_changed_sender, view_changed) = watch::channel(core.sequence);
         // Records are recomputed from the (restored) program, so they are a
         // pure function of the journal and need no storage of their own here.
@@ -79,6 +80,7 @@ impl WorldHandle {
             event_sender: events.clone(),
             revocation_sender: revocations.clone(),
             records_sender,
+            replication_sender,
             view_changed: view_changed_sender,
             presence: HashMap::new(),
             active_subs: BTreeSet::new(),
@@ -123,6 +125,26 @@ impl WorldHandle {
             .await
             .context("world actor dropped metrics response")
     }
+
+    /// Capture a durable journal image and subscribe atomically to later
+    /// appends and reset snapshots. The returned receiver is live before the
+    /// snapshot is taken, preventing a write from being missed between them.
+    pub(crate) async fn replication_feed(
+        &self,
+    ) -> Result<(
+        String,
+        JournalSnapshot,
+        broadcast::Receiver<ReplicationUpdate>,
+    )> {
+        let (reply, response) = oneshot::channel();
+        self.commands
+            .send(WorldCommand::ReplicationFeed { reply })
+            .await
+            .map_err(|_| anyhow::anyhow!("world actor is closed"))?;
+        response
+            .await
+            .context("world actor dropped replication feed response")?
+    }
     /// The latest `(sequence, records)` published by the world program.
     /// Callers must authorize the reader first (see [`Self::participant_id`]).
     #[must_use]
@@ -801,10 +823,8 @@ impl WorldHandle {
         installed_by: Option<u64>,
         program: Box<dyn WorldProgram>,
     ) -> Result<WorldEvent> {
-        self.install_program_with_trust(
-            module_hash, seed, provenance, installed_by, None, program,
-        )
-        .await
+        self.install_program_with_trust(module_hash, seed, provenance, installed_by, None, program)
+            .await
     }
 
     /// As [`Self::install_program_by`], also recording the module trust

pkgs/id/src/world/mod.rs

@@ -994,6 +994,19 @@ pub trait WorldJournal: Send + 'static {
     /// Any I/O failure; the actor then refuses further mutations.
     fn append(&mut self, entry: &JournalEntry) -> Result<()>;
 
+    /// Capture the exact durable journal bytes and their header version.
+    /// Implementations that do not own a readable journal may leave this
+    /// unsupported; the replication feed is available only for durable
+    /// journals that can provide an atomic snapshot from the world actor.
+    fn replication_snapshot(&mut self) -> Result<JournalSnapshot> {
+        bail!("this journal cannot provide a replication snapshot")
+    }
+
+    /// Current journal header version, when the implementation can report it.
+    fn replication_header_version(&self) -> Option<u32> {
+        None
+    }
+
     /// Atomically replace the whole journal with `entries` (a header, the
     /// participants and a [`JournalEntry::Checkpoint`]). Either the old or
     /// the new journal survives a crash, never a mixture.
@@ -1007,6 +1020,31 @@ pub trait WorldJournal: Send + 'static {
     }
 }
 
+/// A byte-exact, newline-terminated journal image captured by the actor.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct JournalSnapshot {
+    /// Version in the journal's `Created` header.
+    pub header_version: u32,
+    /// Exact JSONL bytes, including every entry's newline.
+    pub bytes: Vec<u8>,
+}
+
+/// A durable change delivered to replication subscribers.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum ReplicationUpdate {
+    /// One entry appended and fsynced after the subscriber's snapshot.
+    Append {
+        /// Header version the appended entry is encoded under.
+        header_version: u32,
+        /// Exact serialized JSONL entry bytes, including its newline.
+        entry: Vec<u8>,
+    },
+    /// The journal was compacted or its header version changed.
+    Reset(JournalSnapshot),
+    /// A reset became necessary but the journal image could not be captured.
+    ResetFailed,
+}
+
 /// The program a journal says was running, and where to resume it from.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct RestoredProgram {