src/chunker.rs

@@ -1,3 +1,4 @@
+use crate::filter::{IpFilter, LINKTYPE_ETHERNET};
 use anyhow::{Result, bail};
 use std::time::{Duration, Instant};
 
@@ -146,10 +147,37 @@ pub struct RecordAlignedChunker {
     header_ready: bool,
     endianness: Option<Endianness>,
     snaplen: u32,
+    /// Link-layer type from global-header bytes [20..24], needed to know
+    /// where the IP header starts. Only meaningful once `header_done`.
+    linktype: u32,
+    /// Source bytes read but not yet confirmed as complete records. Every
+    /// record that `rescan` confirms is consumed out of here immediately,
+    /// so this holds only the unconfirmed tail -- bounded by one read
+    /// window, no matter how little the filter keeps.
     pending: Vec<u8>,
-    complete_len: usize,
+    /// Cumulative source bytes consumed (confirmed and removed from
-    emitted_len: u64,
+    /// `pending`), NOT bytes emitted downstream -- the two differ whenever
-    last_growth: Instant,
+    /// the filter drops a record. Source-offset accounting depends only on
+    /// this.
+    consumed_len: u64,
+    /// Confirmed records the filter kept, accumulating across many
+    /// `pending` recycles until they reach `target`. This is what gets
+    /// emitted, so chunk (and therefore zstd frame) size is measured in
+    /// KEPT bytes -- a filter keeping 1% of the traffic still produces
+    /// full-size frames rather than 1%-size ones.
+    kept: Vec<u8>,
+    filter: IpFilter,
+    /// When the oldest byte currently in `kept` was confirmed, i.e. how
+    /// long the pending output has been waiting. `None` when `kept` is
+    /// empty.
+    ///
+    /// This deliberately tracks the age of the OUTPUT, not of the last
+    /// source growth: with a selective filter the source can grow
+    /// continuously (resetting any source-based timer forever) while the
+    /// handful of matching records sit here unsent. Timing from the kept
+    /// bytes is what keeps `flush_after` an actual latency bound rather
+    /// than an idle-only one.
+    kept_since: Option<Instant>,
     target: usize,
     flush_after: Duration,
     /// Whether this source is a still-growing (partial) file. Gates every
@@ -161,17 +189,34 @@ pub struct RecordAlignedChunker {
 }
 
 impl RecordAlignedChunker {
+    /// Unfiltered: keeps every record. Equivalent to `with_filter` with
+    /// `IpFilter::match_all()` -- which is not a separate code path, just
+    /// a filter that says yes, so this and the filtered case exercise the
+    /// same record-confirmation logic. Production goes through
+    /// `with_filter`; this is the convenience the tests use.
+    #[allow(dead_code)]
     pub fn new(target: usize, flush_after: Duration, partial: bool) -> Self {
+        Self::with_filter(target, flush_after, partial, IpFilter::match_all())
+    }
+
+    pub fn with_filter(
+        target: usize,
+        flush_after: Duration,
+        partial: bool,
+        filter: IpFilter,
+    ) -> Self {
         Self {
             header_buf: Vec::with_capacity(GLOBAL_HEADER_LEN),
             header_done: false,
             header_ready: false,
             endianness: None,
             snaplen: 0,
+            linktype: 0,
             pending: Vec::new(),
-            complete_len: 0,
+            consumed_len: 0,
-            emitted_len: 0,
+            kept: Vec::new(),
-            last_growth: Instant::now(),
+            filter,
+            kept_since: None,
             target,
             flush_after,
             partial,
@@ -191,19 +236,49 @@ impl RecordAlignedChunker {
         }
         let e = detect_endianness(&self.header_buf)?;
         self.snaplen = read_u32(&self.header_buf[16..20], e);
+        self.linktype = read_u32(&self.header_buf[20..24], e);
         self.endianness = Some(e);
         self.header_done = true;
         self.header_ready = true;
+        // A filter can only find addresses in an Ethernet frame, so on any
+        // other link type it keeps nothing and the transfer silently
+        // produces a header-only pcap. `tcpdump -i any` (LINUX_SLL, 113) is
+        // the easy way to hit this, so say so out loud. stderr, never
+        // stdout: in --server mode stdout is the wire.
+        if !self.filter.is_match_all() && self.linktype != LINKTYPE_ETHERNET {
+            eprintln!(
+                "scpcap: warning: --ip filtering needs Ethernet (link type {LINKTYPE_ETHERNET}), \
+                 but this capture is link type {}; no packets will match",
+                self.linktype
+            );
+        }
         Ok(())
     }
 
+    /// Walks `pending` confirming whole records, hands each confirmed
+    /// record to the filter (keeping the matches in `kept`), and consumes
+    /// the confirmed bytes out of `pending`.
+    ///
+    /// Confirming/consuming and *emitting* are deliberately separate. If
+    /// consumption were tied to emission, a filter keeping 1% of the
+    /// traffic would hold the other 99% in `pending` for as long as `kept`
+    /// stayed under `target` -- the whole un-emitted source accumulating in
+    /// memory, gigabytes on a large capture. Consuming here caps `pending`
+    /// at the unconfirmed tail while `kept` fills up across many passes.
+    ///
+    /// `cursor` is deliberately a local, not a field: every confirmed byte
+    /// leaves `pending` before this returns, so there is no persistent
+    /// "confirmed but still buffered" region for the source-offset
+    /// accounting (or `refresh_tail`) to have to reason about.
     fn rescan(&mut self, trust_trailing_record: bool) {
         let e = match self.endianness {
             Some(e) => e,
             None => return,
         };
+        let max_incl_len = self.max_incl_len();
+        let mut cursor = 0usize;
         loop {
-            let remaining = &self.pending[self.complete_len..];
+            let remaining = &self.pending[cursor..];
             if remaining.len() < RECORD_HEADER_LEN {
                 break;
             }
@@ -224,7 +299,7 @@ impl RecordAlignedChunker {
                 if ts_sec == 0 {
                     break;
                 }
-                if incl_len > self.max_incl_len() {
+                if incl_len > max_incl_len {
                     break;
                 }
 
@@ -235,13 +310,13 @@ impl RecordAlignedChunker {
                 // 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
+                // confirmed_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();
+                    && (read_u32(&after[8..12], e) as usize) <= max_incl_len;
 
                 // Escape hatch (only when the caller is certain nothing
                 // more will EVER arrive): a record that is itself plausible
@@ -259,15 +334,30 @@ impl RecordAlignedChunker {
                     break;
                 }
             }
-            self.complete_len += record_total;
+
+            // Confirmed. Show the captured bytes (everything after the
+            // record header) to the filter, and keep the whole record --
+            // header included -- if it matches, so `kept` stays a valid
+            // sequence of pcap records.
+            let start = cursor;
+            let end = cursor + record_total;
+            if self.filter.matches(&self.pending[start + RECORD_HEADER_LEN..end], self.linktype) {
+                self.kept.extend_from_slice(&self.pending[start..end]);
+            }
+            cursor = end;
+        }
+        if cursor > 0 {
+            self.pending.drain(..cursor);
+            self.consumed_len += cursor as u64;
+        }
+        if !self.kept.is_empty() && self.kept_since.is_none() {
+            self.kept_since = Some(Instant::now());
         }
     }
 
-    fn drain_complete(&mut self) -> (ChunkKind, Vec<u8>) {
+    fn take_kept(&mut self) -> (ChunkKind, Vec<u8>) {
-        let chunk: Vec<u8> = self.pending.drain(..self.complete_len).collect();
+        self.kept_since = None;
-        self.emitted_len += self.complete_len as u64;
+        (ChunkKind::Data, std::mem::take(&mut self.kept))
-        self.complete_len = 0;
-        (ChunkKind::Data, chunk)
     }
 }
 
@@ -296,7 +386,6 @@ impl Chunker for RecordAlignedChunker {
         if !rest.is_empty() {
             self.pending.extend_from_slice(rest);
             self.rescan(false);
-            self.last_growth = Instant::now();
         }
         Ok(())
     }
@@ -306,19 +395,27 @@ impl Chunker for RecordAlignedChunker {
             self.header_ready = false;
             return Some((ChunkKind::Header, std::mem::take(&mut self.header_buf)));
         }
-        if self.complete_len >= self.target {
+        // All three cut points gate on `kept`, never on how much source was
-            return Some(self.drain_complete());
+        // consumed: a stretch of source where the filter matched nothing
+        // must produce no chunk at all, rather than an empty one. An empty
+        // chunk would become a zero-content zstd frame, which
+        // STREAMING_ZSTD_FORMAT.md §3 does not allow for.
+        if self.kept.len() >= self.target {
+            return Some(self.take_kept());
         }
         None
     }
 
     fn should_flush(&self, now: Instant) -> bool {
-        self.complete_len > 0 && now.duration_since(self.last_growth) >= self.flush_after
+        // `kept_since` is Some exactly when `kept` is non-empty, so this
+        // can never ask for a flush that would produce an empty chunk.
+        self.kept_since
+            .is_some_and(|since| now.duration_since(since) >= self.flush_after)
     }
 
     fn flush(&mut self) -> Option<(ChunkKind, Vec<u8>)> {
-        if self.complete_len > 0 {
+        if !self.kept.is_empty() {
-            Some(self.drain_complete())
+            Some(self.take_kept())
         } else {
             None
         }
@@ -328,8 +425,8 @@ impl Chunker for RecordAlignedChunker {
         if !self.header_done {
             bail!("source ended before a complete pcap global header was read");
         }
-        if self.complete_len > 0 {
+        if !self.kept.is_empty() {
-            Ok(Some(self.drain_complete()))
+            Ok(Some(self.take_kept()))
         } else {
             Ok(None)
         }
@@ -339,7 +436,7 @@ impl Chunker for RecordAlignedChunker {
         if !self.header_done {
             self.header_buf.len() as u64
         } else {
-            GLOBAL_HEADER_LEN as u64 + self.emitted_len + self.pending.len() as u64
+            GLOBAL_HEADER_LEN as u64 + self.consumed_len + self.pending.len() as u64
         }
     }
 
@@ -352,7 +449,14 @@ impl Chunker for RecordAlignedChunker {
             // treating the stuck zero bytes as already consumed.
             0
         } else {
-            GLOBAL_HEADER_LEN as u64 + self.emitted_len + self.complete_len as u64
+            // Source-offset accounting, entirely independent of what the
+            // filter kept: every confirmed record is consumed the moment
+            // `rescan` proves it, so `consumed_len` alone is the confirmed
+            // prefix. (This is the same number the old
+            // `emitted_len + complete_len` produced; consuming on
+            // confirmation just moves the quantity from the second term to
+            // the first.)
+            GLOBAL_HEADER_LEN as u64 + self.consumed_len
         }
     }
 
@@ -373,30 +477,24 @@ impl Chunker for RecordAlignedChunker {
             if !self.header_done {
                 return Ok(());
             }
-            // header_done just flipped true, so complete_len is still 0 here
+            // header_done just flipped true, so nothing beyond the header
-            // (nothing beyond the header has been considered yet) --
+            // has been considered yet -- pending is simply the fresh
-            // pending is simply the fresh post-header bytes.
+            // post-header bytes.
             self.pending = bytes[GLOBAL_HEADER_LEN..].to_vec();
             self.rescan(trust_trailing_record);
-            self.last_growth = Instant::now();
             return Ok(());
         }
-        // `pending[..complete_len]` is CONFIRMED content that just hasn't
+        // `pending` holds only the UNCONFIRMED tail -- `rescan` consumes
-        // been drained (emitted) yet -- e.g. because it's still under
+        // every record it confirms, so nothing confirmed is still sitting
-        // `target`. `bytes` starts exactly at `confirmed_len()` (checked
+        // here waiting to be emitted. `bytes` is a fresh read starting at
-        // above), i.e. exactly where that confirmed prefix ends, so it must
+        // `confirmed_len()` (checked above), i.e. exactly where that
-        // be appended after it, never used to replace the whole buffer --
+        // unconfirmed tail begins, so replacing the buffer wholesale is
-        // doing that would silently discard already-confirmed, already
+        // correct and discards nothing: anything already confirmed lives in
-        // real, not-yet-emitted content on every single re-peek, which is
+        // `consumed_len` (for source accounting) and, if the filter kept
-        // catastrophic for a long-running growing transfer (each poll would
+        // it, in `kept` (for emission). Neither is touched here.
-        // re-derive from scratch instead of building on prior confirmed
+        self.pending.clear();
-        // progress, and once the fresh read window is too small to
-        // re-establish the same progress alone -- e.g. capped by
-        // READ_WINDOW -- confirmed_len can never recover).
-        self.pending.truncate(self.complete_len);
         self.pending.extend_from_slice(bytes);
         self.rescan(trust_trailing_record);
-        self.last_growth = Instant::now();
         Ok(())
     }
 }
@@ -771,4 +869,213 @@ mod tests {
             "confirmed_len must never regress below what was already established"
         );
     }
+
+    // --- IP filtering -------------------------------------------------
+    //
+    // Note that every test above this point already exercises the
+    // filtering code path, with a match-all filter: there is no separate
+    // unfiltered implementation to regress against.
+
+    /// A global header declaring Ethernet (link type 1) at bytes [20..24],
+    /// which the filter needs in order to find an IP header at all.
+    fn eth_header_bytes() -> Vec<u8> {
+        let mut h = header_bytes(0xA1B2C3D4);
+        h[20..24].copy_from_slice(&1u32.to_le_bytes());
+        h
+    }
+
+    /// A pcap record holding an Ethernet/IPv4 packet between `src` and `dst`,
+    /// padded out to `payload_len` total captured bytes.
+    fn ip_record(src: [u8; 4], dst: [u8; 4], payload_len: usize) -> Vec<u8> {
+        let mut pkt = vec![0u8; payload_len.max(34)];
+        pkt[12..14].copy_from_slice(&0x0800u16.to_be_bytes()); // ethertype IPv4
+        pkt[14] = 0x45; // version 4, IHL 5
+        pkt[26..30].copy_from_slice(&src);
+        pkt[30..34].copy_from_slice(&dst);
+        record_bytes_with_ts(&pkt, false, 1_700_000_000)
+    }
+
+    fn filter_for(specs: &[&str]) -> IpFilter {
+        IpFilter::parse(&specs.iter().map(|s| s.to_string()).collect::<Vec<_>>()).unwrap()
+    }
+
+    fn drain_all(c: &mut RecordAlignedChunker) -> Vec<(ChunkKind, Vec<u8>)> {
+        let mut out = Vec::new();
+        while let Some(chunk) = c.ready_chunk() {
+            out.push(chunk);
+        }
+        out
+    }
+
+    const WANTED: [u8; 4] = [1, 1, 1, 1];
+    const OTHER: [u8; 4] = [9, 9, 9, 9];
+    const THIRD: [u8; 4] = [8, 8, 8, 8];
+
+    #[test]
+    fn keeps_only_matching_records_and_keeps_them_whole() {
+        let mut c = RecordAlignedChunker::with_filter(1, DEFAULT_FLUSH, false, filter_for(&["1.1.1.1"]));
+        let hit_src = ip_record(WANTED, OTHER, 60);
+        let hit_dst = ip_record(OTHER, WANTED, 60);
+        let miss = ip_record(OTHER, THIRD, 60);
+
+        let mut stream = eth_header_bytes();
+        stream.extend_from_slice(&miss);
+        stream.extend_from_slice(&hit_src);
+        stream.extend_from_slice(&miss);
+        stream.extend_from_slice(&hit_dst);
+        stream.extend_from_slice(&miss);
+        c.feed(&stream).unwrap();
+
+        let chunks = drain_all(&mut c);
+        assert_eq!(chunks[0].0, ChunkKind::Header);
+        assert_eq!(chunks[0].1, eth_header_bytes(), "header is never filtered");
+
+        // Records come out whole (16-byte header included) and in order.
+        let data: Vec<u8> = chunks[1..].iter().flat_map(|(_, d)| d.clone()).collect();
+        let mut expected = hit_src.clone();
+        expected.extend_from_slice(&hit_dst);
+        assert_eq!(data, expected);
+    }
+
+    #[test]
+    fn target_counts_kept_bytes_not_source_bytes() {
+        // The reason filtering lives at record confirmation rather than in a
+        // post-pass over emitted chunks: with 1% of traffic kept, chunks
+        // (and therefore zstd frames) must still reach `target`, not shrink
+        // to 1% of it.
+        let hit = ip_record(WANTED, OTHER, 100);
+        let miss = ip_record(OTHER, THIRD, 100);
+        let target = 10 * hit.len();
+        let mut c = RecordAlignedChunker::with_filter(target, DEFAULT_FLUSH, false, filter_for(&["1.1.1.1"]));
+
+        c.feed(&eth_header_bytes()).unwrap();
+        assert_eq!(c.ready_chunk().unwrap().0, ChunkKind::Header);
+        // 1 kept in every 100 records, 1000 records total => 10 kept.
+        for i in 0..1000 {
+            c.feed(if i % 100 == 0 { &hit } else { &miss }).unwrap();
+        }
+
+        let chunks = drain_all(&mut c);
+        assert_eq!(chunks.len(), 1, "expected one full-size chunk, got {}", chunks.len());
+        assert!(
+            chunks[0].1.len() >= target,
+            "chunk is {} bytes, below the {target}-byte target",
+            chunks[0].1.len()
+        );
+        assert_eq!(chunks[0].1.len(), 10 * hit.len());
+    }
+
+    #[test]
+    fn source_buffer_does_not_grow_when_nothing_matches() {
+        // Confirmed records are consumed from `pending` immediately rather
+        // than when a chunk is emitted. Without that, a filter that keeps
+        // nothing would accumulate the entire source in memory while
+        // waiting for `kept` to reach `target`.
+        let miss = ip_record(OTHER, THIRD, 100);
+        let mut c = RecordAlignedChunker::with_filter(
+            DEFAULT_TARGET_BYTES,
+            DEFAULT_FLUSH,
+            false,
+            filter_for(&["1.1.1.1"]),
+        );
+        c.feed(&eth_header_bytes()).unwrap();
+        let mut source_len = GLOBAL_HEADER_LEN as u64;
+        for _ in 0..5000 {
+            c.feed(&miss).unwrap();
+            source_len += miss.len() as u64;
+        }
+
+        assert!(
+            c.pending.len() < miss.len() * 2,
+            "unconfirmed tail grew to {} bytes; confirmed records are not being consumed",
+            c.pending.len()
+        );
+        assert!(c.kept.is_empty());
+        // ...and the source is nonetheless fully accounted for.
+        assert_eq!(c.confirmed_len(), source_len);
+        assert_eq!(c.read_from(), source_len);
+    }
+
+    #[test]
+    fn a_fully_dropped_region_yields_no_chunk_not_an_empty_one() {
+        // An empty chunk would be compressed into a zero-content zstd frame,
+        // which STREAMING_ZSTD_FORMAT.md §3 does not provide for.
+        let miss = ip_record(OTHER, THIRD, 100);
+        let mut c = RecordAlignedChunker::with_filter(1, DEFAULT_FLUSH, false, filter_for(&["1.1.1.1"]));
+        c.feed(&eth_header_bytes()).unwrap();
+        assert_eq!(c.ready_chunk().unwrap().0, ChunkKind::Header);
+        for _ in 0..20 {
+            c.feed(&miss).unwrap();
+        }
+        // target is 1 byte, so anything kept at all would be emitted.
+        assert!(c.ready_chunk().is_none());
+        assert!(c.flush().is_none());
+        assert!(!c.should_flush(Instant::now() + Duration::from_secs(60)));
+        assert!(c.finish().unwrap().is_none());
+    }
+
+    #[test]
+    fn source_accounting_is_unaffected_by_what_the_filter_keeps() {
+        // The direct statement of the invariant the rest of the system
+        // depends on: confirmed_len()/read_from() are SOURCE offsets, which
+        // tail.rs compares against the real file length to decide finalize.
+        // Dropping packets must not move them by a single byte.
+        let mut stream = eth_header_bytes();
+        for i in 0..40u8 {
+            let src = if i % 3 == 0 { WANTED } else { OTHER };
+            stream.extend_from_slice(&ip_record(src, THIRD, 60 + i as usize));
+        }
+
+        let mut all = RecordAlignedChunker::with_filter(1 << 20, DEFAULT_FLUSH, false, IpFilter::match_all());
+        let mut some = RecordAlignedChunker::with_filter(1 << 20, DEFAULT_FLUSH, false, filter_for(&["1.1.1.1"]));
+        let mut none = RecordAlignedChunker::with_filter(1 << 20, DEFAULT_FLUSH, false, filter_for(&["203.0.113.7"]));
+        for c in [&mut all, &mut some, &mut none] {
+            c.feed(&stream).unwrap();
+        }
+
+        for c in [&some, &none] {
+            assert_eq!(c.confirmed_len(), all.confirmed_len());
+            assert_eq!(c.read_from(), all.read_from());
+        }
+        assert_eq!(all.confirmed_len(), stream.len() as u64);
+
+        // Meanwhile the kept content really does differ.
+        assert_eq!(all.kept.len(), stream.len() - GLOBAL_HEADER_LEN);
+        assert!(!some.kept.is_empty() && some.kept.len() < all.kept.len());
+        assert!(none.kept.is_empty());
+    }
+
+    #[test]
+    fn filtering_a_growing_source_confirms_and_drops_across_refreshes() {
+        // The partial path reaches records through refresh_tail, which
+        // replaces the unconfirmed tail wholesale. A record must be filtered
+        // exactly once even though the bytes around it are re-read.
+        let hit = ip_record(WANTED, OTHER, 60);
+        let miss = ip_record(OTHER, THIRD, 60);
+        let mut c = RecordAlignedChunker::with_filter(1 << 20, DEFAULT_FLUSH, true, filter_for(&["1.1.1.1"]));
+
+        let mut stream = eth_header_bytes();
+        stream.extend_from_slice(&miss);
+        stream.extend_from_slice(&hit);
+        stream.extend_from_slice(&miss);
+        stream.extend_from_slice(&hit);
+
+        // Feed it in growing prefixes, each time re-reading from the current
+        // confirmed offset, exactly as tail.rs does.
+        for end in [40, 100, stream.len() - 10, stream.len()] {
+            let offset = c.confirmed_len() as usize;
+            c.refresh_tail(offset as u64, &stream[offset..end], false).unwrap();
+        }
+        // The trailing record has no next-record proof yet, so only the
+        // first `hit` is confirmed.
+        assert_eq!(c.kept, hit);
+
+        // Source stops growing and the caller vouches for the tail.
+        let offset = c.confirmed_len() as usize;
+        c.refresh_tail(offset as u64, &stream[offset..], true).unwrap();
+        assert_eq!(c.confirmed_len(), stream.len() as u64);
+        let mut both = hit.clone();
+        both.extend_from_slice(&hit);
+        assert_eq!(c.kept, both, "each kept record appears exactly once");
+    }
 }

src/cli.rs

@@ -33,6 +33,21 @@ pub struct ClientCli {
     #[arg(long, default_value = "scpcap")]
     pub remote_exe: String,
 
+    /// How long a partial chunk may wait before being sent anyway, bounding
+    /// latency on a still-growing source: "100ms", "2s", or a bare number of
+    /// milliseconds. Chunks that reach --chunk sooner are sent immediately;
+    /// this only governs the leftover. Matters most with --ip, where a
+    /// selective filter would otherwise take a long time to fill a chunk.
+    #[arg(long, value_name = "DURATION", default_value = "100ms")]
+    pub flush_interval: String,
+
+    /// Only forward packets whose source or destination IP matches. May be
+    /// repeated; a packet is kept if it matches any entry. Accepts a bare
+    /// address ("1.1.1.1", "2606:4700::1111") or a CIDR prefix
+    /// ("10.0.0.0/8"). Requires an Ethernet-linktype pcap source.
+    #[arg(long = "ip", value_name = "ADDR")]
+    pub ip: Vec<String>,
+
     /// Append a diagnostic trace of this (local) process's transfer events
     /// to this file -- positions, rename detection, finalize -- for
     /// figuring out where a stalled transfer is stuck.
@@ -80,6 +95,16 @@ pub struct ServerCli {
     #[arg(long, default_value_t = crate::zstd_frame::DEFAULT_LEVEL)]
     pub level: i32,
 
+    /// Flush interval for a partial chunk (--send mode). See the client's
+    /// --flush-interval.
+    #[arg(long, value_name = "DURATION", default_value = "100ms")]
+    pub flush_interval: String,
+
+    /// IP filter for --send mode; repeated, one per address/prefix. See the
+    /// client's --ip.
+    #[arg(long = "ip", value_name = "ADDR")]
+    pub ip: Vec<String>,
+
     /// Destination final path (--recv mode).
     #[arg(long)]
     pub dest_final: Option<String>,

src/client.rs

@@ -1,5 +1,6 @@
 use crate::channel::{ChannelSink, ChannelSource};
 use crate::cli::ClientCli;
+use crate::filter::IpFilter;
 use crate::naming::{self, resolve_dest_paths, strip_name, Target};
 use crate::pipeline::{run_recv, run_send, RecvOptions, SendOptions};
 use crate::protocol::{MessageSink, MessageSource, WireSink, WireSource};
@@ -35,6 +36,28 @@ pub fn run(cli: ClientCli) -> Result<()> {
     let is_partial = source_stripped.was_partial;
     let effective_compress = cli.compress.is_some() && !source_stripped.was_zst;
     let chunk_target = naming::parse_size(&cli.chunk)?;
+    let flush_after = naming::parse_duration(&cli.flush_interval)?;
+    // Parsed up front so a bad --ip fails before anything is spawned or
+    // opened. Only the SENDING side filters, so this is used directly when
+    // we are the sender (local source) and forwarded as args when the
+    // remote is (see the Remote->Local branch). It is never passed to a
+    // --recv server: the receiver writes bytes verbatim and never parses
+    // them.
+    let filter = IpFilter::parse(&cli.ip)?;
+    // Filtering means reading pcap records, which an already-compressed
+    // source doesn't expose. Without this the transfer would still pick the
+    // record-aligned chunker (a filter forces it) and abort deep inside with
+    // "unrecognized pcap global header magic: 0xfd2fb528" -- the zstd magic.
+    // Such a copy is verbatim and works fine unfiltered, so refuse the
+    // combination rather than the source.
+    if !filter.is_match_all() && source_stripped.was_zst {
+        bail!(
+            "cannot apply --ip to an already-compressed source ({}): \
+             filtering needs to read pcap records. Copy it without --ip, \
+             or filter the uncompressed capture.",
+            cli.source
+        );
+    }
 
     let dest = resolve_dest_paths(dest_target.path(), &source_stripped, effective_compress, &cli.extension);
 
@@ -46,7 +69,9 @@ pub fn run(cli: ClientCli) -> Result<()> {
                 final_source_path: PathBuf::from(&final_source_path),
                 compress: effective_compress,
                 chunk_target,
+                flush_after,
                 level: cli.level,
+                filter: filter.clone(),
                 log: log.clone(),
             };
             let recv_opts = RecvOptions {
@@ -83,7 +108,9 @@ pub fn run(cli: ClientCli) -> Result<()> {
                 final_source_path: PathBuf::from(&final_source_path),
                 compress: effective_compress,
                 chunk_target,
+                flush_after,
                 level: cli.level,
+                filter: filter.clone(),
                 log: log.clone(),
             };
             let sink: Box<dyn MessageSink> = Box::new(WireSink(stdin));
@@ -104,12 +131,18 @@ pub fn run(cli: ClientCli) -> Result<()> {
                 final_source_path.clone(),
                 "--chunk".to_string(),
                 cli.chunk.clone(),
+                "--flush-interval".to_string(),
+                cli.flush_interval.clone(),
                 "--level".to_string(),
                 cli.level.to_string(),
             ];
             if is_partial {
                 server_args.push("--partial".to_string());
             }
+            for spec in &cli.ip {
+                server_args.push("--ip".to_string());
+                server_args.push(spec.clone());
+            }
             if effective_compress {
                 server_args.push("--compress".to_string());
                 server_args.push("zstd".to_string());

src/filter.rs

@@ -0,0 +1,317 @@
+use anyhow::{Result, bail};
+use std::net::IpAddr;
+
+/// The only link-layer type the IP peek understands. Anything else matches
+/// nothing -- see `matches`.
+pub const LINKTYPE_ETHERNET: u32 = 1;
+
+const ETH_HEADER_LEN: usize = 14;
+const ETHERTYPE_IPV4: u16 = 0x0800;
+const ETHERTYPE_IPV6: u16 = 0x86DD;
+const ETHERTYPE_VLAN: u16 = 0x8100;
+const ETHERTYPE_QINQ: u16 = 0x88A8;
+const ETHERTYPE_QINQ_ALT: u16 = 0x9100;
+/// 802.1Q and one layer of QinQ. Deeper stacking is real but vanishingly
+/// rare in capture files, and the bound keeps a corrupt/truncated packet
+/// from walking off into the payload looking for an ethertype.
+const MAX_VLAN_TAGS: usize = 2;
+
+const IPV4_HEADER_MIN: usize = 20;
+const IPV6_HEADER_LEN: usize = 40;
+
+/// A set of IP addresses/prefixes to keep. A packet matches if *either* its
+/// source or destination address falls in *any* entry.
+///
+/// Sized for the expected 1-4 entries: two plain vectors, split by address
+/// family so a v4 packet never compares against a v6 entry, scanned
+/// linearly. No sorting (a prefix match isn't an equality test, so ordering
+/// buys nothing) and no hashing.
+#[derive(Debug, Clone)]
+pub struct IpFilter {
+    /// Set when no `--ip` was given: every packet is kept, without looking
+    /// at it at all. This is the default so that the filtering code path is
+    /// the *only* code path -- see the chunker.
+    match_all: bool,
+    /// (network, mask) pairs, network already masked at parse time.
+    v4: Vec<(u32, u32)>,
+    v6: Vec<(u128, u128)>,
+}
+
+impl IpFilter {
+    /// The no-`--ip` default: keeps everything, matches without inspecting.
+    pub fn match_all() -> Self {
+        Self { match_all: true, v4: Vec::new(), v6: Vec::new() }
+    }
+
+    /// Parses `--ip` specs: a bare address (`1.1.1.1`, `2606:4700::1111`)
+    /// or a CIDR prefix (`10.0.0.0/8`). An empty list yields `match_all`.
+    pub fn parse(specs: &[String]) -> Result<Self> {
+        if specs.is_empty() {
+            return Ok(Self::match_all());
+        }
+        let mut f = Self { match_all: false, v4: Vec::new(), v6: Vec::new() };
+        for spec in specs {
+            let (addr_str, prefix_str) = match spec.split_once('/') {
+                Some((a, p)) => (a, Some(p)),
+                None => (spec.as_str(), None),
+            };
+            let addr: IpAddr = addr_str
+                .parse()
+                .map_err(|_| anyhow::anyhow!("--ip {spec:?}: not an IP address"))?;
+            let width = if addr.is_ipv4() { 32u32 } else { 128u32 };
+            let prefix = match prefix_str {
+                Some(p) => p
+                    .parse::<u32>()
+                    .map_err(|_| anyhow::anyhow!("--ip {spec:?}: prefix length is not a number"))?,
+                None => width,
+            };
+            if prefix > width {
+                bail!("--ip {spec:?}: prefix length {prefix} exceeds {width} bits");
+            }
+            match addr {
+                IpAddr::V4(v4) => {
+                    let mask = if prefix == 0 { 0 } else { u32::MAX << (32 - prefix) };
+                    f.v4.push((u32::from(v4) & mask, mask));
+                }
+                IpAddr::V6(v6) => {
+                    let mask = if prefix == 0 { 0 } else { u128::MAX << (128 - prefix) };
+                    f.v6.push((u128::from(v6) & mask, mask));
+                }
+            }
+        }
+        Ok(f)
+    }
+
+    /// Whether this filter keeps everything, i.e. no `--ip` was given.
+    /// Callers use it to skip work (and warnings) that only a real filter
+    /// needs.
+    pub fn is_match_all(&self) -> bool {
+        self.match_all
+    }
+
+    /// Whether to keep `packet` -- the captured bytes of one pcap record,
+    /// starting at the link-layer header.
+    ///
+    /// Everything unmatched is dropped: non-IP frames (ARP, LLDP), packets
+    /// truncated by snaplen before the addresses, and any link type other
+    /// than Ethernet.
+    pub fn matches(&self, packet: &[u8], linktype: u32) -> bool {
+        if self.match_all {
+            return true;
+        }
+        if linktype != LINKTYPE_ETHERNET || packet.len() < ETH_HEADER_LEN {
+            return false;
+        }
+        let mut ethertype = be16(&packet[12..14]);
+        let mut off = ETH_HEADER_LEN;
+        let mut tags = 0;
+        // A VLAN tag is TPID(2) + TCI(2); the TPID is the ethertype we just
+        // read, so the *next* type field sits 2 bytes into the tag.
+        while matches!(ethertype, ETHERTYPE_VLAN | ETHERTYPE_QINQ | ETHERTYPE_QINQ_ALT) {
+            tags += 1;
+            if tags > MAX_VLAN_TAGS || packet.len() < off + 4 {
+                return false;
+            }
+            ethertype = be16(&packet[off + 2..off + 4]);
+            off += 4;
+        }
+        match ethertype {
+            ETHERTYPE_IPV4 => {
+                if packet.len() < off + IPV4_HEADER_MIN || packet[off] >> 4 != 4 {
+                    return false;
+                }
+                let src = be32(&packet[off + 12..off + 16]);
+                let dst = be32(&packet[off + 16..off + 20]);
+                self.v4.iter().any(|&(net, mask)| src & mask == net || dst & mask == net)
+            }
+            ETHERTYPE_IPV6 => {
+                if packet.len() < off + IPV6_HEADER_LEN || packet[off] >> 4 != 6 {
+                    return false;
+                }
+                let src = be128(&packet[off + 8..off + 24]);
+                let dst = be128(&packet[off + 24..off + 40]);
+                self.v6.iter().any(|&(net, mask)| src & mask == net || dst & mask == net)
+            }
+            _ => false,
+        }
+    }
+}
+
+fn be16(b: &[u8]) -> u16 {
+    u16::from_be_bytes(b.try_into().unwrap())
+}
+
+fn be32(b: &[u8]) -> u32 {
+    u32::from_be_bytes(b.try_into().unwrap())
+}
+
+fn be128(b: &[u8]) -> u128 {
+    u128::from_be_bytes(b.try_into().unwrap())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn specs(list: &[&str]) -> Vec<String> {
+        list.iter().map(|s| s.to_string()).collect()
+    }
+
+    fn filter(list: &[&str]) -> IpFilter {
+        IpFilter::parse(&specs(list)).unwrap()
+    }
+
+    /// An Ethernet frame with `vlan_tags` 802.1Q tags in front of `ethertype`.
+    fn eth(vlan_tags: usize, ethertype: u16, payload: &[u8]) -> Vec<u8> {
+        let mut f = vec![0u8; 12]; // dst + src MAC
+        for _ in 0..vlan_tags {
+            f.extend_from_slice(&ETHERTYPE_VLAN.to_be_bytes());
+            f.extend_from_slice(&[0x00, 0x64]); // TCI: VLAN 100
+        }
+        f.extend_from_slice(&ethertype.to_be_bytes());
+        f.extend_from_slice(payload);
+        f
+    }
+
+    fn ipv4(src: &str, dst: &str) -> Vec<u8> {
+        let mut p = vec![0u8; IPV4_HEADER_MIN];
+        p[0] = 0x45; // version 4, IHL 5
+        let s: std::net::Ipv4Addr = src.parse().unwrap();
+        let d: std::net::Ipv4Addr = dst.parse().unwrap();
+        p[12..16].copy_from_slice(&s.octets());
+        p[16..20].copy_from_slice(&d.octets());
+        p
+    }
+
+    fn ipv6(src: &str, dst: &str) -> Vec<u8> {
+        let mut p = vec![0u8; IPV6_HEADER_LEN];
+        p[0] = 0x60; // version 6
+        let s: std::net::Ipv6Addr = src.parse().unwrap();
+        let d: std::net::Ipv6Addr = dst.parse().unwrap();
+        p[8..24].copy_from_slice(&s.octets());
+        p[24..40].copy_from_slice(&d.octets());
+        p
+    }
+
+    fn v4_packet(src: &str, dst: &str) -> Vec<u8> {
+        eth(0, ETHERTYPE_IPV4, &ipv4(src, dst))
+    }
+
+    #[test]
+    fn match_all_keeps_everything_including_non_ip_and_odd_linktypes() {
+        let f = IpFilter::match_all();
+        assert!(f.is_match_all());
+        assert!(f.matches(&[], 113));
+        assert!(f.matches(&eth(0, 0x0806, &[1, 2, 3]), LINKTYPE_ETHERNET));
+        // An empty --ip list is the same thing.
+        assert!(IpFilter::parse(&[]).unwrap().is_match_all());
+    }
+
+    #[test]
+    fn exact_v4_matches_either_direction_only() {
+        let f = filter(&["1.1.1.1"]);
+        assert!(!f.is_match_all());
+        assert!(f.matches(&v4_packet("1.1.1.1", "9.9.9.9"), LINKTYPE_ETHERNET));
+        assert!(f.matches(&v4_packet("9.9.9.9", "1.1.1.1"), LINKTYPE_ETHERNET));
+        assert!(!f.matches(&v4_packet("9.9.9.9", "8.8.8.8"), LINKTYPE_ETHERNET));
+        // Neighbouring address must not match a bare (/32) entry.
+        assert!(!f.matches(&v4_packet("1.1.1.2", "9.9.9.9"), LINKTYPE_ETHERNET));
+    }
+
+    #[test]
+    fn exact_v6_matches_either_direction() {
+        let f = filter(&["2606:4700::1111"]);
+        let pkt = eth(0, ETHERTYPE_IPV6, &ipv6("2001:db8::1", "2606:4700::1111"));
+        assert!(f.matches(&pkt, LINKTYPE_ETHERNET));
+        let miss = eth(0, ETHERTYPE_IPV6, &ipv6("2001:db8::1", "2606:4700::1112"));
+        assert!(!f.matches(&miss, LINKTYPE_ETHERNET));
+    }
+
+    #[test]
+    fn cidr_boundaries_are_exact() {
+        let f = filter(&["10.0.0.0/8"]);
+        assert!(f.matches(&v4_packet("10.0.0.0", "9.9.9.9"), LINKTYPE_ETHERNET));
+        assert!(f.matches(&v4_packet("10.255.255.255", "9.9.9.9"), LINKTYPE_ETHERNET));
+        assert!(!f.matches(&v4_packet("11.0.0.0", "9.9.9.9"), LINKTYPE_ETHERNET));
+        assert!(!f.matches(&v4_packet("9.255.255.255", "9.9.9.9"), LINKTYPE_ETHERNET));
+
+        // /0 keeps every packet of that family, but still not non-IP.
+        let all = filter(&["0.0.0.0/0"]);
+        assert!(all.matches(&v4_packet("203.0.113.9", "9.9.9.9"), LINKTYPE_ETHERNET));
+        assert!(!all.matches(&eth(0, 0x0806, &[0u8; 28]), LINKTYPE_ETHERNET));
+
+        let v6 = filter(&["2606:4700::/32"]);
+        assert!(v6.matches(&eth(0, ETHERTYPE_IPV6, &ipv6("2606:4700:dead::1", "::1")), LINKTYPE_ETHERNET));
+        assert!(!v6.matches(&eth(0, ETHERTYPE_IPV6, &ipv6("2606:4701::1", "::1")), LINKTYPE_ETHERNET));
+    }
+
+    #[test]
+    fn families_do_not_cross_match() {
+        // A v4-only filter must never keep a v6 packet, and vice versa --
+        // the split vectors mean the comparison is never even attempted.
+        let v4only = filter(&["1.1.1.1"]);
+        assert!(!v4only.matches(&eth(0, ETHERTYPE_IPV6, &ipv6("::1", "::2")), LINKTYPE_ETHERNET));
+        let v6only = filter(&["::1"]);
+        assert!(!v6only.matches(&v4_packet("1.1.1.1", "9.9.9.9"), LINKTYPE_ETHERNET));
+    }
+
+    #[test]
+    fn vlan_tags_are_skipped_single_and_double() {
+        let f = filter(&["1.1.1.1"]);
+        let one = eth(1, ETHERTYPE_IPV4, &ipv4("1.1.1.1", "9.9.9.9"));
+        let two = eth(2, ETHERTYPE_IPV4, &ipv4("9.9.9.9", "1.1.1.1"));
+        assert!(f.matches(&one, LINKTYPE_ETHERNET));
+        assert!(f.matches(&two, LINKTYPE_ETHERNET));
+        // Beyond the tag bound we stop rather than hunt through the payload.
+        let three = eth(3, ETHERTYPE_IPV4, &ipv4("1.1.1.1", "9.9.9.9"));
+        assert!(!f.matches(&three, LINKTYPE_ETHERNET));
+    }
+
+    #[test]
+    fn unparseable_packets_are_dropped() {
+        let f = filter(&["1.1.1.1"]);
+        // ARP.
+        assert!(!f.matches(&eth(0, 0x0806, &[0u8; 28]), LINKTYPE_ETHERNET));
+        // Truncated before the addresses (snaplen too small).
+        let full = v4_packet("1.1.1.1", "9.9.9.9");
+        assert!(!f.matches(&full[..ETH_HEADER_LEN + 12], LINKTYPE_ETHERNET));
+        // Shorter than an Ethernet header, and empty.
+        assert!(!f.matches(&full[..8], LINKTYPE_ETHERNET));
+        assert!(!f.matches(&[], LINKTYPE_ETHERNET));
+        // Right ethertype, wrong version nibble.
+        let mut bad_version = full.clone();
+        bad_version[ETH_HEADER_LEN] = 0x65;
+        assert!(!f.matches(&bad_version, LINKTYPE_ETHERNET));
+    }
+
+    #[test]
+    fn non_ethernet_linktypes_match_nothing() {
+        let f = filter(&["1.1.1.1"]);
+        let pkt = v4_packet("1.1.1.1", "9.9.9.9");
+        // 113 = LINUX_SLL, what `tcpdump -i any` produces; 101 = RAW; 0 = NULL.
+        for linktype in [0u32, 101, 113, 276] {
+            assert!(!f.matches(&pkt, linktype), "linktype {linktype} should match nothing");
+        }
+    }
+
+    #[test]
+    fn bad_specs_are_rejected_at_parse() {
+        for spec in ["1.1.1.1/33", "::1/129", "not-an-ip", "1.1.1.1/x", "1.1.1.256"] {
+            assert!(
+                IpFilter::parse(&specs(&[spec])).is_err(),
+                "{spec:?} should be rejected"
+            );
+        }
+        // /0 and full-width prefixes are fine.
+        assert!(IpFilter::parse(&specs(&["1.1.1.1/32", "0.0.0.0/0", "::/0", "::1/128"])).is_ok());
+    }
+
+    #[test]
+    fn multiple_entries_are_all_considered() {
+        let f = filter(&["1.1.1.1", "8.8.8.8", "10.0.0.0/8", "2606:4700::/32"]);
+        assert!(f.matches(&v4_packet("8.8.8.8", "9.9.9.9"), LINKTYPE_ETHERNET));
+        assert!(f.matches(&v4_packet("9.9.9.9", "10.1.2.3"), LINKTYPE_ETHERNET));
+        assert!(f.matches(&eth(0, ETHERTYPE_IPV6, &ipv6("::1", "2606:4700::1")), LINKTYPE_ETHERNET));
+        assert!(!f.matches(&v4_packet("9.9.9.9", "203.0.113.1"), LINKTYPE_ETHERNET));
+    }
+}

src/main.rs

@@ -2,6 +2,7 @@ mod channel;
 mod chunker;
 mod cli;
 mod client;
+mod filter;
 mod naming;
 mod pipeline;
 mod protocol;

src/naming.rs

@@ -1,4 +1,5 @@
 use anyhow::{bail, Result};
+use std::time::Duration;
 use std::path::Path;
 
 /// Either a local filesystem path or a `[user@]host:path` remote target.
@@ -128,6 +129,28 @@ pub fn parse_size(s: &str) -> Result<usize> {
     Ok(n * mult)
 }
 
+/// Parses a flush/idle interval: a bare number is milliseconds, or an
+/// explicit "ms"/"s" suffix. "100", "100ms" and "0.1s" are not all spellings
+/// of the same thing -- only integers are accepted, so use "100ms" or "1s".
+pub fn parse_duration(s: &str) -> Result<Duration> {
+    let s = s.trim();
+    if s.is_empty() {
+        bail!("empty interval");
+    }
+    let (digits, mult_ms) = if let Some(d) = s.strip_suffix("ms") {
+        (d, 1)
+    } else if let Some(d) = s.strip_suffix('s') {
+        (d, 1000)
+    } else {
+        (s, 1)
+    };
+    let n: u64 = digits
+        .trim()
+        .parse()
+        .map_err(|_| anyhow::anyhow!("invalid interval: {s:?} (try \"100ms\" or \"1s\")"))?;
+    Ok(Duration::from_millis(n * mult_ms))
+}
+
 /// Single-quote a string for safe inclusion as one argument in a remote shell
 /// command line, e.g. for `ssh host sh -c '...'`-style invocations.
 pub fn shell_quote(s: &str) -> String {
@@ -251,6 +274,19 @@ mod tests {
         assert_eq!(d.temp_path, "cap.pcap.partial");
     }
 
+    #[test]
+    fn duration_parsing() {
+        assert_eq!(parse_duration("100").unwrap(), Duration::from_millis(100));
+        assert_eq!(parse_duration("100ms").unwrap(), Duration::from_millis(100));
+        assert_eq!(parse_duration("2s").unwrap(), Duration::from_secs(2));
+        assert_eq!(parse_duration(" 250ms ").unwrap(), Duration::from_millis(250));
+        // 0 is legal: cut a chunk as soon as anything is kept.
+        assert_eq!(parse_duration("0").unwrap(), Duration::ZERO);
+        for bad in ["", "abc", "0.1s", "100m", "-5"] {
+            assert!(parse_duration(bad).is_err(), "{bad:?} should be rejected");
+        }
+    }
+
     #[test]
     fn size_parsing() {
         assert_eq!(parse_size("256k").unwrap(), 256 * 1024);

src/pipeline.rs

@@ -1,4 +1,5 @@
-use crate::chunker::{Chunker, RawChunker, RecordAlignedChunker, DEFAULT_FLUSH};
+use crate::chunker::{Chunker, RawChunker, RecordAlignedChunker};
+use crate::filter::IpFilter;
 use crate::protocol::{Message, MessageSink, MessageSource};
 use crate::tail::{run_tail, SourceEvent};
 use crate::tracelog::Logger;
@@ -8,6 +9,7 @@ use std::fs::File;
 use std::io::Write;
 use std::path::PathBuf;
 use std::sync::{mpsc, Arc};
+use std::time::Duration;
 use std::thread;
 
 /// Bound on the reader->compressor and compressor->sender channels: enough to
@@ -22,7 +24,14 @@ pub struct SendOptions {
     pub final_source_path: PathBuf,
     pub compress: bool,
     pub chunk_target: usize,
+    /// How long kept-but-undersized output may wait before being sent
+    /// anyway. See `Chunker::should_flush`.
+    pub flush_after: Duration,
     pub level: i32,
+    /// Which packets to forward. `IpFilter::match_all()` when no `--ip` was
+    /// given -- not a disabled filter, just one that says yes, so there is
+    /// only ever one code path through the chunker.
+    pub filter: IpFilter,
     pub log: Arc<Logger>,
 }
 
@@ -39,8 +48,32 @@ 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 || opts.is_partial {
+    // Two independent reasons to parse records, either of which forces the
-        Box::new(RecordAlignedChunker::new(opts.chunk_target, DEFAULT_FLUSH, opts.is_partial))
+    // record-aligned chunker:
+    //   * `compress`: chunk boundaries become zstd frame boundaries, and the
+    //     format requires every frame to hold a whole number of records
+    //     (STREAMING_ZSTD_FORMAT.md invariant 3). The cut point *is* the format.
+    //   * `is_partial`: a growing source needs `confirmed_len()` -- record
+    //     structure is the only way to tell real content from a preallocating
+    //     writer's not-yet-written tail. Here the chunker is a validator, not
+    //     a framer.
+    //   * a real `--ip` filter: deciding whether to forward a packet means
+    //     finding the packet, and record boundaries are the only way to know
+    //     where one starts. Here the chunker is a demultiplexer.
+    // A final, uncompressed copy has neither: `run_recv` writes every Data
+    // payload verbatim and never inspects a boundary, so alignment would buy
+    // nothing, while requiring pcap structure would break copying pcapng,
+    // truncated, or otherwise non-pcap files that transfer fine today. So it
+    // gets the passthrough chunker, framed on the wire only by tail.rs's
+    // READ_WINDOW.
+    let chunker: Box<dyn Chunker> = if opts.compress || opts.is_partial || !opts.filter.is_match_all()
+    {
+        Box::new(RecordAlignedChunker::with_filter(
+            opts.chunk_target,
+            opts.flush_after,
+            opts.is_partial,
+            opts.filter.clone(),
+        ))
     } else {
         Box::new(RawChunker::new())
     };
@@ -238,7 +271,9 @@ mod tests {
             final_source_path: src.clone(),
             compress: false,
             chunk_target: crate::chunker::DEFAULT_TARGET_BYTES,
+            flush_after: crate::chunker::DEFAULT_FLUSH,
             level: crate::zstd_frame::DEFAULT_LEVEL,
+            filter: IpFilter::match_all(),
             log: Arc::new(Logger::none()),
         };
         run_pipe(opts, dest_final.clone(), dest_temp.clone()).unwrap();
@@ -274,7 +309,9 @@ mod tests {
             final_source_path: final_src.clone(),
             compress: false,
             chunk_target: crate::chunker::DEFAULT_TARGET_BYTES,
+            flush_after: crate::chunker::DEFAULT_FLUSH,
             level: crate::zstd_frame::DEFAULT_LEVEL,
+            filter: IpFilter::match_all(),
             log: Arc::new(Logger::none()),
         };
 
@@ -321,7 +358,9 @@ mod tests {
             final_source_path: src.clone(),
             compress: true,
             chunk_target: 512, // small target to force multiple frames
+            flush_after: crate::chunker::DEFAULT_FLUSH,
             level: crate::zstd_frame::DEFAULT_LEVEL,
+            filter: IpFilter::match_all(),
             log: Arc::new(Logger::none()),
         };
         run_pipe(opts, dest_final.clone(), dest_temp.clone()).unwrap();
@@ -357,10 +396,138 @@ mod tests {
             final_source_path: src.clone(),
             compress: true,
             chunk_target: crate::chunker::DEFAULT_TARGET_BYTES,
+            flush_after: crate::chunker::DEFAULT_FLUSH,
             level: crate::zstd_frame::DEFAULT_LEVEL,
+            filter: IpFilter::match_all(),
             log: Arc::new(Logger::none()),
         };
         assert!(run_pipe(opts, dest_final.clone(), dest_temp).is_err());
         assert!(!dest_final.exists());
     }
+
+    fn eth_header_bytes() -> Vec<u8> {
+        let mut h = header_bytes(0xA1B2C3D4);
+        h[20..24].copy_from_slice(&1u32.to_le_bytes()); // LINKTYPE_ETHERNET
+        h
+    }
+
+    /// A record holding an Ethernet/IPv4 packet, padded to `len` captured bytes.
+    fn ip_record(src: [u8; 4], dst: [u8; 4], len: usize) -> Vec<u8> {
+        let mut pkt = vec![0u8; len.max(34)];
+        pkt[12..14].copy_from_slice(&0x0800u16.to_be_bytes());
+        pkt[14] = 0x45;
+        pkt[26..30].copy_from_slice(&src);
+        pkt[30..34].copy_from_slice(&dst);
+        record_bytes(&pkt, 1_700_000_000)
+    }
+
+    fn ip_filter(specs: &[&str]) -> IpFilter {
+        IpFilter::parse(&specs.iter().map(|s| s.to_string()).collect::<Vec<_>>()).unwrap()
+    }
+
+    /// Source stream plus the exact pcap a `--ip 1.1.1.1` transfer should
+    /// produce from it: same global header, only the matching records.
+    fn filtered_fixture(n: usize) -> (Vec<u8>, Vec<u8>) {
+        let mut source = eth_header_bytes();
+        let mut expected = eth_header_bytes();
+        for i in 0..n {
+            if i % 5 == 0 {
+                let hit = ip_record([1, 1, 1, 1], [9, 9, 9, 9], 100 + i);
+                source.extend_from_slice(&hit);
+                expected.extend_from_slice(&hit);
+            } else {
+                source.extend_from_slice(&ip_record([9, 9, 9, 9], [8, 8, 8, 8], 100 + i));
+            }
+        }
+        (source, expected)
+    }
+
+    #[test]
+    fn filtered_transfer_carries_only_matching_packets() {
+        let dir = tempdir().unwrap();
+        let src = dir.path().join("cap.pcap");
+        let (source, expected) = filtered_fixture(50);
+        std::fs::write(&src, &source).unwrap();
+        let dest_final = dir.path().join("out.pcap");
+        let dest_temp = dir.path().join("out.pcap.partial");
+
+        let opts = SendOptions {
+            source_path: src.clone(),
+            is_partial: false,
+            final_source_path: src.clone(),
+            compress: false,
+            chunk_target: crate::chunker::DEFAULT_TARGET_BYTES,
+            flush_after: crate::chunker::DEFAULT_FLUSH,
+            level: crate::zstd_frame::DEFAULT_LEVEL,
+            filter: ip_filter(&["1.1.1.1"]),
+            log: Arc::new(Logger::none()),
+        };
+        run_pipe(opts, dest_final.clone(), dest_temp).unwrap();
+
+        // A filter forces the record-aligned chunker even though this is a
+        // final, uncompressed copy -- so the output is a real pcap, not a
+        // byte-for-byte copy of the source.
+        assert_eq!(std::fs::read(&dest_final).unwrap(), expected);
+        assert!(expected.len() < source.len());
+    }
+
+    #[test]
+    fn filtered_compressed_transfer_round_trips_to_the_filtered_pcap() {
+        let dir = tempdir().unwrap();
+        let src = dir.path().join("cap.pcap");
+        let (source, expected) = filtered_fixture(200);
+        std::fs::write(&src, &source).unwrap();
+        let dest_final = dir.path().join("out.pcap.zst");
+        let dest_temp = dir.path().join("out.pcap.zst.partial");
+
+        let opts = SendOptions {
+            source_path: src.clone(),
+            is_partial: false,
+            final_source_path: src.clone(),
+            compress: true,
+            chunk_target: 512,
+            flush_after: crate::chunker::DEFAULT_FLUSH,
+            level: crate::zstd_frame::DEFAULT_LEVEL,
+            filter: ip_filter(&["1.1.1.1"]),
+            log: Arc::new(Logger::none()),
+        };
+        run_pipe(opts, dest_final.clone(), dest_temp).unwrap();
+
+        let out_bytes = std::fs::read(&dest_final).unwrap();
+        assert_eq!(&out_bytes[0..14], &MARKER_FRAME[..]);
+        // The whole point of the container: decompressing yields a valid
+        // pcap, here the filtered one.
+        assert_eq!(zstd::stream::decode_all(out_bytes.as_slice()).unwrap(), expected);
+        // Header frame still declares exactly the 24-byte global header.
+        let header_content_size = zstd::zstd_safe::get_frame_content_size(&out_bytes[14..])
+            .unwrap()
+            .unwrap();
+        assert_eq!(header_content_size, 24);
+    }
+
+    #[test]
+    fn a_filter_matching_nothing_produces_a_header_only_pcap() {
+        // Not an error: the transfer completes and the destination is a
+        // valid, empty capture.
+        let dir = tempdir().unwrap();
+        let src = dir.path().join("cap.pcap");
+        let (source, _) = filtered_fixture(20);
+        std::fs::write(&src, &source).unwrap();
+        let dest_final = dir.path().join("out.pcap");
+        let dest_temp = dir.path().join("out.pcap.partial");
+
+        let opts = SendOptions {
+            source_path: src.clone(),
+            is_partial: false,
+            final_source_path: src.clone(),
+            compress: false,
+            chunk_target: crate::chunker::DEFAULT_TARGET_BYTES,
+            flush_after: crate::chunker::DEFAULT_FLUSH,
+            level: crate::zstd_frame::DEFAULT_LEVEL,
+            filter: ip_filter(&["203.0.113.7"]),
+            log: Arc::new(Logger::none()),
+        };
+        run_pipe(opts, dest_final.clone(), dest_temp).unwrap();
+        assert_eq!(std::fs::read(&dest_final).unwrap(), eth_header_bytes());
+    }
 }

src/server.rs

@@ -1,4 +1,5 @@
 use crate::cli::{ServerCli, ServerRole};
+use crate::filter::IpFilter;
 use crate::naming;
 use crate::pipeline::{run_recv, run_send, RecvOptions, SendOptions};
 use crate::protocol::{MessageSink, MessageSource, WireSink, WireSource};
@@ -35,6 +36,11 @@ fn run_send_server(cli: ServerCli) -> Result<()> {
         None => false,
     };
     let chunk_target = naming::parse_size(&cli.chunk)?;
+    let flush_after = naming::parse_duration(&cli.flush_interval)?;
+    // The sending side is the only side that filters -- dropping packets
+    // before they hit the wire is the whole point. `run_recv` never parses
+    // what it writes, so --recv has no --ip.
+    let filter = IpFilter::parse(&cli.ip)?;
 
     let opts = SendOptions {
         source_path: PathBuf::from(source),
@@ -42,7 +48,9 @@ fn run_send_server(cli: ServerCli) -> Result<()> {
         final_source_path: PathBuf::from(final_source),
         compress,
         chunk_target,
+        flush_after,
         level: cli.level,
+        filter,
         log,
     };
     // `Stdout`/`Stdin` (not the `.lock()` guards) are used here: the guards

src/tail.rs

@@ -170,10 +170,6 @@ fn drain(
                         last_stuck_peek = Some(preview.to_vec());
                     }
                 }
-            } 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;
@@ -193,13 +189,24 @@ fn drain(
                     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)?;
             }
         }
 
+        // Checked on EVERY iteration, not just ones that read nothing. The
+        // flush timer measures how long output has been waiting, so it has
+        // to be able to fire while the source is still growing: with a
+        // selective filter, a busy source produces a steady trickle of
+        // matching records that would otherwise sit here until they reached
+        // the size target -- which at a 1% match rate means ~100x the target
+        // in source bytes, seconds of added latency on a live capture.
+        // `should_flush` is false whenever there is nothing kept, so this
+        // never cuts an empty chunk.
+        if chunker.should_flush(Instant::now())
+            && let Some((kind, data)) = chunker.flush()
+        {
+            send_chunk(events, kind, data)?;
+        }
+
         if !complete_signal && final_path.try_exists()? {
             // Rename observed -- signal now, well before finalization, per
             // section 7.4's repoint-then-drain-then-finalize ordering.