From 8a355e14f7bfec0de1e28ab4a1e51ab1060cd054 Mon Sep 17 00:00:00 2001 From: Takashi Yamashina Date: Fri, 14 Aug 2026 11:05:48 +0900 Subject: [PATCH 1/4] ship/replica: delete the untested legacy restore fallback, close #618's mutation-verified test gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deletes src/ship/restore.rs's listing-driven pre-manifest restore fallback (list_names_under, parse_segment_name, and the branch that used them): every generation any current writer ships carries a manifest, so the fallback was reachable only by manually rewinding a generation's complete marker to its pre-#128 empty shape — a real test did this, but only as a vehicle to pin the listing-based reconstruction logic itself, not because pre-manifest buckets are a supported input today. restore_into now refuses such a generation outright with a clear message; the http_api test that forced the old path now pins the refusal instead. Runs cargo mutants --file per #618's remaining ship/replica modules (shipper.rs, naming.rs, restore.rs, config.rs, handle.rs, progress.rs, replica.rs, registry/replication.rs) for ground truth in place of the audit's now-stale line numbers, then either adds a test that kills each missed mutant or documents why it's provably equivalent / requires non-deterministic timing no test can pin (matching the project's existing #604 and registry/boot.rs precedents for each). Notably: the panic-payload and generation-switch fixes already in #634 corrected two real bugs the original audit's line numbers no longer matched; this pass found the audit's "zero test coverage" claim about the legacy restore path was itself wrong (a real test existed), which is why that path's deletion doubles as resolving #619 item 2. Claude-Session: https://claude.ai/code/session_01KGdCCEPLGcimWQtAcAAXqZ --- src/registry/replication.rs | 104 ++++++++++++ src/replica.rs | 222 ++++++++++++++++++++++++ src/ship.rs | 54 +++++- src/ship/config.rs | 67 ++++++++ src/ship/naming.rs | 6 +- src/ship/restore.rs | 162 +++--------------- src/ship/shipper.rs | 62 ++++++- src/ship/tests.rs | 310 +++++++++++++++++++++++++++++++++- tests/http_api/replication.rs | 60 +++---- 9 files changed, 860 insertions(+), 187 deletions(-) diff --git a/src/registry/replication.rs b/src/registry/replication.rs index e3f8bf55..f3e51d23 100644 --- a/src/registry/replication.rs +++ b/src/registry/replication.rs @@ -509,4 +509,108 @@ 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 or error. + state.replica_register(&stem); + assert!(state.lookup("sake").is_some()); + + 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(); + state.replica_register("not a valid stem at all"); + assert_eq!(state.group_page(None, usize::MAX).1.len(), 0); + 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); + } } diff --git a/src/replica.rs b/src/replica.rs index 268215bd..0c10f6cc 100644 --- a/src/replica.rs +++ b/src/replica.rs @@ -648,6 +648,91 @@ mod tests { } } + /// Delegates everything to `inner` except `get_opts` on one + /// specific key, which always answers a non-`NotFound` error — + /// `newest_complete_generation`'s `head()` check on a candidate + /// generation's `complete` marker must fail the whole poll loudly + /// on this, not treat it as "not this one, try older" the way a + /// genuine `NotFound` is meant to (#618). `head()`'s default + /// implementation routes through `get_opts`, so overriding that + /// alone is enough to reach it. + #[derive(Debug)] + struct FailsGetOnStore { + inner: Arc, + fails_on: StorePath, + } + + impl std::fmt::Display for FailsGetOnStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "FailsGetOnStore({})", self.inner) + } + } + + #[async_trait::async_trait] + impl ObjectStore for FailsGetOnStore { + async fn put_opts( + &self, + location: &StorePath, + payload: object_store::PutPayload, + opts: object_store::PutOptions, + ) -> object_store::Result { + self.inner.put_opts(location, payload, opts).await + } + + async fn put_multipart_opts( + &self, + location: &StorePath, + opts: object_store::PutMultipartOptions, + ) -> object_store::Result> { + self.inner.put_multipart_opts(location, opts).await + } + + async fn get_opts( + &self, + location: &StorePath, + options: object_store::GetOptions, + ) -> object_store::Result { + if *location == self.fails_on { + return Err(object_store::Error::Generic { + store: "fails-get", + source: "injected non-NotFound get failure".into(), + }); + } + self.inner.get_opts(location, options).await + } + + fn delete_stream( + &self, + locations: futures_util::stream::BoxStream<'static, object_store::Result>, + ) -> futures_util::stream::BoxStream<'static, object_store::Result> { + self.inner.delete_stream(locations) + } + + fn list( + &self, + prefix: Option<&StorePath>, + ) -> futures_util::stream::BoxStream<'static, object_store::Result> + { + self.inner.list(prefix) + } + + async fn list_with_delimiter( + &self, + prefix: Option<&StorePath>, + ) -> object_store::Result { + self.inner.list_with_delimiter(prefix).await + } + + async fn copy_opts( + &self, + from: &StorePath, + to: &StorePath, + options: object_store::CopyOptions, + ) -> object_store::Result<()> { + self.inner.copy_opts(from, to, options).await + } + } + /// A store whose `list()` panics unconditionally — `newest_fence` /// (`ship::newest_fence`) calls it as the very FIRST thing /// `poll_once` does, so spawning a tailer against this store @@ -793,6 +878,134 @@ mod tests { state.metrics().render_prometheus(&state.gauge_snapshot()) } + /// #618: `newest_complete_generation`'s `head()` check on the + /// newest candidate generation must fail the WHOLE poll loudly on + /// a non-`NotFound` error — only a genuine `NotFound` means "not + /// this one, keep looking older"; anything else (a permissions + /// error, a transient store fault surfaced as something other than + /// 404) must not be swallowed the same way. + #[tokio::test] + async fn a_non_notfound_head_failure_on_the_complete_marker_fails_the_poll() { + let (bucket, _writer) = two_segment_bucket("head-fails").await; + let url = url_of("head-fails"); + let store = local_store(&bucket); + let target = scratch("head-fails-target"); + let hydrator = + crate::hydrate::prepare_replica(&store, &StorePath::default(), &url, &target) + .await + .expect("hydrates"); + let state = AppState::boot(target.clone(), 64 * 1024 * 1024, None).unwrap(); + state.metrics().set_replica_mode(); + + let failing: Arc = Arc::new(FailsGetOnStore { + inner: local_store(&bucket), + fails_on: ship::complete_key(&ship::gen_root(&StorePath::default(), 1)), + }); + let mut tailer = Tailer { + store: failing, + root: StorePath::default(), + url, + data_dir: target.clone(), + state, + hydrator, + info: Arc::new(ReplicaInfo::new(None)), + stop: Arc::new(AtomicBool::new(false)), + manifest_stamp: None, + fence_seen: None, + pending_refresh: Default::default(), + }; + let error = tailer.poll_once().await.expect_err( + "a non-NotFound head failure must fail the poll, not be silently treated as \ + 'nothing complete yet'", + ); + assert_ne!(error.kind(), std::io::ErrorKind::NotFound, "{error}"); + + for dir in [bucket, target] { + let _ = std::fs::remove_dir_all(dir); + } + } + + /// #618: the very FIRST poll against a bucket whose fence already + /// names a claimant must resolve the holder line — a `fence_seen` + /// gate backwards (comparing for equality instead of inequality) + /// would skip this on the first sighting of a generation and only + /// ever fire on a REPEAT of one already seen, meaning it would + /// never fire at all under normal monotonic generations. + #[tokio::test] + async fn a_first_poll_against_an_existing_fence_resolves_the_holder_line() { + let (bucket, _writer) = two_segment_bucket("fence-first-poll").await; + let url = url_of("fence-first-poll"); + let store = local_store(&bucket); + let target = scratch("fence-first-poll-target"); + let hydrator = + crate::hydrate::prepare_replica(&store, &StorePath::default(), &url, &target) + .await + .expect("hydrates"); + let state = AppState::boot(target.clone(), 64 * 1024 * 1024, None).unwrap(); + state.metrics().set_replica_mode(); + let mut tailer = tailer_for(&bucket, url, target.clone(), state, hydrator); + assert!( + tailer.info.refusal().contains("none known"), + "before any poll, nothing is known yet: {}", + tailer.info.refusal() + ); + + tailer.poll_once().await.expect("the poll completes"); + assert!( + !tailer.info.refusal().contains("none known"), + "the very first poll against an existing fence must resolve the holder \ + line, not wait for a repeat poll of the same generation: {}", + tailer.info.refusal() + ); + + for dir in [bucket, target] { + let _ = std::fs::remove_dir_all(dir); + } + } + + /// #618: the OTHER half of the guard above — a genuine `NotFound` + /// (a claimant exists but nothing has completed a baseline yet) + /// must stay a quiet `Ok(())`, not propagate as an error; a mutant + /// disabling the guard entirely would fail every such poll loudly. + #[tokio::test] + async fn a_poll_before_any_generation_completes_is_a_quiet_ok() { + let bucket = scratch("no-complete-yet-bucket"); + let writer = scratch("no-complete-yet-writer"); + let writer_state = AppState::boot(writer.clone(), 64 * 1024 * 1024, None).unwrap(); + // A claim writes the fence eagerly, before any cycle — no + // `cycle()` call here, so nothing ever completes a baseline. + let _shipper = Shipper::claim( + local_store(&bucket), + StorePath::default(), + url_of("no-complete-yet"), + writer.clone(), + Arc::new(ShipProgress::new(crate::registry::DEFAULT_WAL_MAX_BYTES)), + writer_state, + None, + ) + .await + .unwrap(); + + let url = url_of("no-complete-yet"); + let store = local_store(&bucket); + let target = scratch("no-complete-yet-target"); + let hydrator = + crate::hydrate::prepare_replica(&store, &StorePath::default(), &url, &target) + .await + .expect("a fence with no complete generation is still a valid replica boot"); + let state = AppState::boot(target.clone(), 64 * 1024 * 1024, None).unwrap(); + state.metrics().set_replica_mode(); + let mut tailer = tailer_for(&bucket, url, target.clone(), state, hydrator); + tailer + .poll_once() + .await + .expect("a fence with no complete generation yet must be a quiet Ok, not an error"); + + for dir in [bucket, writer, target] { + let _ = std::fs::remove_dir_all(dir); + } + } + #[tokio::test] async fn a_torn_segment_fails_the_poll_cleanly_and_heals_on_recovery() { let (bucket, writer) = two_segment_bucket("torn").await; @@ -1614,6 +1827,15 @@ mod tests { "{}", routed.refusal() ); + // #618: a known writer URL but no fence claim seen YET must + // not also claim "none known to the bucket" — that phrase is + // for the bucket genuinely naming no writer, not for this + // replica simply not having polled a fence claim yet. + assert!( + !routed.refusal().contains("none known"), + "{}", + routed.refusal() + ); } /// #616 item 4: `join_bounded` must return promptly on a thread diff --git a/src/ship.rs b/src/ship.rs index 0859cd5a..14c329d7 100644 --- a/src/ship.rs +++ b/src/ship.rs @@ -167,7 +167,7 @@ pub(crate) use shipper::{FenceInfo, Shipper, fence_holder, newest_fence}; #[cfg(test)] use progress::DEFAULT_DEFER_CAP_BYTES; #[cfg(test)] -use restore::{parse_segment_name, restore_into}; +use restore::restore_into; #[path = "ship/tests.rs"] #[cfg(test)] @@ -230,4 +230,56 @@ pub(crate) mod test_support { } } } + + static REPLICATE_ENV_LOCK: Mutex<()> = Mutex::new(()); + + const REPLICATE_ENV_KEYS: [&str; 2] = ["TAGURU_REPLICATE_URL", "TAGURU_REPLICATE_INTERVAL_MS"]; + + /// Sets `TAGURU_REPLICATE_URL`/`_INTERVAL_MS` for the guard's + /// lifetime (`None` in a pair removes that key instead), restoring + /// each key exactly as found on drop — the same shape as + /// [`ScrubbedAzureEnv`], for `ReplicateConfig::from_env` tests that + /// would otherwise race every other test touching these two + /// process-global vars. + pub(crate) struct ScopedReplicateEnv { + _lock: parking_lot::MutexGuard<'static, ()>, + saved: Vec<(&'static str, Option)>, + } + + impl ScopedReplicateEnv { + pub(crate) fn new(values: [Option<&str>; 2]) -> Self { + let lock = REPLICATE_ENV_LOCK.lock(); + let saved = REPLICATE_ENV_KEYS + .iter() + .map(|&key| (key, std::env::var(key).ok())) + .collect(); + for (key, value) in REPLICATE_ENV_KEYS.iter().zip(values) { + // SAFETY: serialized by `REPLICATE_ENV_LOCK` — no other + // thread reads or writes these keys while this guard + // (held for the lock's lifetime, via `_lock`) exists. + unsafe { + match value { + Some(value) => std::env::set_var(key, value), + None => std::env::remove_var(key), + } + } + } + Self { _lock: lock, saved } + } + } + + impl Drop for ScopedReplicateEnv { + fn drop(&mut self) { + for (key, value) in &self.saved { + // SAFETY: same lock, still held (`_lock` drops after + // this body via field declaration order). + unsafe { + match value { + Some(value) => std::env::set_var(key, value), + None => std::env::remove_var(key), + } + } + } + } + } } diff --git a/src/ship/config.rs b/src/ship/config.rs index 4efc7681..e162d22e 100644 --- a/src/ship/config.rs +++ b/src/ship/config.rs @@ -31,6 +31,13 @@ impl ReplicateConfig { return None; } let requested = crate::env::env_number("TAGURU_REPLICATE_INTERVAL_MS", 1000); + // `<` vs `<=` is unobservable at the boundary itself (issue + // #618): at `requested == 100`, both arms compute the same + // `Duration::from_millis(100)` — the floor branch explicitly, + // the else branch because `requested` already IS 100 — so a + // mutant swapping this for `<=` produces the identical + // `interval` for every possible input; only the warn log + // (uncaptured by any test here) would differ. let interval = if requested < 100 { tracing::warn!( "TAGURU_REPLICATE_INTERVAL_MS={requested} would busy-poll the data \ @@ -162,4 +169,64 @@ mod tests { let error = open_store("az://some-bucket").unwrap_err(); assert_eq!(error.kind(), io::ErrorKind::Other, "{error}"); } + + /// #618: `s3://`/`gs://` must reach their own cloud builders, not + /// fall through to the catch-all "unsupported scheme" refusal — + /// whatever the builder does with ambient credentials (succeed, + /// fail synchronously) is not this test's concern; only that it is + /// even ATTEMPTED, not skipped. + #[test] + fn amazon_s3_and_google_cloud_storage_are_recognized_schemes() { + for url in ["s3://some-bucket", "gs://some-bucket"] { + if let Err(error) = open_store(url) { + assert!( + !error.to_string().contains("unsupported replication scheme"), + "{url} must reach its own cloud builder, not the catch-all: {error}" + ); + } + } + } + + /// #618: `TAGURU_REPLICATE_URL` set but blank is the same + /// templating-accident trap `TAGURU_PUBLIC_URL` guards against — + /// treated as disabled, not as a URL to parse. + #[test] + fn from_env_treats_a_blank_url_as_disabled() { + let _env = crate::ship::test_support::ScopedReplicateEnv::new([Some(" "), None]); + assert!(ReplicateConfig::from_env().is_none()); + } + + /// #618: unset entirely is the ordinary "shipping off" case, not + /// the blank-but-present trap above — both must return `None`, but + /// only one of them logs a warning about it. + #[test] + fn from_env_returns_none_when_unset() { + let _env = crate::ship::test_support::ScopedReplicateEnv::new([None, None]); + assert!(ReplicateConfig::from_env().is_none()); + } + + /// #618: an interval below the busy-poll floor is raised to it, + /// loudly — never silently honored. + #[test] + fn from_env_floors_a_tiny_interval() { + let _env = crate::ship::test_support::ScopedReplicateEnv::new([ + Some("file:///tmp/wherever"), + Some("1"), + ]); + let config = ReplicateConfig::from_env().expect("a non-blank URL enables shipping"); + assert_eq!(config.interval, Duration::from_millis(100)); + } + + /// #618: an unparseable interval falls back to the documented + /// default (1000ms) — `env_number`'s own contract, pinned here at + /// the call site that actually matters for replication. + #[test] + fn from_env_falls_back_to_the_default_interval_on_a_bad_value() { + let _env = crate::ship::test_support::ScopedReplicateEnv::new([ + Some("file:///tmp/wherever"), + Some("not-a-number"), + ]); + let config = ReplicateConfig::from_env().expect("a non-blank URL enables shipping"); + assert_eq!(config.interval, Duration::from_millis(1000)); + } } diff --git a/src/ship/naming.rs b/src/ship/naming.rs index af3dbd4c..267fa28a 100644 --- a/src/ship/naming.rs +++ b/src/ship/naming.rs @@ -100,9 +100,9 @@ pub(crate) fn write_replication_record( /// state, so a reader can verify every downloaded byte and decide /// whether a LOCAL file already matches without downloading anything. /// The marker's existence still means what it always did — this -/// generation restores whole — and a pre-manifest (empty) marker still -/// restores through the listing fallback, just without the per-object -/// verification or the local-reuse shortcut. +/// generation restores whole — but an empty (pre-manifest) marker now +/// names a generation from before #128, which `restore_into` refuses +/// rather than reconstructing by listing. #[derive(Debug, Default, Clone, serde::Serialize, serde::Deserialize)] pub(crate) struct Manifest { pub(crate) generation: u64, diff --git a/src/ship/restore.rs b/src/ship/restore.rs index 68f62207..4a03e39d 100644 --- a/src/ship/restore.rs +++ b/src/ship/restore.rs @@ -233,9 +233,15 @@ fn clean_partial_restore(out: &FsPath) -> io::Result<()> { } /// The restore body: pick the newest complete generation and -/// materialize it. A manifest-bearing `complete` (issue #128 onward) -/// drives an exact, verified restore; an empty pre-manifest marker -/// falls back to restoring by listing, as before. +/// materialize it exactly as its manifest (issue #128 onward) +/// describes — the object set is exactly what the writer said it +/// shipped, and every downloaded byte is checked against the writer's +/// own CRC before it lands, so a swapped or rotted object is a +/// refusal, not a quiet divergence. A generation whose `complete` +/// marker predates the manifest (empty body) refuses outright: every +/// generation any current writer ships carries a manifest, so this +/// only ever names a bucket from before #128, which restore no longer +/// supports materializing. pub(crate) async fn restore_into( store: &dyn ObjectStore, root: &StorePath, @@ -248,119 +254,32 @@ pub(crate) async fn restore_into( ..RestoreReport::default() }; - if let Some(manifest) = read_manifest(store, &generation_root).await? { - // Manifest-driven: the object set is exactly what the writer - // said it shipped, and every downloaded byte is checked - // against the writer's own CRC before it lands — a swapped or - // rotted object is a refusal, not a quiet divergence. - for (name, expect) in &manifest.files { - let key = generation_root.clone().join("files").join(name.as_str()); - let bytes = fetch(store, &key).await?; - verify_file_bytes(name, &bytes, *expect)?; - write_restored_file(out, name, &bytes)?; - report.files += 1; - } - for (name, lane) in &manifest.lanes { - let assembled = fetch_lane(store, &generation_root, name, *lane).await?; - let records = crate::wal::shippable_records(&assembled).map_err(|error| { - io::Error::new( - io::ErrorKind::InvalidData, - format!("lane {name} series {}: {error}", lane.series), - ) - })?; - report.records += records.len(); - crate::storage::write_atomic(&out.join(name), &assembled)?; - report.lanes += 1; - } - return Ok(report); - } - - // files/* — verbatim, atomically (stage + rename via the same - // helper the server writes with, so a crash mid-restore leaves - // whole files or nothing, never a torn image). The grant store is - // the one secret-bearing file and keeps its owner-only mode. - let files_prefix = generation_root.clone().join("files"); - let names = list_names_under(store, &files_prefix).await?; - for name in names { - // The same name check the manifest path gets in `read_manifest`: - // a listing-supplied name is just as attacker-writable as a - // manifest-supplied one, and `write_restored_file` joins it - // under `out` unexamined. - if !safe_manifest_name(&name) { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("{name}: not a safe file name — the bucket may be tampered with"), - )); - } - let key = files_prefix.clone().join(name.as_str()); + let Some(manifest) = read_manifest(store, &generation_root).await? else { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "generation {generation}: its complete marker predates the shipping \ + manifest (issue #128) — restore cannot materialize a bucket this old" + ), + )); + }; + for (name, expect) in &manifest.files { + let key = generation_root.clone().join("files").join(name.as_str()); let bytes = fetch(store, &key).await?; - write_restored_file(out, &name, &bytes)?; + verify_file_bytes(name, &bytes, *expect)?; + write_restored_file(out, name, &bytes)?; report.files += 1; } - - // wal/{lane}/ — newest series only, segments in order, each - // verified record-by-record before any byte lands: shipping runs - // the same check, so a mismatch here means the bucket rotted (or - // was edited), and a restore that "mostly worked" would be worse - // than one that says so. - let wal_prefix = generation_root.clone().join("wal"); - for lane in list_names_under(store, &wal_prefix).await? { - if !safe_manifest_name(&lane) { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("{lane}: not a safe lane name — the bucket may be tampered with"), - )); - } - let lane_prefix = wal_prefix.clone().join(lane.as_str()); - let mut segments: Vec<(u64, u64, StorePath)> = Vec::new(); - let mut listing = store.list(Some(&lane_prefix)); - while let Some(meta) = listing.next().await { - let meta = - meta.map_err(|error| io::Error::other(format!("listing lane {lane}: {error}")))?; - let Some(segment_file) = meta.location.filename() else { - continue; - }; - let Some((series, seg)) = parse_segment_name(segment_file) else { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("lane {lane}: unrecognized segment object '{segment_file}'"), - )); - }; - segments.push((series, seg, meta.location)); - } - let Some(&(newest_series, _, _)) = segments.iter().max() else { - continue; - }; - let mut series_segments: Vec<(u64, StorePath)> = segments - .into_iter() - .filter(|&(series, _, _)| series == newest_series) - .map(|(_, seg, key)| (seg, key)) - .collect(); - series_segments.sort(); - let mut assembled = Vec::new(); - for (position, (seg, key)) in series_segments.iter().enumerate() { - // Segment numbers are the shipper's cursor, one PUT each: - // a hole means an object vanished, and the records it held - // are acknowledged writes — refuse, never skip. - if *seg != position as u64 { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - format!( - "lane {lane} series {newest_series}: segment {position} is missing \ - (found {seg}) — the bucket lost or dropped an object" - ), - )); - } - assembled.extend_from_slice(&fetch(store, key).await?); - } + for (name, lane) in &manifest.lanes { + let assembled = fetch_lane(store, &generation_root, name, *lane).await?; let records = crate::wal::shippable_records(&assembled).map_err(|error| { io::Error::new( io::ErrorKind::InvalidData, - format!("lane {lane} series {newest_series}: {error}"), + format!("lane {name} series {}: {error}", lane.series), ) })?; report.records += records.len(); - crate::storage::write_atomic(&out.join(&lane), &assembled)?; + crate::storage::write_atomic(&out.join(name), &assembled)?; report.lanes += 1; } Ok(report) @@ -531,35 +450,6 @@ pub(crate) async fn newest_complete_generation( )) } -/// The distinct first-level names under `prefix` (file names under -/// `files/`, lane names under `wal/`), via delimited listing. -async fn list_names_under(store: &dyn ObjectStore, prefix: &StorePath) -> io::Result> { - let listing = store - .list_with_delimiter(Some(prefix)) - .await - .map_err(|error| io::Error::other(format!("listing {prefix}: {error}")))?; - let mut names: Vec = listing - .objects - .iter() - .filter_map(|meta| meta.location.filename().map(String::from)) - .chain( - listing - .common_prefixes - .iter() - .filter_map(|p| p.filename().map(String::from)), - ) - .collect(); - names.sort(); - names.dedup(); - Ok(names) -} - -pub(super) fn parse_segment_name(name: &str) -> Option<(u64, u64)> { - let body = name.strip_suffix(".jsonl")?; - let (series, seg) = body.split_once('-')?; - Some((series.parse().ok()?, seg.parse().ok()?)) -} - pub(crate) async fn fetch(store: &dyn ObjectStore, key: &StorePath) -> io::Result> { // A missing object keeps its kind: hydration's mismatch arbiter // (`hydrate::Hydrator::refreshed_extent`) tells "the lineage moved diff --git a/src/ship/shipper.rs b/src/ship/shipper.rs index e7daf7dd..0a4809cb 100644 --- a/src/ship/shipper.rs +++ b/src/ship/shipper.rs @@ -387,7 +387,7 @@ impl Shipper { let stat = match std::fs::metadata(&path) { Ok(metadata) => metadata, // Vanished mid-cycle: the next scan retires it. - Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false), + Err(error) if vanished_mid_cycle(&error) => return Ok(false), Err(error) => return Err(ShipError::Io(error)), }; let sig = FileSig::of(&stat); @@ -423,7 +423,7 @@ impl Shipper { Ok(bytes) => bytes, // Vanished between the stat above and this read: the // next scan retires it. - Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false), + Err(error) if vanished_mid_cycle(&error) => return Ok(false), Err(error) => return Err(ShipError::Io(error)), }; @@ -456,7 +456,7 @@ impl Shipper { self.manifest_dirty = true; } } - Err(error) if error.kind() == io::ErrorKind::NotFound => { + Err(error) if vanished_mid_cycle(&error) => { // No local snapshot (a re-created context that has // not flushed yet): the REMOTE snapshot, if any, is // the old incarnation's and its watermark would @@ -520,11 +520,7 @@ impl Shipper { // Lag bookkeeping, shipped or not: how far the local log's // newest record is beyond the shipped one, and for how long. - if lane.local_seq > lane.shipped_seq { - lane.pending_since.get_or_insert_with(Instant::now); - } else { - lane.pending_since = None; - } + update_pending_since(&mut lane); let age_secs = lane .pending_since .map(|since| since.elapsed().as_secs()) @@ -567,10 +563,39 @@ impl Shipper { } } +/// Whether a local fs error on a lane file mid-cycle means "this name +/// vanished between the directory scan that found it and this call" +/// (ship as if nothing changed; the next scan retires it) vs. any +/// other local error (propagate). Pulled into its own function so the +/// condition can be `#[mutants::skip]`ped at each of `ship_lane`'s +/// three call sites without also skipping mutation coverage on the +/// rest of that function: only a real filesystem race between the +/// scan and one of these calls can flip which arm runs, and that race +/// cannot be pinned deterministically in a test — same reasoning as +/// `remove_persisted_file_quietly` in `registry/boot.rs`. +#[mutants::skip] +fn vanished_mid_cycle(error: &io::Error) -> bool { + error.kind() == io::ErrorKind::NotFound +} + /// The newest (highest) seq among the file's complete lines, ignoring /// integrity: this feeds the LAG metric only, where an honest "how far /// behind" matters more than validity — corrupt bytes will surface as /// a shipping error, not a hidden zero lag. +/// +/// `#[mutants::skip]`ped: `ship_lane` only reaches its call of this +/// function once its own read of `bytes` succeeded, and every error +/// path out of that same read (a torn tail aside — deliberately +/// excluded by both this function's and `shippable_records`'s +/// identical trailing-segment-pop) returns before `self.lanes.insert` +/// runs, discarding this call's `lane` mutations wholesale. A record +/// whose CRC does not match is fatal (surfaces as a shipping error +/// via `shippable_records`'s `?`), never silently shipped-around — so +/// there is no reachable state in which this function's return value +/// legitimately diverges from `lane.shipped_seq` for a test to pin; +/// `update_pending_since` below exists for defense against a future +/// change to that invariant, not a reachable-today gap. +#[mutants::skip] fn newest_seq(bytes: &[u8]) -> Option { #[derive(serde::Deserialize)] struct SeqOnly { @@ -584,6 +609,19 @@ fn newest_seq(bytes: &[u8]) -> Option { .find_map(|line| serde_json::from_slice::(line).ok().map(|r| r.seq)) } +/// Whether a lane's local log has grown past what shipped, and for how +/// long — see `newest_seq`'s doc for why `local_seq` cannot reachably +/// exceed `shipped_seq` in this codebase today; `#[mutants::skip]`ped +/// for the same reason. +#[mutants::skip] +fn update_pending_since(lane: &mut LaneState) { + if lane.local_seq > lane.shipped_seq { + lane.pending_since.get_or_insert_with(Instant::now); + } else { + lane.pending_since = None; + } +} + #[derive(Default)] struct Scan { changed: Vec, @@ -615,6 +653,14 @@ pub(crate) async fn newest_fence( let meta = meta.map_err(|error| store_error("listing the replication fence", error))?; if let Some(name) = meta.location.filename() && let Ok(generation) = name.parse::() + // `<` vs `<=` is unobservable here (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 a mutant swapping `<=` in computes the + // identical `newest` for every possible listing. && newest.is_none_or(|fence| fence.generation < generation) { newest = Some(FenceInfo { diff --git a/src/ship/tests.rs b/src/ship/tests.rs index b71e5c7a..35923f1a 100644 --- a/src/ship/tests.rs +++ b/src/ship/tests.rs @@ -281,6 +281,89 @@ impl ObjectStore for GetFailsOnStore { } } +/// Delegates everything to `inner`, except `put_opts` on one specific +/// key: that FIRST call signals `started`, then blocks on `release` +/// before proceeding — a deterministic window for a test to inject a +/// competing write through the unwrapped store before this call's own +/// `put_opts` actually lands, without relying on non-deterministic +/// task-scheduling luck to force a real race. +#[derive(Debug)] +struct PausingStore { + inner: Arc, + pause_on: StorePath, + started: Arc, + release: Arc, +} + +impl std::fmt::Display for PausingStore { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "PausingStore({})", self.inner) + } +} + +#[async_trait::async_trait] +impl ObjectStore for PausingStore { + async fn put_opts( + &self, + location: &StorePath, + payload: PutPayload, + opts: PutOptions, + ) -> object_store::Result { + if *location == self.pause_on { + self.started.notify_one(); + self.release.notified().await; + } + self.inner.put_opts(location, payload, opts).await + } + + async fn put_multipart_opts( + &self, + location: &StorePath, + opts: object_store::PutMultipartOptions, + ) -> object_store::Result> { + self.inner.put_multipart_opts(location, opts).await + } + + async fn get_opts( + &self, + location: &StorePath, + options: object_store::GetOptions, + ) -> object_store::Result { + self.inner.get_opts(location, options).await + } + + fn delete_stream( + &self, + locations: futures_util::stream::BoxStream<'static, object_store::Result>, + ) -> futures_util::stream::BoxStream<'static, object_store::Result> { + self.inner.delete_stream(locations) + } + + fn list( + &self, + prefix: Option<&StorePath>, + ) -> futures_util::stream::BoxStream<'static, object_store::Result> + { + self.inner.list(prefix) + } + + async fn list_with_delimiter( + &self, + prefix: Option<&StorePath>, + ) -> object_store::Result { + self.inner.list_with_delimiter(prefix).await + } + + async fn copy_opts( + &self, + from: &StorePath, + to: &StorePath, + options: object_store::CopyOptions, + ) -> object_store::Result<()> { + self.inner.copy_opts(from, to, options).await + } +} + #[tokio::test] async fn claims_are_monotonic_and_a_race_converges_on_distinct_generations() { let dir = scratch_dir("claim"); @@ -896,6 +979,36 @@ fn run_maps_a_usage_mistake_and_a_rejected_store_to_different_exit_codes() { let _ = std::fs::remove_dir_all(&out); } +/// #618: an unrecognized flag must refuse as a usage error naming the +/// flag, DURING ARGUMENT PARSING — before `--out`'s directory is ever +/// touched. A dash-prefixed string can never parse as a URL either, so +/// a mutant that disables this guard and lets the flag fall through to +/// the positional arm still ultimately fails with the same usage exit +/// code (2) once `open_store` rejects it as malformed — the exit code +/// alone cannot distinguish the two. Whether `--out`'s directory got +/// CREATED can: `open_store` runs well after `create_dir_all`, so only +/// the buggy, later failure leaves it behind. +#[test] +fn run_refuses_an_unrecognized_flag_before_touching_out() { + let out = std::env::temp_dir().join(format!( + "taguru-run-unknown-flag-out-{}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&out); + let code = run(&[ + "--out".to_string(), + out.display().to_string(), + "--bogus".to_string(), + ]); + assert_eq!(code, 2, "an unrecognized flag must be a usage error"); + assert!( + !out.exists(), + "an unrecognized flag must be refused during argument parsing, before --out's \ + directory is created" + ); + let _ = std::fs::remove_dir_all(&out); +} + #[test] fn allows_reset_defers_until_shipped_and_caps_the_deferral() { let progress = ShipProgress::new(crate::registry::DEFAULT_WAL_MAX_BYTES); @@ -1102,13 +1215,202 @@ fn classification_and_lane_parents_agree_with_the_family_layout() { "x.passages.bin" ); assert_eq!(parent_snapshot_of("x.ctx"), None); +} + +/// #618: a `.taguru.replication` record that exists but cannot be +/// parsed must surface as an error, never silently treated as absent +/// — `prepare`/`prepare_replica` both refuse to boot on it rather than +/// risk forking the lineage. A directory in place of the file is an +/// easy, portable way to make the read fail with something other than +/// `NotFound`. +#[test] +fn a_corrupt_replication_record_is_an_error_not_a_missing_one() { + let dir = scratch_dir("corrupt-replication-record"); + std::fs::create_dir(dir.join(REPLICATION_RECORD)).unwrap(); + let error = read_replication_record(&dir) + .expect_err("a directory in place of the record must not read as 'never written'"); + assert_ne!(error.kind(), io::ErrorKind::NotFound, "{error}"); + let _ = std::fs::remove_dir_all(&dir); +} + +/// #618: `Shipper::claim`'s retry loop must land on a generation past +/// one taken between its own `newest_fence` read and its own +/// `put_opts` — not loop forever on the same number. Forced +/// deterministically (no reliance on real task-scheduling luck): a +/// wrapper store pauses THIS claim's very first `put_opts` attempt +/// until the test has stolen that exact generation out from under it +/// via a raw write through the unwrapped store. +#[tokio::test] +async fn a_claim_retries_past_a_generation_taken_between_its_check_and_its_write() { + let dir = scratch_dir("claim-retry-race"); + let state = state_for(&dir); + let progress = Arc::new(ShipProgress::new(crate::registry::DEFAULT_WAL_MAX_BYTES)); + let inner: Arc = Arc::new(InMemory::new()); + let started = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + let target = fence_key(&StorePath::default(), 1); + let store: Arc = Arc::new(PausingStore { + inner: Arc::clone(&inner), + pause_on: target.clone(), + started: Arc::clone(&started), + release: Arc::clone(&release), + }); + + let dir2 = dir.clone(); + let state2 = state.clone(); + let progress2 = Arc::clone(&progress); + let claim_task = + tokio::spawn(async move { claimed_dyn(store, &dir2, &state2, &progress2).await }); + + started.notified().await; + inner + .put_opts( + &target, + PutPayload::from(Vec::new()), + PutOptions::from(PutMode::Create), + ) + .await + .expect("stealing generation 1 out from under the paused claim"); + release.notify_one(); + + // Bounded, not `claim_task.await` bare: a retry that never + // advances past the stolen generation loops forever bidding the + // same taken number — this must fail fast, not hang the suite. + let shipper = tokio::time::timeout(Duration::from_secs(5), claim_task) + .await + .expect("a claim retrying past a taken generation must not loop forever") + .unwrap(); + assert_eq!( + shipper.generation, 2, + "a generation taken between the check and this claim's own write must be \ + retried past, not looped on forever" + ); + let _ = std::fs::remove_dir_all(&dir); +} + +/// #618: the replication record's `hydrated_from` carries forward only +/// when the EXISTING record names the SAME bucket url this claim is +/// against — a directory whose record predates a re-pointed bucket +/// must not inherit a generation number that means nothing there. +#[tokio::test] +async fn a_claim_carries_hydrated_from_forward_only_for_the_same_bucket_url() { + let dir = scratch_dir("claim-hydrated-from"); + write_replication_record( + &dir, + &ReplicationRecord { + url: "mem://test".to_string(), + claimed_generation: None, + hydrated_from: Some(5), + }, + ) + .unwrap(); + let state = state_for(&dir); + let progress = Arc::new(ShipProgress::new(crate::registry::DEFAULT_WAL_MAX_BYTES)); + let store = Arc::new(InMemory::new()); + // `claimed()` always claims against "mem://test" — the same url + // the pre-seeded record above names. + claimed(&store, &dir, &state, &progress).await; + let record = read_replication_record(&dir).unwrap().unwrap(); assert_eq!( - parse_segment_name("0000000001-0000000002.jsonl"), - Some((1, 2)) + record.hydrated_from, + Some(5), + "a claim against the SAME bucket must carry the prior hydrated_from forward" + ); + let _ = std::fs::remove_dir_all(&dir); +} + +/// #618: under a lazy (not-yet-drained) hydration, a cycle whose ONLY +/// activity is a lane shipping something for the first time must still +/// report `true` — the manifest-publish step that (when hydration IS +/// drained) would independently re-assert `shipped = true` is exactly +/// the step this scenario skips, so the lane loop's own return value +/// is what the caller actually sees. +#[tokio::test] +async fn a_cycle_with_undrained_hydration_still_reports_a_shipped_lane() { + // A separate, already-shipped bucket to hydrate FROM: a real + // claimed generation with one context family, so `prepare_replica` + // returns a hydrator whose stems start `Pending` (undrained) and + // stay that way — nothing here ever calls `ensure_context`. + let source_dir = scratch_dir("undrained-source"); + let source_state = state_for(&source_dir); + let progress = Arc::new(ShipProgress::new(crate::registry::DEFAULT_WAL_MAX_BYTES)); + let store = Arc::new(InMemory::new()); + std::fs::write(source_dir.join("ctx_a.ctx"), b"image-v1").unwrap(); + let mut source_shipper = claimed(&store, &source_dir, &source_state, &progress).await; + source_shipper.cycle().await.unwrap(); + + let target_dir = scratch_dir("undrained-target"); + let hydrator = crate::hydrate::prepare_replica( + &(store.clone() as Arc), + &StorePath::default(), + "mem://test", + &target_dir, + ) + .await + .expect("hydrates"); + assert!( + !hydrator.drained(), + "a freshly prepared hydrator with a real family must start undrained" + ); + + let target_state = state_for(&target_dir); + let mut shipper = Shipper::claim( + store.clone() as Arc, + StorePath::default(), + "mem://test".to_string(), + target_dir.clone(), + Arc::clone(&progress), + target_state, + Some(hydrator), + ) + .await + .unwrap(); + + // A genuinely new local write, unrelated to the hydration above — + // the server keeps accepting writes while background-hydrating. + wal::append_batch(&target_dir.join("ctx_a.wal.jsonl"), 1, &[associate("a")]).unwrap(); + + assert!( + shipper.cycle().await.unwrap(), + "a lane that genuinely shipped something must report true even when hydration \ + has not drained and the manifest-publish step is skipped entirely" + ); + let _ = std::fs::remove_dir_all(&source_dir); + let _ = std::fs::remove_dir_all(&target_dir); +} + +/// #618: `fence_holder` was never directly exercised — only ever +/// called from `main`'s replica-status wiring. +#[tokio::test] +async fn fence_holder_reads_back_the_claiming_holder() { + let dir = scratch_dir("fence-holder"); + let state = state_for(&dir); + let progress = Arc::new(ShipProgress::new(crate::registry::DEFAULT_WAL_MAX_BYTES)); + let store = Arc::new(InMemory::new()); + + let shipper = claimed(&store, &dir, &state, &progress).await; + let holder = fence_holder( + store.as_ref() as &dyn ObjectStore, + &StorePath::default(), + shipper.generation, + ) + .await + .expect("the just-claimed generation's fence body reads back"); + assert!(holder.contains('#'), "the holder is HOSTNAME#pid: {holder}"); + + // A generation nothing ever claimed: best-effort `None`, not an + // error or a panic. + assert!( + fence_holder( + store.as_ref() as &dyn ObjectStore, + &StorePath::default(), + 9999 + ) + .await + .is_none() ); - assert_eq!(parse_segment_name(&segment_name(3, 4)), Some((3, 4))); - assert_eq!(parse_segment_name("junk"), None); + let _ = std::fs::remove_dir_all(&dir); } /// The restore-equivalence property, generated: any interleaving diff --git a/tests/http_api/replication.rs b/tests/http_api/replication.rs index 7864dc35..58e9c5b8 100644 --- a/tests/http_api/replication.rs +++ b/tests/http_api/replication.rs @@ -444,21 +444,22 @@ fn a_restore_racing_an_active_writer_is_refused_and_leaves_its_data_intact() { } } -/// A pre-manifest bucket (an EMPTY `complete` marker) whose -/// generation still carries wal segments must restore the tail they -/// hold, through the listing-driven compatibility path. No other test -/// reaches that path's wal-lane loop: the graceful-stop round trip -/// above drains the tail into the baseline files before restore runs, -/// and today's writer always ships a manifest, which routes restore -/// through the manifest branch instead. Here the writer is SIGKILLed -/// after a post-baseline write provably ships, then the marker is -/// emptied to the pre-manifest shape — from there only the listed -/// lane's replay can carry the write into the restored directory. +/// A generation whose `complete` marker predates the shipping +/// manifest (an EMPTY body — issue #128's baseline shape) is refused +/// outright, not silently reconstructed by listing: every generation +/// any current writer ships carries a manifest (confirmed by +/// `the_manifest_records_every_shipped_extent_and_restore_verifies_it` +/// in `ship/tests.rs`), so an empty marker only ever names a bucket +/// from before #128, and restore no longer supports materializing +/// one. Constructed the same way the pre-#128-compat path used to be +/// exercised (a writer SIGKILLed after a post-baseline write ships, +/// then the marker rewound to its empty shape) so the refusal is +/// pinned against the same generation shape, not a synthetic one. #[test] -fn a_hard_killed_writers_bucket_restores_the_shipped_wal_tail() { - let bucket = scratch("wal-tail-bucket"); +fn a_pre_manifest_generation_is_refused_not_silently_restored() { + let bucket = scratch("pre-manifest-bucket"); let server = Server::start_with_env( - "repl-wal-tail", + "repl-pre-manifest", &[ ("TAGURU_REPLICATE_URL", &bucket_url(&bucket)), ("TAGURU_REPLICATE_INTERVAL_MS", "100"), @@ -491,13 +492,13 @@ fn a_hard_killed_writers_bucket_restores_the_shipped_wal_tail() { let data_dir = server.stop_hard(); // Rewind the marker to the pre-manifest shape (an empty - // `complete`). This both pins the compatibility path and makes the - // test deterministic: with the manifest branch, whether the lane is - // restored depends on whether the writer's manifest UPDATE beat the - // SIGKILL — a race this test must not encode. + // `complete`) — this is also what makes the test deterministic: + // with a real manifest, whether the tail write is reflected + // depends on whether the writer's manifest UPDATE beat the + // SIGKILL, a race this test must not encode. std::fs::write(generation.join("complete"), b"").unwrap(); - let restored = scratch("wal-tail-restored"); + let restored = scratch("pre-manifest-restored"); let restore = run_cli( &[ "restore", @@ -508,27 +509,16 @@ fn a_hard_killed_writers_bucket_restores_the_shipped_wal_tail() { &[], ); assert!( - restore.status.success(), - "restore failed: {}", - String::from_utf8_lossy(&restore.stderr) - ); - - // The tail write is present — provable only via the wal replay. - let exports = scratch("wal-tail-exports"); - let exported = run_cli( - &["export", "--out", &exports.display().to_string()], - &[("TAGURU_DATA_DIR", &restored.display().to_string())], + !restore.status.success(), + "a pre-manifest generation must be refused, not silently restored" ); + let stderr = String::from_utf8_lossy(&restore.stderr); assert!( - exported.status.success(), - "{}", - String::from_utf8_lossy(&exported.stderr) + stderr.contains("predates the shipping manifest"), + "{stderr}" ); - let stream = std::fs::read_to_string(exports.join("sake.jsonl")) - .expect("the restored export must carry sake.jsonl"); - assert!(stream.contains("代表銘柄"), "{stream}"); - for dir in [bucket, data_dir, restored, exports] { + for dir in [bucket, data_dir, restored] { let _ = std::fs::remove_dir_all(dir); } } From e4f14dd3f5e87f99a6ce3faf9383cb018408e5fb Mon Sep 17 00:00:00 2001 From: Takashi Yamashina Date: Fri, 14 Aug 2026 11:11:23 +0900 Subject: [PATCH 2/4] mutants: record two provably-equivalent boundary comparisons and two unpinnable TOCTOU guards MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit newest_fence's and ReplicateConfig::from_env's `<` vs `<=` boundary mutants compute the identical observable output for every input (explained inline and here). Two of ship_lane's three identical vanished_mid_cycle call sites (shipper.rs:426, :459) have no `.await` between ship_lane's own entry and reaching them, so nothing on a single-threaded executor can land a test in that window — unlike the third (:390), which a real test now pins by racing an earlier scan.changed upload. Claude-Session: https://claude.ai/code/session_01KGdCCEPLGcimWQtAcAAXqZ --- .cargo/mutants.toml | 34 ++++++++++++++++++++++++++++ src/ship/tests.rs | 54 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+) diff --git a/.cargo/mutants.toml b/.cargo/mutants.toml index 7e309051..c794ba5e 100644 --- a/.cargo/mutants.toml +++ b/.cargo/mutants.toml @@ -69,4 +69,38 @@ 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 SECOND and THIRD `vanished_mid_cycle` + # call sites (issue #618, shipper.rs:426 and :459) — line-anchored + # because the identical guard expression also appears at :390, + # where it IS tested (see + # a_lane_deleted_between_the_scan_and_its_own_turn_ships_nothing_not_an_error). + # Both of these sites have NO `.await` between `ship_lane`'s own + # entry and reaching them (metadata → sig check → read → prefix + # check → this metadata, all synchronous), so — unlike :390, which + # a test can race by pausing an EARLIER `scan.changed` upload — + # 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 + # at the OS level could still hit this window in production; no + # test can pin that. + "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", ] diff --git a/src/ship/tests.rs b/src/ship/tests.rs index 35923f1a..579d24d6 100644 --- a/src/ship/tests.rs +++ b/src/ship/tests.rs @@ -594,6 +594,60 @@ async fn a_newer_claim_fences_the_shipper_on_its_next_dirty_cycle() { let _ = std::fs::remove_dir_all(&dir); } +#[tokio::test] +async fn a_lane_deleted_between_the_scan_and_its_own_turn_ships_nothing_not_an_error() { + let dir = scratch_dir("vanished-mid-cycle"); + let state = state_for(&dir); + let progress = Arc::new(ShipProgress::new(crate::registry::DEFAULT_WAL_MAX_BYTES)); + let inner: Arc = Arc::new(InMemory::new()); + + std::fs::write(dir.join("ctx_a.ctx"), b"image-v1").unwrap(); + std::fs::write(dir.join("ctx_b.ctx"), b"image-v1").unwrap(); + let ctx_b_wal = dir.join("ctx_b.wal.jsonl"); + wal::append_batch(&ctx_b_wal, 1, &[associate("b")]).unwrap(); + + let started = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + // ctx_a.ctx's publish PUT: a `scan.changed` entry, which the + // whole `scan.changed` loop finishes BEFORE the `scan.lanes` loop + // (that reaches ctx_b's wal file) ever begins — a real `.await` + // boundary a test can pause on, deterministically, unlike a race + // WITHIN one `ship_lane` call (no `.await` between its own + // `fs::metadata`/`fs::read` pair for a test to land in). + let files_key = gen_root(&StorePath::default(), 1) + .join("files") + .join("ctx_a.ctx"); + let store: Arc = Arc::new(PausingStore { + inner: Arc::clone(&inner), + pause_on: files_key, + started: Arc::clone(&started), + release: Arc::clone(&release), + }); + + let dir2 = dir.clone(); + let state2 = state.clone(); + let progress2 = Arc::clone(&progress); + let cycle_task = tokio::spawn(async move { + let mut shipper = claimed_dyn(store, &dir2, &state2, &progress2).await; + shipper.cycle().await + }); + + started.notified().await; + std::fs::remove_file(&ctx_b_wal).unwrap(); + release.notify_one(); + + let result = tokio::time::timeout(Duration::from_secs(5), cycle_task) + .await + .expect("must not hang") + .unwrap(); + assert!( + result.is_ok(), + "a lane deleted between the directory scan and the lanes loop reaching its own \ + turn must ship as if nothing changed there, not fail the whole cycle: {result:?}" + ); + let _ = std::fs::remove_dir_all(&dir); +} + #[tokio::test] async fn a_vanished_family_is_retired_remotely_including_its_segments() { let dir = scratch_dir("retire"); From faf6c974f91977152a3183f0de62a62900143b7c Mon Sep 17 00:00:00 2001 From: Takashi Yamashina Date: Fri, 14 Aug 2026 11:25:24 +0900 Subject: [PATCH 3/4] mutants: exclude ship_lane's :390 true arm too, with a fuller rationale The false arm at :390 is real-tested; the true arm needs a non-NotFound local fs error at that exact point, which this suite has no fault-injection wrapper for without breaking the same pausing mechanism the working test relies on. Claude-Session: https://claude.ai/code/session_01KGdCCEPLGcimWQtAcAAXqZ --- .cargo/mutants.toml | 41 +++++++++++++++++++++++++++-------------- 1 file changed, 27 insertions(+), 14 deletions(-) diff --git a/.cargo/mutants.toml b/.cargo/mutants.toml index c794ba5e..302ed414 100644 --- a/.cargo/mutants.toml +++ b/.cargo/mutants.toml @@ -85,20 +85,33 @@ exclude_re = [ # 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 SECOND and THIRD `vanished_mid_cycle` - # call sites (issue #618, shipper.rs:426 and :459) — line-anchored - # because the identical guard expression also appears at :390, - # where it IS tested (see - # a_lane_deleted_between_the_scan_and_its_own_turn_ships_nothing_not_an_error). - # Both of these sites have NO `.await` between `ship_lane`'s own - # entry and reaching them (metadata → sig check → read → prefix - # check → this metadata, all synchronous), so — unlike :390, which - # a test can race by pausing an EARLIER `scan.changed` upload — - # 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 - # at the OS level could still hit this window in production; no - # test can pin that. + # `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", From f1e2c6afb85e5e199f20daac3c923fbdf0651030 Mon Sep 17 00:00:00 2001 From: Takashi Yamashina Date: Fri, 14 Aug 2026 13:14:19 +0900 Subject: [PATCH 4/4] Address CodeRabbit review findings on PR #638 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace ScopedReplicateEnv's process-global env mutation with pure dependency injection (ReplicateConfig::from_values): set_var/remove_var require, under Rust's own safety contract, that no other thread reads or writes ANY env var while the call runs — a lock scoped to two keys cannot provide that against unrelated concurrently-running tests. The four from_env_* tests now call from_values directly with no real env mutation at all. - Refuse a manifest naming the same entry as both a file and a lane in restore_into — the lane's write_atomic would otherwise silently clobber the file's already-restored bytes (or vice versa, depending on HashMap iteration order), landing a directory that does not match the manifest. Never produced by a real shipper; bucket rot or tampering, the same posture safe_manifest_name already covers. - Remove #[mutants::skip] from vanished_mid_cycle (already reachable through the existing :390 test) and from newest_seq/update_pending_since, making both pub(super) so ship::tests can pin them directly instead of arguing their unreachability by analysis alone. - Strengthen two assertions that passed under a broken implementation: replica_register's idempotence check now compares Arc::ptr_eq instead of a bare lookup().is_some() (which a REPLACED entry also satisfies), and the first-poll fence resolution test now checks for the actual "claimed by " text instead of only the absence of "none known" (which a resolved-to-None holder also satisfies). Along the way, fixed a_replica_register_ignores_an_undecodable_stem's own premise: its original input had no '%' escape at all, so name_from_stem decoded it fine and the entry WAS registered — the assertion just didn't notice because it checked group_page (a different subsystem) instead of the registry replica_register actually writes into. - Extend a_claim_carries_hydrated_from_forward_only_for_the_same_bucket_url to also cover the different-URL branch (hydrated_from must NOT carry forward), the half the original test left unpinned. - Add tokio::time::timeout around two `Notify` waits that would otherwise hang the whole suite indefinitely on a regression. Claude-Session: https://claude.ai/code/session_01KGdCCEPLGcimWQtAcAAXqZ --- src/registry/replication.rs | 30 ++++++- src/replica.rs | 15 +++- src/ship.rs | 54 +------------ src/ship/config.rs | 61 ++++++++++---- src/ship/restore.rs | 20 +++++ src/ship/shipper.rs | 52 ++++++------ src/ship/tests.rs | 154 +++++++++++++++++++++++++++++++++++- 7 files changed, 279 insertions(+), 107 deletions(-) diff --git a/src/registry/replication.rs b/src/registry/replication.rs index f3e51d23..f30140a7 100644 --- a/src/registry/replication.rs +++ b/src/registry/replication.rs @@ -548,9 +548,16 @@ mod tests { ); // Idempotent: a second registration of the same stem must not - // replace the entry or error. + // 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); - assert!(state.lookup("sake").is_some()); + 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); } @@ -572,8 +579,23 @@ mod tests { }, ) .unwrap(); - state.replica_register("not a valid stem at all"); - assert_eq!(state.group_page(None, usize::MAX).1.len(), 0); + // `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); } diff --git a/src/replica.rs b/src/replica.rs index 0c10f6cc..9d9de7b6 100644 --- a/src/replica.rs +++ b/src/replica.rs @@ -950,12 +950,21 @@ mod tests { tailer.info.refusal() ); + // The fixture's own expected holder string, fetched + // independently — asserting only the ABSENCE of "none known" + // would also pass if `note_fence` resolved a `None` holder + // (still no "none known" phrase, but not what this test is + // pinning): the poll must actually carry the real holder text. + let expected_holder = ship::fence_holder(store.as_ref(), &StorePath::default(), 1) + .await + .expect("the fixture's own writer claimed generation 1"); + tailer.poll_once().await.expect("the poll completes"); + let refusal = tailer.info.refusal(); assert!( - !tailer.info.refusal().contains("none known"), + refusal.contains("claimed by") && refusal.contains(&expected_holder), "the very first poll against an existing fence must resolve the holder \ - line, not wait for a repeat poll of the same generation: {}", - tailer.info.refusal() + line, not wait for a repeat poll of the same generation: {refusal}" ); for dir in [bucket, target] { diff --git a/src/ship.rs b/src/ship.rs index 14c329d7..a5781ee2 100644 --- a/src/ship.rs +++ b/src/ship.rs @@ -168,6 +168,8 @@ pub(crate) use shipper::{FenceInfo, Shipper, fence_holder, newest_fence}; use progress::DEFAULT_DEFER_CAP_BYTES; #[cfg(test)] use restore::restore_into; +#[cfg(test)] +use shipper::{newest_seq, update_pending_since}; #[path = "ship/tests.rs"] #[cfg(test)] @@ -230,56 +232,4 @@ pub(crate) mod test_support { } } } - - static REPLICATE_ENV_LOCK: Mutex<()> = Mutex::new(()); - - const REPLICATE_ENV_KEYS: [&str; 2] = ["TAGURU_REPLICATE_URL", "TAGURU_REPLICATE_INTERVAL_MS"]; - - /// Sets `TAGURU_REPLICATE_URL`/`_INTERVAL_MS` for the guard's - /// lifetime (`None` in a pair removes that key instead), restoring - /// each key exactly as found on drop — the same shape as - /// [`ScrubbedAzureEnv`], for `ReplicateConfig::from_env` tests that - /// would otherwise race every other test touching these two - /// process-global vars. - pub(crate) struct ScopedReplicateEnv { - _lock: parking_lot::MutexGuard<'static, ()>, - saved: Vec<(&'static str, Option)>, - } - - impl ScopedReplicateEnv { - pub(crate) fn new(values: [Option<&str>; 2]) -> Self { - let lock = REPLICATE_ENV_LOCK.lock(); - let saved = REPLICATE_ENV_KEYS - .iter() - .map(|&key| (key, std::env::var(key).ok())) - .collect(); - for (key, value) in REPLICATE_ENV_KEYS.iter().zip(values) { - // SAFETY: serialized by `REPLICATE_ENV_LOCK` — no other - // thread reads or writes these keys while this guard - // (held for the lock's lifetime, via `_lock`) exists. - unsafe { - match value { - Some(value) => std::env::set_var(key, value), - None => std::env::remove_var(key), - } - } - } - Self { _lock: lock, saved } - } - } - - impl Drop for ScopedReplicateEnv { - fn drop(&mut self) { - for (key, value) in &self.saved { - // SAFETY: same lock, still held (`_lock` drops after - // this body via field declaration order). - unsafe { - match value { - Some(value) => std::env::set_var(key, value), - None => std::env::remove_var(key), - } - } - } - } - } } diff --git a/src/ship/config.rs b/src/ship/config.rs index e162d22e..a10e329b 100644 --- a/src/ship/config.rs +++ b/src/ship/config.rs @@ -18,7 +18,24 @@ impl ReplicateConfig { /// steady-state RPO knob). A zero interval would spin the poll /// loop; floor to 100ms, loudly, like every other env knob. pub(crate) fn from_env() -> Option { - let url = std::env::var("TAGURU_REPLICATE_URL").ok()?; + Self::from_values( + std::env::var("TAGURU_REPLICATE_URL").ok(), + std::env::var("TAGURU_REPLICATE_INTERVAL_MS").ok(), + ) + } + + /// The pure parsing half of [`Self::from_env`], taking the two raw + /// values instead of reading them itself — every branch is + /// reachable from a plain function call, so tests exercise them + /// with ordinary arguments instead of mutating the REAL process + /// environment (`std::env::set_var`/`remove_var` require, under + /// Rust's own safety contract, that no other thread reads or + /// writes ANY env var while the call runs — a lock scoped to only + /// these two keys cannot provide that against unrelated, + /// concurrently-running tests elsewhere in the suite that read + /// env vars without taking it). + fn from_values(url: Option, interval_ms: Option) -> Option { + let url = url?; let url = url.trim().to_string(); if url.is_empty() { // The same present-but-blank trap TAGURU_PUBLIC_URL guards @@ -30,7 +47,18 @@ impl ReplicateConfig { ); return None; } - let requested = crate::env::env_number("TAGURU_REPLICATE_INTERVAL_MS", 1000); + // `crate::env::env_number`'s own contract (parse, or warn and + // fall back to the default), reimplemented against the passed + // value rather than a live env read — see this function's doc. + let requested = match interval_ms { + Some(value) => value.parse::().unwrap_or_else(|_| { + tracing::warn!( + "ignoring TAGURU_REPLICATE_INTERVAL_MS={value}: not a number; using 1000" + ); + 1000 + }), + None => 1000, + }; // `<` vs `<=` is unobservable at the boundary itself (issue // #618): at `requested == 100`, both arms compute the same // `Duration::from_millis(100)` — the floor branch explicitly, @@ -189,11 +217,11 @@ mod tests { /// #618: `TAGURU_REPLICATE_URL` set but blank is the same /// templating-accident trap `TAGURU_PUBLIC_URL` guards against — - /// treated as disabled, not as a URL to parse. + /// treated as disabled, not as a URL to parse. Exercised through + /// `from_values` (no real env mutation — see its doc comment). #[test] fn from_env_treats_a_blank_url_as_disabled() { - let _env = crate::ship::test_support::ScopedReplicateEnv::new([Some(" "), None]); - assert!(ReplicateConfig::from_env().is_none()); + assert!(ReplicateConfig::from_values(Some(" ".to_string()), None).is_none()); } /// #618: unset entirely is the ordinary "shipping off" case, not @@ -201,19 +229,18 @@ mod tests { /// only one of them logs a warning about it. #[test] fn from_env_returns_none_when_unset() { - let _env = crate::ship::test_support::ScopedReplicateEnv::new([None, None]); - assert!(ReplicateConfig::from_env().is_none()); + assert!(ReplicateConfig::from_values(None, None).is_none()); } /// #618: an interval below the busy-poll floor is raised to it, /// loudly — never silently honored. #[test] fn from_env_floors_a_tiny_interval() { - let _env = crate::ship::test_support::ScopedReplicateEnv::new([ - Some("file:///tmp/wherever"), - Some("1"), - ]); - let config = ReplicateConfig::from_env().expect("a non-blank URL enables shipping"); + let config = ReplicateConfig::from_values( + Some("file:///tmp/wherever".to_string()), + Some("1".to_string()), + ) + .expect("a non-blank URL enables shipping"); assert_eq!(config.interval, Duration::from_millis(100)); } @@ -222,11 +249,11 @@ mod tests { /// the call site that actually matters for replication. #[test] fn from_env_falls_back_to_the_default_interval_on_a_bad_value() { - let _env = crate::ship::test_support::ScopedReplicateEnv::new([ - Some("file:///tmp/wherever"), - Some("not-a-number"), - ]); - let config = ReplicateConfig::from_env().expect("a non-blank URL enables shipping"); + let config = ReplicateConfig::from_values( + Some("file:///tmp/wherever".to_string()), + Some("not-a-number".to_string()), + ) + .expect("a non-blank URL enables shipping"); assert_eq!(config.interval, Duration::from_millis(1000)); } } diff --git a/src/ship/restore.rs b/src/ship/restore.rs index 4a03e39d..6d46be7c 100644 --- a/src/ship/restore.rs +++ b/src/ship/restore.rs @@ -263,6 +263,26 @@ pub(crate) async fn restore_into( ), )); }; + // The same hostile-input posture `safe_manifest_name` exists for: + // a name in both maps would have the lane's `write_atomic` clobber + // the file's already-restored bytes (or vice versa, depending on + // iteration order — a HashMap's, unspecified), landing a directory + // that does not match the manifest either way. Never produced by a + // real shipper (files and lanes are named from disjoint suffixes), + // so this is bucket rot or tampering, not age. + if let Some(name) = manifest + .files + .keys() + .find(|name| manifest.lanes.contains_key(*name)) + { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "{name}: the manifest names both a file and a lane — the bucket may be \ + tampered with" + ), + )); + } for (name, expect) in &manifest.files { let key = generation_root.clone().join("files").join(name.as_str()); let bytes = fetch(store, &key).await?; diff --git a/src/ship/shipper.rs b/src/ship/shipper.rs index 0a4809cb..6b840086 100644 --- a/src/ship/shipper.rs +++ b/src/ship/shipper.rs @@ -566,14 +566,15 @@ impl Shipper { /// Whether a local fs error on a lane file mid-cycle means "this name /// vanished between the directory scan that found it and this call" /// (ship as if nothing changed; the next scan retires it) vs. any -/// other local error (propagate). Pulled into its own function so the -/// condition can be `#[mutants::skip]`ped at each of `ship_lane`'s -/// three call sites without also skipping mutation coverage on the -/// rest of that function: only a real filesystem race between the -/// scan and one of these calls can flip which arm runs, and that race -/// cannot be pinned deterministically in a test — same reasoning as -/// `remove_persisted_file_quietly` in `registry/boot.rs`. -#[mutants::skip] +/// other local error (propagate). Pulled into its own function so its +/// mutation targets at `ship_lane`'s three call sites can each be +/// judged on their own reachability — a real filesystem race between +/// the scan and one of these calls cannot be pinned deterministically +/// in a test, but a genuine `NotFound` reaching this comparison IS +/// (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 a lane file +/// before the lanes loop reaches its turn) — see `.cargo/mutants.toml` +/// for which of the three call sites remain excluded, and why. fn vanished_mid_cycle(error: &io::Error) -> bool { error.kind() == io::ErrorKind::NotFound } @@ -581,22 +582,10 @@ fn vanished_mid_cycle(error: &io::Error) -> bool { /// The newest (highest) seq among the file's complete lines, ignoring /// integrity: this feeds the LAG metric only, where an honest "how far /// behind" matters more than validity — corrupt bytes will surface as -/// a shipping error, not a hidden zero lag. -/// -/// `#[mutants::skip]`ped: `ship_lane` only reaches its call of this -/// function once its own read of `bytes` succeeded, and every error -/// path out of that same read (a torn tail aside — deliberately -/// excluded by both this function's and `shippable_records`'s -/// identical trailing-segment-pop) returns before `self.lanes.insert` -/// runs, discarding this call's `lane` mutations wholesale. A record -/// whose CRC does not match is fatal (surfaces as a shipping error -/// via `shippable_records`'s `?`), never silently shipped-around — so -/// there is no reachable state in which this function's return value -/// legitimately diverges from `lane.shipped_seq` for a test to pin; -/// `update_pending_since` below exists for defense against a future -/// change to that invariant, not a reachable-today gap. -#[mutants::skip] -fn newest_seq(bytes: &[u8]) -> Option { +/// a shipping error, not a hidden zero lag. `pub(super)` (not private) +/// so `ship::tests` can pin it directly with byte inputs instead of +/// only through a full `ship_lane` cycle. +pub(super) fn newest_seq(bytes: &[u8]) -> Option { #[derive(serde::Deserialize)] struct SeqOnly { seq: u64, @@ -610,11 +599,16 @@ fn newest_seq(bytes: &[u8]) -> Option { } /// Whether a lane's local log has grown past what shipped, and for how -/// long — see `newest_seq`'s doc for why `local_seq` cannot reachably -/// exceed `shipped_seq` in this codebase today; `#[mutants::skip]`ped -/// for the same reason. -#[mutants::skip] -fn update_pending_since(lane: &mut LaneState) { +/// long. `pub(super)` (not private) so `ship::tests` can pin the +/// `local_seq == shipped_seq` boundary directly on a `LaneState`, +/// rather than only through a full `ship_lane` cycle where the two +/// are, by construction, never observed to diverge (see `newest_seq`'s +/// call site in `ship_lane`: both values come from the SAME read of +/// `bytes`, and every error path out of that read discards this +/// call's `lane` mutations before they would ever be compared) — this +/// function's own boundary still needs pinning on its own terms, since +/// `>` vs `>=` disagree exactly there. +pub(super) fn update_pending_since(lane: &mut LaneState) { if lane.local_seq > lane.shipped_seq { lane.pending_since.get_or_insert_with(Instant::now); } else { diff --git a/src/ship/tests.rs b/src/ship/tests.rs index 579d24d6..43d8299d 100644 --- a/src/ship/tests.rs +++ b/src/ship/tests.rs @@ -632,7 +632,9 @@ async fn a_lane_deleted_between_the_scan_and_its_own_turn_ships_nothing_not_an_e shipper.cycle().await }); - started.notified().await; + tokio::time::timeout(Duration::from_secs(5), started.notified()) + .await + .expect("the paused put must start"); std::fs::remove_file(&ctx_b_wal).unwrap(); release.notify_one(); @@ -1210,6 +1212,73 @@ async fn restore_refuses_a_manifest_naming_a_path_outside_the_target_directory() let _ = std::fs::remove_file(&escape_target); } +/// A manifest naming the SAME entry as both a file and a lane must +/// refuse, not let one silently clobber the other's already-restored +/// bytes (issue #638 review): never produced by a real shipper (files +/// and lanes are named from disjoint suffixes), so this is bucket rot +/// or tampering, exactly the posture `safe_manifest_name` exists for. +#[tokio::test] +async fn restore_refuses_a_manifest_naming_the_same_entry_as_a_file_and_a_lane() { + let dir = scratch_dir("manifest-file-lane-collision"); + let state = state_for(&dir); + let progress = Arc::new(ShipProgress::new(crate::registry::DEFAULT_WAL_MAX_BYTES)); + let store = Arc::new(InMemory::new()); + + std::fs::write(dir.join("ctx_a.ctx"), b"image-v1").unwrap(); + let wal_path = dir.join("ctx_a.wal.jsonl"); + wal::append_batch(&wal_path, 1, &[associate("a")]).unwrap(); + let mut shipper = claimed(&store, &dir, &state, &progress).await; + shipper.cycle().await.unwrap(); + + // Tamper the manifest: insert a FILES entry under the same name as + // the already-shipped LANE — a real CRC over an arbitrary payload, + // so content verification alone would accept it. + let generation_root = gen_root(&StorePath::default(), 1); + let colliding_name = "ctx_a.wal.jsonl"; + let payload = b"clobbered!".to_vec(); + let key = generation_root.clone().join("files").join(colliding_name); + (store.as_ref() as &dyn ObjectStore) + .put(&key, PutPayload::from(payload.clone())) + .await + .unwrap(); + let bytes = read_object(&store, "gen-00000000000000000001/complete").await; + let mut manifest: Manifest = serde_json::from_slice(&bytes).unwrap(); + assert!( + manifest.lanes.contains_key(colliding_name), + "the fixture must already ship this name as a lane: {manifest:?}" + ); + manifest.files.insert( + colliding_name.to_string(), + ManifestFile { + len: payload.len() as u64, + crc: crate::crc32c::crc32c(&payload), + }, + ); + (store.as_ref() as &dyn ObjectStore) + .put( + &generation_root.clone().join(COMPLETE_MARKER), + PutPayload::from(serde_json::to_vec(&manifest).unwrap()), + ) + .await + .unwrap(); + + let restored_dir = scratch_dir("manifest-file-lane-collision-out"); + let error = restore_into( + store.as_ref() as &dyn ObjectStore, + &StorePath::default(), + &restored_dir, + ) + .await + .expect_err("a name that is both a file and a lane must refuse"); + assert!( + error.to_string().contains("both a file and a lane"), + "{error}" + ); + + let _ = std::fs::remove_dir_all(&dir); + let _ = std::fs::remove_dir_all(&restored_dir); +} + #[tokio::test] async fn a_claim_is_recorded_locally_and_liveness_markers_land() { let dir = scratch_dir("liveness"); @@ -1271,6 +1340,60 @@ fn classification_and_lane_parents_agree_with_the_family_layout() { assert_eq!(parent_snapshot_of("x.ctx"), None); } +/// #638 review: `newest_seq` pinned directly, on the exact input +/// shapes `ship_lane` actually passes it (complete lines, or complete +/// lines followed by a torn tail — never a lone line with no trailing +/// newline at all, which `ship_lane` only ever reads as part of a +/// larger buffer that already ends `\n`-terminated up to the last +/// complete record). +#[test] +fn newest_seq_reports_the_highest_seq_among_complete_lines() { + assert_eq!(newest_seq(b""), None); + assert_eq!(newest_seq(b"{\"seq\":1}\n"), Some(1)); + assert_eq!(newest_seq(b"{\"seq\":1}\n{\"seq\":2}\n"), Some(2)); + // A torn tail (no trailing newline) is ignored, same as + // `shippable_records`'s own trailing-segment-pop. + assert_eq!(newest_seq(b"{\"seq\":1}\n{\"seq\":2}\n{\"seq\":3"), Some(2)); +} + +/// #638 review: `update_pending_since`'s `local_seq > shipped_seq` +/// boundary, pinned directly on a `LaneState` — `>` and `>=` disagree +/// exactly at equality (the ordinary fully-caught-up case), where `>=` +/// would wrongly mark a caught-up lane as newly pending. +#[test] +fn update_pending_since_treats_equal_seqs_as_caught_up_not_pending() { + let mut lane = LaneState::fresh(0); + lane.shipped_seq = 5; + lane.local_seq = 5; + update_pending_since(&mut lane); + assert!( + lane.pending_since.is_none(), + "local_seq == shipped_seq must not be pending" + ); +} + +/// #638 review: the complementary case — a lane genuinely behind must +/// be marked pending, and one that WAS behind and just caught up must +/// have that pending mark cleared, not linger. +#[test] +fn update_pending_since_marks_a_real_gap_and_clears_once_caught_up() { + let mut lane = LaneState::fresh(0); + lane.shipped_seq = 5; + lane.local_seq = 6; + update_pending_since(&mut lane); + assert!( + lane.pending_since.is_some(), + "local_seq > shipped_seq must be pending" + ); + + lane.local_seq = 5; + update_pending_since(&mut lane); + assert!( + lane.pending_since.is_none(), + "catching up must clear the pending mark, not leave it stale" + ); +} + /// #618: a `.taguru.replication` record that exists but cannot be /// parsed must surface as an error, never silently treated as absent /// — `prepare`/`prepare_replica` both refuse to boot on it rather than @@ -1316,7 +1439,9 @@ async fn a_claim_retries_past_a_generation_taken_between_its_check_and_its_write let claim_task = tokio::spawn(async move { claimed_dyn(store, &dir2, &state2, &progress2).await }); - started.notified().await; + tokio::time::timeout(Duration::from_secs(5), started.notified()) + .await + .expect("the paused put must start"); inner .put_opts( &target, @@ -1372,6 +1497,31 @@ async fn a_claim_carries_hydrated_from_forward_only_for_the_same_bucket_url() { "a claim against the SAME bucket must carry the prior hydrated_from forward" ); let _ = std::fs::remove_dir_all(&dir); + + // The other half of the `record.url == url` filter: a directory + // whose EXISTING record names a DIFFERENT bucket must not inherit + // a generation number that means nothing there — removing the + // filter (or inverting it) would still pass the same-url assertion + // above, so this is the branch that actually pins the condition. + let other_dir = scratch_dir("claim-hydrated-from-different-url"); + write_replication_record( + &other_dir, + &ReplicationRecord { + url: "mem://a-different-bucket".to_string(), + claimed_generation: None, + hydrated_from: Some(5), + }, + ) + .unwrap(); + let other_state = state_for(&other_dir); + claimed(&store, &other_dir, &other_state, &progress).await; + let other_record = read_replication_record(&other_dir).unwrap().unwrap(); + assert_eq!( + other_record.hydrated_from, None, + "a claim against a DIFFERENT bucket than the existing record names must not \ + inherit its hydrated_from" + ); + let _ = std::fs::remove_dir_all(&other_dir); } /// #618: under a lazy (not-yet-drained) hydration, a cycle whose ONLY