[Detail Bug] Async Producer: submit/flush/close ignore task cancellation when cancelling linger task

[detail-app]

Detail Bug Report

https://app.detail.dev/org_89d327b3-b883-4365-b6a3-46b6701342a9/bugs/bug_8cb3b15b-57bc-4021-afd3-be6ffc0c8b65

Introduced in #78 by @quettabit on Jun 13, 2026

Summary

• Context: Producer._cancel_linger_task cancels a pending linger background task and awaits its termination whenever submit/flush/close routes through _submit_batch_now.
• Bug: An external cancellation of the calling submit/flush/close task that arrives while it is suspended inside await linger_task is silently swallowed by the surrounding with suppress(asyncio.CancelledError).
• Actual vs. expected: the caller's task should raise CancelledError; instead submit returns a RecordSubmitTicket, flush returns None, and close returns None — the caller's cancellation is dropped on the floor and the caller is returned a success result as if the operation completed normally.
• Impact: standard asyncio cancellation/timeout mechanisms (task.cancel(), asyncio.timeout, asyncio.wait_for with T>0) cannot reliably abort submit/flush/close once execution reaches _cancel_linger_task; the call continues and returns a normal success value.

Code with Bug

In src/s2_sdk/_producer.py:

async def _cancel_linger_task(self) -> None:
    linger_task = self._linger_task
    if linger_task is None:
        return
    self._linger_task = None
    if linger_task is asyncio.current_task():
        return
    linger_task.cancel()
    with suppress(asyncio.CancelledError):
        await linger_task  # <-- BUG 🔴 suppress also swallows the caller task's external cancellation

Explanation

with suppress(asyncio.CancelledError): await linger_task cannot distinguish between:

• the CancelledError raised because this code cancelled linger_task, and
• a CancelledError injected into the current coroutine because an external caller cancelled/timeouted the surrounding submit/flush/close task while it was suspended at that await.

If the external cancellation lands during that await, it is caught and discarded, clearing the task’s cancellation state. The method then continues and returns normally (RecordSubmitTicket/None), so the caller observes “success” rather than CancelledError/TimeoutError.

Codebase Inconsistency

ReadSession.close in src/s2_sdk/_read_session.py uses the same suppress/await pattern but includes a Task.cancelling() delta guard to re-raise if the surrounding task was cancelled:

current = asyncio.current_task()
cancellation_count = current.cancelling() if current is not None else 0
self._task.cancel()
with suppress(asyncio.CancelledError):
    await self._task
if current is not None and current.cancelling() > cancellation_count:
    raise asyncio.CancelledError

Recommended Fix

Port the Task.cancelling() delta guard pattern to Producer._cancel_linger_task, but ensure in-flight accumulated records/tickets are not stranded if cancellation is re-raised.

A verbatim port re-raises correctly but can skip _submit_accumulated_records(), leaving previously-issued RecordSubmitTicket ack futures unresolved (hang). When the guard detects external cancellation, also resolve any already-issued ack futures (e.g., set CancelledError on _indexed_ack_futs and clear the accumulator) before re-raising.

History

This bug was introduced in commit 4d4ac80 ("fix: unawaited asyncio.Task cancellations (#78)"): it added await linger_task wrapped in a bare with suppress(asyncio.CancelledError) inside _cancel_linger_task, which unintentionally suppresses external cancellation of the caller task.