[Detail Bug] Producer drops records and becomes unusable when a batch exceeds the 1MiB server limit

[detail-app]

Detail Bug Report

https://app.detail.dev/org_89d327b3-b883-4365-b6a3-46b6701342a9/bugs/bug_d6e9610d-d0c8-4758-b8ca-9daf505a8047

Introduced in #1 by @quettabit on Apr 7, 2026

Summary

• Context: The Producer class in _producer.py batches records using BatchAccumulator before submitting them to AppendSession.
• Bug: BatchAccumulator.add() appends records without checking if the total metered bytes would exceed ONE_MIB, allowing the accumulator to overflow the server's hard batch size limit.
• Actual vs. expected: When overflow occurs, accumulator.take() clears records BEFORE validation in AppendSession.submit(). Validation then fails, permanently poisoning the Producer. The records are lost - not in accumulator, not submitted, not recoverable.
• Impact: Records are permanently lost when valid individual records are combined into a batch exceeding ONE_MIB. The Producer becomes permanently unusable.

Code with Bug

# src/s2_sdk/_batching.py
def add(self, record: Record) -> None:
    self._records.append(record)  # <-- BUG 🔴 no size check; can push total bytes over max_bytes
    self._bytes += metered_bytes((record,))

def is_full(self) -> bool:
    return (
        len(self._records) >= self._batching.max_records
        or self._bytes >= self._batching.max_bytes
    )

# src/s2_sdk/_producer.py
first_in_batch = self._accumulator.is_empty()
self._accumulator.add(record)  # <-- BUG 🔴 overflow possible before is_full() check
if self._accumulator.is_full():
    await self._submit_batch_now()

# src/s2_sdk/_producer.py
records = self._accumulator.take()  # <-- BUG 🔴 clears accumulator before submit() validates batch size
batch = AppendInput(
    records=records,
    fencing_token=self._fencing_token,
    match_seq_num=self._match_seq_num,
)

try:
    ticket = await self._session.submit(batch)  # validation happens here
except BaseException as e:
    e = normalize_exception(e)
    self._error = e  # Producer poisoned
    for ack_fut in indexed_ack_futs:
        set_and_retrieve_future_exception(ack_fut, e)
    raise e

Explanation

• BatchAccumulator.add() always appends and increments _bytes, so the accumulator can exceed Batching.max_bytes by the size of the last record added.
• Producer.submit() checks is_full() only after adding; if the last record pushed the batch over the limit, _submit_batch_now() runs with an already-invalid batch.
• _submit_batch_now() calls accumulator.take() (clearing the accumulator) before AppendSession.submit() validates the batch size. If validation rejects the batch (e.g., num_bytes > ONE_MIB), the producer sets _error and fails all outstanding tickets, but the cleared records are not retained anywhere for retry.

Codebase Inconsistency

# src/s2_sdk/_types.py
max_bytes: int = ONE_MIB  # Default batching limit

# src/s2_sdk/_validators.py
if 1 <= num_records <= 1000 and num_bytes <= ONE_MIB:  # Server hard limit

The default Batching.max_bytes equals the server hard limit (ONE_MIB) with no safety margin, so a batch that exceeds the limit by even 1 byte will be rejected.

Failing Test

# tests/test_batch_overflow_bug.py
import pytest

from s2_sdk._producer import Producer

# helpers create_record_with_metered_size(), make_producer_with_fake_session()
# build records whose metered_bytes() matches the requested size.


@pytest.mark.asyncio
async def test_producer_overflow_poisons_and_loses_data():
    producer, submitted_batches = make_producer_with_fake_session()

    record_500k = create_record_with_metered_size(500_000)  # valid individually
    record_550k = create_record_with_metered_size(550_000)  # valid individually
    # Total: 1,050,000 bytes > ONE_MIB (1,048,576)

    ticket1 = await producer.submit(record_500k)
    with pytest.raises(Exception):
        await producer.submit(record_550k)

    assert producer._error is not None
    assert producer._accumulator.is_empty()
    assert len(submitted_batches) == 0
    with pytest.raises(Exception):
        await ticket1

Recommended Fix

Check for overflow before adding a record in Producer.submit(): if the new record would push the accumulator above max_bytes, flush the current batch first (when non-empty), then add the record.

History

This bug was introduced in commit 3dc9795. The initial implementation of the SDK contained the flawed design from day one: BatchAccumulator.add() appends records unconditionally, is_full() checks after the fact, and Producer._flush() clears the accumulator via take() before AppendSession.submit() validates the batch size. When valid individual records combine to exceed ONE_MIB, records are permanently lost because they're cleared from the accumulator before validation fails.