fix: record_ack_and_advance uses sentinel 0 for unknown sources #431

Summary

In crates/logfwd-core/src/pipeline/lifecycle.rs, record_ack_and_advance() uses unwrap_or(0) for old_offset when a source has never been committed. If a spurious ack arrives for an unknown source, this inserts 0 into the committed map — making committed_offset() return Some(0) instead of None.

Current behavior

let old_offset = committed.get(&source).copied().unwrap_or(0);
// ... advance loop may not run ...
committed.insert(source, current); // inserts 0 for unknown source

Expected behavior

old_offset should be Option<u64>, and committed.insert() should only happen when the offset actually advanced. This keeps committed_offset() returning None for sources that were never properly committed.

Impact

Low — AckReceipt fields are pub(crate), so external code can't forge receipts. But the internal invariant is violated if a spurious ack is processed.

Fix

Change record_ack_and_advance to track old_offset as Option<u64> and only insert into committed when current != old_offset. The CommitAdvance struct already uses Option<u64> fields — just needs the internal logic to match.

Refs #270