fix: unblock Repository producers when a subscriber disconnects mid-replay

[kinyoklion]

Problem

eventsource is an SSE server. When a Repository is registered for a channel, Server.run() calls repo.Replay(channel, lastEventID) <-chan Event on each new subscription, wraps the returned channel as an eventBatch, and hands it to the per-connection HTTP handler goroutine, which drains it and writes each event to the client.

The handler detects client disconnect immediately via req.Context().Done() (both a clean FIN and an abrupt reset fire it). But on disconnect it does break ReadLoop and abandons the batch channel without draining it — and it cannot close it, because it holds the receiving end. Repository.Replay has no context/cancellation hook (two strings in, a bare channel out), so the producer goroutine inside Replay blocks forever on its next out <- event: a clean shutdown and a dirty one are indistinguishable to it, because "my send never completes" is all the API exposes.

Consequence: a mid-replay disconnect strands the Replay producer goroutine and its payload until the process exits. (Once events are flowing, disconnects are already caught on the write path via enc.Encode errors — only the initial producer to handler hop is blind.)

Fix — two levels

1. Background drain (no API change, covers every Repository). When the handler exits its read loop while still consuming a batch (readBatchCh non-nil), it drains the remaining events in a throwaway goroutine (go func(){ for range ch {} }()). The producer unblocks as fast as it can emit and releases in milliseconds. This alone fixes the forever-leak for any Repository, including third-party ones.

2. RepositoryWithContext (optional extension, clean propagation). A Repository may additionally implement:

ReplayWithContext(ctx context.Context, channel, id string) <-chan Event

The Server type-asserts the registered repository for it and, when present, calls it with the subscription's request context, so the producer can select { case out <- event: case <-ctx.Done(): return } and abort promptly and cleanly. Replay remains required (the new interface embeds Repository); repositories that implement only Replay are served exactly as before via the drain safety net.

Together: #1 guarantees no producer blocks past the handler's exit even for repos that never adopt the context; #2 lets adopters stop immediately on disconnect.

Backward compatibility

No exported symbol changed or was removed. Repository.Replay is unchanged; RepositoryWithContext is new and optional and is selected purely by type assertion. SliceRepository and any existing consumer continue to work unmodified.

Tests

server_replay_disconnect_test.go:

• Repro/regression: a producer blocked on a channel send is stranded on disconnect before the fix; after the fix it unblocks well under the deadline — verified for both the plain-Replay drain path and the ReplayWithContext path.
• Clean FIN and abrupt RST (SetLinger(0)) disconnects.
• Normal in-order delivery still works (plain and context repos) when the subscriber stays connected.
• ReplayWithContext observes ctx.Done() on disconnect, and is preferred over Replay when both are implemented.
• Empty batch (nil channel and already-closed channel) does not hang and normal publishing still works afterward.
• Multiple concurrent subscriptions all unblock.
• Server close during replay does not strand the producer.

Full suite passes under go test -race ./...; gofmt and go vet clean.