MessageStore::ClosedError can escape ensure_expire_fiber via an off-lock @closed check (latent under multi-threaded scheduler)

[viktorerlingsson]

Summary

ensure_expire_fiber checks !@closed outside @msg_store_lock, then
should_start_expire_fiber? touches the message store. If a concurrent queue
delete closes the store in between, @msg_store.first? raises
MessageStore::ClosedError, which escapes from two call sites. Safe on today's
single-threaded default context; latent under a multi-threaded one. Filing as
hardening.

Mechanism

• should_start_expire_fiber? calls @msg_store.first? under @msg_store_lock
  (queue.cr:277), which raises MessageStore::ClosedError on a closed store.
• ensure_expire_fiber (queue.cr:263-267) guards with !@closed, but that
  check is outside the lock (TOCTOU).
• Two call sites are unguarded against the resulting raise:
  • rm_consumer (queue.cr:1013), reached from basic.cancel / channel close.
    The connection read loop's catch-all logs "Unexpected error" and breaks the
    connection, so a healthy client connection is dropped.
  • the expire fiber's own ensure (queue.cr:195), where it would surface as a
    stack trace (cosmetic, the queue is tearing down anyway).

The publish path already rescues this exact error (queue.cr:636-640) and has
regression specs, which shows the hazard is known; these two sites were missed.

Why it is latent today

Queue#close sets queue @closed = true (queue.cr:512) before closing the
store (queue.cr:526-528), both entry points guard on @closed
(rm_consumer 989, ensure_expire_fiber 264), and MessageStore#close (sole
caller queue.cr:527) performs no fiber yield or evented IO inside
@msg_store_lock. On the single-threaded default context no cooperative
interleaving can reach first? on a closed store. The TOCTOU only becomes
reachable under true parallelism.

Suggested fix

Rescue MessageStore::ClosedError inside ensure_expire_fiber (a single choke
point that covers all callers), treating a concurrently-closed store as "nothing
to expire". The publish-path rescue then becomes redundant.

Notes

Part of a broader "audit shared state before enabling a multi-threaded default
context" theme (see also #2085). Not a user-facing bug on the current runtime.