From d3e56b13486fdbd49bdfa46c5f03c4f5e52c5c77 Mon Sep 17 00:00:00 2001 From: Takashi Yamashina Date: Sun, 9 Aug 2026 18:58:45 +0900 Subject: [PATCH 1/2] =?UTF-8?q?audit:=20close=20the=20second-pass=20findin?= =?UTF-8?q?gs=20=E2=80=94=20replica=20refresh=20debt,=20migration=20float?= =?UTF-8?q?=20equality,=20and=20five=20sibling=20asymmetries?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Findings from the focused second-pass review (issue #522), each with a test that kills its mutants: - replica: the tailer now carries a pending_refresh debt across polls — a stem whose hydration a per-request loader completed after the tailer's own failed attempt never turns stale again, yet its in-memory meta (pinned, description, revision bookkeeping) was never re-read; the remembered debt pays out on the next poll instead of freezing the meta until an unrelated manifest change. - replication: replica_refresh re-stats both WAL gauges — on a replica the bytes arrive as tailed file copies, never through the writer's live accounting, so a cold unpinned context understated taguru_wal_bytes indefinitely. - context/image: the pre-v5 migration's sourceless-call detection compares the two summation orders under a first-order rounding bound (extracted as summation_gap_is_real, boundary-pinned) instead of exact f64 equality — regrouping noise no longer credits a phantom attribution that retract_source can never remove. - context/consolidation: contradiction-group ObjectRow.sources drops zero-sum records, the same posture sign_conflicts already takes — a source that cancelled its own assertion no longer reads as attesting evidence to the judge. - api/consolidation: `checks` passes the same overlong() ceiling every sibling list input enforces (dedup folds only consecutive repeats, so an alternating list dodged both selector guards at any length the body cap admits). - extract: --source-id trims its emptiness check like --tag's. - mcp/schema: search_passages/explain_search declare `minimum: 0` on since/until like every sibling tool. Closes #522 Claude-Session: https://claude.ai/code/session_011NozdDS9JqgCpi9Z3wo4Pd --- src/api/consolidation.rs | 8 +- src/context/consolidation.rs | 47 +++++++++++ src/context/image.rs | 134 ++++++++++++++++++++++++++++---- src/extract/args.rs | 4 +- src/extract/tests.rs | 4 + src/mcp/schema.rs | 8 +- src/registry/replication.rs | 41 ++++++++++ src/replica.rs | 128 +++++++++++++++++++++++++++++- tests/http_api/consolidation.rs | 15 ++++ 9 files changed, 366 insertions(+), 23 deletions(-) diff --git a/src/api/consolidation.rs b/src/api/consolidation.rs index b04c7a84..7bb8fb38 100644 --- a/src/api/consolidation.rs +++ b/src/api/consolidation.rs @@ -25,7 +25,7 @@ use crate::registry::{AccessError, AppState}; use super::vocabulary::vocabulary_audit; use super::{ AppJson, AppPath, ErrorCode, MAX_MATCH_LIMIT, access_error, clamp, deadline_exceeded, error, - not_found, ok, + not_found, ok, overlong, }; /// The detector stamp (the `louvain-cc/1` precedent): fingerprints are @@ -185,6 +185,12 @@ pub async fn audit_consolidation( AppJson(request): AppJson, ) -> Response { let started_at = Instant::now(); + // Before the dedup: `dedup` folds only CONSECUTIVE repeats, so an + // alternating list would otherwise pass both guards below at any + // length the body cap admits. + if let Some(refusal) = overlong("checks", request.checks.len(), started_at) { + return refusal; + } let mut checks: Vec<&str> = request.checks.iter().map(String::as_str).collect(); checks.dedup(); if checks.is_empty() { diff --git a/src/context/consolidation.rs b/src/context/consolidation.rs index 01051d24..b6a4cadf 100644 --- a/src/context/consolidation.rs +++ b/src/context/consolidation.rs @@ -276,8 +276,16 @@ impl Context { object: self.concept_name(edge.object).to_string(), weight: edge.sum / edge.count as f64, count: edge.count, + // Zero-sum records stay chain-linked (only a + // retraction unlinks) but attest nothing — + // the same posture `sign_conflicts` takes, + // whose zero-sum sources sit on neither side + // of a dispute. Listing one here would hand + // the judge a source that never net-asserted + // the triple. sources: self .attribution_chain(edge.first_attribution) + .filter(|(_, record)| record.sum != 0.0) .map(|(_, record)| self.source_name(record.source).to_string()) .collect(), } @@ -507,6 +515,45 @@ mod tests { assert!(after.iter().all(|g| g.label != "杜氏"), "{after:?}"); } + /// A source that asserted a triple and then cancelled its own + /// assertion (net sum exactly zero, record still chain-linked — + /// only a retraction unlinks) attests nothing: the evidence rows + /// handed to the judge must not list it, the same posture + /// `sign_conflicts` already takes for its zero-sum records. + #[test] + fn contradiction_group_sources_exclude_a_source_whose_own_sum_cancelled_to_zero() { + let mut context = Context::default(); + context + .associate_from("蔵A", "杜氏", "高瀬", 1.0, "old", None) + .unwrap(); + context + .associate_from("蔵A", "杜氏", "青山", 1.0, "new", None) + .unwrap(); + // 撤回記事 asserts 高瀬, then cancels itself with the exact + // negative — chain-linked at sum 0.0, attesting nothing. + context + .associate_from("蔵A", "杜氏", "高瀬", 2.0, "撤回記事", None) + .unwrap(); + context + .associate_from("蔵A", "杜氏", "高瀬", -2.0, "撤回記事", None) + .unwrap(); + // Tendency needs company to make 杜氏 a candidate. + context.associate("蔵B", "杜氏", "田中", 1.0).unwrap(); + + let groups = context.contradiction_groups(Deadline::unbounded()).unwrap(); + let toji = groups.iter().find(|g| g.label == "杜氏").unwrap(); + let takase = toji + .objects + .iter() + .find(|row| row.object == "高瀬") + .unwrap(); + assert_eq!( + takase.sources, + vec!["old"], + "the cancelled-out source must not read as attesting: {takase:?}" + ); + } + #[test] fn sign_conflicts_split_the_dispute_by_source() { let mut context = Context::default(); diff --git a/src/context/image.rs b/src/context/image.rs index 616eddf1..7f5b4b1f 100644 --- a/src/context/image.rs +++ b/src/context/image.rs @@ -276,19 +276,20 @@ impl Context { // pathological image whose every edge points at one long shared // chain walks that chain once per edge — O(edges × chain) — // during a migration that holds the write lock and never yields. - let mut chains: HashMap = HashMap::new(); + let mut chains: HashMap = HashMap::new(); for legacy in &legacy_edges { - let (chain_len, attributed_sum) = match chains.get(&legacy.first_attribution) { - Some(&cached) => cached, - None => { - let computed = legacy_attribution_chain_len( - &legacy_attributions, - legacy.first_attribution, - )?; - chains.insert(legacy.first_attribution, computed); - computed - } - }; + let (chain_len, attributed_sum, attributed_magnitude) = + match chains.get(&legacy.first_attribution) { + Some(&cached) => cached, + None => { + let computed = legacy_attribution_chain_len( + &legacy_attributions, + legacy.first_attribution, + )?; + chains.insert(legacy.first_attribution, computed); + computed + } + }; // An empty chain (first_attribution == NIL) is ambiguous in // the legacy format: it means either an edge that was always // sourceless or one that was fully retracted (weight zeroed @@ -325,7 +326,12 @@ impl Context { // in context.rs. let count = if chain_len == 0 && legacy.weight == 0.0 { 0 - } else if legacy.weight != attributed_sum { + } else if summation_gap_is_real( + legacy.weight - attributed_sum, + chain_len, + attributed_magnitude, + legacy.weight, + ) { chain_len + 1 } else { chain_len.max(1) @@ -731,12 +737,39 @@ fn checked_arena_str(arena: &[u8], offset: u32, len: u32) -> Result<&str, Corrup /// trusting: this runs before `index_attributions` has ever looked at the /// chain, so a hostile or truncated pre-v5 image must not send it out of /// bounds or looping forever on a cycle. +/// Whether a gap between a legacy edge's cumulative weight and its +/// chain's re-summed total proves a sourceless call, or is only +/// summation rounding. Not an exact `!=`: the edge accumulated its +/// calls in chronological order, the chain re-adds the same values in +/// chain order, and two orderings of the same f64 additions can differ +/// by rounding alone (a large-magnitude cancellation makes it +/// visible). An exact comparison would read that noise as proof of a +/// phantom sourceless call — weight `retract_source` can then never +/// remove. The bound is first-order naive-summation error for both +/// orders; a REAL sourceless call under it goes uncredited, the same +/// accepted false negative as the migration's zero-weight case. +/// Residual limitation, also accepted: a magnitude that vanished from +/// BOTH sides before migration (a large weight asserted then +/// retracted — its record unlinked, its rounding footprint still in +/// the edge weight) leaves nothing here to bound against, so that +/// shape can still credit a phantom; no tolerance derivable from the +/// image can tell it from a real sourceless call. +fn summation_gap_is_real(gap: f64, chain_len: u64, magnitude: f64, edge_weight: f64) -> bool { + let tolerance = 2.0 * (chain_len as f64 + 1.0) * f64::EPSILON * (magnitude + edge_weight.abs()); + gap.abs() > tolerance +} + fn legacy_attribution_chain_len( attributions: &[LegacyAttributionRecord], mut cursor: AttributionId, -) -> Result<(u64, f64), CorruptImage> { +) -> Result<(u64, f64, f64), CorruptImage> { let mut len = 0u64; let mut sum = 0.0f64; + // Magnitude alongside the sum: the caller's sourceless-gap test + // needs an error bound for the two accumulation orders it + // compares, and first-order summation error scales with Σ|w|, + // not with the (possibly cancelled-to-nothing) net sum. + let mut magnitude = 0.0f64; let mut steps: usize = 0; while cursor != NIL { steps += 1; @@ -750,9 +783,10 @@ fn legacy_attribution_chain_len( .ok_or(CorruptImage("legacy attribution link is out of range"))?; len += 1; accumulate_saturating(&mut sum, record.weight); + accumulate_saturating(&mut magnitude, record.weight.abs()); cursor = record.next; } - Ok((len, sum)) + Ok((len, sum, magnitude)) } /// Checks that one linked chain of edges is exactly `count` records long, @@ -1458,6 +1492,76 @@ mod tests { assert!(after.attributions.is_empty()); } + /// Pins the tolerance formula itself, `2·(chain_len+1)·ε·(magnitude + /// plus |edge_weight|)` under a strict `>`, against hardcoded + /// boundary values, so no factor can silently change scale: for + /// chain_len 2, magnitude 1.0, weight −0.5 the bound is exactly + /// 1.9984014443252818e-15, and the gap must EXCEED it (the bound + /// itself is still attributable to rounding). + #[test] + fn the_summation_gap_bound_sits_exactly_at_first_order_rounding_error() { + let bound = 1.9984014443252818e-15; + assert!(!summation_gap_is_real(bound, 2, 1.0, -0.5)); + assert!(!summation_gap_is_real(-bound, 2, 1.0, -0.5)); + assert!(summation_gap_is_real( + 1.998_401_444_325_282e-15, + 2, + 1.0, + -0.5 + )); + assert!(summation_gap_is_real( + -1.998_401_444_325_282e-15, + 2, + 1.0, + -0.5 + )); + // Well clear of the bound on both sides. + assert!(!summation_gap_is_real(bound / 2.0, 2, 1.0, -0.5)); + assert!(summation_gap_is_real(bound * 2.0, 2, 1.0, -0.5)); + // An empty chain still tolerates rounding on the edge weight + // alone (2·1·ε·|w|), rather than collapsing to zero width. + assert!(!summation_gap_is_real(f64::EPSILON, 0, 0.0, 1.0)); + assert!(summation_gap_is_real(3.0 * f64::EPSILON, 0, 0.0, 1.0)); + } + + /// A fully sourced edge whose chain regroups the same additions + /// differently than the edge's own chronological accumulation + /// (`(0.1⊕0.3)⊕0.1⊕0.1 = 0.6` vs `(0.1⊕0.1⊕0.1)⊕0.3 = + /// 0.6000000000000001`) must NOT read that 1-ulp gap as a phantom + /// sourceless call: the phantom's count can never be retracted, so + /// it would keep this edge alive after every real source retracts. + #[test] + fn migrating_a_pre_v5_image_does_not_read_summation_rounding_as_a_phantom_sourceless_call() { + let mut context = Context::default(); + context + .associate_from("私", "好き", "りんご", 0.1, "A", None) + .unwrap(); + context + .associate_from("私", "好き", "りんご", 0.3, "B", None) + .unwrap(); + context + .associate_from("私", "好き", "りんご", 0.1, "A", None) + .unwrap(); + context + .associate_from("私", "好き", "りんご", 0.1, "A", None) + .unwrap(); + + let v4 = context.to_bytes_as_version(4); + let mut restored = Context::from_bytes(&v4).expect("v4 image must load"); + + // One record per source in the downgraded chain — no phantom + // third contribution from the rounding gap. + assert_eq!(restored.recall("私")[0].count, 2); + assert_eq!(restored.retract_source("A"), Some(1)); + assert_eq!(restored.retract_source("B"), Some(1)); + assert_eq!( + restored.dead_edges(), + 1, + "with every real source retracted the edge must die — a phantom \ + credit would hold it alive on rounding noise alone" + ); + } + #[test] fn image_roundtrip_preserves_every_read_path() { let mut context = Context::default(); diff --git a/src/extract/args.rs b/src/extract/args.rs index 040ccea2..35d2a5c4 100644 --- a/src/extract/args.rs +++ b/src/extract/args.rs @@ -300,10 +300,10 @@ impl Args { } }, "--source-id" => match rest.next() { - Some(id) if source_id.is_none() && !id.is_empty() => { + Some(id) if source_id.is_none() && !id.trim().is_empty() => { source_id = Some(id.clone()); } - Some(id) if id.is_empty() => { + Some(id) if id.trim().is_empty() => { return Err(crate::config::subcommand_usage_error( "extract", "--source-id must not be empty", diff --git a/src/extract/tests.rs b/src/extract/tests.rs index 98e87d46..770db7d7 100644 --- a/src/extract/tests.rs +++ b/src/extract/tests.rs @@ -3699,6 +3699,10 @@ fn runbook_flags_parse_and_their_contradictions_are_usage_errors() { let mut empty = base.to_vec(); empty.extend(["--source-id", "", "doc.md"]); assert!(matches!(parse(&empty), Err(2))); + // Whitespace-only is the same emptiness — trimmed like --tag's. + let mut blank = base.to_vec(); + blank.extend(["--source-id", " ", "doc.md"]); + assert!(matches!(parse(&blank), Err(2))); let mut bad_date = base.to_vec(); bad_date.extend(["--date", "yesterday", "doc.md"]); assert!(matches!(parse(&bad_date), Err(2))); diff --git a/src/mcp/schema.rs b/src/mcp/schema.rs index 835f08e2..60e24a22 100644 --- a/src/mcp/schema.rs +++ b/src/mcp/schema.rs @@ -620,8 +620,8 @@ pub(super) fn tool_definitions() -> Vec { "limit": { "type": "integer", "minimum": 0, "description": "default 5" }, "semantic_floor": { "type": "number", "description": "one-call override of the vector lane's cosine floor (0-1); floors only the semantic lane — BM25-only hits still return" }, "tags": { "type": "array", "items": { "type": "string" }, "description": "only sources carrying at least one of these tags may answer" }, - "since": { "type": "integer", "description": "only sources whose date ?? stored_at is at or after this (epoch seconds)" }, - "until": { "type": "integer", "description": "only sources whose date ?? stored_at is strictly before this (epoch seconds)" } + "since": { "type": "integer", "minimum": 0, "description": "only sources whose date ?? stored_at is at or after this (epoch seconds)" }, + "until": { "type": "integer", "minimum": 0, "description": "only sources whose date ?? stored_at is strictly before this (epoch seconds)" } }), &["query"], ), @@ -696,8 +696,8 @@ pub(super) fn tool_definitions() -> Vec { "limit": { "type": "integer", "minimum": 0, "description": "the search call being explained (default 5)" }, "semantic_floor": { "type": "number", "description": "the floor override of the search call being explained — pass the same value" }, "tags": { "type": "array", "items": { "type": "string" }, "description": "the tag filter of the search call being explained — pass the same values" }, - "since": { "type": "integer", "description": "the time window's inclusive start (epoch seconds) of the search call being explained" }, - "until": { "type": "integer", "description": "the time window's exclusive end (epoch seconds) of the search call being explained" } + "since": { "type": "integer", "minimum": 0, "description": "the time window's inclusive start (epoch seconds) of the search call being explained" }, + "until": { "type": "integer", "minimum": 0, "description": "the time window's exclusive end (epoch seconds) of the search call being explained" } }), &["context", "query", "source"], ), diff --git a/src/registry/replication.rs b/src/registry/replication.rs index c9c902fb..7abd89b6 100644 --- a/src/registry/replication.rs +++ b/src/registry/replication.rs @@ -105,6 +105,18 @@ impl AppState { // old bytes unreachable (see `EntryInner::cache_identity`). inner.invalidate_cache_identity(); inner.load_failure = None; + // Re-stat both WAL gauges: on a replica the bytes arrive as + // tailed file copies, never through the writer's live + // increments, and `ensure_hot`'s own re-stat runs only for + // pinned entries below (or on the next local read) — without + // this, a cold unpinned context the tailer keeps growing + // understates `taguru_wal_bytes` indefinitely. + inner.wal_bytes = std::fs::metadata(wal_path(&self.0.data_dir, &stem)) + .map(|meta| meta.len()) + .unwrap_or(0); + inner.passages_wal_bytes = std::fs::metadata(passages_wal_path(&self.0.data_dir, &stem)) + .map(|meta| meta.len()) + .unwrap_or(0); if matches!(inner.slot, Slot::Hot(_)) { inner.slot = Slot::Cold; // The same bump eviction does: a flush that staged this @@ -240,6 +252,35 @@ mod tests { let _ = fs::remove_dir_all(dir); } + /// On a replica the WAL grows by tailed file copies, never through + /// the writer's live byte accounting — the refresh must re-stat + /// both WAL gauges itself, or a cold unpinned context the tailer + /// keeps growing understates `taguru_wal_bytes` until some local + /// read happens to run `ensure_hot`. + #[test] + fn a_replica_refresh_restats_both_wal_gauges() { + let dir = scratch_dir("replica-refresh-wal-bytes"); + let state = AppState::boot(dir.clone(), usize::MAX, None).unwrap(); + state.create("sake", ContextMeta::default()).unwrap(); + let stem = file_stem("sake"); + // The tailer's shape: bytes land as plain file writes. + fs::write(wal_path(&dir, &stem), b"{\"tailed\":1}\n").unwrap(); + fs::write(passages_wal_path(&dir, &stem), b"{\"tailed\":2}\n").unwrap(); + state.replica_refresh("sake"); + let entry = state.lookup("sake").unwrap(); + let inner = entry.inner.read(); + assert_eq!( + inner.wal_bytes, + fs::metadata(wal_path(&dir, &stem)).unwrap().len() + ); + assert_eq!( + inner.passages_wal_bytes, + fs::metadata(passages_wal_path(&dir, &stem)).unwrap().len() + ); + drop(inner); + let _ = fs::remove_dir_all(dir); + } + /// A tailed refresh of a HOT entry drops the slot and bumps the /// image generation exactly like an eviction — a staged flush must /// see the slot it captured is gone (vacuous on today's replica, diff --git a/src/replica.rs b/src/replica.rs index 47daf7da..22885d5b 100644 --- a/src/replica.rs +++ b/src/replica.rs @@ -175,6 +175,7 @@ pub(crate) fn spawn( stop: stopping, manifest_stamp: None, fence_seen: None, + pending_refresh: Default::default(), }; loop { match runtime.block_on(tailer.poll_once()) { @@ -216,6 +217,16 @@ struct Tailer { /// The newest fence generation whose body was fetched (one GET /// per new claimant, for the refusal's holder string). fence_seen: Option, + /// Stems owed a [`AppState::replica_refresh`]: added when + /// `retarget` reports them stale, removed only once a poll + /// actually refreshes them. The debt outlives the staleness + /// signal — when this tailer's own hydration attempt fails and a + /// per-request loader (`ensure_hot`, the passage first touch) + /// completes the same stem before the next poll, `retarget` sees + /// the family signature already current and never reports the + /// stem stale again, yet the entry's in-memory meta (pinned, + /// description, revision/cache bookkeeping) was never re-read. + pending_refresh: std::collections::BTreeSet, } impl Tailer { @@ -300,6 +311,7 @@ impl Tailer { tracing::info!(context = %name, "the lineage no longer carries this context; dropping it"); self.state.replica_deregister(&name); self.state.metrics().forget_replica_context(&name); + self.pending_refresh.remove(stem); } // Shared files (groups, the grant store, every sidecar meta) // next, so the per-family passes below see fresh metas and the @@ -307,8 +319,16 @@ impl Tailer { self.hydrator.hydrate_shared().await?; self.state.replica_reload_groups(); + // Stale stems join the refresh debt; the worklist is the whole + // debt, not this retarget's report — a stem whose hydration a + // per-request loader completed after this tailer's own failed + // attempt never turns stale again, but its refresh is still + // owed (`ensure_context` on a settled stem is O(1), so paying + // the debt late costs one meta re-read, not a re-hydration). + self.pending_refresh.extend(report.stale.iter().cloned()); + let worklist: Vec = self.pending_refresh.iter().cloned().collect(); let mut failed: std::collections::BTreeSet<&str> = std::collections::BTreeSet::new(); - for stem in &report.stale { + for stem in &worklist { if self.stop.load(Ordering::Relaxed) { return Ok(()); } @@ -331,6 +351,7 @@ impl Tailer { continue; } self.state.replica_refresh(&name); + self.pending_refresh.remove(stem); } // Lag rows for every lane the manifest carries. A family that // landed (this pass or any earlier one — retarget reported @@ -463,6 +484,7 @@ mod tests { stop: Arc::new(AtomicBool::new(false)), manifest_stamp: None, fence_seen: None, + pending_refresh: Default::default(), } } @@ -550,6 +572,110 @@ mod tests { } } + /// The refresh debt outlives the staleness signal: when the + /// tailer's own hydration attempt fails and a per-request loader + /// completes the same stem before the next poll, `retarget` sees + /// the family signature already current and never reports the stem + /// stale again — without the remembered debt, the entry's + /// in-memory meta (description, pinned) would stay frozen at the + /// pre-failure state indefinitely while the served data is fresh. + #[tokio::test] + async fn a_refresh_owed_from_a_failed_poll_lands_even_when_a_reader_hydrates_first() { + let bucket = scratch("owed-bucket"); + let writer = scratch("owed-writer"); + std::fs::write(writer.join("ctx_a.ctx"), b"image-v1").unwrap(); + std::fs::write( + writer.join("ctx_a.meta.json"), + br#"{"description":"old","pinned":false}"#, + ) + .unwrap(); + wal::append_batch(&writer.join("ctx_a.wal.jsonl"), 1, &[associate("a")]).unwrap(); + let writer_state = AppState::boot(writer.clone(), 64 * 1024 * 1024, None).unwrap(); + let mut shipper = Shipper::claim( + local_store(&bucket), + StorePath::default(), + url_of("owed"), + writer.clone(), + Arc::new(ShipProgress::new()), + writer_state, + None, + ) + .await + .unwrap(); + shipper.cycle().await.unwrap(); + + let url = url_of("owed"); + let store = local_store(&bucket); + let target = scratch("owed-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.clone(), + target.clone(), + state.clone(), + hydrator.clone(), + ); + tailer.poll_once().await.expect("the first manifest lands"); + assert_eq!(state.directory_entry("ctx_a").unwrap().description, "old"); + + // The writer moves the meta and ships a second segment. + std::fs::write( + writer.join("ctx_a.meta.json"), + br#"{"description":"new","pinned":false}"#, + ) + .unwrap(); + wal::append_batch(&writer.join("ctx_a.wal.jsonl"), 2, &[associate("b")]).unwrap(); + shipper.cycle().await.unwrap(); + shipper.retire_generation().await; + + // Tear the new segment: the tailer's own attempt fails, the + // refresh is skipped, and the manifest stamp does not advance. + let lane_dir = bucket + .join("gen-00000000000000000001") + .join("wal") + .join("ctx_a.wal.jsonl"); + let segments: Vec<_> = std::fs::read_dir(&lane_dir) + .unwrap() + .map(|entry| entry.unwrap().path()) + .collect(); + let torn = lane_dir.join("torn-aside"); + std::fs::rename(segments.iter().max().unwrap(), &torn).unwrap(); + tailer + .poll_once() + .await + .expect_err("the torn lane fails this poll"); + + // The transient clears and a per-request loader (ensure_hot's + // shape) hydrates the stem first: the family signature is now + // current, so the next retarget will not report it stale. + std::fs::rename(&torn, segments.iter().max().unwrap()).unwrap(); + hydrator + .ensure_context("ctx_a") + .expect("the reader's own hydration lands"); + assert_eq!( + state.directory_entry("ctx_a").unwrap().description, + "old", + "hydration alone re-reads no meta — the refresh is still owed" + ); + + tailer.poll_once().await.expect("the owed refresh lands"); + assert_eq!( + state.directory_entry("ctx_a").unwrap().description, + "new", + "the remembered debt must pay out even though retarget \ + reported nothing stale" + ); + + for dir in [bucket, writer, target] { + let _ = std::fs::remove_dir_all(dir); + } + } + #[tokio::test] async fn a_restart_mid_apply_re_verifies_the_cache_and_heals() { let (bucket, writer) = two_segment_bucket("midapply").await; diff --git a/tests/http_api/consolidation.rs b/tests/http_api/consolidation.rs index 11fd96a0..d8c6d0ea 100644 --- a/tests/http_api/consolidation.rs +++ b/tests/http_api/consolidation.rs @@ -166,6 +166,21 @@ fn sections_detect_join_and_fingerprint_their_candidates() { Some(json!({"checks": ["typo"]})), ); assert_eq!(status, 400, "{body}"); + // Past the shared list ceiling: `dedup` folds only consecutive + // repeats, so an alternating list this long would otherwise pass + // both selector guards. + let alternating: Vec<&str> = ["merge", "contradiction"] + .into_iter() + .cycle() + .take(1001) + .collect(); + let (status, body) = server.call( + "POST", + "/contexts/sake/consolidation/audit", + Some(json!({"checks": alternating})), + ); + assert_eq!(status, 400, "{body}"); + assert_eq!(body["code"], json!("over_limit"), "{body}"); let (status, _) = server.call( "POST", "/contexts/nope/consolidation/audit", From 8c208b060a420da805144c91571806d654b5d98b Mon Sep 17 00:00:00 2001 From: Takashi Yamashina Date: Sun, 9 Aug 2026 19:27:16 +0900 Subject: [PATCH 2/2] promote review fixes: record the refresh debt before the shared pass; zero WAL gauges only on NotFound MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit on #523: - replica: report.stale joins pending_refresh BEFORE hydrate_shared can fail the poll — a debt recorded only after the shared pass was lost with the error, reopening the same reader-hydrates-first freeze one step earlier. Regression test tears the published meta.json object so the shared pass itself aborts the poll. - replication: the WAL gauge re-stat zeroes only on NotFound (no WAL shipped yet); any other stat failure keeps the last-known value instead of walking a live gauge to nothing on a transient error. Claude-Session: https://claude.ai/code/session_011NozdDS9JqgCpi9Z3wo4Pd --- src/registry/replication.rs | 72 +++++++++++++++++------ src/replica.rs | 114 +++++++++++++++++++++++++++++++++--- 2 files changed, 162 insertions(+), 24 deletions(-) diff --git a/src/registry/replication.rs b/src/registry/replication.rs index 7abd89b6..69ec7d88 100644 --- a/src/registry/replication.rs +++ b/src/registry/replication.rs @@ -110,13 +110,20 @@ impl AppState { // increments, and `ensure_hot`'s own re-stat runs only for // pinned entries below (or on the next local read) — without // this, a cold unpinned context the tailer keeps growing - // understates `taguru_wal_bytes` indefinitely. - inner.wal_bytes = std::fs::metadata(wal_path(&self.0.data_dir, &stem)) - .map(|meta| meta.len()) - .unwrap_or(0); - inner.passages_wal_bytes = std::fs::metadata(passages_wal_path(&self.0.data_dir, &stem)) - .map(|meta| meta.len()) - .unwrap_or(0); + // understates `taguru_wal_bytes` indefinitely. `NotFound` is + // the one honest zero (no WAL shipped for this lane yet); any + // other stat failure keeps the last-known value rather than + // walking a live gauge down to nothing on a transient error. + let restat = |path: &std::path::Path, last: u64| match std::fs::metadata(path) { + Ok(meta) => meta.len(), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => 0, + Err(_) => last, + }; + inner.wal_bytes = restat(&wal_path(&self.0.data_dir, &stem), inner.wal_bytes); + inner.passages_wal_bytes = restat( + &passages_wal_path(&self.0.data_dir, &stem), + inner.passages_wal_bytes, + ); if matches!(inner.slot, Slot::Hot(_)) { inner.slot = Slot::Cold; // The same bump eviction does: a flush that staged this @@ -268,16 +275,47 @@ mod tests { fs::write(passages_wal_path(&dir, &stem), b"{\"tailed\":2}\n").unwrap(); state.replica_refresh("sake"); let entry = state.lookup("sake").unwrap(); - let inner = entry.inner.read(); - assert_eq!( - inner.wal_bytes, - fs::metadata(wal_path(&dir, &stem)).unwrap().len() - ); - assert_eq!( - inner.passages_wal_bytes, - fs::metadata(passages_wal_path(&dir, &stem)).unwrap().len() - ); - drop(inner); + { + let inner = entry.inner.read(); + assert_eq!( + inner.wal_bytes, + fs::metadata(wal_path(&dir, &stem)).unwrap().len() + ); + assert_eq!( + inner.passages_wal_bytes, + fs::metadata(passages_wal_path(&dir, &stem)).unwrap().len() + ); + } + + // A vanished WAL is the one honest zero. + fs::remove_file(wal_path(&dir, &stem)).unwrap(); + fs::remove_file(passages_wal_path(&dir, &stem)).unwrap(); + state.replica_refresh("sake"); + { + let inner = entry.inner.read(); + assert_eq!(inner.wal_bytes, 0); + assert_eq!(inner.passages_wal_bytes, 0); + } + + // Any OTHER stat failure keeps the last-known value instead of + // walking a live gauge down to nothing on a transient error. + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + fs::write(wal_path(&dir, &stem), b"{\"tailed\":3}\n").unwrap(); + fs::write(passages_wal_path(&dir, &stem), b"{\"tailed\":4}\n").unwrap(); + state.replica_refresh("sake"); + let before = { + let inner = entry.inner.read(); + (inner.wal_bytes, inner.passages_wal_bytes) + }; + assert_ne!(before, (0, 0)); + fs::set_permissions(&dir, fs::Permissions::from_mode(0o000)).unwrap(); + state.replica_refresh("sake"); + fs::set_permissions(&dir, fs::Permissions::from_mode(0o755)).unwrap(); + let inner = entry.inner.read(); + assert_eq!((inner.wal_bytes, inner.passages_wal_bytes), before); + } let _ = fs::remove_dir_all(dir); } diff --git a/src/replica.rs b/src/replica.rs index 22885d5b..c3c06c45 100644 --- a/src/replica.rs +++ b/src/replica.rs @@ -304,6 +304,16 @@ impl Tailer { // Applied seqs are per-lineage: see `reset_replica_lanes`. self.state.metrics().reset_replica_lanes(); } + // Stale stems join the refresh debt BEFORE anything below can + // fail the poll; the worklist is the whole debt, not this + // retarget's report — a stem whose hydration a per-request + // loader completed after this tailer's own failed attempt (a + // family fetch below, or `hydrate_shared` erroring out of the + // whole poll) never turns stale again, but its refresh is + // still owed (`ensure_context` on a settled stem is O(1), so + // paying the debt late costs one meta re-read, not a + // re-hydration). + self.pending_refresh.extend(report.stale.iter().cloned()); for stem in &report.vanished { let Some(name) = crate::registry::name_from_stem(stem) else { continue; @@ -319,13 +329,6 @@ impl Tailer { self.hydrator.hydrate_shared().await?; self.state.replica_reload_groups(); - // Stale stems join the refresh debt; the worklist is the whole - // debt, not this retarget's report — a stem whose hydration a - // per-request loader completed after this tailer's own failed - // attempt never turns stale again, but its refresh is still - // owed (`ensure_context` on a settled stem is O(1), so paying - // the debt late costs one meta re-read, not a re-hydration). - self.pending_refresh.extend(report.stale.iter().cloned()); let worklist: Vec = self.pending_refresh.iter().cloned().collect(); let mut failed: std::collections::BTreeSet<&str> = std::collections::BTreeSet::new(); for stem in &worklist { @@ -676,6 +679,103 @@ mod tests { } } + /// The debt must be recorded before `hydrate_shared` can fail the + /// poll: a torn shared object (here the published meta.json) + /// aborts the poll before any family applies — and if a + /// per-request loader hydrates the stem before the next poll, + /// retarget never reports it stale again, so a debt recorded only + /// after the shared pass would have been lost with the error. + #[tokio::test] + async fn a_refresh_owed_survives_a_shared_hydration_failure() { + let bucket = scratch("shared-owed-bucket"); + let writer = scratch("shared-owed-writer"); + std::fs::write(writer.join("ctx_a.ctx"), b"image-v1").unwrap(); + std::fs::write( + writer.join("ctx_a.meta.json"), + br#"{"description":"old","pinned":false}"#, + ) + .unwrap(); + wal::append_batch(&writer.join("ctx_a.wal.jsonl"), 1, &[associate("a")]).unwrap(); + let writer_state = AppState::boot(writer.clone(), 64 * 1024 * 1024, None).unwrap(); + let mut shipper = Shipper::claim( + local_store(&bucket), + StorePath::default(), + url_of("shared-owed"), + writer.clone(), + Arc::new(ShipProgress::new()), + writer_state, + None, + ) + .await + .unwrap(); + shipper.cycle().await.unwrap(); + + let url = url_of("shared-owed"); + let store = local_store(&bucket); + let target = scratch("shared-owed-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.clone(), + target.clone(), + state.clone(), + hydrator.clone(), + ); + tailer.poll_once().await.expect("the first manifest lands"); + assert_eq!(state.directory_entry("ctx_a").unwrap().description, "old"); + + std::fs::write( + writer.join("ctx_a.meta.json"), + br#"{"description":"new","pinned":false}"#, + ) + .unwrap(); + wal::append_batch(&writer.join("ctx_a.wal.jsonl"), 2, &[associate("b")]).unwrap(); + shipper.cycle().await.unwrap(); + shipper.retire_generation().await; + + // Tear the published meta object: the shared pass fails the + // whole poll before any family is even attempted. + let meta_object = bucket + .join("gen-00000000000000000001") + .join("files") + .join("ctx_a.meta.json"); + let torn = meta_object.with_extension("torn-aside"); + std::fs::rename(&meta_object, &torn).unwrap(); + tailer + .poll_once() + .await + .expect_err("a torn shared object fails the poll"); + + // The transient clears and a reader hydrates the family first: + // the next retarget will not report the stem stale. + std::fs::rename(&torn, &meta_object).unwrap(); + hydrator + .ensure_context("ctx_a") + .expect("the reader's own hydration lands"); + assert_eq!( + state.directory_entry("ctx_a").unwrap().description, + "old", + "hydration alone re-reads no meta — the refresh is still owed" + ); + + tailer.poll_once().await.expect("the owed refresh lands"); + assert_eq!( + state.directory_entry("ctx_a").unwrap().description, + "new", + "a debt recorded only after the shared pass would have been \ + lost with the error" + ); + + for dir in [bucket, writer, target] { + let _ = std::fs::remove_dir_all(dir); + } + } + #[tokio::test] async fn a_restart_mid_apply_re_verifies_the_cache_and_heals() { let (bucket, writer) = two_segment_bucket("midapply").await;