test: deflake test_encode_and_write_as_completed_cancels_stray_writes_on_failure

[d-v-b]

AI PR that makes a flaky test a bit more reliable.

🤖 AI text below 🤖

Summary

tests/test_fused_pipeline.py::test_encode_and_write_as_completed_cancels_stray_writes_on_failure (added in #4205) fails intermittently on assert write_started.is_set(). It has failed five times in CI since 2026-07-30:

Date	Run	Job
2026-07-31	30624113397	windows, py3.14, optional
2026-09-09	34382421806 (main)	macos, py3.14, optional
2026-09-12	34707555000	macos, py3.14, optional
2026-09-20	35536950892	macos, py3.14, optional
2026-09-25	36104239476 (#4407)	ubuntu, py3.13, minimal

Cause

The test assumes the slow write is always in flight when the failing write raises. It isn't when the failing chunk's encode lands first (encodes run in the thread pool, so order varies):

1. The failing write task is created and raises while _encode_and_write_as_completed is still awaiting the other encode.
2. The slow encode lands, and the slow write task is created, but it hasn't run a step yet.
3. asyncio.gather runs the done-callbacks of already-finished children synchronously (the "finished eagerly" optimization). With return_exceptions=False, the failed child makes the outer future fail immediately.
4. await gather(...) therefore doesn't suspend. The finally cancels the slow write before its first step, so write_started is never set.

Instrumented trace of a failing run:

task created 1          # failing write
task 1 first step
fail.set raise
task created 0          # slow write, not yet stepped
gather called: done=True, done=False
gather outer done=True  # already failed; await won't suspend
drain: Task(done=True), Task(done=False), ...

The pipeline behaves correctly here: no write is left running, since it's cancelled before it starts. The test is what's wrong: it only means to check cancellation of an in-flight write.

Fix

Test-only. The failing write now waits (bounded, 5 s) until the slow write has started before raising, so every run exercises the in-flight case. The docstring explains why.

Verification

• Reproduced with pytest-repeat and 16 parallel workers: 8/400 failed on py3.13, 3/400 on py3.14.
• With the fix: 2000/2000 pass on both py3.13 and py3.14.
• Still catches the bug it guards: with the _cancel_and_drain call in _encode_and_write_as_completed replaced by pass, the test fails on assert not write_finished.
• Full file: tests/test_fused_pipeline.py passes.

No changelog entry, since this is a test-only change.

🤖 Generated with Claude Code