streaming: prevent reader Close from deadlocking on a fatal read error

[raphael]

Summary

Follow-up to #77. That PR fixed Close deadlocking on a stalled subscriber; this fixes a sibling deadlock in the same area that I spotted while reviewing it.

On a fatal read error — handleReadError returns the error when Redis reports the stream key no longer exists — the reader's read loop calls Close() on itself:

if err := handleReadError(err, r.logger); err != nil {
    r.logger.Error(...)
    r.Close()
    return
}

But Close does r.wait.Wait(), which waits for the read goroutine to run its deferred cleanup() (wait.Done()). Since that goroutine is the one blocked inside Close, it deadlocks: the read goroutine and its Redis connection leak, and any external Close() blocks forever too. Same goroutine + connection leak as #77, different trigger.

Fix

Trigger the shutdown asynchronously (pulse.Go(r.logger, r.Close)) and let the read goroutine return so cleanup releases the wait group. Close is already idempotent via sync.Once (from #77), so this is safe alongside a concurrent external Close.

Only Reader is affected — Sink.read logs and continues on read errors rather than closing itself.

Note on the test

The fatal path is only reached when Redis returns "stream key no longer exists", which it does not emit deterministically for a blocking XREAD (deleting the key doesn't unblock the call, so a Redis-driven test would be timing-dependent). To get a deterministic regression test that exercises the real read() loop, I added a small package-level xreadFn seam that defaults to (*Reader).xread and is only overridden in tests to inject the fatal error. Happy to drop it if you'd rather not carry the seam.

Test plan

• go test -race ./streaming -run TestReaderCloseOnFatalReadError -count=10 — new regression test; hangs on pre-fix code (verified by reverting the fix), passes here.
• go test -race ./streaming -count=1