warehack.ing/mcspeak
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Drain the queue on shutdown instead of dropping it

Browse Source 
On graceful stop (SIGTERM from docker compose up/restart/down), stop accepting
new speech and play out what's already queued before exiting — rather than
finishing only the current message and dropping the rest.

Split the single _stopped flag into two: _accepting gates new enqueues
(rejected the moment shutdown begins) and _stopped tells the consumer to exit
(only after the queue drains). stop() now: set _accepting=False, wait for the
consumer to play the queue empty (bounded by shutdown_timeout), then stop the
consumer; anything past the budget is dropped/reaped as before. No external
endpoint or deploy-time coordination needed — uvicorn's existing SIGTERM ->
lifespan -> queue.stop() path carries it.
This commit is contained in:
Ryan Malloy 2026-07-04 21:50:47 -06:00
parent 08f40ae2dc
commit ec98b6958e
2 changed files with 50 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
CLAUDE.md
View File

@ -262,14 +262,14 @@ When a currently-playing item is explicitly cancelled, pw-play is killed immedia
## Graceful Shutdown
On `docker compose down` or `make restart`, the server lets the currently-playing speech finish before stopping — no more mid-sentence cutoffs.
On `docker compose down`, `make up`, or `make restart`, the server **stops accepting new speech and plays out what's already queued** before stopping — not just the currently-playing message. No dropped messages on a redeploy.
**How it works:** SIGTERM triggers a two-phase shutdown that shares Docker's wall-clock budget:
1. **Phase 1 — Handler drain (3s fixed):** Uvicorn cancels in-flight `speak()` handlers. The handlers re-raise `CancelledError` without touching the consumer — audio keeps playing.
2. **Phase 2 — Queue drain (`shutdown_timeout`, default 30s):** Lifespan finalizer calls `queue.stop()`, which waits for the consumer to finish the current audio. If the timeout expires, it force-cancels.
1. **Phase 1 — Handler drain (3s fixed):** Uvicorn cancels in-flight `speak()` handlers. The handlers re-raise `CancelledError` without touching the consumer — already-enqueued audio keeps playing.
2. **Phase 2 — Queue drain (`shutdown_timeout`, default 30s):** Lifespan finalizer calls `queue.stop()`, which (a) sets `_accepting=False` so new `speak()` calls are rejected immediately, then (b) lets the consumer keep playing queued utterances normally until the queue is empty and nothing's playing, then (c) stops the consumer. Whatever doesn't fit the budget is dropped and reaped. The `_accepting` flag (reject new) is deliberately separate from `_stopped` (consumer exit) so the drain window sits between them.
Docker's `stop_grace_period` must exceed the total: `3s + shutdown_timeout + 5s safety margin` (default 38s). If the max expected audio length exceeds `shutdown_timeout`, increase both `TTS_SHUTDOWN_TIMEOUT` and `stop_grace_period`.
Docker's `stop_grace_period` must exceed the total: `3s + shutdown_timeout + safety margin` (default 38s). Because the whole queue now drains (not just the current message), a deep queue can hit the budget — raise both `TTS_SHUTDOWN_TIMEOUT` and `stop_grace_period` if you want longer drains.
**Entry tone timing:** The Nextel chirp fires immediately when `speak()` is called (before synthesis), acting as an audible "I heard you" that covers the 1-3s synthesis latency. Exit/standby tones still play from the consumer (they depend on queue state after playback).

72
src/mcspeak/queue.py
View File

