Track confirmed pcap content instead of raw file length for growth/finalize

A preallocating capture writer (ftruncate-extend, zero-filled) reports the
full reserved file size immediately and fills real data into it in place,
without file length ever reflecting real progress until a final truncate at
rotation. The tailer trusted raw file length for both "is there new data"
and "are we done", so it read (and, in raw mode, forwarded) preallocated
zero padding as real packet data, then could never satisfy pos == file_len
once the file was truncated down at rotation -- a permanent hang, and worse,
a corrupted destination even when the hang didn't bite.

Replace file-length-based tracking with confirmed_len(), derived from
actually walking pcap records (RecordAlignedChunker), for any partial
(still-growing) source -- non-compress non-partial transfers are untouched,
preserving today's "arbitrary content" guarantee there. Growth detection now
re-reads from the last confirmed boundary every poll rather than only ever
reading forward past what's already been read, since a preallocating writer
can flip a byte from zero to real content without file length ever changing.

Two correctness properties enforced by the new record scanner, both gated to
partial sources only:
- A record's header being fully present and plausible is not proof its
  payload is real (header and payload aren't necessarily written
  atomically) -- a candidate record is only confirmed once the *next*
  record's header has also been observed and looks real, proof the writer
  moved past it. This holds complete_len one record behind by construction.
- That rule alone would starve the true last record of any capture forever,
  so a narrow escape hatch trusts the trailing record on its own
  plausibility once the rename has been observed and the file's length has
  been stable across continuous re-checks for a grace period -- backed by
  the writer's own "I'm done" signal (the rename), not a timing guess alone.

Also: an all-zero global header on a partial source is now treated as "not
written yet" rather than a hard error, for the same preallocation reason.

Verified against a live preallocation simulation over ssh (locl.sh): a
source truncated to a padded size well beyond its real content, with a
writer catch-up (in-place record write with no length change) before the
final truncate-and-rename, transfers with no hang and a byte-exact
destination -- confirmed_len stalls precisely at the real/padding boundary
and only advances once content, not file length, proves growth.

src/chunker.rs

@@ -6,6 +6,10 @@ pub const RECORD_HEADER_LEN: usize = 16;
 pub const DEFAULT_FLUSH: Duration = Duration::from_millis(100);
 #[allow(dead_code)]
 pub const DEFAULT_TARGET_BYTES: usize = 256 * 1024;
