fix: prevent stream shutdown deadlock when Close races with event delivery

[aaron-zeisler]

Summary

Fixes a deadlock in the stream goroutine's shutdown path that can leave stream.Events permanently open. A consumer that drains stream.Events after calling Close (the documented shutdown pattern) then hangs forever.

Background

When the stream is torn down — via Close, Restart, or a read error — discardCurrentStream shuts down the current connection's decoder goroutine and drains its two internal channels so the goroutine can exit:

_ = r.Close()
r = nil
for range errs {   // drain errs first...
}
for range events { // ...then events
}

The decoder goroutine is a simple loop: decode, and either send the event on events, or on error send on errs and close both channels.

for {
    ev, err := dec.Decode()
    if err != nil {
        errs <- err
        close(errs)
        close(events)
        return
    }
    events <- ev
}

The drain assumes that after r.Close() the decoder is (or will shortly be) in the error path, so draining errs first lets it reach the close(errs); close(events). But if the decoder had already decoded an event and was blocked on events <- ev when we closed, r.Close() doesn't unblock it — it's parked on a send, not inside Decode. The drain waits on errs (nothing will ever arrive there), and the decoder waits for someone to receive its events send. Neither moves, so events/errs are never closed, stream.stream never reaches close(stream.Events), and any consumer ranging over stream.Events on shutdown blocks forever.

It's timing-dependent — it needs Close to land while an event is mid-delivery — so it shows up as a rare, intermittent hang. It surfaced downstream in ld-relay's auto-config stream (relay shutdown intermittently hanging until the process/test timed out).

Fix

Drain both internal channels concurrently. The decoder can be blocked sending on either one, and once unblocked it always closes both — so draining them together lets it terminate regardless of which send it was parked on.

_ = r.Close()
r = nil
var wg sync.WaitGroup
wg.Add(2)
go func() { defer wg.Done(); for range errs { } }()   // may be blocked here
go func() { defer wg.Done(); for range events { } }()  // ...or here
wg.Wait()

Verification

• New regression test TestStreamCloseDuringEventDeliveryDoesNotDeadlock floods the stream and closes mid-delivery, repeatedly. It hangs → fails on the old code (deadlock stack in discardCurrentStream) and passes with the fix.
• Full suite green under -race: go test -race ./....
• Validated end-to-end against ld-relay: with this fix pulled in via a local replace and no changes to ld-relay, its auto-config shutdown regression test passes; against stock v1.11.0 the same test hangs.