@ -154,6 +154,10 @@ class SpeechQueue:
self._item_cancelled = False
self._counter = 0
self._max_depth = max_depth
# _accepting gates NEW enqueues; _stopped tells the consumer to exit.
# On graceful shutdown they're set at different times: stop accepting
# first, keep playing until the queue drains, THEN stop the consumer.
self._accepting = True
self._stopped = False
self._exit_tone = exit_tone
self._standby_tone = standby_tone
@ -237,6 +241,7 @@ class SpeechQueue:
"""Launch the consumer coroutine."""
if self._consumer_task is None or self._consumer_task.done():
self._stopped = False
self._accepting = True
self._consumer_task = asyncio.create_task(
self._consumer(), name="speech-queue-consumer"
)
@ -282,46 +287,61 @@ class SpeechQueue:
self._consumer_task.cancel()
async def stop(self) -> None:
"""Gracefully stop the consumer — let current speech finish, then drain.
"""Graceful shutdown: stop accepting new speech, DRAIN the queue, then stop.
Waits up to shutdown_timeout seconds for the currently-playing utterance
to complete. If it doesn't finish in time, falls back to hard cancel.
Drain always runs, even if stop() itself is cancelled.
Step 1: reject new enqueues immediately (`_accepting=False`).
Step 2: let the consumer keep playing queued utterances normally (it's
still running because `_stopped` is false) until the queue is
empty and nothing is playing — bounded by `shutdown_timeout`.
Step 3: tell the consumer to exit; anything that didn't fit the budget
is dropped and reaped by `_drain_remaining()`.
Runs its cleanup even if stop() itself is cancelled.
"""
self._stopped = True
self._accepting = False # stop accepting new requests right away
loop = asyncio.get_running_loop()
deadline = loop.time() + self._shutdown_timeout
try:
if self._consumer_task and not self._consumer_task.done():
if self._current:
elapsed = time.time() - self._current.enqueued_at
remaining = max(0, self._current.duration_estimate - elapsed)
print(
f"Speech queue: waiting for current audio "
f"(~{remaining:.0f}s remaining, budget {self._shutdown_timeout:.0f}s)",
file=sys.stderr,
)
pending = self._queue.qsize() + (1 if self._current else 0)
print(
f"Speech queue: draining {pending} item(s) before shutdown "
f"(budget {self._shutdown_timeout:.0f}s)",
file=sys.stderr,
)
# Wait for the consumer to play out the queue. It runs normally
# (not stopped), so queued messages finish rather than dropping.
while (not self._queue.empty()) or self._current is not None:
if loop.time() >= deadline:
print(
f"Speech queue: drain budget expired with "
f"{self._queue.qsize()} queued; dropping the rest",
file=sys.stderr,
)
break
await asyncio.sleep(0.2)
else:
print("Speech queue: consumer idle, shutting down", file=sys.stderr)
print("Speech queue: queue drained", file=sys.stderr)
# Consumer polls _stopped every 0.5s between utterances, so it
# exits the while loop after finishing the current one.
# Now tell the consumer to exit and wait for it within whatever
# of the budget remains (keeps total stop() <= shutdown_timeout).
self._stopped = True
exit_wait = max(0.5, deadline - loop.time())
try:
await asyncio.wait_for(
asyncio.shield(self._consumer_task),
timeout=self._shutdown_timeout,
asyncio.shield(self._consumer_task), timeout=exit_wait
)
print("Speech queue: graceful shutdown complete", file=sys.stderr)
except asyncio.TimeoutError:
print(
f"Speech queue: shutdown timeout ({self._shutdown_timeout:.0f}s) "
f"expired, force-cancelling consumer",
file=sys.stderr,
)
self._consumer_task.cancel()
try:
await self._consumer_task
except asyncio.CancelledError:
pass
else:
self._stopped = True
except asyncio.CancelledError:
self._stopped = True
self._force_cancel_consumer()
raise
finally:
@ -330,7 +350,7 @@ class SpeechQueue:
await self._ducker.unduck()
except Exception:
pass # Non-fatal
self._drain_remaining()
self._drain_remaining() # drops anything that didn't fit the budget
def _resolve_outcome(self, utt: _Utterance, outcome: dict) -> None:
"""Set the future result and record the outcome."""
@ -516,11 +536,11 @@ class SpeechQueue:
arrive, so a slow synth just delays that one message — it never lets a
faster producer's audio interleave.
"""
if self._stopped:
if not self._accepting:
_unlink_quietly(announce_audio)
return None, {
"queued": False,
"error": "Queue is shut down",
"error": "Queue is shutting down (not accepting new speech)",
"engine": engine,
"voice": voice,
}