Fix confirmed_len regression bug in refresh_tail; add verbose tail logging

Root cause of the production hang reported after the previous fix:
refresh_tail unconditionally replaced the chunker's entire `pending` buffer
with the fresh read, discarding pending[..complete_len] -- content already
confirmed real but not yet drained (anything under the 256KiB emit target).
Since a fresh read always starts exactly at confirmed_len (past that
already-confirmed prefix), the next rescan() started from complete_len=0
within a buffer that no longer contained the records needed to reprove it,
silently regressing confirmed_len. In a small buffer this just caused
wasteful oscillation (each poll's full remaining read happened to
re-derive the same progress); at production scale, once a poll's window
failed to independently re-establish the same high-water mark (e.g. capped
by READ_WINDOW, or landing on an unlucky boundary), confirmed_len could
regress and then get permanently wedged well behind the real write cursor,
manifesting as a "torn/corrupt" bail with confirmed_len frozen at a value
far below the real content size, despite the source visibly still growing.

Fix: refresh_tail now truncates pending to complete_len (keeping the
confirmed-but-undrained prefix intact) before appending the fresh bytes,
instead of replacing pending wholesale. Added a regression test
(refresh_tail_never_discards_already_confirmed_undrained_content) that
fails against the old behavior and passes against the fix -- verified by
temporarily reverting the fix and confirming the test catches it.

Also add verbose (deliberately noisy for now) tail.rs diagnostics: a log
line on every confirmed_len advance, and a throttled "stuck" line
(including a byte preview of what's at the confirmed boundary and whether
it changed since the last check) whenever there's more to read but nothing
validates -- both before and after rename, where previously there was no
progress visibility at all before rename was observed. This is what
surfaced the bug: real production logs showed confirmed_len permanently
frozen at a fixed byte count for 4+ minutes while the source kept growing,
which is inconsistent with the "held back by one record" design and
pointed straight at a state-management bug rather than a writer-side or
filesystem-caching issue.

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();
+        // `pending[..complete_len]` is CONFIRMED content that just hasn't
-        self.complete_len = 0;
+        // 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"
+        );
+    }
 }

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<Instant> = None;
+    let mut last_stuck_peek: Option<Vec<u8>> = 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()