+/// Sanity bound on a record's declared payload length, per
+/// STREAMING_ZSTD_FORMAT.md §8/§9. Only enforced for partial (still-growing)
+/// sources -- see `RecordAlignedChunker::partial`.
+pub const MAX_INCL_LEN: usize = 262_144;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum ChunkKind {
@@ -37,8 +41,9 @@ fn read_u32(bytes: &[u8], e: Endianness) -> u32 {
 }
 
 /// Turns a stream of raw source bytes into chunks ready to hand to the wire
-/// protocol. `RawChunker` is a passthrough for the non-compress path;
+/// protocol. `RawChunker` is a passthrough for the non-compress, non-partial
-/// `RecordAlignedChunker` cuts on pcap record boundaries for the compress path.
+/// path; `RecordAlignedChunker` cuts on pcap record boundaries for the
+/// compress path and/or any partial (still-growing) source.
 pub trait Chunker: Send {
     fn feed(&mut self, bytes: &[u8]) -> Result<()>;
     /// Pop a chunk that's ready by size/target. Call repeatedly until `None`.
@@ -47,14 +52,50 @@ pub trait Chunker: Send {
     fn should_flush(&self, now: Instant) -> bool;
     /// Force-cut whatever's pending (still respecting record alignment).
     fn flush(&mut self) -> Option<(ChunkKind, Vec<u8>)>;
-    /// Called once, after the source is confirmed fully drained. Errors if
+    /// Called once, after the source is confirmed fully drained (per the
-    /// what's left doesn't cleanly end at a valid boundary.
+    /// caller's own `confirmed_len() == real final length` check -- this no
+    /// longer re-derives that itself). Only errors if a complete global
+    /// header was never read.
     fn finish(&mut self) -> Result<Option<(ChunkKind, Vec<u8>)>>;
+
+    /// Next source offset the caller should read from to get new bytes via
+    /// `feed`. Monotonically non-decreasing.
+    fn read_from(&self) -> u64;
+
+    /// Cumulative leading source bytes confirmed to form a complete,
+    /// plausible global header and/or pcap records. Monotonically
+    /// non-decreasing. Equal to `read_from()` for a chunker with no
+    /// "unconfirmed tail" concept.
+    fn confirmed_len(&self) -> u64;
+
+    /// Re-validating feed for a source that may still be overwriting
+    /// not-yet-confirmed bytes in place (a preallocating writer). `bytes`
+    /// MUST be a fresh disk read starting exactly at `confirmed_len()`; it
+    /// replaces whatever was buffered beyond the last confirmed boundary and
+    /// re-validates from scratch. Only called by tail.rs's partial-source
+    /// path.
+    ///
+    /// `trust_trailing_record`: normally a candidate record is only
+    /// confirmed once the *next* record's header has also been observed and
+    /// looks real (proof the writer moved past it). When the caller is
+    /// certain no more bytes will EVER arrive (rename observed and file
+    /// length stable across repeated checks), it passes `true` to relax
+    /// that requirement for exactly the trailing record: a record that is
+    /// itself plausible AND is followed by nothing at all gets confirmed
+    /// without needing a next-record proof, since there provably isn't one.
+    ///
+    /// Default delegates to `feed` -- correct for a chunker with no
+    /// unconfirmed-tail/next-record concept (RawChunker).
+    fn refresh_tail(&mut self, offset: u64, bytes: &[u8], trust_trailing_record: bool) -> Result<()> {
+        let _ = (offset, trust_trailing_record);
+        self.feed(bytes)
+    }
 }
 
 #[derive(Default)]
 pub struct RawChunker {
     pending: Vec<u8>,
+    total_fed: u64,
 }
 
 impl RawChunker {
@@ -66,6 +107,7 @@ impl RawChunker {
 impl Chunker for RawChunker {
     fn feed(&mut self, bytes: &[u8]) -> Result<()> {
         self.pending.extend_from_slice(bytes);
+        self.total_fed += bytes.len() as u64;
         Ok(())
     }
 
@@ -88,6 +130,14 @@ impl Chunker for RawChunker {
     fn finish(&mut self) -> Result<Option<(ChunkKind, Vec<u8>)>> {
         Ok(self.ready_chunk())
     }
+
+    fn read_from(&self) -> u64 {
+        self.total_fed
+    }
+
+    fn confirmed_len(&self) -> u64 {
+        self.total_fed
+    }
 }
 
 pub struct RecordAlignedChunker {
@@ -95,29 +145,59 @@ pub struct RecordAlignedChunker {
     header_done: bool,
     header_ready: bool,
     endianness: Option<Endianness>,
+    snaplen: u32,
     pending: Vec<u8>,
     complete_len: usize,
+    emitted_len: u64,
     last_growth: Instant,
     target: usize,
     flush_after: Duration,
+    /// Whether this source is a still-growing (partial) file. Gates every
+    /// heuristic below (zero-header wait, zero-ts_sec rejection, incl_len
+    /// bound, next-record proof requirement) -- a non-partial source is
+    /// stable and final, so none of these preallocation-writer defenses are
+    /// needed or safe to apply (they could reject genuinely final content).
+    partial: bool,
 }
 
 impl RecordAlignedChunker {
-    pub fn new(target: usize, flush_after: Duration) -> Self {
+    pub fn new(target: usize, flush_after: Duration, partial: bool) -> Self {
         Self {
             header_buf: Vec::with_capacity(GLOBAL_HEADER_LEN),
             header_done: false,
             header_ready: false,
             endianness: None,
+            snaplen: 0,
             pending: Vec::new(),
             complete_len: 0,
+            emitted_len: 0,
             last_growth: Instant::now(),
             target,
             flush_after,
+            partial,
         }
     }
 
-    fn rescan(&mut self) {
+    fn max_incl_len(&self) -> usize {
+        (self.snaplen as usize).max(MAX_INCL_LEN)
+    }
+
+    fn validate_header(&mut self) -> Result<()> {
+        // On a partial source, a header that hasn't been written yet reads
+        // back as zero bytes if the writer preallocated the file before
+        // writing into it -- treat that as "not yet", not a hard error.
+        if self.partial && self.header_buf.iter().all(|&b| b == 0) {
+            return Ok(());
+        }
+        let e = detect_endianness(&self.header_buf)?;
+        self.snaplen = read_u32(&self.header_buf[16..20], e);
+        self.endianness = Some(e);
+        self.header_done = true;
+        self.header_ready = true;
+        Ok(())
+    }
+
+    fn rescan(&mut self, trust_trailing_record: bool) {
         let e = match self.endianness {
             Some(e) => e,
             None => return,
@@ -127,17 +207,65 @@ impl RecordAlignedChunker {
             if remaining.len() < RECORD_HEADER_LEN {
                 break;
             }
+            let ts_sec = read_u32(&remaining[0..4], e);
             let incl_len = read_u32(&remaining[8..12], e) as usize;
             let record_total = RECORD_HEADER_LEN + incl_len;
             if remaining.len() < record_total {
                 break;
             }
+
+            if self.partial {
+                // Heuristic: a live capture's timestamp is never the Unix
+                // epoch. An all-zero-looking ts_sec is essentially
+                // certainly unwritten/preallocated disk space, not a real
+                // record -- treat it exactly like "not enough bytes yet"
+                // (pause, no error), so a later re-peek can pick up real
+                // content once the writer's cursor reaches this offset.
+                if ts_sec == 0 {
+                    break;
+                }
+                if incl_len > self.max_incl_len() {
+                    break;
+                }
+
+                // Never trust a record's own header alone: a writer need
+                // not flush header and payload atomically, so a
+                // fully-plausible header can still be sitting in front of a
+                // payload that's still mid-write (or still zero padding).
+                // Require positive proof the writer has moved past this
+                // record -- the start of the NEXT record's header, itself
+                // looking real -- before confirming it. This makes
+                // complete_len always sit (at least) one record behind
+                // whatever's currently visible, by construction, for every
+                // record, not just "the last one" as a special case.
+                let after = &remaining[record_total..];
+                let next_is_proven = after.len() >= RECORD_HEADER_LEN
+                    && read_u32(&after[0..4], e) != 0
+                    && (read_u32(&after[8..12], e) as usize) <= self.max_incl_len();
+
+                // Escape hatch (only when the caller is certain nothing
+                // more will EVER arrive): a record that is itself plausible
+                // AND has truly nothing after it (current_len ends exactly
+                // here) is confirmed without a next-record proof, since
+                // there provably isn't a next record to prove it with. Any
+                // other leftover (a nonempty `after` shorter than a full
+                // header, or one that fails its own plausibility check)
+                // still does NOT confirm -- that's a genuinely torn/corrupt
+                // tail, correctly left unconfirmed so tail.rs's finalize
+                // check reports it as such.
+                let trailing_and_trusted = trust_trailing_record && after.is_empty();
+
+                if !next_is_proven && !trailing_and_trusted {
+                    break;
+                }
+            }
             self.complete_len += record_total;
         }
     }
 
     fn drain_complete(&mut self) -> (ChunkKind, Vec<u8>) {
         let chunk: Vec<u8> = self.pending.drain(..self.complete_len).collect();
+        self.emitted_len += self.complete_len as u64;
         self.complete_len = 0;
         (ChunkKind::Data, chunk)
     }
@@ -155,9 +283,11 @@ impl Chunker for RecordAlignedChunker {
             self.header_buf.extend_from_slice(&rest[..take]);
             rest = &rest[take..];
             if self.header_buf.len() == GLOBAL_HEADER_LEN {
-                self.endianness = Some(detect_endianness(&self.header_buf)?);
+                self.validate_header()?;
-                self.header_done = true;
+                if !self.header_done {
-                self.header_ready = true;
+                    // Partial source, all-zero header -- not written yet.
+                    return Ok(());
+                }
             } else {
                 // Still accumulating the header; nothing else to do.
                 return Ok(());
@@ -165,7 +295,7 @@ impl Chunker for RecordAlignedChunker {
         }
         if !rest.is_empty() {
             self.pending.extend_from_slice(rest);
-            self.rescan();
+            self.rescan(false);
             self.last_growth = Instant::now();
         }
         Ok(())
@@ -198,15 +328,63 @@ impl Chunker for RecordAlignedChunker {
         if !self.header_done {
             bail!("source ended before a complete pcap global header was read");
         }
-        if self.pending.len() != self.complete_len {
-            bail!("truncated trailing pcap record at end of source");
-        }
         if self.complete_len > 0 {
             Ok(Some(self.drain_complete()))
         } else {
             Ok(None)
         }
     }
+
+    fn read_from(&self) -> u64 {
+        if !self.header_done {
+            self.header_buf.len() as u64
+        } else {
+            GLOBAL_HEADER_LEN as u64 + self.emitted_len + self.pending.len() as u64
+        }
+    }
+
+    fn confirmed_len(&self) -> u64 {
+        if !self.header_done {
+            // While not yet validated, nothing is confirmed -- even if
+            // header_buf is "full" but stuck on an all-zero (partial-only)
+            // header. This must be 0, not header_buf.len(), so tail.rs's
+            // partial-branch read anchor re-reads offset 0 instead of
+            // treating the stuck zero bytes as already consumed.
+            0
+        } else {
+            GLOBAL_HEADER_LEN as u64 + self.emitted_len + self.complete_len as u64
+        }
+    }
+
+    fn refresh_tail(&mut self, offset: u64, bytes: &[u8], trust_trailing_record: bool) -> Result<()> {
+        if offset != self.confirmed_len() {
+            bail!(
+                "chunker refresh_tail offset mismatch: got {offset}, expected {}",
+                self.confirmed_len()
+            );
+        }
+        if !self.header_done {
+            if bytes.len() < GLOBAL_HEADER_LEN {
+                self.header_buf = bytes.to_vec();
+                return Ok(());
+            }
+            self.header_buf = bytes[..GLOBAL_HEADER_LEN].to_vec();
+            self.validate_header()?;
+            if !self.header_done {
+                return Ok(());
+            }
+            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;
+        self.rescan(trust_trailing_record);
+        self.last_growth = Instant::now();
+        Ok(())
+    }
 }
 
 #[cfg(test)]
@@ -220,8 +398,17 @@ mod tests {
     }
 
     fn record_bytes(payload: &[u8], big_endian: bool) -> Vec<u8> {
+        record_bytes_with_ts(payload, big_endian, 0)
+    }
+
+    fn record_bytes_with_ts(payload: &[u8], big_endian: bool, ts_sec: u32) -> Vec<u8> {
         let mut r = Vec::with_capacity(RECORD_HEADER_LEN + payload.len());
-        r.extend_from_slice(&[0u8; 8]); // ts_sec, ts_usec
+        if big_endian {
+            r.extend_from_slice(&ts_sec.to_be_bytes());
+        } else {
+            r.extend_from_slice(&ts_sec.to_le_bytes());
+        }
+        r.extend_from_slice(&[0u8; 4]); // ts_usec
         let incl_len = payload.len() as u32;
         if big_endian {
             r.extend_from_slice(&incl_len.to_be_bytes());
@@ -246,7 +433,7 @@ mod tests {
 
     #[test]
     fn header_emitted_as_its_own_chunk() {
-        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH);
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, false);
         c.feed(&header_bytes(0xA1B2C3D4)).unwrap();
         let (kind, data) = c.ready_chunk().unwrap();
         assert_eq!(kind, ChunkKind::Header);
@@ -256,7 +443,7 @@ mod tests {
 
     #[test]
     fn insufficient_header_bytes_is_not_an_error() {
-        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH);
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, false);
         // Only 10 bytes so far -- not enough to validate the magic.
         c.feed(&[0u8; 10]).unwrap();
         assert!(c.ready_chunk().is_none());
@@ -264,7 +451,7 @@ mod tests {
 
     #[test]
     fn bad_magic_is_an_error() {
-        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH);
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, false);
         let mut bad = vec![0xFFu8; GLOBAL_HEADER_LEN];
         bad[0..4].copy_from_slice(&0xDEADBEEFu32.to_le_bytes());
         assert!(c.feed(&bad).is_err());
@@ -272,7 +459,7 @@ mod tests {
 
     #[test]
     fn chunks_always_end_on_record_boundary_when_fed_in_small_increments() {
-        let mut c = RecordAlignedChunker::new(50, DEFAULT_FLUSH); // tiny target to force cuts
+        let mut c = RecordAlignedChunker::new(50, DEFAULT_FLUSH, false); // tiny target to force cuts
         let mut stream = header_bytes(0xA1B2C3D4);
         let mut expected_records: Vec<Vec<u8>> = Vec::new();
         for i in 0..20u8 {
@@ -307,7 +494,7 @@ mod tests {
 
     #[test]
     fn target_is_a_floor_not_a_ceiling_cuts_at_next_boundary() {
-        let mut c = RecordAlignedChunker::new(15, DEFAULT_FLUSH);
+        let mut c = RecordAlignedChunker::new(15, DEFAULT_FLUSH, false);
         c.feed(&header_bytes(0xA1B2C3D4)).unwrap();
         c.ready_chunk(); // drain header
 
@@ -321,7 +508,7 @@ mod tests {
 
     #[test]
     fn flush_predicate_and_flush_are_time_independent_of_sleep() {
-        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, Duration::from_millis(100));
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, Duration::from_millis(100), false);
         c.feed(&header_bytes(0xA1B2C3D4)).unwrap();
         c.ready_chunk();
         let rec = record_bytes(&[9u8; 5], false);
@@ -340,19 +527,25 @@ mod tests {
     }
 
     #[test]
-    fn truncated_trailing_record_at_finish_is_an_error() {
+    fn torn_trailing_record_is_left_unconfirmed_not_an_error() {
-        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH);
+        // finish()'s job narrowed to "was a header ever read" -- detecting a
+        // genuinely unresolvable torn tail at true EOF is tail.rs's job now
+        // (it has the file-length/rename context needed to tell "still
+        // resolving" from "will never resolve"). See tail.rs's
+        // `non_partial_torn_trailing_record_is_an_error` and
+        // `torn_trailing_record_after_rename_is_an_error_not_a_hang`.
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, false);
         c.feed(&header_bytes(0xA1B2C3D4)).unwrap();
         c.ready_chunk();
         let rec = record_bytes(&[1u8; 20], false);
         // Feed everything except the last 3 bytes of the record.
         c.feed(&rec[..rec.len() - 3]).unwrap();
-        assert!(c.finish().is_err());
+        assert_eq!(c.finish().unwrap(), None);
     }
 
     #[test]
     fn big_endian_magic_reads_incl_len_as_big_endian() {
-        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH);
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, false);
         c.feed(&header_bytes(0xD4C3B2A1)).unwrap();
         c.ready_chunk();
         let rec = record_bytes(&[1u8; 30], true);
@@ -363,7 +556,152 @@ mod tests {
 
     #[test]
     fn finish_before_any_header_bytes_is_an_error() {
-        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH);
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, false);
         assert!(c.finish().is_err());
     }
+
+    #[test]
+    fn all_zero_ts_sec_is_treated_as_not_yet_written() {
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true);
+        c.feed(&header_bytes(0xA1B2C3D4)).unwrap();
+        c.ready_chunk();
+        let rec1 = record_bytes_with_ts(&[1u8; 10], false, 1_700_000_000);
+        let rec2 = record_bytes_with_ts(&[2u8; 10], false, 1_700_000_001);
+        c.feed(&rec1).unwrap();
+        c.feed(&rec2).unwrap();
+        // rec1 confirms (proven by rec2's real header); rec2 itself is held
+        // back (nothing proves it yet).
+        assert_eq!(c.confirmed_len(), GLOBAL_HEADER_LEN as u64 + rec1.len() as u64);
+
+        // Now simulate trailing preallocated padding: a large all-zero
+        // block. confirmed_len must not advance into it.
+        let padding = vec![0u8; 4096];
+        c.feed(&padding).unwrap();
+        assert_eq!(c.confirmed_len(), GLOBAL_HEADER_LEN as u64 + rec1.len() as u64);
+    }
+
+    #[test]
+    fn refresh_tail_reexamines_previously_zero_region_once_it_fills_in() {
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true);
+        c.refresh_tail(0, &header_bytes(0xA1B2C3D4), false).unwrap();
+        assert_eq!(c.confirmed_len(), GLOBAL_HEADER_LEN as u64);
+
+        let zero_block = vec![0u8; 64];
+        c.refresh_tail(c.confirmed_len(), &zero_block, false).unwrap();
+        assert_eq!(c.confirmed_len(), GLOBAL_HEADER_LEN as u64);
+
+        // Writer has since filled this offset in with two real records.
+        let rec1 = record_bytes_with_ts(&[1u8; 10], false, 1_700_000_000);
+        let rec2 = record_bytes_with_ts(&[2u8; 10], false, 1_700_000_001);
+        let mut real = rec1.clone();
+        real.extend_from_slice(&rec2);
+        c.refresh_tail(c.confirmed_len(), &real, false).unwrap();
+        assert_eq!(c.confirmed_len(), GLOBAL_HEADER_LEN as u64 + rec1.len() as u64);
+    }
+
+    #[test]
+    fn all_zero_global_header_is_treated_as_not_yet_written() {
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true);
+        let zero_header = vec![0u8; GLOBAL_HEADER_LEN];
+        c.refresh_tail(0, &zero_header, false).unwrap();
+        assert_eq!(c.confirmed_len(), 0);
+
+        c.refresh_tail(0, &header_bytes(0xA1B2C3D4), false).unwrap();
+        assert_eq!(c.confirmed_len(), GLOBAL_HEADER_LEN as u64);
+    }
+
+    #[test]
+    fn nonzero_bad_magic_is_still_a_hard_error_even_when_partial() {
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true);
+        let mut bad = vec![0xFFu8; GLOBAL_HEADER_LEN];
+        bad[0..4].copy_from_slice(&0xDEADBEEFu32.to_le_bytes());
+        assert!(c.feed(&bad).is_err());
+    }
+
+    #[test]
+    fn incl_len_exceeding_bound_is_rejected() {
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true);
+        c.feed(&header_bytes(0xA1B2C3D4)).unwrap();
+        c.ready_chunk();
+        // incl_len far exceeds MAX_INCL_LEN, but we only need the header's
+        // 16 bytes physically present for the check to fire.
+        let mut rec = vec![0u8; RECORD_HEADER_LEN];
+        rec[0..4].copy_from_slice(&1_700_000_000u32.to_le_bytes());
+        rec[8..12].copy_from_slice(&(MAX_INCL_LEN as u32 + 1).to_le_bytes());
+        c.feed(&rec).unwrap();
+        assert_eq!(c.confirmed_len(), GLOBAL_HEADER_LEN as u64);
+    }
+
+    #[test]
+    fn record_not_confirmed_without_proof_of_next_record() {
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true);
+        c.feed(&header_bytes(0xA1B2C3D4)).unwrap();
+        c.ready_chunk();
+        let rec = record_bytes_with_ts(&[1u8; 10], false, 1_700_000_000);
+        c.feed(&rec).unwrap();
+        // A single fully-present, individually-plausible record with
+        // nothing after it: ordinary feed() (trust=false) must not confirm
+        // it -- no proof yet that the writer has moved past it.
+        assert_eq!(c.confirmed_len(), GLOBAL_HEADER_LEN as u64);
+    }
+
+    #[test]
+    fn record_confirmed_once_next_record_header_is_real() {
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true);
+        c.feed(&header_bytes(0xA1B2C3D4)).unwrap();
+        c.ready_chunk();
+        let rec1 = record_bytes_with_ts(&[1u8; 10], false, 1_700_000_000);
+        let rec2 = record_bytes_with_ts(&[2u8; 10], false, 1_700_000_001);
+        c.feed(&rec1).unwrap();
+        assert_eq!(c.confirmed_len(), GLOBAL_HEADER_LEN as u64);
+        c.feed(&rec2).unwrap();
+        assert_eq!(c.confirmed_len(), GLOBAL_HEADER_LEN as u64 + rec1.len() as u64);
+    }
+
+    #[test]
+    fn torn_payload_followed_by_zero_padding_is_not_confirmed() {
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true);
+        c.feed(&header_bytes(0xA1B2C3D4)).unwrap();
+        c.ready_chunk();
+        // A record whose own header looks plausible (real ts_sec, sane
+        // incl_len) but whose declared payload reaches into all-zero bytes.
+        let rec = record_bytes_with_ts(&[0u8; 10], false, 1_700_000_000);
+        c.feed(&rec).unwrap();
+        c.feed(&[0u8; 64]).unwrap(); // more zero padding after it
+        assert_eq!(c.confirmed_len(), GLOBAL_HEADER_LEN as u64);
+
+        // Real content lands there instead.
+        let rec2 = record_bytes_with_ts(&[1u8; 10], false, 1_700_000_001);
+        // Replace: simulate the writer having overwritten what followed rec
+        // by feeding a fresh, corrected view via refresh_tail.
+        c.refresh_tail(c.confirmed_len(), &[&rec[..], &rec2[..]].concat(), false)
+            .unwrap();
+        assert_eq!(c.confirmed_len(), GLOBAL_HEADER_LEN as u64 + rec.len() as u64);
+    }
+
+    #[test]
+    fn trailing_record_confirmed_only_when_trust_granted_and_truly_last() {
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true);
+        c.refresh_tail(0, &header_bytes(0xA1B2C3D4), false).unwrap();
+        let rec = record_bytes_with_ts(&[1u8; 10], false, 1_700_000_000);
+        let offset = c.confirmed_len();
+        c.refresh_tail(offset, &rec, false).unwrap();
+        assert_eq!(c.confirmed_len(), offset, "must not confirm without trust");
+
+        c.refresh_tail(offset, &rec, true).unwrap();
+        assert_eq!(c.confirmed_len(), offset + rec.len() as u64, "must confirm once trusted");
+    }
+
+    #[test]
+    fn trust_does_not_confirm_a_genuinely_torn_tail() {
+        let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true);
+        c.refresh_tail(0, &header_bytes(0xA1B2C3D4), false).unwrap();
+        let rec = record_bytes_with_ts(&[1u8; 10], false, 1_700_000_000);
+        let mut bytes = rec.clone();
+        // Nonempty garbage after the record -- not a full header, not empty.
+        bytes.extend_from_slice(&[0xAB, 0xCD, 0xEF]);
+        let offset = c.confirmed_len();
+        c.refresh_tail(offset, &bytes, true).unwrap();
+        assert_eq!(c.confirmed_len(), offset, "trust must not rescue a nonempty torn fragment");
+    }
 }

src/pipeline.rs

@@ -39,8 +39,8 @@ enum CompressedItem {
 /// serializing -- see the plan's Performance section. Non-compress mode
 /// collapses to 2 stages since there's no compression work to overlap.
 pub fn run_send(opts: SendOptions, mut sink: Box<dyn MessageSink>) -> Result<()> {
-    let chunker: Box<dyn Chunker> = if opts.compress {
+    let chunker: Box<dyn Chunker> = if opts.compress || opts.is_partial {
-        Box::new(RecordAlignedChunker::new(opts.chunk_target, DEFAULT_FLUSH))
+        Box::new(RecordAlignedChunker::new(opts.chunk_target, DEFAULT_FLUSH, opts.is_partial))
     } else {
         Box::new(RawChunker::new())
     };
@@ -205,9 +205,10 @@ mod tests {
         h
     }
 
-    fn record_bytes(payload: &[u8]) -> Vec<u8> {
+    fn record_bytes(payload: &[u8], ts_sec: u32) -> Vec<u8> {
         let mut r = Vec::new();
-        r.extend_from_slice(&[0u8; 8]);
+        r.extend_from_slice(&ts_sec.to_le_bytes());
+        r.extend_from_slice(&[0u8; 4]); // ts_usec
         let incl_len = payload.len() as u32;
         r.extend_from_slice(&incl_len.to_le_bytes());
         r.extend_from_slice(&incl_len.to_le_bytes());
@@ -251,10 +252,19 @@ mod tests {
 
     #[test]
     fn non_compress_partial_source_waits_for_rename() {
+        // Under the new chunker-selection (compress || is_partial), a raw
+        // partial transfer now routes through RecordAlignedChunker, which
+        // requires genuine pcap structure -- see the plan's "behavior
+        // change requiring sign-off". A real global header plus a real
+        // record is required; the second record is appended before rename
+        // so the first one gets its next-record proof quickly rather than
+        // waiting out the trailing-record grace period.
         let dir = tempdir().unwrap();
         let partial = dir.path().join("cap.pcap.partial");
         let final_src = dir.path().join("cap.pcap");
-        std::fs::write(&partial, b"growing-data").unwrap();
+        let mut stream = header_bytes(0xA1B2C3D4);
+        stream.extend_from_slice(&record_bytes(&[1u8; 10], 1_700_000_000));
+        std::fs::write(&partial, &stream).unwrap();
         let dest_final = dir.path().join("out.pcap");
         let dest_temp = dir.path().join("out.pcap.partial");
 
@@ -276,13 +286,21 @@ mod tests {
         });
         let send_handle = thread::spawn(move || run_send(opts, Box::new(ChannelSink(tx))));
 
+        thread::sleep(std::time::Duration::from_millis(60));
+        let rec2 = record_bytes(&[2u8; 10], 1_700_000_001);
+        {
+            let mut f = std::fs::OpenOptions::new().append(true).open(&partial).unwrap();
+            f.write_all(&rec2).unwrap();
+            f.sync_all().unwrap();
+        }
+        stream.extend_from_slice(&rec2);
         thread::sleep(std::time::Duration::from_millis(60));
         std::fs::rename(&partial, &final_src).unwrap();
 
         send_handle.join().unwrap().unwrap();
         recv_handle.join().unwrap().unwrap();
 
-        assert_eq!(std::fs::read(&dest_final).unwrap(), b"growing-data");
+        assert_eq!(std::fs::read(&dest_final).unwrap(), stream);
     }
 
     #[test]
@@ -291,7 +309,7 @@ mod tests {
         let src = dir.path().join("cap.pcap");
         let mut stream = header_bytes(0xA1B2C3D4);
         for i in 0..50u8 {
-            stream.extend_from_slice(&record_bytes(&[i; 100]));
+            stream.extend_from_slice(&record_bytes(&[i; 100], 0));
         }
         std::fs::write(&src, &stream).unwrap();
 

src/tail.rs

@@ -11,6 +11,16 @@ use std::time::{Duration, Instant};
 pub const POLL_INTERVAL: Duration = Duration::from_millis(20);
 const READ_WINDOW: usize = 1024 * 1024;
 
+/// How long a partial source's file length must sit unchanged, after the
+/// rename to final has been observed, before a trailing record that has no
+/// "next record" proof is trusted on its own plausibility. This is a
+/// defensive allowance for the rename and a separate truncate call becoming
+/// visible via `fstat` slightly apart from each other -- NOT a guess about
+/// whether the writer finished; the authoritative "writer is done" signal
+/// is the rename itself (`complete_signal`). Re-checked continuously (reset
+/// on every observed length change), not a one-shot timer.
+const FINALIZE_GRACE_PERIOD: Duration = Duration::from_millis(500);
+
 pub enum SourceEvent {
     Chunk(ChunkKind, Vec<u8>),
     Eof,
@@ -41,7 +51,7 @@ impl EventSink for std::sync::mpsc::SyncSender<SourceEvent> {
 /// after the rename below is observed) and feeds bytes to `chunker`, sending
 /// ready chunks to `events`. Implements STREAMING_ZSTD_FORMAT.md section 7:
 /// finalizes only once the rename to `final_path` has been observed AND every
-/// byte up to that point has been drained. If `source_path` was not itself a
+/// byte up to that point has been confirmed. If `source_path` was not itself a
 /// partial-suffixed name, it's treated as "born complete" (section 7.6).
 ///
 /// Per the confirmed design: if `source_path` doesn't exist at all when this
@@ -86,44 +96,72 @@ fn drain(
     log: &Logger,
 ) -> Result<()> {
     let file = File::open(source_path)?;
-    let mut pos: u64 = 0;
     let mut complete_signal = !is_partial;
-    let mut shrunk_below_pos = false;
+    let mut shrunk_below_anchor = false;
     let mut last_wait_log: Option<Instant> = None;
     let mut buf = vec![0u8; READ_WINDOW];
 
+    // Partial-branch-only: track how long current_len has sat unchanged, to
+    // gate trusting an unproven trailing record (see FINALIZE_GRACE_PERIOD).
+    let mut prev_current_len: Option<u64> = None;
+    let mut last_len_change = Instant::now();
+    // Whether the most recent refresh_tail call was made with trust granted
+    // 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;
+
     loop {
         let file_len = file.metadata()?.len();
         // Whether this iteration did real work -- if so, loop again
         // immediately rather than sleeping, so a large already-available
         // file drains at disk/network speed instead of being capped at
         // READ_WINDOW / POLL_INTERVAL. Only an iteration that found nothing
-        // to do (genuinely waiting for growth or the rename) sleeps.
+        // to do (genuinely waiting for growth, the rename, or a next-record
+        // proof) sleeps.
         let mut made_progress = false;
-        if file_len > pos {
+
-            let want = ((file_len - pos) as usize).min(buf.len());
+        if is_partial {
-            let n = file.read_at(&mut buf[..want], pos)?;
+            let confirmed = chunker.confirmed_len();
-            if n > 0 {
+            let trust_trailing = complete_signal && last_len_change.elapsed() >= FINALIZE_GRACE_PERIOD;
-                chunker.feed(&buf[..n])?;
+            if file_len > confirmed {
-                pos += n as u64;
+                let want = ((file_len - confirmed) as usize).min(buf.len());
-                emit_ready(chunker, events)?;
+                let n = file.read_at(&mut buf[..want], confirmed)?;
-                made_progress = true;
+                if n > 0 {
-                shrunk_below_pos = false;
+                    chunker.refresh_tail(confirmed, &buf[..n], trust_trailing)?;
+                    emit_ready(chunker, events)?;
+                    let advanced = chunker.confirmed_len() > confirmed;
+                    made_progress = advanced;
+                    trust_attempted_without_progress = trust_trailing && !advanced;
+                    shrunk_below_anchor = false;
+                }
+            } else if chunker.should_flush(Instant::now())
+                && let Some((kind, data)) = chunker.flush()
+            {
+                send_chunk(events, kind, data)?;
+            }
+            if file_len < confirmed && !shrunk_below_anchor {
+                shrunk_below_anchor = true;
+                log.log(format_args!(
+                    "tail: file shrank below confirmed_len (confirmed={confirmed}, file_len={file_len}); \
+                     waiting for regrowth -- if the file never grows past confirmed_len again, this \
+                     transfer will never finish"
+                ));
+            }
+        } else {
+            let read_pos = chunker.read_from();
+            if file_len > read_pos {
+                let want = ((file_len - read_pos) as usize).min(buf.len());
+                let n = file.read_at(&mut buf[..want], read_pos)?;
+                if n > 0 {
+                    chunker.feed(&buf[..n])?;
+                    emit_ready(chunker, events)?;
+                    made_progress = true;
+                }
+            } else if chunker.should_flush(Instant::now())
+                && let Some((kind, data)) = chunker.flush()
+            {
+                send_chunk(events, kind, data)?;
             }
-        } else if chunker.should_flush(Instant::now())
-            && let Some((kind, data)) = chunker.flush()
-        {
-            send_chunk(events, kind, data)?;
-        }
-        // If file_len < pos, the writer truncated/restarted mid-file
-        // (STREAMING_ZSTD_FORMAT.md section 7.7): do nothing, never rewind
-        // `pos`, just wait for it to regrow past `pos` on a later poll.
-        if file_len < pos && !shrunk_below_pos {
-            shrunk_below_pos = true;
-            log.log(format_args!(
-                "tail: file shrank below pos (pos={pos}, file_len={file_len}); waiting for regrowth -- \
-                 if the file never grows past pos again, this transfer will never finish"
-            ));
         }
 
         if !complete_signal && final_path.try_exists()? {
@@ -131,28 +169,55 @@ fn drain(
             // section 7.4's repoint-then-drain-then-finalize ordering.
             complete_signal = true;
             log.log(format_args!(
-                "tail: rename to {} observed at pos={pos}, file_len={file_len}",
+                "tail: rename to {} observed at confirmed_len={}, file_len={file_len}",
-                final_path.display()
+                final_path.display(),
+                chunker.confirmed_len()
             ));
         }
 
         let current_len = file.metadata()?.len();
-        if complete_signal && pos == current_len {
+        if is_partial {
+            if prev_current_len != Some(current_len) {
+                prev_current_len = Some(current_len);
+                last_len_change = Instant::now();
+            }
+        }
+
+        let anchor = if is_partial { chunker.confirmed_len() } else { chunker.read_from() };
+
+        if complete_signal && anchor == current_len {
+            if !is_partial && chunker.confirmed_len() < current_len {
+                bail!(
+                    "truncated or corrupt trailing pcap record: only {} of {} bytes validated",
+                    chunker.confirmed_len(),
+                    current_len
+                );
+            }
             if let Some((kind, data)) = chunker.finish()? {
                 send_chunk(events, kind, data)?;
             }
-            log.log(format_args!("tail: finalized at pos={pos}"));
+            log.log(format_args!("tail: finalized at {anchor} bytes"));
             return Ok(());
         }
-        if complete_signal
+
-            && pos != current_len
+        if complete_signal && anchor != current_len {
-            && last_wait_log.is_none_or(|t| t.elapsed() >= Duration::from_secs(1))
+            if is_partial && trust_attempted_without_progress {
-        {
+                bail!(
-            last_wait_log = Some(Instant::now());
+                    "torn/corrupt trailing pcap record: only {} of {current_len} bytes validated after rename",
-            log.log(format_args!(
+                    chunker.confirmed_len()
-                "tail: waiting to finalize: pos={pos} != current_len={current_len}{}",
+                );
-                if current_len < pos { " (final file is smaller than what was already read -- this will never catch up)" } else { "" }
+            }
-            ));
+            if last_wait_log.is_none_or(|t| t.elapsed() >= Duration::from_secs(1)) {
+                last_wait_log = Some(Instant::now());
+                log.log(format_args!(
+                    "tail: waiting to finalize: confirmed={anchor} != current_len={current_len}{}",
+                    if current_len < anchor {
+                        " (final file is smaller than what was already read -- this will never catch up)"
+                    } else {
+                        ""
+                    }
+                ));
+            }
         }
 
         if !made_progress {
@@ -177,7 +242,7 @@ fn send_chunk(events: &impl EventSink, kind: ChunkKind, data: Vec<u8>) -> Result
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::chunker::RawChunker;
+    use crate::chunker::{RawChunker, RecordAlignedChunker, DEFAULT_FLUSH, DEFAULT_TARGET_BYTES, GLOBAL_HEADER_LEN, RECORD_HEADER_LEN};
     use std::io::Write;
     use std::sync::mpsc;
     use tempfile::tempdir;
@@ -195,6 +260,36 @@ mod tests {
         (data, got_eof)
     }
 
+    fn try_collect(rx: mpsc::Receiver<SourceEvent>) -> (Vec<u8>, bool) {
+        let mut data = Vec::new();
+        let mut got_eof = false;
+        for ev in rx {
+            match ev {
+                SourceEvent::Chunk(_, d) => data.extend_from_slice(&d),
+                SourceEvent::Eof => got_eof = true,
+                SourceEvent::Error(_) => {}
+            }
+        }
+        (data, got_eof)
+    }
+
+    fn header_bytes(magic_le: u32) -> Vec<u8> {
+        let mut h = vec![0u8; GLOBAL_HEADER_LEN];
+        h[0..4].copy_from_slice(&magic_le.to_le_bytes());
+        h
+    }
+
+    fn record_bytes(payload: &[u8], ts_sec: u32) -> Vec<u8> {
+        let mut r = Vec::with_capacity(RECORD_HEADER_LEN + payload.len());
+        r.extend_from_slice(&ts_sec.to_le_bytes());
+        r.extend_from_slice(&[0u8; 4]); // ts_usec
+        let incl_len = payload.len() as u32;
+        r.extend_from_slice(&incl_len.to_le_bytes());
+        r.extend_from_slice(&incl_len.to_le_bytes()); // orig_len
+        r.extend_from_slice(payload);
+        r
+    }
+
     #[test]
     fn already_final_source_is_born_complete() {
         let dir = tempdir().unwrap();
@@ -309,7 +404,7 @@ mod tests {
     fn truncation_then_regrowth_does_not_error_or_rewind() {
         // Simulates a writer restart (section 7.7): the file shrinks below
         // what we've already read, then regrows past it with the corrected
-        // continuation. The reader must not error and must not rewind `pos`.
+        // continuation. The reader must not error and must not rewind.
         let dir = tempdir().unwrap();
         let partial_path = dir.path().join("cap.pcap.partial");
         let final_path = dir.path().join("cap.pcap");
@@ -344,10 +439,232 @@ mod tests {
         handle.join().unwrap().unwrap();
         let (data, eof) = collect(rx);
         // The reader never rewinds: it was already past byte 10 ("0123456789")
-        // when the truncate+regrow happened, so it waits at pos=10 until the
+        // when the truncate+regrow happened, so it waits at that boundary
-        // file regrows past that point, then reads the correct continuation
+        // until the file regrows past it, then reads the correct
-        // ("abcdefghijklmnop"[10..] == "klmnop").
+        // continuation ("abcdefghijklmnop"[10..] == "klmnop").
         assert_eq!(data, b"0123456789klmnop");
         assert!(eof);
     }
+
+    #[test]
+    fn preallocated_zero_padding_does_not_hang_and_is_not_forwarded() {
+        let dir = tempdir().unwrap();
+        let partial_path = dir.path().join("cap.pcap.partial");
+        let final_path = dir.path().join("cap.pcap");
+
+        let mut stream = header_bytes(0xA1B2C3D4);
+        let rec1 = record_bytes(&[1u8; 10], 1_700_000_000);
+        let rec2 = record_bytes(&[2u8; 10], 1_700_000_001);
+        stream.extend_from_slice(&rec1);
+        stream.extend_from_slice(&rec2);
+        let real_len = stream.len() as u64;
+
+        std::fs::write(&partial_path, &stream).unwrap();
+        // Simulate a preallocating writer: extend the file with zero
+        // padding well beyond the real content, with no further growth.
+        {
+            let f = std::fs::OpenOptions::new().write(true).open(&partial_path).unwrap();
+            f.set_len(real_len + 8192).unwrap();
+        }
+
+        let (tx, rx) = mpsc::channel();
+        let handle = std::thread::spawn({
+            let partial_path = partial_path.clone();
+            let final_path = final_path.clone();
+            move || {
+                run_tail(
+                    partial_path,
+                    true,
+                    final_path,
+                    Box::new(RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true)),
+                    tx,
+                    Arc::new(Logger::none()),
+                )
+            }
+        });
+
+        std::thread::sleep(Duration::from_millis(100));
+        // Truncate back down to just the real content, then rename.
+        {
+            let f = std::fs::OpenOptions::new().write(true).open(&partial_path).unwrap();
+            f.set_len(real_len).unwrap();
+        }
+        std::fs::rename(&partial_path, &final_path).unwrap();
+
+        handle.join().unwrap().unwrap();
+        let (data, eof) = collect(rx);
+        assert_eq!(data, stream, "no padding should have been forwarded");
+        assert!(eof);
+    }
+
+    #[test]
+    fn writer_fills_in_previously_zero_region_before_truncating() {
+        let dir = tempdir().unwrap();
+        let partial_path = dir.path().join("cap.pcap.partial");
+        let final_path = dir.path().join("cap.pcap");
+
+        let mut stream = header_bytes(0xA1B2C3D4);
+        let rec1 = record_bytes(&[1u8; 10], 1_700_000_000);
+        stream.extend_from_slice(&rec1);
+        let initial_len = stream.len() as u64;
+
+        std::fs::write(&partial_path, &stream).unwrap();
+        {
+            let f = std::fs::OpenOptions::new().write(true).open(&partial_path).unwrap();
+            f.set_len(initial_len + 8192).unwrap();
+        }
+
+        let (tx, rx) = mpsc::channel();
+        let handle = std::thread::spawn({
+            let partial_path = partial_path.clone();
+            let final_path = final_path.clone();
+            move || {
+                run_tail(
+                    partial_path,
+                    true,
+                    final_path,
+                    Box::new(RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true)),
+                    tx,
+                    Arc::new(Logger::none()),
+                )
+            }
+        });
+
+        std::thread::sleep(Duration::from_millis(100));
+        // Writer catches up: overwrite the previously-zero region in place
+        // with a real second record, no length change.
+        let rec2 = record_bytes(&[2u8; 10], 1_700_000_001);
+        {
+            let f = std::fs::OpenOptions::new().write(true).open(&partial_path).unwrap();
+            f.write_at(&rec2, initial_len).unwrap();
+        }
+        std::thread::sleep(Duration::from_millis(100));
+
+        let final_len = initial_len + rec2.len() as u64;
+        {
+            let f = std::fs::OpenOptions::new().write(true).open(&partial_path).unwrap();
+            f.set_len(final_len).unwrap();
+        }
+        std::fs::rename(&partial_path, &final_path).unwrap();
+
+        handle.join().unwrap().unwrap();
+        let (data, eof) = collect(rx);
+        let mut expected = stream.clone();
+        expected.extend_from_slice(&rec2);
+        assert_eq!(data, expected, "the writer's in-place catch-up must be observed");
+        assert!(eof);
+    }
+
+    #[test]
+    fn torn_trailing_record_after_rename_is_an_error_not_a_hang() {
+        let dir = tempdir().unwrap();
+        let partial_path = dir.path().join("cap.pcap.partial");
+        let final_path = dir.path().join("cap.pcap");
+
+        let mut stream = header_bytes(0xA1B2C3D4);
+        let rec1 = record_bytes(&[1u8; 10], 1_700_000_000);
+        stream.extend_from_slice(&rec1);
+        // Genuine non-zero garbage that never forms a plausible record.
+        stream.extend_from_slice(&[0xAB, 0xCD, 0xEF]);
+
+        std::fs::write(&partial_path, &stream).unwrap();
+
+        let (tx, rx) = mpsc::channel();
+        let handle = std::thread::spawn({
+            let partial_path = partial_path.clone();
+            let final_path = final_path.clone();
+            move || {
+                run_tail(
+                    partial_path,
+                    true,
+                    final_path,
+                    Box::new(RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true)),
+                    tx,
+                    Arc::new(Logger::none()),
+                )
+            }
+        });
+
+        std::thread::sleep(Duration::from_millis(60));
+        std::fs::rename(&partial_path, &final_path).unwrap();
+
+        let result = handle.join().unwrap();
+        try_collect(rx);
+        assert!(result.is_err(), "a genuinely torn/corrupt tail must error, not hang");
+    }
+
+    #[test]
+    fn last_record_of_a_healthy_capture_confirms_after_grace_period() {
+        // The ordinary "last packet of a rotation" case, no corruption at
+        // all: without the trust escape hatch, this would hang forever,
+        // since there is no "next record" after the last one.
+        let dir = tempdir().unwrap();
+        let partial_path = dir.path().join("cap.pcap.partial");
+        let final_path = dir.path().join("cap.pcap");
+
+        let mut stream = header_bytes(0xA1B2C3D4);
+        let rec1 = record_bytes(&[1u8; 10], 1_700_000_000);
+        stream.extend_from_slice(&rec1);
+
+        std::fs::write(&partial_path, &stream).unwrap();
+
+        // The file's length is stable from the moment it's written, before
+        // the tailer even opens it -- so the FINALIZE_GRACE_PERIOD clock
+        // (measuring "length hasn't changed"), not "time since rename",
+        // starts effectively at thread spawn, not at the rename below.
+        let start = Instant::now();
+        let (tx, rx) = mpsc::channel();
+        let handle = std::thread::spawn({
+            let partial_path = partial_path.clone();
+            let final_path = final_path.clone();
+            move || {
+                run_tail(
+                    partial_path,
+                    true,
+                    final_path,
+                    Box::new(RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true)),
+                    tx,
+                    Arc::new(Logger::none()),
+                )
+            }
+        });
+
+        std::thread::sleep(Duration::from_millis(60));
+        std::fs::rename(&partial_path, &final_path).unwrap();
+
+        handle.join().unwrap().unwrap();
+        let elapsed = start.elapsed();
+        let (data, eof) = collect(rx);
+        assert_eq!(data, stream);
+        assert!(eof);
+        assert!(
+            elapsed >= FINALIZE_GRACE_PERIOD,
+            "must not finalize before the grace period elapses: took {elapsed:?}"
+        );
+    }
+
+    #[test]
+    fn non_partial_torn_trailing_record_is_an_error() {
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("cap.pcap");
+
+        let mut stream = header_bytes(0xA1B2C3D4);
+        let rec = record_bytes(&[1u8; 20], 0);
+        // Feed everything except the last 3 bytes of the record -- torn at
+        // true EOF, non-partial (no polling-for-resolution possible).
+        stream.extend_from_slice(&rec[..rec.len() - 3]);
+        std::fs::write(&path, &stream).unwrap();
+
+        let (tx, rx) = mpsc::channel();
+        let result = run_tail(
+            path.clone(),
+            false,
+            path,
+            Box::new(RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, false)),
+            tx,
+            Arc::new(Logger::none()),
+        );
+        try_collect(rx);
+        assert!(result.is_err());
+    }
 }