diff --git a/.cargo/mutants.toml b/.cargo/mutants.toml index 7e309051..302ed414 100644 --- a/.cargo/mutants.toml +++ b/.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", ] diff --git a/src/registry/replication.rs b/src/registry/replication.rs index e3f8bf55..f30140a7 100644 --- a/src/registry/replication.rs +++ b/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); + } } diff --git a/src/replica.rs b/src/replica.rs index 268215bd..9d9de7b6 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,143 @@ 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() + ); + + // 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!( + 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: {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 +1836,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..a5781ee2 100644 --- a/src/ship.rs +++ b/src/ship.rs @@ -167,7 +167,9 @@ 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; +#[cfg(test)] +use shipper::{newest_seq, update_pending_since}; #[path = "ship/tests.rs"] #[cfg(test)] diff --git a/src/ship/config.rs b/src/ship/config.rs index 4efc7681..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,25 @@ 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, + // 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 +197,63 @@ 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. Exercised through + /// `from_values` (no real env mutation — see its doc comment). + #[test] + fn from_env_treats_a_blank_url_as_disabled() { + assert!(ReplicateConfig::from_values(Some(" ".to_string()), None).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() { + 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 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)); + } + + /// #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 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/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..6d46be7c 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,52 @@ 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); + 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" + ), + )); + }; + // 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" + ), + )); } - - // 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()); + 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 +470,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..6b840086 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,11 +563,29 @@ 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 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 +} + /// 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. -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, @@ -584,6 +598,24 @@ 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. `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 { + lane.pending_since = None; + } +} + #[derive(Default)] struct Scan { changed: Vec, @@ -615,6 +647,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..43d8299d 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"); @@ -511,6 +594,62 @@ 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 + }); + + 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(); + + 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"); @@ -896,6 +1035,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); @@ -1043,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"); @@ -1102,13 +1338,283 @@ fn classification_and_lane_parents_agree_with_the_family_layout() { "x.passages.bin" ); 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 +/// 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 }); + + tokio::time::timeout(Duration::from_secs(5), started.notified()) + .await + .expect("the paused put must start"); + 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!( - parse_segment_name("0000000001-0000000002.jsonl"), - Some((1, 2)) + shipper.generation, 2, + "a generation taken between the check and this claim's own write must be \ + retried past, not looped on forever" ); - 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); +} + +/// #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!( + 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); + + // 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 +/// 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() + ); + 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); } }