Fix two crashes in the aiokafka threaded producer - #759
Conversation
Both were found by the type checker in #758 and left marked `XXX` there because fixing them changes runtime behaviour. `ThreadedProducer._shutdown_thread` was a plain `def` overriding `mode.threads.ServiceThread._shutdown_thread`, which is `async def` and is awaited by `_serve()` in a `finally:`. Every shutdown of the producer thread therefore evaluated `await None` and raised TypeError. The thread only recovered because `_start_thread` catches that exception and calls `set_shutdown()` before re-raising -- so mode's teardown (`on_thread_stop`, stopping children, futures and exit stacks) never ran, and the thread died with a traceback instead of stopping cleanly. The override also scheduled `on_thread_stop()` with `asyncio.run_coroutine_threadsafe` onto `self.thread_loop` -- the loop that was about to stop, and the loop already running `_serve()`. Because the TypeError tore down `run_until_complete` immediately, that coroutine never got a chance to run, so the producer was never flushed or stopped on this path. Make it `async def` and await `super()._shutdown_thread()`, which runs `on_thread_stop()` and the rest of mode's teardown in order. The once-only guard is kept; when shutdown has already been initiated the shutdown event is still set, matching what the old TypeError path ended up doing via `_start_thread`. `ThreadedProducer.publish_message(wait=True)` called `fut.message.channel._on_published(message=..., state=..., producer=...)`. `Topic._on_published` takes the send future as a required *positional* `fut` and reads the result off it, so the call raised `TypeError: Topic._on_published() missing 1 required positional argument`. The waiting branch has no such future -- `send_and_wait` has already resolved -- so complete the message directly instead: report the sensor, set the result, and invoke the callback, which is what `Topic.publish_message(wait=True)` does via `_finalize_message`. The non-waiting branch keeps using `_on_published` as a done-callback, where `add_done_callback` supplies `fut`. `test_publish_message_with_wait` did not catch this because its channel is a bare `Mock`, which accepts any call; the new test uses a real topic and fails with the TypeError above against the previous code. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012N2yysiNbzzVcvYhtrpsM7
805596d to
02a772c
Compare
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #759 +/- ##
==========================================
- Coverage 96.20% 96.20% -0.01%
==========================================
Files 110 110
Lines 11790 11791 +1
Branches 1281 1281
==========================================
Hits 11343 11343
- Misses 350 351 +1
Partials 97 97 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
I'm not pushing a change for it, because there is no uncovered line to cover. Codecov's own patch check agrees: "All modified and coverable lines are covered by tests." The project delta comes from its summary reporting That is the only file whose numbers move anywhere in the package — 3 new statements, all 3 covered, misses flat. The missing lines are also the same set on both sides, just shifted by two line numbers. So the -0.01% is a single line out of ~11,084 appearing as uncovered in the merged 15-leg report while every individual leg covers it. Worth noting this is structural rather than specific to this PR: the repo has no codecov:
notify:
after_n_builds: 15 # 5 pythons x 2 cython (aiokafka) + 5 confluent.
# Not 16 - the PyPy leg is continue-on-error and
# uploads conditionally, so waiting on it can hang.
coverage:
status:
project:
default:
threshold: 0.5% # tolerate cross-leg varianceHappy to open that as a separate PR if wanted — it doesn't belong in this one. Generated by Claude Code |
The wait=True branch now awaits ``channel._finalize_message(fut, ret)``, but this test still passed a bare ``Mock()`` as the channel. A plain Mock returns a Mock, which is not awaitable, so every aiokafka test leg failed with ``TypeError: object Mock can't be used in 'await' expression``. Give the stand-in channel an ``AsyncMock`` ``_finalize_message``, matching the coroutine every real channel implements. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01A5Hzidb2taZ7ci6AeBUuJT
The 3.14t free-threaded leg failed teardown with:
DirtyTest: ('Left over tasks', ...
"<Task pending coro=<sleep()> cb=[_chain_future._call_set_state()]>")
That task is mode's, not faust's: ServiceThread._wakeup_timer_in_thread
ends every keepalive tick with
run_coroutine_threadsafe(asyncio.sleep(0), self.parent_loop)
and fires one last tick as the thread stops. The task completes on the
next iteration of the parent loop -- but the autouse tasks_not_lingering
fixture snapshots tasks as soon as the test coroutine returns, so if the
loop never runs again it is still pending and gets reported.
It surfaced here because _shutdown_thread now performs mode's real
teardown instead of raising TypeError out of _serve(), which gives the
keepalive room to tick before the thread goes away.
Yield once after stop() so the transient task settles.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01A5Hzidb2taZ7ci6AeBUuJT
2 participants
Both bugs were surfaced by the type checker in #758 and left marked
XXXthere, because fixing them changes runtime behaviour and that PR was annotation-only. This PR fixes them, each with a regression test.1.
_shutdown_threadraised TypeError on every producer-thread shutdownThreadedProducer._shutdown_threadwas a plaindefoverridingmode.threads.ServiceThread._shutdown_thread, which isasync defand is awaited by_serve(). The fix makes it async and awaitssuper()._shutdown_thread(), preserving mode's teardown and the existing once-only guard.2.
publish_message(wait=True)could never succeedThe waiting branch called
Topic._on_publishedwithout its required send-future argument. Real topics therefore raisedTypeError.The fix follows Faust's channel abstraction:
ChannelTdeclares_finalize_message.channel._finalize_message.Why the existing test missed it
The existing test used a bare
Mock()channel, which accepts the malformed call. The regression test uses a real topic and verifies the future result, callback, and sensor completion.Tests
Added regression coverage for the real-topic wait path and all shutdown paths.