ship/replica: delete untested legacy restore fallback, close #618's mutation-verified test gaps

.cargo/mutants.toml

@@ -69,4 +69,51 @@ exclude_re = [
     # mutants here (which DO change the addend when `tf > 0.0` and
     # are caught), only the `>=` variant is this tautology.
     "replace > with >= in Bm25Index::explain",
+    # `newest_fence`'s max-so-far comparison (issue #618): every fence
+    # object's name is that generation's decimal key, written at most
+    # once each (`claim`'s `PutMode::Create` refuses a second write to
+    # the same name), so one listing pass can never present the same
+    # generation twice — `fence.generation == generation` never holds
+    # mid-loop, and `<=` computes the identical `newest` to `<` for
+    # every possible listing.
+    "replace < with <= in newest_fence",
+    # `ReplicateConfig::from_env`'s interval floor (issue #618): at
+    # `requested == 100` (the only value where `<` and `<=` disagree),
+    # both arms compute the same `Duration::from_millis(100)` — the
+    # floor branch explicitly, the else branch because `requested`
+    # already IS 100 — so the mutant produces the identical `interval`
+    # for every possible input; only the warn log (uncaptured by any
+    # test here) would differ.
+    "replace < with <= in ReplicateConfig::from_env",
+    # `Shipper::ship_lane`'s THREE identical `vanished_mid_cycle` call
+    # sites (issue #618, shipper.rs:390, :426, :459) — line-anchored
+    # since the guard expression is byte-identical at all three.
+    # :390's "false" arm (a genuine NotFound reaching this guard IS
+    # the vanished-mid-cycle case) is real-tested — see
+    # a_lane_deleted_between_the_scan_and_its_own_turn_ships_nothing_not_an_error,
+    # which races an earlier `scan.changed` upload to delete the lane
+    # file before the lanes loop reaches it. Excluded here:
+    # - :390 "true": distinguishing this needs a LOCAL fs error at
+    #   this exact point that is NOT NotFound (permission denied,
+    #   ELOOP, ...) — this suite has no fault-injection wrapper for
+    #   `std::fs` calls (only for the `ObjectStore` trait), and the
+    #   two portable ways to fake one — chmod-ing the containing
+    #   directory, or a self-referential symlink — either break the
+    #   SAME pausing mechanism :390's own test relies on (chmod blocks
+    #   every other file under it, including the upload being paused
+    #   on) or are platform-fragile.
+    # - :426 and :459: no `.await` exists between `ship_lane`'s own
+    #   entry and reaching them (metadata → sig check → read → prefix
+    #   check → this metadata, all synchronous) — unlike :390, nothing
+    #   running on this cooperative single-threaded executor can ever
+    #   land between two back-to-back non-yielding `std::fs` calls
+    #   within the SAME `ship_lane` invocation.
+    # A real concurrent unlink/permission change at the OS level could
+    # still hit any of these windows in production; no test here can
+    # pin that.
+    "src/ship/shipper\\.rs:390:27: replace match guard vanished_mid_cycle\\(&error\\) with true in Shipper::ship_lane",
+    "src/ship/shipper\\.rs:426:31: replace match guard vanished_mid_cycle\\(&error\\) with true in Shipper::ship_lane",
+    "src/ship/shipper\\.rs:426:31: replace match guard vanished_mid_cycle\\(&error\\) with false in Shipper::ship_lane",
+    "src/ship/shipper\\.rs:459:35: replace match guard vanished_mid_cycle\\(&error\\) with true in Shipper::ship_lane",
+    "src/ship/shipper\\.rs:459:35: replace match guard vanished_mid_cycle\\(&error\\) with false in Shipper::ship_lane",
 ]

src/registry/replication.rs

@@ -509,4 +509,130 @@ mod tests {
 
         let _ = fs::remove_dir_all(dir);
     }
