Give every subscription its own cursor, payload, and worker queue so an old blocked waiter cannot advance a replacement listener. Serialize lifecycle handling and guard cancellation.
Add a restart regression with an injected waiter; verified with Thread Sanitizer.
