streaming decoder: stray fields in later records, and "malformed at offset N" on valid bytes, only with small chunk sizes

I treat Incomplete as "not an error, just wait"

It is, but your parser has already moved on by then. pos has advanced and scratch holds the fields parsed so far. Your drain then throws away those half-record bytes, so the next call starts mid-record. That's one bug with two symptoms: stale keys, or a random byte read as a count.

Wrap it so a failed attempt rolls back:

pub fn next_record(&mut self, buf: &[u8]) -> Result<Record, ParseError> {
    let start = self.pos;
    let r = self.parse_record(buf);
    if matches!(r, Err(ParseError::Incomplete)) {
        self.pos = start;
        self.scratch.clear();
    }
    r
}

Rename your current function to parse_record. Try chunk size 1 too, as it's the quickest way to check.

scratch holds the fields parsed so far.

Why not local Vec?

pos has advanced and scratch holds the fields parsed so far.

The pos half is the real fix. The scratch half is only needed because the field is shared state. If the Vec is local, a failed attempt drops it on the early return and only pos needs restoring:

fn parse_record(&mut self, buf: &[u8]) -> Result<Record, ParseError> {
    let mut fields = Vec::new();
    let count = self.take(buf, 1)?[0] as usize;
    // ... count check as before ...
    for _ in 0..count {
        let k = self.take_str(buf)?;
        let v = self.take_str(buf)?;
        fields.push((k, v));
    }
    Ok(Record { fields })
}

One caveat for alden. After the drain, pos counts from the start of the remaining buffer, not the stream. So the "malformed at offset N" numbers will differ between chunk sizes even for the same bad byte. I once spent an evening chasing an offset that was just relative to the last drain. Keep a running total of drained bytes if you want stable numbers.

Guessing max_fields is set fairly high in your parser? Try chunk size 1 and diff the output against the one-chunk run.

I think the high max_fields also explains why alden sees stray fields at some chunk sizes and malformed at others. A misaligned byte only fails the count check if it's above the limit. Otherwise it passes as a count, and the parser then reads on and usually ends in Incomplete, so the desync stays silent. It only becomes Malformed when the garbage happens to trip the check or fail UTF-8.

For the diff, log where each record starts, counted from the stream rather than the buffer:

eprintln!("record {} starts at {}", seen + 1, drained + start);

Here drained is the running total of bytes removed so far, and start is pos before the call.

thanks brin40, and yes, max_fields is high (64 i think, i'd have to check the constant). the client's sample logs never go past 6 or 7 fields per record, so 64 was just "comfortably above anything real".

Guessing max_fields is set fairly high in your parser?

if i follow the idea, any stray byte under 64 passes the count == 0 || count > self.max_fields check as a plausible count. the parser then reads on, runs out of buffer, and returns Incomplete, so nothing looks wrong. only a byte above 64 (or a bad UTF-8 slice) gives Malformed. that would explain why the two symptoms land on different chunk sizes. it's one desync that fails loudly or quietly depending on which byte got misread. that fits better than my "two bugs" guess.

for the chunk size 1 check i'm running it like this (the one-chunk run is just a chunk size bigger than the file):

cargo run -- 1 > one.txt 2>&1
cargo run -- 100000 > all.txt 2>&1
diff one.txt all.txt

with the rollback from #2 in place the diff should be empty. if it isn't, i've got a second bug hiding behind the first.

one question, since the drain was where i started blaming things. would it be cleaner for next_record to return the bytes it consumed alongside the Record, so main calls Vec::drain with that count? then pos and reset_pos wouldn't be shared state in the parser at all. or is the extra return value just moving the same bookkeeping somewhere else?

would it be cleaner for next_record to return the bytes it consumed alongside the Record

I'd say yes. It's still bookkeeping, but it lives in one place. Make the parse a function of &[u8] with a local cursor starting at 0, returning (Record, usize). Incomplete then rolls back for free, because nothing was stored. reset_pos goes away too.

I'm assuming the real thing reads from a socket that can close mid-record. If so, check for leftover bytes in buf at EOF and report them as truncated. Right now that ends quietly, same as the silent desync.

For the diff, I'd also print the stream offset of each record start, as in #6. Then the first differing line points at a byte, not just a record number.

so main calls Vec::drain with that count

I think it's cleaner, but there's one catch with the offsets. With a local cursor starting at 0, the offset inside Malformed is relative to the slice you passed in, not to the buffer and not to the stream. So main has to add the running drained total to it before printing. If you also drain only once after the inner loop, it needs the bytes consumed by earlier records in that same pass. Otherwise the numbers are wrong again, just differently wrong.

The simplest way is to drain straight after each Ok, inside the loop:

Ok((rec, used)) => {
    buf.drain(..used);
    drained += used;
    // ...
}
Err(ParseError::Malformed { offset }) => {
    eprintln!("malformed at offset {}", drained + offset);
    return;
}

Now buf always starts at a record boundary, and drained + offset is a true stream offset. I had a length-prefixed decoder once where I summed the consumed counts and drained at the end of the pass. The error offsets were off by exactly one record's length, and it took me far too long to see why.

Also, in your diff the Malformed line goes through eprintln!, so check it lands in the right place in the file with 2>&1.

Instead of trying 7, 10 and 1 by hand, I'd loop over every chunk size in a test. Then you find the smallest size that breaks, not just two that happen to. I guess decode_all has to return the records instead of printing them, but that's a tiny change.

#[test]
fn same_for_every_chunk_size() {
    let want = decode_all(DATA, DATA.len());
    for n in 1..=DATA.len() {
        assert_eq!(decode_all(DATA, n), want, "chunk size {n}");
    }
}