diff --git a/src/chunker.rs b/src/chunker.rs index 81eb7d5..e65e913 100644 --- a/src/chunker.rs +++ b/src/chunker.rs @@ -373,14 +373,28 @@ impl Chunker for RecordAlignedChunker { if !self.header_done { return Ok(()); } + // header_done just flipped true, so complete_len is still 0 here + // (nothing beyond the header has been considered yet) -- + // pending is simply the fresh post-header bytes. self.pending = bytes[GLOBAL_HEADER_LEN..].to_vec(); - self.complete_len = 0; self.rescan(trust_trailing_record); self.last_growth = Instant::now(); return Ok(()); } - self.pending = bytes.to_vec(); - self.complete_len = 0; + // `pending[..complete_len]` is CONFIRMED content that just hasn't + // been drained (emitted) yet -- e.g. because it's still under + // `target`. `bytes` starts exactly at `confirmed_len()` (checked + // above), i.e. exactly where that confirmed prefix ends, so it must + // be appended after it, never used to replace the whole buffer -- + // doing that would silently discard already-confirmed, already + // real, not-yet-emitted content on every single re-peek, which is + // catastrophic for a long-running growing transfer (each poll would + // re-derive from scratch instead of building on prior confirmed + // progress, and once the fresh read window is too small to + // re-establish the same progress alone -- e.g. capped by + // READ_WINDOW -- confirmed_len can never recover). + self.pending.truncate(self.complete_len); + self.pending.extend_from_slice(bytes); self.rescan(trust_trailing_record); self.last_growth = Instant::now(); Ok(()) @@ -704,4 +718,57 @@ mod tests { c.refresh_tail(offset, &bytes, true).unwrap(); assert_eq!(c.confirmed_len(), offset, "trust must not rescue a nonempty torn fragment"); } + + #[test] + fn refresh_tail_never_discards_already_confirmed_undrained_content() { + // Regression test for a real production incident: refresh_tail used + // to unconditionally replace `pending` with the fresh read, which + // silently discarded already-confirmed-but-not-yet-drained content + // (anything below `target`, i.e. never emitted via ready_chunk) any + // time a LATER refresh_tail call's fresh window started past it and + // didn't happen to re-derive the same progress on its own. This + // caused confirmed_len to oscillate or, worse, get permanently + // stuck once the fresh window was too small (e.g. capped by + // READ_WINDOW in tail.rs) to ever re-establish the lost progress. + // + // Use a target far bigger than the data so nothing ever drains via + // ready_chunk -- everything must survive purely via the confirmed + // prefix being preserved across refresh_tail calls. + let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true); + c.refresh_tail(0, &header_bytes(0xA1B2C3D4), false).unwrap(); + + let rec1 = record_bytes_with_ts(&[1u8; 10], false, 1_700_000_001); + let rec2 = record_bytes_with_ts(&[2u8; 10], false, 1_700_000_002); + let rec3 = record_bytes_with_ts(&[3u8; 10], false, 1_700_000_003); + + // First call: feed rec1+rec2 together. rec1 confirms (proven by + // rec2's header); rec2 itself is held back (nothing proves it yet). + let offset = c.confirmed_len(); + let mut first_batch = rec1.clone(); + first_batch.extend_from_slice(&rec2); + c.refresh_tail(offset, &first_batch, false).unwrap(); + assert_eq!( + c.confirmed_len(), + offset + rec1.len() as u64, + "rec1 should confirm, proven by rec2's header" + ); + let after_first = c.confirmed_len(); + + // Second call: a FRESH read starting exactly at confirmed_len() + // (per the trait contract) -- this does NOT include rec1, which is + // already confirmed but still sitting undrained in `pending`. Only + // rec2's own bytes (already seen) plus rec3 (new) are read this + // time, simulating a subsequent, narrower re-peek. + let mut second_batch = rec2.clone(); + second_batch.extend_from_slice(&rec3); + c.refresh_tail(after_first, &second_batch, false).unwrap(); + + // rec1's confirmation must NOT have been discarded, and rec2 should + // now also confirm (proven by rec3's header). + assert_eq!( + c.confirmed_len(), + offset + rec1.len() as u64 + rec2.len() as u64, + "confirmed_len must never regress below what was already established" + ); + } } diff --git a/src/tail.rs b/src/tail.rs index cb347ed..a713f76 100644 --- a/src/tail.rs +++ b/src/tail.rs @@ -111,6 +111,16 @@ fn drain( // and still made no progress -- if so, the outcome is deterministic and // we bail immediately rather than waiting further. let mut trust_attempted_without_progress = false; + // Diagnostic-only (partial branch): throttled "we have bytes to look at + // but confirmed_len isn't advancing" logging, both before and after + // rename, with a byte preview of what's actually at the boundary each + // time -- this is deliberately verbose for now (see git log), to + // distinguish "the writer hasn't produced a valid next record yet" from + // "repeated reads keep returning the exact same stale bytes", which + // would point at a page-cache/read-coherency issue rather than the + // writer's own timing. + let mut last_stuck_log: Option = None; + let mut last_stuck_peek: Option> = None; loop { let file_len = file.metadata()?.len(); @@ -131,10 +141,34 @@ fn drain( if n > 0 { chunker.refresh_tail(confirmed, &buf[..n], trust_trailing)?; emit_ready(chunker, events)?; - let advanced = chunker.confirmed_len() > confirmed; + let new_confirmed = chunker.confirmed_len(); + let advanced = new_confirmed > confirmed; made_progress = advanced; trust_attempted_without_progress = trust_trailing && !advanced; shrunk_below_anchor = false; + + if advanced { + log.log(format_args!( + "tail: progress confirmed {confirmed} -> {new_confirmed} \ + (+{} bytes), file_len={file_len}, trust_trailing={trust_trailing}", + new_confirmed - confirmed + )); + last_stuck_log = None; + last_stuck_peek = None; + } else { + let preview = &buf[..n.min(32)]; + if last_stuck_log.is_none_or(|t| t.elapsed() >= Duration::from_secs(1)) { + last_stuck_log = Some(Instant::now()); + let changed = last_stuck_peek.as_deref() != Some(preview); + let hex: String = preview.iter().map(|b| format!("{b:02x}")).collect(); + log.log(format_args!( + "tail: stuck at confirmed={confirmed}, file_len={file_len}, read_n={n}, \ + trust_trailing={trust_trailing}, complete_signal={complete_signal}, \ + peek_changed_since_last_check={changed}, peek={hex}" + )); + } + last_stuck_peek = Some(preview.to_vec()); + } } } else if chunker.should_flush(Instant::now()) && let Some((kind, data)) = chunker.flush()