Fix #606: never commit past an in-flight (unacked) message

[wbarnha]

What

Under concurrency, faust could commit an offset past a message that was still being processed, silently losing it on the next restart.

Reproduction from the issue (concurrency=2): message 1 is slow (sleep(100)) while messages 2–5 process in ~1s and ack. faust commits the offset of message 5, and after a SIGKILL/restart the consumer resumes after message 1 — message 1 is never processed.

Fixes #606.

Root cause

In Consumer._new_offset, the head-gap guard only withholds when there's a gap larger than one:

if self._committed_offset[tp]:
    if min(acked) - self._committed_offset[tp] > 1:
        return None

When the still-processing message sits at exactly the committed head, the acked run starts one past it, so min(acked) - committed == 1 — not > 1 — and faust commits straight over it. It's an off-by-one against the in-flight case: a single missing offset at the head is treated as a compacted-topic gap (safe to skip) when it's actually an event still being processed (must not skip).

Deterministic reproduction (before the fix): _new_offset with committed=2585, acked=[2586,2587,2588,2589] returns 2590 → offset 2585 lost.

Fix

Enforce the at-least-once invariant directly rather than inferring it from the acked run: cap the committable batch at the smallest offset that has been read but not yet acked for the topic-partition (_smallest_unacked_offset, computed from the _unacked_messages faust already tracks).

• The in-flight offset and everything after it are withheld until it acks.
• Offsets acked before an in-flight message are still committed (committed=100, acked=[100,101,102], 101 in-flight → commits 100, returns 101).
• Compacted-topic gaps are unaffected — a skipped offset is never in _unacked_messages, so those still advance as before.

Tests

• test_new_offset__withheld_for_inflight_message — the exact Wrong offset commit in case of gaps #606 state; asserts _new_offset returns None instead of 2590 (verified to fail before the fix with assert 2590 is None).
• test_new_offset__commits_up_to_inflight_message — earlier acked offsets before the in-flight one are still committable.
• All existing test_new_offset / test_new_offset_with_gaps / _add_gap tests pass unchanged (they carry no in-flight messages, so the cap is inert). Full tests/unit/transport/test_consumer.py: 115 passed, 2 skipped.

Note on end-to-end verification

The bug lives entirely in _new_offset's arithmetic, which these deterministic tests exercise at exactly the failing state. A full live-broker end-to-end reproduction (produce → slow-consume head → kill → restart → assert no message lost), on top of the new Kafka-in-CI harness from #706, is planned as a follow-up once the in-process faust App E2E fixture is hardened.

🤖 Generated with Claude Code

---

Generated by Claude Code

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 94.15%. Comparing base (c95a518) to head (7c3e622).
⚠️ Report is 2 commits behind head on master.

Additional details and impacted files

@@           Coverage Diff           @@
##           master     #707   +/-   ##
=======================================
  Coverage   94.15%   94.15%           
=======================================
  Files         104      104           
  Lines       11136    11144    +8     
  Branches     1201     1203    +2     
=======================================
+ Hits        10485    10493    +8     
  Misses        550      550           
  Partials      101      101           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.