// Qwen3 Coder 30B

▮ PILLAR BREAKDOWN

▮ CRITICAL BUGS

• submit() raises Exception('Queue is full') instead of awaiting a slot (no backpressure) — submitting past queue_limit aborts the whole pipeline; this is why Test c crashes with an uncaught Queue-is-full error.
• Bounded concurrency is enforced two ways (manual len(_active_tasks)<max_concurrent check in _start_worker AND an unused asyncio.Semaphore acquire in _worker) while the actual queue (collections.deque) is never awaited, so there is no flow-control backpressure anywhere.
• Test b asserts the broken raise behavior as correct ('assert False, Expected queue limit to be exceeded'), codifying the bug as a feature — the model misunderstood backpressure as reject-on-overflow.
• drain() busy-polls with sleep(0.01) then sets _shutdown_event after, rather than awaiting the active task set; on shutdown workers are never explicitly cancelled so in-flight _mock_synthesize sleeps can outlive the pipeline.
• Cancellation of in-flight jobs is only a flag (job.cancelled=True) checked cooperatively; a job blocked in asyncio.sleep inside _mock_synthesize cannot be interrupted promptly.

▮ RECOMMENDED USE

▮ REFACTORED PATCH

# Fix 1: real bounded backpressure via asyncio.Queue; submit awaits a slot
import asyncio
class TTSJobPipeline:
    def __init__(self, max_concurrent=4, queue_limit=100):
        self.max_concurrent = max_concurrent
        self.queue = asyncio.Queue(maxsize=queue_limit)
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self._workers = set(); self._stopping = asyncio.Event()
    async def submit(self, text, voice):
        job = Job(id=..., text=text, voice=voice)
        await self.queue.put(job)          # BACKPRESSURE: await free slot, never raise
        self._callback_event(job.id, JobEvent.QUEUED); return job.id
    async def _worker(self):
        while not self._stopping.is_set():
            try: job = await asyncio.wait_for(self.queue.get(), timeout=0.5)
            except asyncio.TimeoutError: continue
            if job.cancelled:
                self._callback_event(job.id, JobEvent.CANCELLED); self.queue.task_done(); continue
            async with self.semaphore:      # the ONLY concurrency gate
                self._callback_event(job.id, JobEvent.STARTED)
                try:
                    await self._process_with_retry(job)
                    self._callback_event(job.id, JobEvent.COMPLETED)
                except Exception as e:
                    self._callback_event(job.id, JobEvent.FAILED, str(e))
                finally: self.queue.task_done()
    async def drain(self): await self.queue.join()   # no busy-poll
    async def aclose(self):                           # clean shutdown
        self._stopping.set()
        for w in self._workers: w.cancel()
        await asyncio.gather(*self._workers, return_exceptions=True)

# Fix 2 (tests): backpressure should be observed as submit() BLOCKING under load,
# not raising. Over-submit and assert it waited, didn't raise.