feat(simulator): drive the workload under network and replica faults

core/binary_protocol/src/lib.rs

@@ -85,7 +85,7 @@ pub use dispatch::{COMMAND_TABLE, CommandMeta, lookup_by_operation, lookup_comma
 pub use error::WireError;
 pub use framing::{RequestFrame, ResponseFrame, STATUS_OK};
 pub use primitives::ack_level::AckLevel;
-pub use primitives::consumer::{KIND_CONSUMER_GROUP, WireConsumer};
+pub use primitives::consumer::{KIND_CONSUMER, KIND_CONSUMER_GROUP, WireConsumer};
 pub use primitives::identifier::{MAX_WIRE_NAME_LENGTH, WireIdentifier, WireName};
 pub use primitives::options::{MAX_OPTIONS, MAX_OPTIONS_BYTES, WireOptions, validate_options};
 pub use primitives::partition_assignment::CreatedPartitionAssignment;

core/binary_protocol/src/primitives/consumer.rs

@@ -20,7 +20,8 @@ use crate::WireIdentifier;
 use crate::codec::{WireDecode, WireEncode, read_u8};
 use bytes::{BufMut, BytesMut};
 
-const KIND_CONSUMER: u8 = 1;
+/// Wire discriminant for a single consumer (vs a `ConsumerGroup`).
+pub const KIND_CONSUMER: u8 = 1;
 /// Wire discriminant for a consumer-group consumer (vs a single `Consumer`).
 /// Public so the server dispatch can match on it by name instead of a raw `2`.
 pub const KIND_CONSUMER_GROUP: u8 = 2;

core/metadata/src/impls/metadata.rs

@@ -338,7 +338,10 @@ impl<M> SnapshotCoordinator<M> {
     /// forced. Must stay >= the prepare-queue depth: the ops already
     /// pipelined while a checkpoint runs skip it and append into this
     /// margin.
-    const CHECKPOINT_MARGIN: usize = 64;
+    ///
+    /// Public so a caller sizing a journal can refuse a slot count at or below it:
+    /// such a journal checkpoints on every commit rather than on occupancy.
+    pub const CHECKPOINT_MARGIN: usize = 64;
 
     #[must_use]
     pub fn new(
@@ -366,7 +369,9 @@ impl<M> SnapshotCoordinator<M> {
     /// transfer serves and installs.
     #[must_use]
     pub fn snapshot_path(&self) -> std::path::PathBuf {
-        self.data_dir.join(super::METADATA_DIR).join("snapshot.bin")
+        self.data_dir
+            .join(super::METADATA_DIR)
+            .join(super::SNAPSHOT_FILE_NAME)
     }
 
     /// The last persisted checkpoint's `(op, checksum)`, `(0, 0)` when none.
@@ -1170,21 +1175,48 @@ where
         // guard, not here.
         self.checkpoint_if_needed(consensus, journal).await;
 
-        // Backup: gap check (op == current_op + 1).
-        // Primary: sequencer pre-advanced by push_prepare_entry (guards
-        // sibling on_request races during journal.append await).
-        // TODO: promote the backup gap warn below to a hard assert or a
-        // repair-session trigger (message repair has landed; the drop-and-
-        // wait-for-retransmit path is the last soft handling left here).
+        // Backup: gap check against the JOURNAL head, not the sequencer.
+        //
+        // The two frontiers can disagree. The sequencer is pre-advanced on the
+        // primary by `push_prepare_entry` and re-synced on a backup only after a
+        // successful append, so a replica can carry a sequencer one ahead of what
+        // its WAL holds. Gating admission on it then rejects the very prepare that
+        // would heal the log: a backup with `last_op = 44` refused op 45 because
+        // its sequencer said to expect 46. The primary retransmits that op for the
+        // life of the process, every backup logs an out-of-order gap, it never
+        // reaches a commit quorum, and its client is never answered.
+        //
+        // `max(last_op, snapshot_op)`, never `last_op` alone. A state transfer
+        // installs a snapshot that IS ops `..=snapshot_op` applied and truncates the
+        // WAL above that floor rather than refilling below it, so `last_op` reads the
+        // receiver as needing an op the snapshot already contains and no peer will
+        // send again. That drop
+        // never heals: an offer built on a quiet cluster carries `commit_op ==
+        // snapshot_seq`, so the install lands `commit_min == commit_max`, and
+        // `maybe_request_metadata_repair`, the only path that refills the head,
+        // arms on `commit_min < commit_max`. With the other backup down the primary
+        // needs this replica's ack to commit anything, so the plane stops on a
+        // cluster still inside its quorum.
+        //
+        // The journal is the only frontier that answers "what can be appended
+        // next", which is what this check is for, and the hash-chain verification
+        // below is stated against it too. A prepare at or below the head that this
+        // replica already holds was re-acked and returned above. What reaches HERE
+        // is the next op or a gap, and not every gap is fillable: metadata repair
+        // covers only `commit_min + 1 ..= commit_max`, so an interior hole below
+        // the head and a forward gap above `commit_max` both sit outside it.
         let is_backup = consensus.is_follower();
         if is_backup {
-            if header.op != current_op + 1 {
+            let handle = journal.handle();
+            let journal_head = handle.last_op().unwrap_or(0).max(handle.snapshot_op());
+            if header.op != journal_head + 1 {
                 warn!(
                     target: "iggy.metadata.diag",
                     plane = "metadata",
                     replica_id = consensus.replica(),
                     op = header.op,
-                    expected = current_op + 1,
+                    expected = journal_head + 1,
+                    sequencer_op = current_op,
                     "on_replicate: dropping out-of-order prepare (gap)"
                 );
                 return;
@@ -4837,6 +4869,201 @@ mod tests {
         );
     }
 
+    /// A backup admits the prepare its JOURNAL needs next, even when its
+    /// sequencer has run ahead of the journal.
+    ///
+    /// The two frontiers legitimately disagree: `on_start_view` sets the sequencer
+    /// to the view's announced head, deliberately ahead of what this replica
+    /// holds, because the bodies arrive afterwards by retransmit or repair. Gating
+    /// admission on the sequencer therefore rejected exactly the prepare that
+    /// would heal the log: a backup with journal head 44 refusing op 45 because
+    /// its adopted head said to expect 46. The primary retransmits that op
+    /// forever, every backup logs an out-of-order gap, it never reaches a commit
+    /// quorum, and its client is never answered. Systematic for any rejoining
+    /// replica, so the deterministic simulator wedged on every metadata workload
+    /// under crash/restart injection until this was gated on the journal.
+    #[compio::test]
+    async fn backup_admits_the_prepare_its_journal_needs_despite_a_leading_sequencer() {
+        const CLIENT: u128 = 1;
+        const SESSION: u64 = 1;
+        const ACTING_USER: u32 = 7;
+        /// Stands in for a head adopted from a `StartView` whose bodies have not
+        /// arrived, so it sits well above the empty journal.
+        const ADOPTED_HEAD: u64 = 5;
+
+        let dir = tempfile::tempdir().unwrap();
+        std::fs::create_dir_all(dir.path().join(crate::impls::METADATA_DIR)).unwrap();
+        let journal =
+            journal::prepare_journal::PrepareJournal::open(&dir.path().join("journal.wal"), 0)
+                .await
+                .unwrap();
+        // Replica 1 of 3 at view 0, so `primary_index(0) == 0` makes this a backup
+        // and `on_replicate` takes the gap-check branch.
+        let consensus = VsrConsensus::new(
+            1,
+            1,
+            3,
+            server_common::sharding::METADATA_GROUP,
+            NoopBus,
+            LocalPipeline::new(),
+        );
+        consensus.init();
+        let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (), TestMux> =
+            IggyMetadata::new(
+                Some(consensus),
+                Some(journal),
+                None,
+                None,
+                TestMux::default(),
+                Some(dir.path().to_path_buf()),
+            );
+        let consensus = md.consensus.as_ref().unwrap();
+        assert!(
+            consensus.is_follower(),
+            "replica 1 of 3 at view 0 must be a backup for this to exercise the gap check"
+        );
+        md.client_table.borrow_mut().commit_register(
+            CLIENT,
+            ACTING_USER,
+            register_reply(CLIENT, SESSION),
+        );
+
+        // Minted while the sequencer is still at 0, so it carries op 1: exactly
+        // what the empty journal needs next.
+        let prepare = md
+            .prepare_request(create_stream_request(CLIENT, 1, "s1"))
+            .expect("CreateStream is client-allowed");
+        assert_eq!(prepare.header().op, 1, "the first prepare must be op 1");
+
+        // Now run the sequencer ahead, as adopting a started view does.
+        consensus.sequencer().set_sequence(ADOPTED_HEAD);
+        let journal = md.journal.as_ref().unwrap();
+        assert_eq!(
+            journal.last_op(),
+            None,
+            "the journal must still be empty, else the divergence under test is absent"
+        );
+
+        md.on_replicate(prepare).await;
+
+        assert!(
+            journal.header(1).is_some(),
+            "backup dropped the prepare its journal needed next because its sequencer \
+             was ahead; the primary's retransmit of this op can never be accepted, so \
+             the op never commits and its client never gets a reply"
+        );
+    }
+
+    /// A state-transfer receiver admits the first live prepare above the floor it
+    /// installed, instead of waiting for an op the snapshot already contains.
+    ///
+    /// `install_state_transfer` moves the snapshot floor, the commit floor, the
+    /// sequencer and `commit_max`, and leaves the WAL head where it was: the
+    /// snapshot IS every op below the floor, so there is nothing left to append
+    /// for them. A gap check reading `last_op` alone therefore has the receiver
+    /// ask for `last_op + 1`, an op inside the snapshot that no peer will send
+    /// again, and it drops every live prepare forever.
+    ///
+    /// Nothing recovers it. An offer built on a quiet cluster carries `commit_op
+    /// == snapshot_seq`, so the install lands `commit_min == commit_max` and
+    /// `maybe_request_metadata_repair`, gated on `commit_min < commit_max`, never
+    /// arms; repair is the only path that could refill the head. With the
+    /// other backup down the primary needs this replica's ack to commit at all,
+    /// so a 3-node cluster still inside its quorum stops serving metadata.
+    #[compio::test]
+    async fn state_transfer_receiver_admits_the_first_prepare_above_the_installed_floor() {
+        const CLIENT: u128 = 1;
+        const SESSION: u64 = 1;
+        const ACTING_USER: u32 = 7;
+        /// The `snapshot_seq` of a transferred offer, far above anything this
+        /// replica's own WAL holds.
+        const INSTALLED_FLOOR: u64 = 400;
+
+        let dir = tempfile::tempdir().unwrap();
+        std::fs::create_dir_all(dir.path().join(crate::impls::METADATA_DIR)).unwrap();
+        let journal =
+            journal::prepare_journal::PrepareJournal::open(&dir.path().join("journal.wal"), 0)
+                .await
+                .unwrap();
+        // Replica 1 of 3 at view 0, so `primary_index(0) == 0` makes this a backup
+        // and `on_replicate` takes the gap-check branch.
+        let consensus = VsrConsensus::new(
+            1,
+            1,
+            3,
+            server_common::sharding::METADATA_GROUP,
+            NoopBus,
+            LocalPipeline::new(),
+        );
+        consensus.init();
+        let md: IggyMetadata<_, journal::prepare_journal::PrepareJournal, (), TestMux> =
+            IggyMetadata::new(
+                Some(consensus),
+                Some(journal),
+                None,
+                None,
+                TestMux::default(),
+                Some(dir.path().to_path_buf()),
+            );
+        let consensus = md.consensus.as_ref().unwrap();
+        assert!(
+            consensus.is_follower(),
+            "replica 1 of 3 at view 0 must be a backup for this to exercise the gap check"
+        );
+        md.client_table.borrow_mut().commit_register(
+            CLIENT,
+            ACTING_USER,
+            register_reply(CLIENT, SESSION),
+        );
+
+        // Give the WAL a head far below the floor about to be installed, which is
+        // what a replica that fell behind its peers' retention actually carries.
+        let first = md
+            .prepare_request(create_stream_request(CLIENT, 1, "s1"))
+            .expect("CreateStream is client-allowed");
+        md.on_replicate(first).await;
+        let journal = md.journal.as_ref().unwrap();
+        assert_eq!(
+            journal.last_op(),
+            Some(1),
+            "the WAL head must sit below the installed floor, else the divergence \
+             under test is absent"
+        );
+
+        // Exactly the frontiers `install_state_transfer` leaves for a quiet-cluster
+        // offer, where the manifest's `commit_op` equals its `snapshot_seq`.
+        journal.set_snapshot_op(INSTALLED_FLOOR);
+        consensus.set_commit_floor(INSTALLED_FLOOR);
+        consensus.sequencer().set_sequence(INSTALLED_FLOOR);
+        consensus.advance_commit_max(INSTALLED_FLOOR);
+        assert_eq!(
+            consensus.commit_min(),
+            consensus.commit_max(),
+            "the wedge needs an install with no repair left to arm; diverged \
+             frontiers heal through `maybe_request_metadata_repair`"
+        );
+
+        let next = md
+            .prepare_request(create_stream_request(CLIENT, 2, "s2"))
+            .expect("CreateStream is client-allowed");
+        assert_eq!(
+            next.header().op,
+            INSTALLED_FLOOR + 1,
+            "the primary numbers the next op off the installed floor"
+        );
+
+        md.on_replicate(next).await;
+
+        assert!(
+            journal
+                .header(usize::try_from(INSTALLED_FLOOR + 1).unwrap())
+                .is_some(),
+            "state-transfer receiver dropped the first prepare above its installed \
+             floor; the ops the snapshot already holds are never re-sent, and with \
+             `commit_min == commit_max` no repair arms, so this op never commits"
+        );
+    }
+
     /// A checkpoint reclaims the WAL prefix the snapshot supersedes, but must
     /// stop one op short of the checkpoint op itself.
     ///

core/metadata/src/impls/mod.rs

@@ -20,3 +20,8 @@ pub mod recovery;
 
 /// Subdirectory under the data root where metadata state is stored.
 pub const METADATA_DIR: &str = "metadata";
+
+/// File under [`METADATA_DIR`] holding the persisted snapshot, and the artifact
+/// state transfer serves. Public so a reader outside this crate (the simulator
+/// asserting a checkpoint landed) can name it.
+pub const SNAPSHOT_FILE_NAME: &str = "snapshot.bin";