+
+    /// #618: `replica_register` was never directly exercised — only
+    /// ever called from `Tailer::poll_once`. The hydrator's shared
+    /// pass lands the meta before this runs, so registration is a
+    /// pure in-memory step reading a file that already exists.
+    #[test]
+    fn a_replica_register_adds_a_new_context_from_its_landed_meta() {
+        let dir = scratch_dir("replica-register-new");
+        let state = AppState::boot_with(
+            dir.clone(),
+            usize::MAX,
+            None,
+            BootOptions {
+                replica: Some(std::sync::Arc::new(crate::replica::ReplicaInfo::new(None))),
+                ..BootOptions::default()
+            },
+        )
+        .unwrap();
+        assert!(
+            state.lookup("sake").is_none(),
+            "nothing registered before the tailer touches it"
+        );
+
+        let stem = file_stem("sake");
+        fs::write(
+            dir.join(format!("{stem}.meta.json")),
+            br#"{"description":"sake","pinned":false}"#,
+        )
+        .unwrap();
+        state.replica_register(&stem);
+        assert_eq!(
+            state
+                .directory_entry("sake")
+                .expect("the stem must register")
+                .description,
+            "sake"
+        );
+
+        // Idempotent: a second registration of the same stem must not
+        // REPLACE the entry — `lookup` alone would pass even if it did
+        // (a fresh entry with the same name still looks up fine), so
+        // this pins the actual identity via `Arc::ptr_eq`.
+        let first = state.lookup("sake").expect("just registered");
+        state.replica_register(&stem);
+        let second = state.lookup("sake").expect("still registered");
+        assert!(
+            std::sync::Arc::ptr_eq(&first, &second),
+            "a repeat registration of an already-registered stem must not replace the entry"
+        );
+
+        let _ = fs::remove_dir_all(dir);
+    }
+
+    /// #618: an undecodable stem (never a name this server itself
+    /// wrote) must be a silent no-op, not a panic — the tailer's own
+    /// worklist loop already filters these via `name_from_stem`, but
+    /// `replica_register` guards against it independently too.
+    #[test]
+    fn a_replica_register_ignores_an_undecodable_stem() {
+        let dir = scratch_dir("replica-register-undecodable");
+        let state = AppState::boot_with(
+            dir.clone(),
+            usize::MAX,
+            None,
+            BootOptions {
+                replica: Some(std::sync::Arc::new(crate::replica::ReplicaInfo::new(None))),
+                ..BootOptions::default()
+            },
+        )
+        .unwrap();
+        // `context_count` is the actual registry `replica_register`
+        // writes into — `group_page` is a different subsystem and
+        // would pass even if this call registered something.
+        //
+        // `name_from_stem` only refuses a `%`-escape it cannot decode
+        // (an odd/invalid hex pair, or a `%` with nothing — or too
+        // little — after it): a plain string with no `%` at all
+        // decodes to itself unchanged, so it is NOT the "undecodable"
+        // case this test means to cover — `"trailing%"` genuinely is,
+        // its dangling `%` running out of input mid-escape.
+        let before = state.context_count();
+        state.replica_register("trailing%");
+        assert_eq!(
+            state.context_count(),
+            before,
+            "an undecodable stem must not register anything"
+        );
+        let _ = fs::remove_dir_all(dir);
+    }
+
+    /// #618: `replica_deregister` was never directly exercised — only
+    /// ever called from `Tailer::poll_once` for a vanished lineage
+    /// member.
+    #[test]
+    fn a_replica_deregister_removes_a_registered_context() {
+        let dir = scratch_dir("replica-deregister");
+        let state = AppState::boot_with(
+            dir.clone(),
+            usize::MAX,
+            None,
+            BootOptions {
+                replica: Some(std::sync::Arc::new(crate::replica::ReplicaInfo::new(None))),
+                ..BootOptions::default()
+            },
+        )
+        .unwrap();
+        let stem = file_stem("sake");
+        fs::write(
+            dir.join(format!("{stem}.meta.json")),
+            br#"{"description":"sake","pinned":false}"#,
+        )
+        .unwrap();
+        state.replica_register(&stem);
+        assert!(state.lookup("sake").is_some());
+
+        state.replica_deregister("sake");
+        assert!(
+            state.lookup("sake").is_none(),
+            "the lineage no longer carrying this context must drop it in memory"
+        );
+
+        // A name never registered: a no-op, not a panic.
+        state.replica_deregister("never-registered");
+
+        let _ = fs::remove_dir_all(dir);
+    }
 }