diff --git a/src/chunker.rs b/src/chunker.rs index 676c208..81eb7d5 100644 --- a/src/chunker.rs +++ b/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; -/// `RecordAlignedChunker` cuts on pcap record boundaries for the compress path. +/// protocol. `RawChunker` is a passthrough for the non-compress, non-partial +/// 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)>; - /// Called once, after the source is confirmed fully drained. Errors if - /// what's left doesn't cleanly end at a valid boundary. + /// Called once, after the source is confirmed fully drained (per the + /// 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)>>; + + /// 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, + 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)>> { 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, + snaplen: u32, pending: Vec, 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) { let chunk: Vec = 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.header_done = true; - self.header_ready = true; + self.validate_header()?; + if !self.header_done { + // 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 { + record_bytes_with_ts(payload, big_endian, 0) + } + + fn record_bytes_with_ts(payload: &[u8], big_endian: bool, ts_sec: u32) -> Vec { 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::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() { - let mut c = RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH); + fn torn_trailing_record_is_left_unconfirmed_not_an_error() { + // 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"); + } } diff --git a/src/pipeline.rs b/src/pipeline.rs index b71b1ed..0c2ae2b 100644 --- a/src/pipeline.rs +++ b/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) -> Result<()> { - let chunker: Box = if opts.compress { - Box::new(RecordAlignedChunker::new(opts.chunk_target, DEFAULT_FLUSH)) + let chunker: Box = if opts.compress || opts.is_partial { + 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 { + fn record_bytes(payload: &[u8], ts_sec: u32) -> Vec { 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(); diff --git a/src/tail.rs b/src/tail.rs index d6a4bc6..51f75c3 100644 --- a/src/tail.rs +++ b/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), Eof, @@ -41,7 +51,7 @@ impl EventSink for std::sync::mpsc::SyncSender { /// 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 = 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 = 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()); - let n = file.read_at(&mut buf[..want], pos)?; - if n > 0 { - chunker.feed(&buf[..n])?; - pos += n as u64; - emit_ready(chunker, events)?; - made_progress = true; - shrunk_below_pos = false; + + if is_partial { + let confirmed = chunker.confirmed_len(); + let trust_trailing = complete_signal && last_len_change.elapsed() >= FINALIZE_GRACE_PERIOD; + if file_len > confirmed { + let want = ((file_len - confirmed) as usize).min(buf.len()); + let n = file.read_at(&mut buf[..want], confirmed)?; + if n > 0 { + 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}", - final_path.display() + "tail: rename to {} observed at confirmed_len={}, file_len={file_len}", + 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 - && 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: 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 complete_signal && anchor != current_len { + if is_partial && trust_attempted_without_progress { + bail!( + "torn/corrupt trailing pcap record: only {} of {current_len} bytes validated after rename", + chunker.confirmed_len() + ); + } + 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) -> 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) -> (Vec, 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 { + 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 { + 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 - // file regrows past that point, then reads the correct continuation - // ("abcdefghijklmnop"[10..] == "klmnop"). + // when the truncate+regrow happened, so it waits at that boundary + // until the file regrows past it, then reads the correct + // 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()); + } }