Sweep untracked tasks and close thread_loop on ServiceThread shutdown

[wbarnha]

Description

Fixes #54.

ServiceThread._shutdown_thread() only cancels/awaits futures registered via this Service's own add_future()/@Service.task tracking. A driver library running inside the thread (e.g. a Kafka client's internal consumer poll/heartbeat loop) can schedule tasks directly on thread_loop, bypassing that tracking entirely. Left unswept, those tasks were simply abandoned the moment _serve() returned and thread_loop stopped running — producing coroutine ... was never awaited / Task was destroyed but it is pending warnings at some later, unpredictable point (e.g. interpreter shutdown), since the abandoned task/coroutine object is only reported once garbage collected.

mode.Worker already has an equivalent safety net for the main loop: _gather_all() sweeps every task on self.loop via all_tasks(), tracked or not, before closing it. ServiceThread had no equivalent for thread_loop, and never explicitly closed thread_loop at all.

Changes

• Add ServiceThread._gather_remaining_tasks(), called at the end of _serve()'s finally block (after _shutdown_thread(), while thread_loop is still running): cancel and await every task still on thread_loop, tracked or not, mirroring Worker._gather_all().
• Close thread_loop explicitly once _start_thread() returns — but only when ServiceThread created it itself (thread_loop=None, the default). A caller who passed their own thread_loop= may expect to keep managing its lifecycle, so that case is left untouched.

Verification

• A minimal repro (no aiokafka/Kafka broker required): a subclass that schedules an untracked asyncio.ensure_future(...) task in on_thread_started() no longer leaves it pending after stop() — confirmed by checking the task's own .done()/.cancelled() state directly. (A warnings-capture-based version of this test was tried first and found unreliable, since the GC-triggered warning can fire outside any reasonably-sized capture window — the direct task-state assertion is what actually proves the fix.)
• Confirmed both new tests fail against the pre-fix code with the exact predicted symptom, and pass with the fix.
• Full suite: passes on Python 3.11 and 3.13, ruff clean, stable across repeated --random-order runs (relevant given this touches concurrency-sensitive shutdown code).
• Full faust-streaming/faust cross-check (tests/unit + tests/functional, 2128 passed / 32 skipped / 0 failed, unchanged from before this change) — confirms no regression in the AIOKafkaConsumerThread-based driver that originally reported this issue.

🤖 Generated with Claude Code

https://claude.ai/code/session_01HgnKFtXZbCjXNoVNWa5JLd

---

Generated by Claude Code