Add a comments channel, and add exponential backoff and jitter to reconnects - #1
Conversation
| // Any error occuring mid-event is considered non-graceful and will | ||
| // show up as some other error (most likely io.ErrUnexpectedEOF). | ||
| func (dec *Decoder) Decode() (Event, error) { | ||
| func (dec *Decoder) Decode() (Event, *string, error) { |
There was a problem hiding this comment.
Now that there are 3 return values, it would be useful in the comments to describe what the *string is that we're returning.
| } | ||
|
|
||
| // Integer power: compute a**b, from Knuth | ||
| func pow(a, b int) int { |
There was a problem hiding this comment.
:( I just looked up golang integer exponent, and see that the only built-in exponent function is for float64...
| return | ||
| } | ||
|
|
||
| func (stream *Stream) backoffWithJitter(attempts int) time.Duration { |
There was a problem hiding this comment.
Can we make a useful unit test for this method?
| } | ||
|
|
||
| // Integer power: compute a**b, from Knuth | ||
| func pow(a, b int) int { |
| "time" | ||
| ) | ||
|
|
||
| // This particular "benchmark" exists to spit out various jitter values. It's structured |
There was a problem hiding this comment.
tests don't necessarily suppress output. If you use: https://golang.org/pkg/testing/#T.Logf and then run the test with -v you will see the output.
There was a problem hiding this comment.
Sure, but I think this still makes more sense as a benchmark than a test, since it doesn't test anything.
There was a problem hiding this comment.
Either one is fine, just pointing out that output is possible in tests.
There was a problem hiding this comment.
test -v will also show any stderr/std out so you don't need T.Logf. It seems weird to have a loop in a benchmark that doesn't loop b.N times, but whatever.
Maybe a better test would be to get a bunch of jittered backoff values, average them, and assert that it is within some threshold of a mean?
Or, if the point is to show an example of the values it might spit out, maybe an Example would be more appropriate.
| } | ||
|
|
||
| // Integer power: compute a**b, from Knuth | ||
| func pow(a, b int) int { |
There was a problem hiding this comment.
Strictly speaking, b should be uint, as this is only correct for non-negative exponents
There was a problem hiding this comment.
That's kind of an infectious change.
| if strings.HasPrefix(line, ":") { | ||
| continue | ||
| comment := line[1:] | ||
| return nil, &comment, nil |
There was a problem hiding this comment.
It would probably be good to add a test in codec_test for comments in the stream.
…eplay (#63) ## 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: ```go 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.
3 participants
No description provided.