diff --git a/encoder.go b/encoder.go index a78df8a..b595717 100644 --- a/encoder.go +++ b/encoder.go @@ -67,7 +67,10 @@ func (enc *Encoder) Encode(ec eventOrComment) error { return fmt.Errorf("eventsource encode: %v", err) } default: - return fmt.Errorf("unexpected parameter to Encode: %v", ec) + // %T, not %v: an unexpected value must not have its contents -- which + // could include an event payload -- rendered into an error string that + // flows to WriteError consumers and logs. + return fmt.Errorf("unexpected parameter to Encode: %T", ec) } if enc.compressed { return enc.w.(*gzip.Writer).Flush() diff --git a/encoder_test.go b/encoder_test.go index c32e2dc..e5322ff 100644 --- a/encoder_test.go +++ b/encoder_test.go @@ -65,6 +65,16 @@ func TestEncoderComment(t *testing.T) { assert.Equal(t, ":hello\n", string(buf.Bytes())) } +func TestEncoderRejectsUnknownTypeWithoutRenderingValue(t *testing.T) { + buf := bytes.NewBuffer(nil) + err := NewEncoder(buf, false).Encode("do-not-disclose") + assert.Error(t, err) + // The error names the type only: its contents could be a payload, and the + // error flows to WriteError consumers and logs. + assert.NotContains(t, err.Error(), "do-not-disclose") + assert.Contains(t, err.Error(), "string") +} + func TestEncoderGzipCompression(t *testing.T) { uncompressedBuf, compressedBuf, expectedCompressedBuf := bytes.NewBuffer(nil), bytes.NewBuffer(nil), bytes.NewBuffer(nil) diff --git a/interface.go b/interface.go index 7a5a5ba..cb645a9 100644 --- a/interface.go +++ b/interface.go @@ -18,7 +18,9 @@ type Event interface { Id() string // The name of the event. Return empty string if not required. Event() string - // The payload of the event. + // The payload of the event. Repeated calls must return the same value: + // the server may read it more than once, for example when accounting for + // payload sizes in addition to encoding. Data() string } diff --git a/server.go b/server.go index 56a25e5..dcc4530 100644 --- a/server.go +++ b/server.go @@ -5,6 +5,7 @@ import ( "net/http" "strings" "sync" + "sync/atomic" "time" ) @@ -20,6 +21,12 @@ type subscription struct { // event channel, the unsubscribe path can still drain it and unblock the Repository's // producer. Accessed only from the Server.run() goroutine. batch <-chan Event + id uint64 + // closeReason records why the Server closed this subscription. It is written + // on the Server.run() goroutine before out is closed, and read by the handler + // only after it observes out being closed. That channel close is the + // happens-before edge that makes this plain field access race-free. + closeReason SubscriberRemovedReason } type eventOrComment interface{} @@ -51,12 +58,23 @@ type eventBatch struct { // Server manages any number of event-publishing channels and allows subscribers to consume them. // To use it within an HTTP server, create a handler for each channel with Handler(). type Server struct { - AllowCORS bool // Enable all handlers to be accessible from any origin - ReplayAll bool // Replay repository even if there's no Last-Event-Id specified - BufferSize int // How many messages do we let the client get behind before disconnecting - Gzip bool // Enable compression if client can accept it - MaxConnTime time.Duration // If non-zero, HTTP connections will be automatically closed after this time - Logger Logger // Logger is a logger that, when set, will be used for logging debug messages + AllowCORS bool // Enable all handlers to be accessible from any origin + ReplayAll bool // Replay repository even if there's no Last-Event-Id specified + BufferSize int // How many messages do we let the client get behind before disconnecting + Gzip bool // Enable compression if client can accept it + MaxConnTime time.Duration // If non-zero, HTTP connections will be automatically closed after this time + // Logger, when set, receives DEBUG lines for subscriber lifecycle events + // (add, remove, replay drain), a WARN line when a slow subscriber is + // dropped, and write errors. Lines identify connections by an opaque + // subscriber id and never include the channel name, because channel names + // may contain values (such as credentials) that must not appear in logs. + Logger Logger + // Trace, when set, receives callbacks at points in the Server's lifecycle. See + // ServerTrace for the concurrency contract that callbacks must satisfy. + // + // EXPERIMENTAL: this field and the ServerTrace API are subject to change or + // removal in any future release. See ServerTrace. + Trace *ServerTrace registrations chan *registration unregistrations chan *unregistration pub chan *outbound @@ -70,6 +88,7 @@ type Server struct { isClosed bool isClosedMutex sync.RWMutex jitter time.Duration + subCounter atomic.Uint64 } // NewServer creates a new Server instance. @@ -126,6 +145,169 @@ func (srv *Server) writeStreamHeaders(w http.ResponseWriter, req *http.Request) return useGzip } +// handlerState carries the state one Handler invocation shares between its +// read loop and its deferred teardown, so the teardown can live in methods +// rather than closures. exitReason, closedNormally, readBatchCh, the replay +// accounting, and delayedEvent are written by the read loop and read by the +// teardown; every access is on the single handler goroutine. +type handlerState struct { + srv *Server + sub *subscription + ctx context.Context + channel string + eventCh chan eventOrComment + flusher http.Flusher + enc *Encoder + connStart time.Time + + // exitReason is set at each point the read loop can exit, so that + // SubscriberRemoved can report why the connection ended. For a close + // initiated by the Server it is read from the subscription after the + // event channel is observed closed. + exitReason SubscriberRemovedReason + closedNormally bool + reportedAdded bool + + readBatchCh <-chan Event + replayStart time.Time + replayCount int + replayBytes int64 + + // delayedEvent is an event parked by a jitter-enabled server until its + // delay elapses; the teardown accounts for one still parked when the + // connection ends. + delayedEvent eventOrComment +} + +// unsubscribe tells the Server this handler is going away. After the Server has +// shut down nothing consumes unsubs (and its small buffer may already be full), +// so a handler that exits late must not block forever on the send. +func (hs *handlerState) unsubscribe() { + select { + case hs.srv.unsubs <- hs.sub: + case <-hs.srv.stopped: + } +} + +// reportExit is the exit-time reporting half of the teardown; it runs first +// and holds the code that invokes consumer callbacks. A panic here must not +// leak the subscription in the Server's map, strand a Repository producer, or +// leave a SubscriberAdded without its matching SubscriberRemoved -- which is +// why cleanup runs in a separate, earlier-registered defer. +func (hs *handlerState) reportExit() { + replayAborted := false + batchToDrain := hs.readBatchCh + if hs.readBatchCh != nil { + replayAborted = true + if hs.exitReason != ReasonWriteError { + // A batch that fully drained before the connection ended -- only + // its end-of-batch sentinel went unobserved, because the disconnect + // and the batch end raced in the read loop's select -- is a + // completed drain, not an aborted one. A write error is the + // exception: the failing write is what ended the drain, so that + // batch is aborted no matter what the channel state says. (In the + // rare case that a Server shutdown's drain is concurrently consuming + // this abandoned batch, the probe can misread the drain's close as + // completion; Aborted is documented as best-effort for exactly this + // race.) + select { + case _, ok := <-hs.readBatchCh: + if !ok { + replayAborted = false + batchToDrain = nil // fully drained; nothing to hand over + } + // A received event was never written to the connection; it + // belongs to the abandoned batch and is discarded with it. + default: + } + } + } + // Producer liveness before consumer-reachable code: an abandoned replay + // batch is handed to its background drain before the flush and the + // callbacks below, so slow or panicking consumer code can neither delay + // nor strand a Repository producer. + if drainAbandonedBatches(batchToDrain, hs.eventCh) { + // The Server closed the subscription while the handler was already exiting + // for a reason of its own. The receive that observed the closed channel + // orders the read of closeReason -- the same happens-before edge as the + // closedNormally path -- and the reason the Server recorded is the + // authoritative one: without this, a subscriber dropped for buffer overflow + // while parked in a slow write would be reported as client_closed or + // max_conn_time. + hs.closedNormally = true + } + if hs.readBatchCh != nil && !replayAborted { + // The completed batch still gets its end-of-batch flush, matching the + // read loop's sentinel path and the DrainDuration contract. + hs.flusher.Flush() + } + if hs.delayedEvent != nil { + // An event was still parked awaiting its jitter delay when the + // connection ended. Report it so that every event a subscriber + // was sent is accounted for as either sent or discarded. + hs.srv.traceEventDiscarded(hs.ctx, hs.channel, DiscardReasonConnectionEnded) + } + if hs.readBatchCh != nil { + // A ReplayStarted with no batch-end sentinel observed still gets + // its matching ReplayFinished, so the drained-so-far totals are + // not lost. + hs.srv.traceReplayFinished(hs.ctx, hs.sub, hs.replayCount, hs.replayBytes, + sinceOrZero(hs.replayStart), replayAborted) + } +} + +// cleanup is the half of the teardown that must never be skipped: resolving +// the exit reason, unsubscribing, and reporting SubscriberRemoved. It is +// registered before reportExit so that it runs last and still runs while a +// panic from reportExit is unwinding. Unsubscribing a subscription the Server +// never registered is a harmless no-op, so both defers safely precede the +// registration send. +func (hs *handlerState) cleanup() { + if hs.closedNormally { + // Reason precedence when a Server-initiated close races the handler's + // own exit: buffer_overflow outranks everything, because the + // SubscriberDropped callback that already fired promises a removal + // with the matching reason; a write error outranks the remaining + // Server reasons, because the handler has already reported that + // definitive local failure through WriteError; and any Server-recorded + // reason outranks the read loop's speculative client_closed and + // max_conn_time. + if hs.sub.closeReason == ReasonBufferOverflow || hs.exitReason != ReasonWriteError { + hs.exitReason = hs.sub.closeReason + } + } else { + hs.unsubscribe() // the server didn't tell us to close, so we must tell it that we're closing + } + if hs.reportedAdded { + hs.srv.traceSubscriberRemoved(hs.ctx, hs.sub, hs.exitReason, sinceOrZero(hs.connStart)) + } +} + +func (hs *handlerState) writeEventOrComment(ec eventOrComment) bool { + if err := hs.enc.Encode(ec); err != nil { + // No unsubscribe here: the deferred cleanup sends it moments later, and + // an early send would let run()'s unsubscription path start draining a + // mid-drain replay batch underneath the teardown's completion probe. + hs.exitReason = ReasonWriteError + hs.srv.traceWriteError(hs.ctx, hs.channel, err) + if hs.srv.Logger != nil { + hs.srv.Logger.Println(err) + } + return false // if this happens, we'll end the handler early because something's clearly broken + } + return true +} + +func (hs *handlerState) writeEventOrCommentAndFlush(ec eventOrComment) bool { + return hs.srv.writeTraced(hs.ctx, hs.channel, ec, func() bool { + if !hs.writeEventOrComment(ec) { + return false + } + hs.flusher.Flush() + return true + }) +} + // Handler creates a new HTTP handler for serving a specified channel. // // The channel does not have to have been previously registered with Register, but if it has been, the @@ -148,46 +330,64 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { maxConnTimeCh = t.C } + // ctx is the subscriber's request context. It is threaded into the trace + // callbacks that fire on this handler goroutine, so a consumer can correlate + // telemetry with the request span, and it is handed to the subscription for + // the benefit of a Repository that implements RepositoryWithContext. + ctx := req.Context() + eventCh := make(chan eventOrComment, srv.BufferSize) sub := &subscription{ channel: channel, lastEventID: req.Header.Get("Last-Event-ID"), out: eventCh, - ctx: req.Context(), - } - srv.subs <- sub - flusher := w.(http.Flusher) - flusher.Flush() - enc := NewEncoder(w, useGzip) - - // unsubscribe tells the Server this handler is going away. After the Server has - // shut down nothing consumes unsubs (and its small buffer may already be full), - // so a handler that exits late must not block forever on the send. - unsubscribe := func() { - select { - case srv.unsubs <- sub: - case <-srv.stopped: - } + ctx: ctx, } - writeEventOrComment := func(ec eventOrComment) bool { - if err := enc.Encode(ec); err != nil { - unsubscribe() - if srv.Logger != nil { - srv.Logger.Println(err) - } - return false // if this happens, we'll end the handler early because something's clearly broken - } - return true + // measuringReplay gates the replay accounting -- the clock reads and the + // extra per-event Data() call -- on something that will actually consume + // it: the ReplayFinished callback or the Logger's replay line. This + // mirrors shouldMeasureWrite's per-callback gating. + measuringReplay := (srv.Trace != nil && srv.Trace.ReplayFinished != nil) || srv.Logger != nil + + hs := &handlerState{ + srv: srv, + sub: sub, + ctx: ctx, + channel: channel, + eventCh: eventCh, + flusher: w.(http.Flusher), + // connStart is meaningful only when something is observing the + // connection; beginSubscription returns the zero time otherwise, + // which sinceOrZero maps to a zero duration. + connStart: srv.beginSubscription(sub), } - writeEventOrCommentAndFlush := func(ec eventOrComment) bool { - if !writeEventOrComment(ec) { - return false - } - flusher.Flush() - return true + defer hs.cleanup() + defer hs.reportExit() + + hs.flusher.Flush() + // reportedAdded is set before the callback so that a panic inside + // SubscriberAdded itself still produces the balancing SubscriberRemoved. + hs.reportedAdded = true + // SubscriberAdded fires before the subscription is registered with the + // Server, so no other callback -- in particular SubscriberDropped, which + // can fire on the dispatch goroutine as soon as the Server knows the + // subscription -- can precede it. The HTTP response was already started + // by writeStreamHeaders above. + srv.traceSubscriberAdded(ctx, sub) + + // If the Server closed while this handler was starting up, nothing will + // ever receive the registration; without the escape the handler would + // park on this send forever, leaking the goroutine and pinning the + // connection open. + select { + case srv.subs <- sub: + case <-srv.stopped: + hs.exitReason = ReasonServerClosed + return } + hs.enc = NewEncoder(w, useGzip) // The logic below works as follows: // - Normally, the handler is reading from eventCh. Server.run() accesses this channel through sub.out @@ -207,9 +407,7 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { // the Server to stop publishing events to it. var readMainCh <-chan eventOrComment = eventCh - var readBatchCh <-chan Event - closedNormally := false - closeNotify := req.Context().Done() + closeNotify := ctx.Done() // The handler consumes events in two different modes -- either as soon as // they arrive, or on some jitter-influenced delay. @@ -222,7 +420,6 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { // functionality. The ping stream sends identical "ping" events, so // discarding intermediate values is a safe operation. - var delayedEvent eventOrComment jitterStrategy := newDefaultJitter(0.5, 0) usingJitter := srv.jitter > 0 @@ -238,26 +435,31 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { for { select { case <-closeNotify: + hs.exitReason = ReasonClientClosed break ReadLoop case <-maxConnTimeCh: // if MaxConnTime was not set, this is a nil channel and has no effect on the select + hs.exitReason = ReasonMaxConnTime break ReadLoop case <-jitterTimer.Channel(): // If the jitter is 0, we may have an initial event that fired before // we could stop the timer. Or maybe the channel is being closed. // Whatever the reason, we can safely discard here. - if !usingJitter || delayedEvent == nil { + if !usingJitter || hs.delayedEvent == nil { continue } - ok := writeEventOrCommentAndFlush(delayedEvent) - delayedEvent = nil + // Cleared before the write so that a panicking EventSent cannot + // leave the event to also be reported as discarded by the + // teardown. + delayed := hs.delayedEvent + hs.delayedEvent = nil - if !ok { + if !hs.writeEventOrCommentAndFlush(delayed) { break ReadLoop } case ev, ok := <-readMainCh: if !ok { - closedNormally = true + hs.closedNormally = true break ReadLoop } @@ -265,24 +467,37 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { // If we receive an event batch, we are meant to switch to this as // our input source. But before we can do that, we need to process // any event that was pending processing. - if delayedEvent != nil { + if hs.delayedEvent != nil { jitterTimer.Stop() - ok := writeEventOrCommentAndFlush(delayedEvent) - delayedEvent = nil + // Cleared before the write, as in the timer case above. + delayed := hs.delayedEvent + hs.delayedEvent = nil - if !ok { + if !hs.writeEventOrCommentAndFlush(delayed) { break ReadLoop } } - readBatchCh = batch.events + hs.readBatchCh = batch.events readMainCh = nil + hs.replayCount = 0 + hs.replayBytes = 0 + // replayStart resets with the counts so that, if ReplayStarted + // panics below, the teardown's abort report cannot measure from + // a previous batch's start. + hs.replayStart = time.Time{} + srv.traceReplayStarted(ctx, channel) + // The drain clock starts after ReplayStarted returns, so the + // consumer's own callback cost is not billed to DrainDuration. + if measuringReplay { + hs.replayStart = time.Now() + } continue } // Write immediately if we aren't using the jitter functionality. if !usingJitter { - if !writeEventOrCommentAndFlush(ev) { + if !hs.writeEventOrCommentAndFlush(ev) { break ReadLoop } continue @@ -290,34 +505,40 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { // If we are using jitter and we have a pending event, then we don't // need to do anything. We can swallow this event. - if delayedEvent != nil { + if hs.delayedEvent != nil { + srv.traceEventDiscarded(ctx, channel, DiscardReasonJitterCoalesce) continue } - delayedEvent = ev + hs.delayedEvent = ev // Figure out the jitter and start the timer. Once this trigger, we // will write the event and clear the way for a new event to come in. delay := jitterStrategy.applyJitter(srv.jitter) jitterTimer.Reset(delay) - case ev, ok := <-readBatchCh: + case ev, ok := <-hs.readBatchCh: if !ok { // end of batch - flusher.Flush() - readBatchCh = nil + hs.flusher.Flush() + hs.readBatchCh = nil readMainCh = eventCh + // DrainDuration is measured after the flush above, so it accounts + // for the batch's single flush rather than excluding it. + srv.traceReplayFinished(ctx, sub, hs.replayCount, hs.replayBytes, sinceOrZero(hs.replayStart), false) continue } - if !writeEventOrComment(ev) { + // Replayed events are not flushed individually, so they report no + // EventSent; replay is observed at batch level via ReplayFinished. + if !hs.writeEventOrComment(ev) { break ReadLoop } + hs.replayCount++ + if measuringReplay { + hs.replayBytes += int64(len(ev.Data())) + } } } - drainAbandonedBatches(readBatchCh, eventCh) - if !closedNormally { - unsubscribe() // the server didn't tell us to close, so we must tell it that we're closing - } } } @@ -338,7 +559,11 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { // the handler's unsubscription; the sweep of the already-buffered values here additionally // covers the case where the Server has shut down and will never process it. Anything the // Server enqueues concurrently with this sweep is still handled by the unsubscription path. -func drainAbandonedBatches(current <-chan Event, eventCh <-chan eventOrComment) { +// +// The return value reports whether the sweep observed eventCh closed. A closed channel means +// the Server ended this subscription and recorded a closeReason before closing, so the caller +// can treat the exit as Server-initiated even if its read loop left for another reason first. +func drainAbandonedBatches(current <-chan Event, eventCh <-chan eventOrComment) bool { if current != nil { go drainReplayedEvents(current) } @@ -346,13 +571,13 @@ func drainAbandonedBatches(current <-chan Event, eventCh <-chan eventOrComment) select { case ev, ok := <-eventCh: if !ok { - return + return true } if batch, isBatch := ev.(eventBatch); isBatch { go drainReplayedEvents(batch.events) } default: - return + return false } } } @@ -436,6 +661,41 @@ func replay(repo Repository, sub *subscription) <-chan Event { return repo.Replay(sub.channel, sub.lastEventID) } +// enqueueReplay hands a newly registered subscription its replay batch, when +// the channel has a Repository and the request calls for a replay. Runs on the +// Server.run() goroutine; subs is run()'s subscription map. +func (srv *Server) enqueueReplay( + subs map[string]map[*subscription]struct{}, + repos map[string]Repository, + sub *subscription, +) { + if !srv.ReplayAll && len(sub.lastEventID) == 0 { + return + } + repo, ok := repos[sub.channel] + if !ok { + return + } + batchCh := replay(repo, sub) + if batchCh == nil { + return + } + if sub.send(eventBatch{events: batchCh}) { + // Remember the batch so that if the subscriber goes away before its + // handler dequeues it, the unsubs path can still drain it. + sub.batch = batchCh + } else { + // The send failed because the subscription's buffer was full (send + // closes the subscription in that case). The batch will never be + // consumed and its producer would otherwise block forever; drain it + // in the background. This is a drop like any other, so it reports + // SubscriberDropped just as trySend does. + delete(subs[sub.channel], sub) + srv.traceSubscriberDropped(sub) + go drainReplayedEvents(batchCh) + } +} + func (srv *Server) run() { defer close(srv.stopped) // All access to the subs and repos maps is done from the same goroutine, so modifications are safe. @@ -445,6 +705,7 @@ func (srv *Server) run() { if !sub.send(ec) { sub.close() delete(subs[sub.channel], sub) + srv.traceSubscriberDropped(sub) } } for { @@ -457,6 +718,7 @@ func (srv *Server) run() { delete(subs, unreg.channel) if unreg.forceDisconnect { for s := range previousSubs { + s.closeReason = ReasonUnregistered s.close() // Unlike the unsubscription and shutdown cases, no batch drain is needed // here: the server keeps running. A buffered channel delivers its queued @@ -469,12 +731,14 @@ func (srv *Server) run() { case sub := <-srv.unsubs: delete(subs[sub.channel], sub) if sub.batch != nil { - // The handler has exited. If it never dequeued the replay batch from its event - // channel -- or exited partway through consuming it -- the Repository's producer - // may still be blocked sending on it; drain it so the producer can finish. If the - // batch was fully consumed, the channel is already closed and this goroutine exits - // immediately. Draining concurrently with the handler's own exit-time drain is - // safe: both simply receive until the channel is closed. + // The handler has exited. If it never dequeued the replay batch from its + // event channel -- or exited partway through consuming it -- the + // Repository's producer may still be blocked sending on it; drain it so + // the producer can finish. Draining concurrently with the handler's own + // exit-time drain is safe: both simply receive until the channel is + // closed. This unsubscription arrives only after the handler's teardown + // has finished its exit-time reporting, so the drain cannot race the + // teardown's completion probe. go drainReplayedEvents(sub.batch) sub.batch = nil } @@ -497,47 +761,29 @@ func (srv *Server) run() { subs[sub.channel] = make(map[*subscription]struct{}) } subs[sub.channel][sub] = struct{}{} - if srv.ReplayAll || len(sub.lastEventID) > 0 { - repo, ok := repos[sub.channel] - if ok { - batchCh := replay(repo, sub) - if batchCh != nil { - if sub.send(eventBatch{events: batchCh}) { - // Remember the batch so that if the subscriber goes away before its - // handler dequeues it, the unsubs case below can still drain it. - sub.batch = batchCh - } else { - // The send failed because the subscription's buffer was full (send - // closes the subscription in that case). The batch will never be - // consumed and its producer would otherwise block forever; drain it - // in the background. - delete(subs[sub.channel], sub) - go drainReplayedEvents(batchCh) - } - } - } - } + srv.enqueueReplay(subs, repos, sub) case <-srv.quit: // We are about to stop processing unsubscriptions, so first handle any that are // already queued: their handlers have exited, and a handler that swept its event // channel before the replay batch was enqueued is relying on this path to drain it. // Subscriptions handled here are deliberately not removed from the map -- the loop - // below revisits them, but with batch already nil and close() being idempotent - // that revisit is a no-op. + // below revisits them, recording a close reason and closing their channel. That is + // harmless: their handlers have already exited and read neither again. DrainUnsubs: for { select { case sub := <-srv.unsubs: if sub.batch != nil { go drainReplayedEvents(sub.batch) - sub.batch = nil } + sub.batch = nil default: break DrainUnsubs } } for _, sub := range subs { for s := range sub { + s.closeReason = ReasonServerClosed s.close() // If the subscriber is already gone, its handler can no longer be relied on // to consume or drain a batch, and its unsubscription may never be seen @@ -584,6 +830,7 @@ func (s *subscription) send(e eventOrComment) bool { case s.out <- e: return true default: + s.closeReason = ReasonBufferOverflow s.close() return false } diff --git a/server_trace.go b/server_trace.go new file mode 100644 index 0000000..f937515 --- /dev/null +++ b/server_trace.go @@ -0,0 +1,502 @@ +package eventsource + +import ( + "context" + "runtime/debug" + "time" +) + +// ServerTrace is a set of optional callbacks that a Server invokes at points in +// its lifecycle. It is modeled on net/http/httptrace.ClientTrace: a struct of +// optional function fields, where new fields can be added without breaking +// existing users. Set it on the Server's Trace field before serving requests. +// +// Callbacks are invoked synchronously from internal Server goroutines, +// including the single goroutine that dispatches events to every channel. A +// callback that blocks stalls event delivery for all subscribers -- and a +// blocked SubscriberDropped wedges the dispatch goroutine itself, which also +// blocks Close and any handler still starting up -- so each callback must +// return promptly and must not call back into the Server. A nil ServerTrace, +// or a nil field within it, disables the corresponding hook. +// +// Callbacks must not panic. A panic in SubscriberDropped, which fires on the +// dispatch goroutine, is recovered -- and reported through the Logger when one +// is set -- so that it cannot take down the process; a panic in any other +// callback propagates on the subscriber's handler goroutine like a panic in +// any HTTP handler, ending that connection. +// +// Every callback that fires on a subscriber's connection handler goroutine +// receives that subscriber's request context as its first argument. The +// context is provided for telemetry correlation only -- for example attaching +// a child span or span event to the request span -- and must not be used to +// control cancellation. Being the request's own context, it also reaches +// everything a request context carries, such as net/http's ambient values and +// any consumer middleware values; do not use it to recover data the info +// structs deliberately omit. The context may already be canceled by the time a +// late callback such as SubscriberRemoved runs, because the client closing the +// connection is itself what ends the subscription; that is expected and does +// not prevent creating a span from it. +// +// EXPERIMENTAL: this type and every type it references -- the *Info structs, the +// SubscriberRemovedReason and EventDiscardedReason values, and the Server.Trace +// field itself -- are experimental. Callbacks, fields, and reason values may be +// added, renamed, or removed in any future release, and the points at which +// callbacks fire may change. It is for use by +// LaunchDarkly libraries ONLY. No guarantee is made about backwards +// compatibility or future support, and code outside LaunchDarkly's own libraries +// should not depend on it. +type ServerTrace struct { + // SubscriberAdded is called once for each new subscriber, after the HTTP + // response has been started and before the subscription is registered with + // the Server, so it precedes every other callback for that subscriber. + SubscriberAdded func(context.Context, SubscriberAddedInfo) + + // SubscriberRemoved is called when a subscriber's connection has ended, for + // any reason. It is called once for each SubscriberAdded. + SubscriberRemoved func(context.Context, SubscriberRemovedInfo) + + // SubscriberDropped is called when a subscriber is forcibly disconnected + // because it could not keep up and fell more than BufferSize events behind. + // SubscriberRemoved is also called for the same subscriber with reason + // ReasonBufferOverflow, so a consumer that only needs a disconnect signal + // can implement SubscriberRemoved alone. (The one exception is a drop that + // lands after the handler has already finished removing the connection for + // another cause; the removal that paired with it then carries that earlier + // reason.) + // + // Unlike the other callbacks, this one takes no context: it fires on the + // Server's dispatch goroutine, which has no association with any single + // subscriber's request, so no request context is available here. + SubscriberDropped func(SubscriberDroppedInfo) + + // EventSent is called after an event has been written to a subscriber's + // connection and the write has been flushed, so a client may observe the + // event before this callback runs. The info reports the event type, payload + // size, and the measured write duration, but never the payload itself. + // + // EventSent fires only for events that are written and flushed + // individually. Replayed events are encoded in bulk and not reported + // individually here -- even though gzip compression or a full response + // buffer may deliver some of their bytes before the batch ends -- so use + // ReplayFinished for batch-level replay telemetry. + EventSent func(context.Context, EventSentInfo) + + // CommentSent is called after a comment has been written to a subscriber's + // connection. + CommentSent func(context.Context, CommentSentInfo) + + // EventDiscarded is called when an event is dropped before delivery. This + // currently happens only on jitter-enabled servers, which coalesce events + // that arrive while an earlier event is still pending and discard a pending + // event whose connection ends before its delay elapses. The discarded item + // may be an event or a comment; both are reported here. + EventDiscarded func(context.Context, EventDiscardedInfo) + + // WriteError is called when encoding or writing an event to a subscriber's + // connection fails. The subscriber is removed after this callback. + WriteError func(context.Context, WriteErrorInfo) + + // ReplayStarted is called when a subscriber begins draining a batch of + // replayed events provided by a Repository. Every ReplayStarted is followed + // by exactly one ReplayFinished for the same batch. + ReplayStarted func(context.Context, ReplayInfo) + + // ReplayFinished is called when a subscriber has finished draining a batch + // of replayed events, or when the connection ends while the batch is still + // draining; the info's Aborted field distinguishes the two. + ReplayFinished func(context.Context, ReplayFinishedInfo) +} + +// SubscriberRemovedReason describes why a subscriber's connection ended. The +// set of values may change over time; consult the EXPERIMENTAL note on +// ServerTrace. +type SubscriberRemovedReason string + +// Reasons reported through ServerTrace.SubscriberRemoved. +const ( + // ReasonClientClosed means the client closed the connection. + ReasonClientClosed SubscriberRemovedReason = "client_closed" + // ReasonMaxConnTime means the connection reached the Server's MaxConnTime. + ReasonMaxConnTime SubscriberRemovedReason = "max_conn_time" + // ReasonWriteError means writing to the connection failed. + ReasonWriteError SubscriberRemovedReason = "write_error" + // ReasonServerClosed means the Server was closed. + ReasonServerClosed SubscriberRemovedReason = "server_closed" + // ReasonUnregistered means the channel was unregistered with forceDisconnect. + ReasonUnregistered SubscriberRemovedReason = "unregistered" + // ReasonBufferOverflow means the subscriber fell too far behind and was dropped. + ReasonBufferOverflow SubscriberRemovedReason = "buffer_overflow" +) + +// EventDiscardedReason describes why an event was discarded before delivery. +// The set of values may change over time; consult the EXPERIMENTAL note on +// ServerTrace. +type EventDiscardedReason string + +// Reasons reported through ServerTrace.EventDiscarded. +const ( + // DiscardReasonJitterCoalesce means the event was coalesced away by a + // jitter-enabled server while an earlier event was still pending. + DiscardReasonJitterCoalesce EventDiscardedReason = "jitter_coalesce" + // DiscardReasonConnectionEnded means the event was still waiting out its + // jitter delay when the connection ended. + DiscardReasonConnectionEnded EventDiscardedReason = "connection_ended" +) + +// SubscriberAddedInfo is passed to ServerTrace.SubscriberAdded. +type SubscriberAddedInfo struct { + // Channel is the channel the subscriber connected to. + Channel string + // SubscriberID is an opaque per-connection identifier that can be used to + // correlate this callback with SubscriberRemoved and SubscriberDropped. + // Identifiers are unique within one Server only; distinct Servers issue + // overlapping ids. + SubscriberID uint64 + // HasLastEventID reports whether the subscriber supplied a Last-Event-ID. + HasLastEventID bool +} + +// SubscriberRemovedInfo is passed to ServerTrace.SubscriberRemoved. +type SubscriberRemovedInfo struct { + // Channel is the channel the subscriber was connected to. + Channel string + // SubscriberID matches the value from the corresponding SubscriberAddedInfo. + SubscriberID uint64 + // Reason describes why the connection ended. When a Server-initiated close + // races the handler's own exit, the reasons rank as follows: + // buffer_overflow outranks everything, since the SubscriberDropped + // callback that already fired promises a removal with the matching + // reason; write_error outranks the remaining Server reasons, since the + // handler has already reported that failure through WriteError; and any + // Server-recorded reason outranks client_closed and max_conn_time. Reason + // is empty only if the connection ended because a consumer callback + // panicked on the handler goroutine, which leaves no recorded cause. + Reason SubscriberRemovedReason + // ConnDuration is how long the connection was open. + ConnDuration time.Duration +} + +// SubscriberDroppedInfo is passed to ServerTrace.SubscriberDropped. +type SubscriberDroppedInfo struct { + // Channel is the channel the subscriber was connected to. + Channel string + // SubscriberID matches the value from the corresponding SubscriberAddedInfo. + SubscriberID uint64 + // BufferSize is the number of events the subscriber was allowed to fall + // behind before being dropped. + BufferSize int +} + +// EventSentInfo is passed to ServerTrace.EventSent. +type EventSentInfo struct { + // Channel is the channel the event was sent on. + Channel string + // EventType is the event's type name, which may be empty. + EventType string + // DataSize is the size in bytes of the event's data payload only. It does + // not include the id/event field values or SSE framing, and it measures + // the payload before any compression, so it does not correspond exactly to + // the bytes on the wire. + DataSize int + // WriteDuration is the wall time spent encoding and flushing this event to + // the connection. It is measured only when the EventSent callback is set, + // so a consumer can back-date a span start to WriteDuration before this + // callback fires. + WriteDuration time.Duration +} + +// CommentSentInfo is passed to ServerTrace.CommentSent. +type CommentSentInfo struct { + // Channel is the channel the comment was sent on. + Channel string + // WriteDuration is the wall time spent encoding and flushing this comment to + // the connection. It is measured only when the CommentSent callback is set, + // so a consumer can back-date a span start to WriteDuration before this + // callback fires. + WriteDuration time.Duration +} + +// EventDiscardedInfo is passed to ServerTrace.EventDiscarded. +type EventDiscardedInfo struct { + // Channel is the channel the event would have been sent on. + Channel string + // Reason describes why the event was discarded. + Reason EventDiscardedReason +} + +// WriteErrorInfo is passed to ServerTrace.WriteError. +type WriteErrorInfo struct { + // Channel is the channel the write was attempted on. + Channel string + // Err is the error returned by encoding or writing the event. + Err error +} + +// ReplayInfo is passed to ServerTrace.ReplayStarted. +type ReplayInfo struct { + // Channel is the channel being replayed. + Channel string +} + +// ReplayFinishedInfo is passed to ServerTrace.ReplayFinished. +type ReplayFinishedInfo struct { + // Channel is the channel that was replayed. + Channel string + // EventCount is the number of replayed events written to the connection. + // For a completed batch they have also been flushed, though a flush cannot + // report failure, so this does not guarantee the client received them; for + // an aborted batch they may never have been flushed at all (see Aborted). + EventCount int + // TotalDataSize is the summed size in bytes of the data payloads of all + // replayed events in the batch. Like EventSentInfo.DataSize it counts the + // data payload only -- no id/event field values, no SSE framing, measured + // before any compression -- so it does not correspond exactly to the bytes + // on the wire. + TotalDataSize int64 + // DrainDuration is how long draining the batch took, as observed by the + // connection handler, including the single flush at the end of the batch. + // The measurement starts after the ReplayStarted callback returns, so that + // callback's own cost is not included. A consumer can back-date a span + // start to DrainDuration before the ReplayFinished callback fires. + DrainDuration time.Duration + // Aborted reports that the connection ended -- client disconnect, + // MaxConnTime, or a write error -- before the end of the batch was + // observed. (The handler checks the batch once more at exit, so a batch + // that fully drained just as the connection ended is still reported + // completed.) EventCount and TotalDataSize then reflect only the events + // written before the exit, those events may never have been flushed to the + // connection, and DrainDuration excludes the end-of-batch flush because + // none occurred. + // + // Aborted is best-effort. If Server.Close races a subscriber that is + // itself disconnecting mid-batch, the shutdown's drain of the abandoned + // batch can make the batch appear completed even though the drained events + // never reached the client; treat Aborted as a strong signal, not an exact + // accounting. + // + // Aborted can also only describe what the Server observed. A Repository + // that itself ends its batch early -- for example one that stops producing + // when the request context is canceled -- yields a normal, non-aborted + // ReplayFinished for whatever it produced. + Aborted bool +} + +// observing reports whether anything consumes what the handler measures, so that +// timings are computed only when a Trace callback or the Logger will use them. +func (srv *Server) observing() bool { + return srv.Trace != nil || srv.Logger != nil +} + +// beginSubscription records tracing bookkeeping for a new subscription and +// returns the time the connection was established. The returned time is the +// zero value when nothing is observing the connection; derive durations from +// it with sinceOrZero, which maps that sentinel to a zero duration. +func (srv *Server) beginSubscription(sub *subscription) time.Time { + if !srv.observing() { + return time.Time{} + } + // The id is assigned whenever anything is observing -- a Logger-only + // server needs it too, since log lines identify connections by id. + sub.id = srv.subCounter.Add(1) + return time.Now() +} + +// sinceOrZero is time.Since for measurements that may not have been started: +// it reports zero for the zero time instead of a nonsense duration measured +// from the epoch. +func sinceOrZero(start time.Time) time.Duration { + if start.IsZero() { + return 0 + } + return time.Since(start) +} + +// The trace helpers below also feed srv.Logger. Log lines identify +// connections by their opaque subscriber id and never include the channel +// name: consumers may use sensitive values (such as credentials) as channel +// names, and logs must not contain them. + +func (srv *Server) traceSubscriberAdded(ctx context.Context, sub *subscription) { + hasLastEventID := sub.lastEventID != "" + if t := srv.Trace; t != nil && t.SubscriberAdded != nil { + t.SubscriberAdded(ctx, SubscriberAddedInfo{ + Channel: sub.channel, + SubscriberID: sub.id, + HasLastEventID: hasLastEventID, + }) + } + if srv.Logger != nil { + srv.Logger.Printf("[DEBUG] eventsource: subscriber added (id=%d, has_last_event_id=%t)", + sub.id, hasLastEventID) + } +} + +func (srv *Server) traceSubscriberRemoved( + ctx context.Context, + sub *subscription, + reason SubscriberRemovedReason, + dur time.Duration, +) { + if t := srv.Trace; t != nil && t.SubscriberRemoved != nil { + t.SubscriberRemoved(ctx, SubscriberRemovedInfo{ + Channel: sub.channel, + SubscriberID: sub.id, + Reason: reason, + ConnDuration: dur, + }) + } + if srv.Logger != nil { + srv.Logger.Printf("[DEBUG] eventsource: subscriber removed (id=%d, reason=%s, duration=%s)", + sub.id, reason, dur) + } +} + +// traceSubscriberDropped runs on the Server.run() dispatch goroutine, where an +// unrecovered panic would take down the process rather than a single +// connection, so it contains panics from the consumer-supplied callback and +// logger instead of letting them unwind run(). +func (srv *Server) traceSubscriberDropped(sub *subscription) { + defer func() { + if r := recover(); r != nil && srv.Logger != nil { + // The report itself runs consumer code -- the Logger, which may be + // exactly what just panicked -- and a second panic here would unwind + // run() after all, so it is swallowed. + defer func() { _ = recover() }() + // %T, not %v: the panic value is consumer-controlled and may carry + // data, such as the info struct with its channel, that must never be + // logged. The stack identifies the panic site. + srv.Logger.Printf("[ERROR] eventsource: panic in ServerTrace callback (%T)\n%s", r, debug.Stack()) + } + }() + if t := srv.Trace; t != nil && t.SubscriberDropped != nil { + t.SubscriberDropped(SubscriberDroppedInfo{ + Channel: sub.channel, + SubscriberID: sub.id, + BufferSize: srv.BufferSize, + }) + } + if srv.Logger != nil { + srv.Logger.Printf("[WARN] eventsource: dropped subscriber (id=%d, fell behind buffer size %d)", + sub.id, srv.BufferSize) + } +} + +// shouldMeasureWrite reports whether the trace callback that would receive this +// event or comment is set, so the handler measures write duration only when a +// consumer will actually read it. This preserves the zero-overhead nil path: +// no clock is read when the corresponding callback is nil. +func (srv *Server) shouldMeasureWrite(ec eventOrComment) bool { + t := srv.Trace + if t == nil { + return false + } + switch ec.(type) { + case Event: + return t.EventSent != nil + case comment: + return t.CommentSent != nil + default: + return false + } +} + +// writeTraced performs a write via write, then reports it through the EventSent or +// CommentSent callback along with the wall time the write took. The clock is read +// only when that callback is set, so the nil path stays allocation- and +// syscall-free. It returns whether the write succeeded. +func (srv *Server) writeTraced( + ctx context.Context, + channel string, + ec eventOrComment, + write func() bool, +) bool { + measure := srv.shouldMeasureWrite(ec) + var writeStart time.Time + if measure { + writeStart = time.Now() + } + if !write() { + return false + } + var writeDuration time.Duration + if measure { + writeDuration = time.Since(writeStart) + } + srv.traceSentEventOrComment(ctx, channel, ec, writeDuration) + return true +} + +func (srv *Server) traceSentEventOrComment( + ctx context.Context, + channel string, + ec eventOrComment, + writeDuration time.Duration, +) { + t := srv.Trace + if t == nil { + return + } + switch item := ec.(type) { + case Event: + if t.EventSent != nil { + t.EventSent(ctx, EventSentInfo{ + Channel: channel, + EventType: item.Event(), + DataSize: len(item.Data()), + WriteDuration: writeDuration, + }) + } + case comment: + if t.CommentSent != nil { + t.CommentSent(ctx, CommentSentInfo{Channel: channel, WriteDuration: writeDuration}) + } + } +} + +func (srv *Server) traceEventDiscarded(ctx context.Context, channel string, reason EventDiscardedReason) { + if t := srv.Trace; t != nil && t.EventDiscarded != nil { + t.EventDiscarded(ctx, EventDiscardedInfo{ + Channel: channel, + Reason: reason, + }) + } +} + +func (srv *Server) traceWriteError(ctx context.Context, channel string, err error) { + if t := srv.Trace; t != nil && t.WriteError != nil { + t.WriteError(ctx, WriteErrorInfo{Channel: channel, Err: err}) + } +} + +func (srv *Server) traceReplayStarted(ctx context.Context, channel string) { + if t := srv.Trace; t != nil && t.ReplayStarted != nil { + t.ReplayStarted(ctx, ReplayInfo{Channel: channel}) + } +} + +func (srv *Server) traceReplayFinished( + ctx context.Context, + sub *subscription, + count int, + totalBytes int64, + dur time.Duration, + aborted bool, +) { + if t := srv.Trace; t != nil && t.ReplayFinished != nil { + t.ReplayFinished(ctx, ReplayFinishedInfo{ + Channel: sub.channel, + EventCount: count, + TotalDataSize: totalBytes, + DrainDuration: dur, + Aborted: aborted, + }) + } + if srv.Logger != nil { + // The duration logs in Go's own formatting (like the removed line) + // rather than integer milliseconds, which rounded the common + // sub-millisecond drain down to 0. + srv.Logger.Printf( + "[DEBUG] eventsource: replay drained (id=%d, event_count=%d, total_bytes=%d, duration=%s, aborted=%t)", + sub.id, count, totalBytes, dur, aborted) + } +} diff --git a/server_trace_test.go b/server_trace_test.go new file mode 100644 index 0000000..62816b9 --- /dev/null +++ b/server_trace_test.go @@ -0,0 +1,1786 @@ +package eventsource + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "log" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// traceRecorder captures ServerTrace callbacks from any goroutine. Every field +// is guarded by mu because callbacks fire from both the handler goroutines and +// the Server's dispatch goroutine. The contexts observed by the handler- +// goroutine callbacks are captured so tests can assert they derive from the +// subscriber's request. +type traceRecorder struct { + mu sync.Mutex + added []SubscriberAddedInfo + removed []SubscriberRemovedInfo + dropped []SubscriberDroppedInfo + eventsSent []EventSentInfo + commentsSent []CommentSentInfo + discarded []EventDiscardedInfo + writeErrors []WriteErrorInfo + replayStarted []ReplayInfo + replayFinished []ReplayFinishedInfo + addedCtx []context.Context + eventSentCtx []context.Context +} + +func (r *traceRecorder) trace() *ServerTrace { + return &ServerTrace{ + SubscriberAdded: func(ctx context.Context, i SubscriberAddedInfo) { + r.mu.Lock() + defer r.mu.Unlock() + r.added = append(r.added, i) + r.addedCtx = append(r.addedCtx, ctx) + }, + SubscriberRemoved: func(_ context.Context, i SubscriberRemovedInfo) { + r.mu.Lock() + defer r.mu.Unlock() + r.removed = append(r.removed, i) + }, + SubscriberDropped: func(i SubscriberDroppedInfo) { + r.mu.Lock() + defer r.mu.Unlock() + r.dropped = append(r.dropped, i) + }, + EventSent: func(ctx context.Context, i EventSentInfo) { + r.mu.Lock() + defer r.mu.Unlock() + r.eventsSent = append(r.eventsSent, i) + r.eventSentCtx = append(r.eventSentCtx, ctx) + }, + CommentSent: func(_ context.Context, i CommentSentInfo) { + r.mu.Lock() + defer r.mu.Unlock() + r.commentsSent = append(r.commentsSent, i) + }, + EventDiscarded: func(_ context.Context, i EventDiscardedInfo) { + r.mu.Lock() + defer r.mu.Unlock() + r.discarded = append(r.discarded, i) + }, + WriteError: func(_ context.Context, i WriteErrorInfo) { + r.mu.Lock() + defer r.mu.Unlock() + r.writeErrors = append(r.writeErrors, i) + }, + ReplayStarted: func(_ context.Context, i ReplayInfo) { + r.mu.Lock() + defer r.mu.Unlock() + r.replayStarted = append(r.replayStarted, i) + }, + ReplayFinished: func(_ context.Context, i ReplayFinishedInfo) { + r.mu.Lock() + defer r.mu.Unlock() + r.replayFinished = append(r.replayFinished, i) + }, + } +} + +func (r *traceRecorder) snapshotAddedCtx() []context.Context { + r.mu.Lock() + defer r.mu.Unlock() + return append([]context.Context(nil), r.addedCtx...) +} + +func (r *traceRecorder) snapshotEventSentCtx() []context.Context { + r.mu.Lock() + defer r.mu.Unlock() + return append([]context.Context(nil), r.eventSentCtx...) +} + +func (r *traceRecorder) snapshotAdded() []SubscriberAddedInfo { + r.mu.Lock() + defer r.mu.Unlock() + return append([]SubscriberAddedInfo(nil), r.added...) +} + +func (r *traceRecorder) snapshotRemoved() []SubscriberRemovedInfo { + r.mu.Lock() + defer r.mu.Unlock() + return append([]SubscriberRemovedInfo(nil), r.removed...) +} + +func (r *traceRecorder) snapshotDropped() []SubscriberDroppedInfo { + r.mu.Lock() + defer r.mu.Unlock() + return append([]SubscriberDroppedInfo(nil), r.dropped...) +} + +func (r *traceRecorder) snapshotEventsSent() []EventSentInfo { + r.mu.Lock() + defer r.mu.Unlock() + return append([]EventSentInfo(nil), r.eventsSent...) +} + +func (r *traceRecorder) snapshotCommentsSent() []CommentSentInfo { + r.mu.Lock() + defer r.mu.Unlock() + return append([]CommentSentInfo(nil), r.commentsSent...) +} + +func (r *traceRecorder) snapshotDiscarded() []EventDiscardedInfo { + r.mu.Lock() + defer r.mu.Unlock() + return append([]EventDiscardedInfo(nil), r.discarded...) +} + +func (r *traceRecorder) snapshotWriteErrors() []WriteErrorInfo { + r.mu.Lock() + defer r.mu.Unlock() + return append([]WriteErrorInfo(nil), r.writeErrors...) +} + +func (r *traceRecorder) snapshotReplayStarted() []ReplayInfo { + r.mu.Lock() + defer r.mu.Unlock() + return append([]ReplayInfo(nil), r.replayStarted...) +} + +func (r *traceRecorder) snapshotReplayFinished() []ReplayFinishedInfo { + r.mu.Lock() + defer r.mu.Unlock() + return append([]ReplayFinishedInfo(nil), r.replayFinished...) +} + +// traceTestWriter is a minimal http.ResponseWriter and http.Flusher that a +// Server handler can be driven with directly. Its Write can be gated (to +// simulate a subscriber that is not reading), made to fail (to exercise the +// write-error path), or delayed (to make a measured write duration exceed the +// platform's clock granularity), and its Flush can be gated independently (to +// stall a handler in its initial flush). gate, flushGate, writeErr, and delay +// are configured before the handler starts and are not mutated afterwards, +// apart from closing the gates to release Write and Flush. +type traceTestWriter struct { + mu sync.Mutex + hdr http.Header + buf bytes.Buffer + gate chan struct{} + flushGate chan struct{} + writeErr error + delay time.Duration + flushDelay time.Duration +} + +func (w *traceTestWriter) Header() http.Header { + if w.hdr == nil { + w.hdr = http.Header{} + } + return w.hdr +} + +func (w *traceTestWriter) WriteHeader(int) {} + +func (w *traceTestWriter) Flush() { + if w.flushGate != nil { + <-w.flushGate + } + if w.flushDelay > 0 { + time.Sleep(w.flushDelay) + } +} + +func (w *traceTestWriter) Write(p []byte) (int, error) { + if w.gate != nil { + <-w.gate + } + if w.writeErr != nil { + return 0, w.writeErr + } + if w.delay > 0 { + time.Sleep(w.delay) + } + w.mu.Lock() + defer w.mu.Unlock() + return w.buf.Write(p) +} + +func (w *traceTestWriter) bytesWritten() int { + w.mu.Lock() + defer w.mu.Unlock() + return w.buf.Len() +} + +func (w *traceTestWriter) contains(s string) bool { + w.mu.Lock() + defer w.mu.Unlock() + return bytes.Contains(w.buf.Bytes(), []byte(s)) +} + +// captureLogger records formatted log lines so that the level-prefixed messages +// can be asserted. +type captureLogger struct { + mu sync.Mutex + lines []string +} + +func (l *captureLogger) Println(args ...interface{}) { + l.mu.Lock() + defer l.mu.Unlock() + l.lines = append(l.lines, fmt.Sprintln(args...)) +} + +func (l *captureLogger) Printf(format string, args ...interface{}) { + l.mu.Lock() + defer l.mu.Unlock() + l.lines = append(l.lines, fmt.Sprintf(format, args...)) +} + +func (l *captureLogger) containing(substr string) bool { + l.mu.Lock() + defer l.mu.Unlock() + for _, line := range l.lines { + if bytes.Contains([]byte(line), []byte(substr)) { + return true + } + } + return false +} + +// traceTestCtxKey is used to seed a value on a handler's request context so a +// test can confirm the value reaches the ServerTrace callbacks. +type traceTestCtxKey struct{} + +// startTraceHandler runs a Server handler on its own goroutine, driving it with +// w and a cancelable request. The request context carries a traceTestCtxKey +// value so tests can assert the callback contexts derive from it. The returned +// cancel simulates the client closing the connection; done is closed once the +// handler returns. +func startTraceHandler( + server *Server, + channel string, + w http.ResponseWriter, + headers map[string]string, +) (context.CancelFunc, <-chan struct{}) { + req := httptest.NewRequest(http.MethodGet, "/", nil) + for k, v := range headers { + req.Header.Set(k, v) + } + ctx := context.WithValue(req.Context(), traceTestCtxKey{}, "request-scoped-value") + ctx, cancel := context.WithCancel(ctx) + req = req.WithContext(ctx) + done := make(chan struct{}) + go func() { + defer close(done) + server.Handler(channel)(w, req) + }() + return cancel, done +} + +func waitClosed(t *testing.T, done <-chan struct{}) { + t.Helper() + select { + case <-done: + case <-time.After(2 * time.Second): + require.Fail(t, "timed out waiting for handler to exit") + } +} + +func TestServerTraceSubscriberAdded(t *testing.T) { + t.Run("without Last-Event-ID", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + defer func() { + cancel() + waitClosed(t, done) + }() + + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + added := rec.snapshotAdded() + assert.Equal(t, "test", added[0].Channel) + assert.False(t, added[0].HasLastEventID) + assert.NotZero(t, added[0].SubscriberID) + }) + + t.Run("with Last-Event-ID", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + defer server.Close() + + headers := map[string]string{"Last-Event-ID": "abc"} + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, headers) + defer func() { + cancel() + waitClosed(t, done) + }() + + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + assert.True(t, rec.snapshotAdded()[0].HasLastEventID) + }) +} + +func TestServerTraceSubscriberRemovedReasons(t *testing.T) { + t.Run("client_closed", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + cancel() + waitClosed(t, done) + + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonClientClosed, removed[0].Reason) + assert.Equal(t, rec.snapshotAdded()[0].SubscriberID, removed[0].SubscriberID) + }) + + t.Run("max_conn_time", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.MaxConnTime = 100 * time.Millisecond + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + defer cancel() + + waitClosed(t, done) + + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonMaxConnTime, removed[0].Reason) + }) + + t.Run("server_closed", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + defer cancel() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + server.Close() + waitClosed(t, done) + + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonServerClosed, removed[0].Reason) + }) + + t.Run("unregistered", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + defer cancel() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + server.Unregister("test", true) + waitClosed(t, done) + + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonUnregistered, removed[0].Reason) + }) + + t.Run("buffer_overflow", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.BufferSize = 1 + defer server.Close() + + w := &traceTestWriter{gate: make(chan struct{})} + cancel, done := startTraceHandler(server, "test", w, nil) + defer cancel() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + // The handler is stalled inside Write, so it stops draining its channel. + // Publishing past BufferSize forces the dispatch goroutine to drop it. + for _, data := range []string{"a", "b", "c"} { + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: data}) + } + + require.Eventually(t, func() bool { + return len(rec.snapshotDropped()) >= 1 + }, 2*time.Second, 10*time.Millisecond) + + close(w.gate) // let the handler run to completion + waitClosed(t, done) + + dropped := rec.snapshotDropped() + require.GreaterOrEqual(t, len(dropped), 1) + assert.Equal(t, "test", dropped[0].Channel) + assert.Equal(t, 1, dropped[0].BufferSize) + assert.Equal(t, rec.snapshotAdded()[0].SubscriberID, dropped[0].SubscriberID) + + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonBufferOverflow, removed[0].Reason) + assert.Equal(t, dropped[0].SubscriberID, removed[0].SubscriberID) + }) + + t.Run("write_error", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + defer server.Close() + + w := &traceTestWriter{writeErr: errors.New("boom")} + cancel, done := startTraceHandler(server, "test", w, nil) + defer cancel() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + server.Publish([]string{"test"}, &publication{data: "x"}) + waitClosed(t, done) + + writeErrors := rec.snapshotWriteErrors() + require.Len(t, writeErrors, 1) + assert.Equal(t, "test", writeErrors[0].Channel) + assert.Error(t, writeErrors[0].Err) + + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonWriteError, removed[0].Reason) + }) + + // Race-free pin of the exit-time sweep: the handler is parked mid-batch + // (its main event channel is invisible to the read loop), so the overflow + // drop cannot be observed by any select case; only the teardown's sweep of + // the closed event channel can recover the buffer_overflow reason. + t.Run("buffer_overflow detected by the exit sweep alone", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.BufferSize = 1 + server.ReplayAll = true + repo := gatedReplayRepository{ + before: 1, + after: 0, + gate: make(chan struct{}), + done: make(chan struct{}), + } + server.Register("test", repo) + defer close(repo.gate) + defer server.Close() + + w := &traceTestWriter{} + cancel, done := startTraceHandler(server, "test", w, nil) + require.Eventually(t, func() bool { + return len(rec.snapshotReplayStarted()) == 1 && w.bytesWritten() > 0 + }, 2*time.Second, 10*time.Millisecond) + + // The handler is waiting for the next batch event (the producer is + // parked at its gate). Overflow the invisible main channel to force + // the drop, then disconnect: closeNotify is the only ready case. + for _, data := range []string{"a", "b", "c"} { + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: data}) + } + require.Eventually(t, func() bool { + return len(rec.snapshotDropped()) == 1 + }, 2*time.Second, 10*time.Millisecond) + cancel() + waitClosed(t, done) + + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonBufferOverflow, removed[0].Reason) + }) + + // SubscriberDropped is a promise of a matching buffer_overflow removal, so + // buffer_overflow outranks even a write error that the drop itself caused. + t.Run("buffer_overflow wins over write_error", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.BufferSize = 1 + defer server.Close() + + w := &traceTestWriter{gate: make(chan struct{}), writeErr: errors.New("boom")} + cancel, done := startTraceHandler(server, "test", w, nil) + defer cancel() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + // The handler parks in the write that will fail; the publishes behind + // it overflow the buffer, so the Server drops the subscriber and fires + // SubscriberDropped; then the parked write completes with its error. + for _, data := range []string{"a", "b", "c"} { + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: data}) + } + require.Eventually(t, func() bool { + return len(rec.snapshotDropped()) == 1 + }, 2*time.Second, 10*time.Millisecond) + close(w.gate) + waitClosed(t, done) + + require.Len(t, rec.snapshotWriteErrors(), 1) + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonBufferOverflow, removed[0].Reason, + "SubscriberDropped promises a buffer_overflow removal") + }) + + // A write error the handler has already reported through WriteError is + // definitive: a Server close that lands while the handler is parked in the + // failing write must not mask it. + t.Run("write_error wins over server_closed", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + + w := &traceTestWriter{gate: make(chan struct{}), writeErr: errors.New("boom")} + cancel, done := startTraceHandler(server, "test", w, nil) + defer cancel() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + // The handler parks in the write that will fail; the Server closes + // (recording server_closed and closing the event channel); then the + // write completes with its error. + server.Publish([]string{"test"}, &publication{data: "x"}) + server.Close() + close(w.gate) + waitClosed(t, done) + + require.Len(t, rec.snapshotWriteErrors(), 1) + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonWriteError, removed[0].Reason) + }) + + // A subscriber the Server drops for buffer overflow must be removed with + // ReasonBufferOverflow even when the handler's read loop leaves through a + // different exit first. The dropped subscriber is stalled in a slow write + // with a full buffer behind it, so the handler cannot have observed the + // closed channel before the competing exit condition fires. + t.Run("buffer_overflow wins over max_conn_time", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.BufferSize = 1 + server.MaxConnTime = 20 * time.Millisecond + defer server.Close() + + w := &traceTestWriter{gate: make(chan struct{})} + cancel, done := startTraceHandler(server, "test", w, nil) + defer cancel() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + for _, data := range []string{"a", "b", "c"} { + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: data}) + } + require.Eventually(t, func() bool { + return len(rec.snapshotDropped()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + // Let MaxConnTime elapse while the handler is still stalled, then + // release it so it unwinds with both exit conditions pending. + time.Sleep(50 * time.Millisecond) + close(w.gate) + waitClosed(t, done) + + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonBufferOverflow, removed[0].Reason) + assert.Equal(t, rec.snapshotDropped()[0].SubscriberID, removed[0].SubscriberID) + }) + + t.Run("buffer_overflow wins over client_closed", func(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.BufferSize = 1 + defer server.Close() + + w := &traceTestWriter{gate: make(chan struct{})} + cancel, done := startTraceHandler(server, "test", w, nil) + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + for _, data := range []string{"a", "b", "c"} { + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: data}) + } + require.Eventually(t, func() bool { + return len(rec.snapshotDropped()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + // The client hangs up before the stalled handler has had any chance to + // observe the closed channel, then the handler unwinds. + cancel() + close(w.gate) + waitClosed(t, done) + + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonBufferOverflow, removed[0].Reason) + assert.Equal(t, rec.snapshotDropped()[0].SubscriberID, removed[0].SubscriberID) + }) +} + +// slowDoneContext delays the first Done() call. A Server handler calls Done() +// exactly once, between registering its subscription and entering its read +// loop, so the delay deterministically pins the handler in the window where +// the dispatch goroutine attempts the replay batch enqueue. +type slowDoneContext struct { + context.Context + delay time.Duration + once sync.Once +} + +func (c *slowDoneContext) Done() <-chan struct{} { + c.once.Do(func() { time.Sleep(c.delay) }) + return c.Context.Done() +} + +// With an unbuffered subscription (BufferSize zero), enqueueing the replay +// batch marker itself overflows the subscription and drops it. That drop must +// fire SubscriberDropped exactly like the publish-path overflow drop. +func TestServerTraceSubscriberDroppedOnReplayEnqueueOverflow(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.BufferSize = 0 + server.ReplayAll = true + server.Register("test", &testServerRepository{}) + defer server.Close() + + req := httptest.NewRequest(http.MethodGet, "/", nil) + innerCtx, cancel := context.WithCancel(req.Context()) + defer cancel() + req = req.WithContext(&slowDoneContext{Context: innerCtx, delay: 200 * time.Millisecond}) + done := make(chan struct{}) + go func() { + defer close(done) + server.Handler("test")(&traceTestWriter{}, req) + }() + waitClosed(t, done) + + dropped := rec.snapshotDropped() + require.Len(t, dropped, 1, "the replay-enqueue overflow drop must fire SubscriberDropped") + assert.Equal(t, 0, dropped[0].BufferSize) + + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonBufferOverflow, removed[0].Reason) + assert.Equal(t, dropped[0].SubscriberID, removed[0].SubscriberID) +} + +// A Server that closes while a handler is between its response start and its +// registration send must not leave that handler parked forever on the send; +// the connection ends and the already-fired SubscriberAdded gets its pairing +// SubscriberRemoved. +func TestServerTraceHandlerExitsWhenServerClosesDuringStartup(t *testing.T) { + rec := &traceRecorder{} + gate := make(chan struct{}) + trace := rec.trace() + inner := trace.SubscriberAdded + trace.SubscriberAdded = func(ctx context.Context, info SubscriberAddedInfo) { + inner(ctx, info) + <-gate + } + server := NewServer() + server.Trace = trace + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + defer cancel() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + // The handler is parked inside SubscriberAdded, before the registration + // send. Close the Server, then let it proceed to the send. + server.Close() + close(gate) + waitClosed(t, done) + + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonServerClosed, removed[0].Reason) + assert.Equal(t, rec.snapshotAdded()[0].SubscriberID, removed[0].SubscriberID) +} + +// SubscriberAdded must precede every other callback for its subscriber, even +// when the handler stalls in its initial flush while the dispatch goroutine is +// already delivering (and potentially dropping) events for the subscription. +func TestServerTraceSubscriberAddedPrecedesDropped(t *testing.T) { + var mu sync.Mutex + var order []string + record := func(name string) { + mu.Lock() + defer mu.Unlock() + order = append(order, name) + } + + server := NewServer() + server.BufferSize = 1 + server.Trace = &ServerTrace{ + SubscriberAdded: func(context.Context, SubscriberAddedInfo) { record("added") }, + SubscriberDropped: func(SubscriberDroppedInfo) { record("dropped") }, + } + defer server.Close() + + w := &traceTestWriter{gate: make(chan struct{}), flushGate: make(chan struct{})} + cancel, done := startTraceHandler(server, "test", w, nil) + defer cancel() + + // Publishing past BufferSize while the handler is parked in its initial + // flush must not produce any callback that precedes SubscriberAdded: were + // the subscription registered before SubscriberAdded fired, these would + // overflow it and fire SubscriberDropped first. + for _, data := range []string{"early-a", "early-b", "early-c"} { + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: data}) + } + close(w.flushGate) + + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(order) >= 1 + }, 2*time.Second, 10*time.Millisecond) + + // Now stall the first event write and overrun the buffer to force a drop. + for _, data := range []string{"a", "b", "c"} { + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: data}) + } + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(order) >= 2 + }, 2*time.Second, 10*time.Millisecond) + + close(w.gate) + waitClosed(t, done) + + mu.Lock() + defer mu.Unlock() + require.NotEmpty(t, order) + assert.Equal(t, "added", order[0], + "SubscriberAdded must be the first callback for a subscription, got %v", order) + assert.Contains(t, order, "dropped") +} + +// A consumer callback that panics on the handler goroutine ends that +// connection (net/http recovers the panic), but must not unbalance the trace: +// the subscription still unregisters and SubscriberRemoved still fires. +func TestServerTracePanicInHandlerCallbackStillRemovesSubscriber(t *testing.T) { + rec := &traceRecorder{} + trace := rec.trace() + trace.EventSent = func(context.Context, EventSentInfo) { + panic("consumer bug") + } + server := NewServer() + server.Trace = trace + defer server.Close() + + httpServer := httptest.NewUnstartedServer(server.Handler("test")) + // Silence net/http's stderr report of the recovered panic. + httpServer.Config.ErrorLog = log.New(io.Discard, "", 0) + httpServer.Start() + defer httpServer.Close() + + resp, err := http.Get(httpServer.URL) + require.NoError(t, err) + defer func() { _ = resp.Body.Close() }() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + server.Publish([]string{"test"}, &publication{data: "boom"}) + + require.Eventually(t, func() bool { + return len(rec.snapshotRemoved()) == 1 + }, 2*time.Second, 10*time.Millisecond) + removed := rec.snapshotRemoved()[0] + assert.Equal(t, rec.snapshotAdded()[0].SubscriberID, removed.SubscriberID) + assert.Equal(t, SubscriberRemovedReason(""), removed.Reason, + "a panic exit has no recorded cause, so Reason is empty") +} + +// gatedReplayRepository sends `before` events, parks until gate is closed, +// then sends `after` more and closes the batch. done is closed once the +// producer has sent everything, so a test can prove the producer was not +// stranded by an exiting handler. +type gatedReplayRepository struct { + before, after int + gate chan struct{} + done chan struct{} +} + +func (r gatedReplayRepository) Replay(channel, id string) chan Event { + out := make(chan Event) + go func() { + defer close(out) + defer close(r.done) + for i := 0; i < r.before; i++ { + out <- &publication{id: fmt.Sprintf("id-%d", i), data: "0123456789"} + } + <-r.gate + for i := 0; i < r.after; i++ { + out <- &publication{id: fmt.Sprintf("id-after-%d", i), data: "0123456789"} + } + }() + return out +} + +// A panic in a callback that the exit-time teardown itself invokes (the +// aborted ReplayFinished here) must not skip the rest of the teardown: the +// Repository producer must still be released and SubscriberRemoved must still +// fire. +func TestServerTracePanicInAbortedReplayReportStillTearsDown(t *testing.T) { + rec := &traceRecorder{} + trace := rec.trace() + trace.ReplayFinished = func(context.Context, ReplayFinishedInfo) { + panic("consumer bug") + } + repo := gatedReplayRepository{ + before: 3, + after: 2, + gate: make(chan struct{}), + done: make(chan struct{}), + } + server := NewServer() + server.Trace = trace + server.ReplayAll = true + server.Register("test", repo) + defer server.Close() + + httpServer := httptest.NewUnstartedServer(server.Handler("test")) + // Silence net/http's stderr report of the recovered panic. + httpServer.Config.ErrorLog = log.New(io.Discard, "", 0) + httpServer.Start() + defer httpServer.Close() + + reqCtx, cancel := context.WithCancel(context.Background()) + req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, httpServer.URL, nil) + require.NoError(t, err) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer func() { _ = resp.Body.Close() }() + + // The producer is parked at its gate with the batch mid-drain; the client + // hangs up, so the teardown fires the aborted ReplayFinished, which panics. + require.Eventually(t, func() bool { + return len(rec.snapshotReplayStarted()) == 1 + }, 2*time.Second, 10*time.Millisecond) + cancel() + + require.Eventually(t, func() bool { + return len(rec.snapshotRemoved()) == 1 + }, 2*time.Second, 10*time.Millisecond, "SubscriberRemoved must fire despite the panicking abort report") + + // The abandoned batch was handed to its drain before the callback panicked, + // so releasing the gate lets the producer finish instead of stranding it. + close(repo.gate) + select { + case <-repo.done: + case <-time.After(2 * time.Second): + require.Fail(t, "the replay producer was stranded by the panicking callback") + } +} + +// The same containment for the other teardown-invoked callback: a panic in +// the connection_ended EventDiscarded must not skip SubscriberRemoved. +func TestServerTracePanicInConnectionEndedDiscardStillTearsDown(t *testing.T) { + rec := &traceRecorder{} + var coalesced atomic.Int64 + trace := rec.trace() + trace.EventDiscarded = func(_ context.Context, info EventDiscardedInfo) { + if info.Reason == DiscardReasonJitterCoalesce { + coalesced.Add(1) + return + } + panic("consumer bug") + } + server := NewServerWithJitter(time.Hour) + server.Trace = trace + defer server.Close() + + httpServer := httptest.NewUnstartedServer(server.Handler("test")) + httpServer.Config.ErrorLog = log.New(io.Discard, "", 0) + httpServer.Start() + defer httpServer.Close() + + reqCtx, cancel := context.WithCancel(context.Background()) + req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, httpServer.URL, nil) + require.NoError(t, err) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer func() { _ = resp.Body.Close() }() + + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + // The coalesce discard of the second event proves the first is parked, so + // hanging up now makes the teardown's connection_ended discard panic. + for _, data := range []string{"parked", "coalesced"} { + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: data}) + } + require.Eventually(t, func() bool { + return coalesced.Load() == 1 + }, 2*time.Second, 10*time.Millisecond) + cancel() + + require.Eventually(t, func() bool { + return len(rec.snapshotRemoved()) == 1 + }, 2*time.Second, 10*time.Millisecond, "SubscriberRemoved must fire despite the panicking discard report") +} + +func TestSinceOrZero(t *testing.T) { + assert.Equal(t, time.Duration(0), sinceOrZero(time.Time{}), + "an unmeasured start must map to a zero duration, not an epoch-based one") + d := sinceOrZero(time.Now().Add(-time.Second)) + assert.True(t, d >= time.Second && d < time.Minute, "got %s", d) +} + +// A panic inside the SubscriberAdded callback itself must still produce the +// balancing SubscriberRemoved: the pairing flag is set before the callback. +func TestServerTracePanicInSubscriberAddedStillRemovesSubscriber(t *testing.T) { + rec := &traceRecorder{} + trace := rec.trace() + inner := trace.SubscriberAdded + trace.SubscriberAdded = func(ctx context.Context, info SubscriberAddedInfo) { + inner(ctx, info) + panic("consumer bug") + } + server := NewServer() + server.Trace = trace + defer server.Close() + + httpServer := httptest.NewUnstartedServer(server.Handler("test")) + httpServer.Config.ErrorLog = log.New(io.Discard, "", 0) + httpServer.Start() + defer httpServer.Close() + + resp, err := http.Get(httpServer.URL) + require.NoError(t, err) + defer func() { _ = resp.Body.Close() }() + + require.Eventually(t, func() bool { + return len(rec.snapshotRemoved()) == 1 + }, 2*time.Second, 10*time.Millisecond) + removed := rec.snapshotRemoved()[0] + assert.Equal(t, rec.snapshotAdded()[0].SubscriberID, removed.SubscriberID) + assert.Equal(t, SubscriberRemovedReason(""), removed.Reason, + "a panic exit has no recorded cause") +} + +// An event that was written when its jitter delay elapsed must not ALSO be +// reported as discarded just because the EventSent callback for it panicked: +// the parking slot is cleared before the write, not after. +func TestServerTracePanicInEventSentDoesNotDoubleReportJitterEvent(t *testing.T) { + rec := &traceRecorder{} + trace := rec.trace() + trace.EventSent = func(context.Context, EventSentInfo) { + panic("consumer bug") + } + server := NewServerWithJitter(20 * time.Millisecond) + server.Trace = trace + defer server.Close() + + httpServer := httptest.NewUnstartedServer(server.Handler("test")) + httpServer.Config.ErrorLog = log.New(io.Discard, "", 0) + httpServer.Start() + defer httpServer.Close() + + resp, err := http.Get(httpServer.URL) + require.NoError(t, err) + defer func() { _ = resp.Body.Close() }() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + // The event parks, its delay elapses, the write succeeds, and EventSent + // panics on the way out. + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: "x"}) + require.Eventually(t, func() bool { + return len(rec.snapshotRemoved()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + assert.Empty(t, rec.snapshotDiscarded(), + "the written event must not also be reported as discarded") +} + +// A panic in the SubscriberDropped callback fires on the Server's dispatch +// goroutine, where an unrecovered panic would kill the whole process rather +// than one connection. The Server must contain it and keep dispatching. +func TestServerTracePanicInSubscriberDroppedIsRecovered(t *testing.T) { + // The channel stands in for a credential; a callback that panics with the + // info struct it was handed must not put it in the log. + const channel = "sdk-key-do-not-log" + + rec := &traceRecorder{} + logger := &captureLogger{} + trace := rec.trace() + trace.SubscriberDropped = func(info SubscriberDroppedInfo) { + panic(info) + } + server := NewServer() + server.Trace = trace + server.Logger = logger + server.BufferSize = 1 + defer server.Close() + + w := &traceTestWriter{gate: make(chan struct{})} + cancel, done := startTraceHandler(server, channel, w, nil) + defer cancel() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + // The third publish overruns the stalled subscriber's buffer, so its own + // acknowledgment proves the dispatch goroutine survived the panic that its + // drop triggered. + for _, data := range []string{"a", "b", "c"} { + <-server.PublishWithAcknowledgment([]string{channel}, &publication{data: data}) + } + <-server.PublishWithAcknowledgment([]string{"other"}, &publication{data: "x"}) + + assert.True(t, logger.containing("panic in ServerTrace callback (eventsource.SubscriberDroppedInfo)"), + "the recovered panic should be reported by type") + assert.False(t, logger.containing(channel), + "the panic value is consumer-controlled and must not be rendered into the log") + + close(w.gate) + waitClosed(t, done) + + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonBufferOverflow, removed[0].Reason) +} + +// warnPanickyLogger panics on the drop path's WARN line and on the recover's +// own ERROR report, modelling a broken log sink; other lines pass through. +type warnPanickyLogger struct { + captureLogger +} + +func (l *warnPanickyLogger) Printf(format string, args ...interface{}) { + if strings.HasPrefix(format, "[WARN]") || strings.HasPrefix(format, "[ERROR]") { + panic("log sink broken") + } + l.captureLogger.Printf(format, args...) +} + +// A panicking Logger on the drop path runs on the dispatch goroutine just like +// the SubscriberDropped callback; the recover's own report re-enters that same +// Logger, so both panics must be contained or the process dies. +func TestServerTracePanicInLoggerOnDropPathIsRecovered(t *testing.T) { + logger := &warnPanickyLogger{} + server := NewServer() + server.Logger = logger + server.BufferSize = 1 + defer server.Close() + + w := &traceTestWriter{gate: make(chan struct{})} + cancel, done := startTraceHandler(server, "test", w, nil) + defer cancel() + require.Eventually(t, func() bool { + return logger.containing("subscriber added") + }, 2*time.Second, 10*time.Millisecond) + + for _, data := range []string{"a", "b", "c"} { + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: data}) + } + // The dispatch goroutine survived both the WARN panic and the ERROR-report + // panic: it still acknowledges publishes. + <-server.PublishWithAcknowledgment([]string{"other"}, &publication{data: "x"}) + + close(w.gate) + waitClosed(t, done) + assert.True(t, logger.containing("subscriber removed"), + "the handler still completed its teardown") +} + +func TestServerTraceEventAndCommentSent(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + defer server.Close() + + // Each write is delayed so the measured duration clears the platform's + // monotonic clock granularity. On Windows that clock advances on the system + // timer tick (~15.6ms), so an undelayed in-memory write can correctly measure + // as zero, and "greater than zero" is not a sound assertion there. + // + // The asserted floor is well below the delay on purpose: the same granularity + // that rounds a fast write down to zero can round a 30ms write down to a + // single tick, so asserting the full delay would just move the flake. + // The upper bound catches a measurement that was never started: a duration + // computed from the zero time.Time reads as centuries, not milliseconds. + const ( + writeDelay = 30 * time.Millisecond + minWriteDuration = 5 * time.Millisecond + maxWriteDuration = time.Minute + ) + cancel, done := startTraceHandler(server, "test", &traceTestWriter{delay: writeDelay}, nil) + defer func() { + cancel() + waitClosed(t, done) + }() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{event: "put", data: "hello"}) + server.PublishComment([]string{"test"}, "keepalive") + + require.Eventually(t, func() bool { + return len(rec.snapshotEventsSent()) == 1 && len(rec.snapshotCommentsSent()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + eventsSent := rec.snapshotEventsSent() + assert.Equal(t, "test", eventsSent[0].Channel) + assert.Equal(t, "put", eventsSent[0].EventType) + assert.Equal(t, len("hello"), eventsSent[0].DataSize) + assert.True(t, eventsSent[0].WriteDuration >= minWriteDuration, + "expected write duration >= %s, got %s", minWriteDuration, eventsSent[0].WriteDuration) + assert.True(t, eventsSent[0].WriteDuration < maxWriteDuration, + "expected a measured write duration, got %s", eventsSent[0].WriteDuration) + + commentsSent := rec.snapshotCommentsSent() + assert.Equal(t, "test", commentsSent[0].Channel) + assert.True(t, commentsSent[0].WriteDuration >= minWriteDuration, + "expected comment write duration >= %s, got %s", minWriteDuration, commentsSent[0].WriteDuration) + assert.True(t, commentsSent[0].WriteDuration < maxWriteDuration, + "expected a measured comment write duration, got %s", commentsSent[0].WriteDuration) +} + +func TestServerTraceContextFromRequest(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + defer func() { + cancel() + waitClosed(t, done) + }() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + addedCtx := rec.snapshotAddedCtx() + require.Len(t, addedCtx, 1) + assert.Equal(t, "request-scoped-value", addedCtx[0].Value(traceTestCtxKey{})) + + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{event: "put", data: "hello"}) + require.Eventually(t, func() bool { + return len(rec.snapshotEventSentCtx()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + eventSentCtx := rec.snapshotEventSentCtx() + require.Len(t, eventSentCtx, 1) + assert.Equal(t, "request-scoped-value", eventSentCtx[0].Value(traceTestCtxKey{})) +} + +func TestServerTraceEventDiscardedByJitter(t *testing.T) { + rec := &traceRecorder{} + // A large jitter keeps the first event pending long enough that the events + // published behind it are coalesced away. + server := NewServerWithJitter(2 * time.Second) + server.Trace = rec.trace() + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + defer func() { + cancel() + waitClosed(t, done) + }() + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + for _, data := range []string{"first", "second", "third"} { + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: data}) + } + + require.Eventually(t, func() bool { + return len(rec.snapshotDiscarded()) == 2 + }, 2*time.Second, 10*time.Millisecond) + + for _, info := range rec.snapshotDiscarded() { + assert.Equal(t, "test", info.Channel) + assert.Equal(t, DiscardReasonJitterCoalesce, info.Reason) + } +} + +// An event still parked awaiting its jitter delay when the connection ends +// must be reported as discarded, so that every event delivered to a +// subscription is accounted for as either sent or discarded. +func TestServerTraceEventDiscardedOnConnectionEnd(t *testing.T) { + rec := &traceRecorder{} + server := NewServerWithJitter(time.Hour) + server.Trace = rec.trace() + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + require.Eventually(t, func() bool { + return len(rec.snapshotAdded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + // The first event parks; the coalesce discard for the second proves the + // first is parked before the client hangs up. + for _, data := range []string{"parked", "coalesced"} { + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: data}) + } + require.Eventually(t, func() bool { + return len(rec.snapshotDiscarded()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + cancel() + waitClosed(t, done) + + discarded := rec.snapshotDiscarded() + require.Len(t, discarded, 2) + assert.Equal(t, DiscardReasonJitterCoalesce, discarded[0].Reason) + assert.Equal(t, DiscardReasonConnectionEnded, discarded[1].Reason) + assert.Empty(t, rec.snapshotEventsSent(), + "neither event was written, so both must be accounted for as discards") +} + +func TestServerTraceReplay(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.ReplayAll = true + server.Register("test", &testServerRepository{}) + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + defer func() { + cancel() + waitClosed(t, done) + }() + + require.Eventually(t, func() bool { + return len(rec.snapshotReplayStarted()) == 1 && len(rec.snapshotReplayFinished()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + assert.Equal(t, "test", rec.snapshotReplayStarted()[0].Channel) + + finished := rec.snapshotReplayFinished() + assert.Equal(t, "test", finished[0].Channel) + assert.Equal(t, 1, finished[0].EventCount) + // testServerRepository replays a single event whose data is "example". + assert.Equal(t, int64(len("example")), finished[0].TotalDataSize) + // The upper bound catches a drain duration computed from an unmeasured + // start, which reads as centuries. + assert.True(t, finished[0].DrainDuration >= 0 && finished[0].DrainDuration < time.Minute, + "expected a measured drain duration, got %s", finished[0].DrainDuration) + + // Replayed events are flushed once per batch, so EventSent must not fire for + // them; replay is reported only through ReplayStarted/ReplayFinished. + assert.Empty(t, rec.snapshotEventsSent()) + assert.False(t, finished[0].Aborted) +} + +// listReplayRepository replays one event per element of data, so a test can +// control the exact payload sizes in a batch. +type listReplayRepository struct { + data []string +} + +func (r listReplayRepository) Replay(channel, id string) chan Event { + out := make(chan Event, len(r.data)) + for i, d := range r.data { + out <- &publication{id: fmt.Sprintf("id-%d", i), data: d} + } + close(out) + return out +} + +// TotalDataSize must be the sum over the whole batch; distinct sizes make an +// implementation that reports only one event's size (first, last, or count +// times either) come out wrong. +func TestServerTraceReplayTotalDataSizeSumsDistinctSizes(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.ReplayAll = true + data := []string{"a", "bcd", "efghi"} + server.Register("test", listReplayRepository{data: data}) + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + defer func() { + cancel() + waitClosed(t, done) + }() + + require.Eventually(t, func() bool { + return len(rec.snapshotReplayFinished()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + finished := rec.snapshotReplayFinished()[0] + assert.Equal(t, len(data), finished.EventCount) + assert.Equal(t, int64(len("a")+len("bcd")+len("efghi")), finished.TotalDataSize) + assert.False(t, finished.Aborted) +} + +// DrainDuration is documented to include the single flush at the end of the +// batch; a flush that takes real time must show up in the reported duration. +func TestServerTraceReplayDrainDurationIncludesBatchFlush(t *testing.T) { + const flushDelay = 50 * time.Millisecond + + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.ReplayAll = true + server.Register("test", &testServerRepository{}) + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{flushDelay: flushDelay}, nil) + defer func() { + cancel() + waitClosed(t, done) + }() + + require.Eventually(t, func() bool { + return len(rec.snapshotReplayFinished()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + finished := rec.snapshotReplayFinished()[0] + assert.True(t, finished.DrainDuration >= flushDelay, + "DrainDuration must include the end-of-batch flush: expected >= %s, got %s", + flushDelay, finished.DrainDuration) + assert.True(t, finished.DrainDuration < time.Minute, + "expected a measured drain duration, got %s", finished.DrainDuration) +} + +// writeEntryGateWriter signals when its first Write is entered and then blocks +// until released, so a test can act at the precise moment the handler is +// mid-write. +type writeEntryGateWriter struct { + traceTestWriter + entered chan struct{} + once sync.Once + release chan struct{} +} + +func (w *writeEntryGateWriter) Write(p []byte) (int, error) { + w.once.Do(func() { close(w.entered) }) + <-w.release + return w.traceTestWriter.Write(p) +} + +// A batch that fully drained before the client disconnected must be reported +// as completed even when the disconnect and the end-of-batch sentinel race in +// the read loop's select; mislabeling it aborted would inflate the aborted +// dimension for the common "client takes the payload and drops" pattern. +func TestServerTraceReplayCompletedDespiteDisconnectRace(t *testing.T) { + // After the write is released, the read loop's select chooses between the + // disconnect and the already-closed batch channel by coin flip, so a + // single run only exercises the exit-time reclassification about half the + // time; iterating makes missing it vanishingly unlikely, and the asserted + // outcome is identical on both paths. + for i := 0; i < 20; i++ { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.ReplayAll = true + // listReplayRepository closes its batch channel up front, so once the + // single event is consumed only the sentinel remains. + server.Register("test", listReplayRepository{data: []string{"solitary"}}) + + w := &writeEntryGateWriter{entered: make(chan struct{}), release: make(chan struct{})} + cancel, done := startTraceHandler(server, "test", w, nil) + + // The handler is mid-write of the last event; hanging up now puts the + // disconnect and the already-closed batch channel in the same select. + <-w.entered + cancel() + close(w.release) + waitClosed(t, done) + + finished := rec.snapshotReplayFinished() + require.Len(t, finished, 1) + assert.False(t, finished[0].Aborted, + "the batch fully drained; losing the select race to the disconnect must not mislabel it aborted") + assert.Equal(t, 1, finished[0].EventCount) + assert.Equal(t, int64(len("solitary")), finished[0].TotalDataSize) + server.Close() + } +} + +// A replay drain that ends in a write error is aborted by definition, even +// when the batch channel is already closed and empty (the failing write was +// the batch's last event): the events the report counts never all reached the +// connection. +func TestServerTraceReplayWriteErrorIsAlwaysAborted(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.ReplayAll = true + server.Register("test", listReplayRepository{data: []string{"solitary"}}) + defer server.Close() + + w := &traceTestWriter{writeErr: errors.New("boom")} + cancel, done := startTraceHandler(server, "test", w, nil) + defer cancel() + waitClosed(t, done) + + require.Len(t, rec.snapshotWriteErrors(), 1) + finished := rec.snapshotReplayFinished() + require.Len(t, finished, 1) + assert.True(t, finished[0].Aborted, + "a write-error exit must never be reported as a completed drain") + assert.Equal(t, 0, finished[0].EventCount) + removed := rec.snapshotRemoved() + require.Len(t, removed, 1) + assert.Equal(t, ReasonWriteError, removed[0].Reason) +} + +// With the Server already closed there is no unsubscription path left to +// rescue a stranded producer, so the teardown itself must hand the abandoned +// batch to its drain before running the consumer callback that panics. +func TestServerTracePanicInAbortReportWithClosedServerStillFreesProducer(t *testing.T) { + rec := &traceRecorder{} + trace := rec.trace() + trace.ReplayFinished = func(context.Context, ReplayFinishedInfo) { + panic("consumer bug") + } + repo := gatedReplayRepository{ + before: 3, + after: 2, + gate: make(chan struct{}), + done: make(chan struct{}), + } + server := NewServer() + server.Trace = trace + server.ReplayAll = true + server.Register("test", repo) + + httpServer := httptest.NewUnstartedServer(server.Handler("test")) + httpServer.Config.ErrorLog = log.New(io.Discard, "", 0) + httpServer.Start() + defer httpServer.Close() + + reqCtx, cancel := context.WithCancel(context.Background()) + req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, httpServer.URL, nil) + require.NoError(t, err) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer func() { _ = resp.Body.Close() }() + + require.Eventually(t, func() bool { + return len(rec.snapshotReplayStarted()) == 1 + }, 2*time.Second, 10*time.Millisecond) + + // Close the Server while the client is still connected (its shutdown path + // leaves the batch to the handler), then hang up: the teardown's aborted + // ReplayFinished panics with no run() left to drain the batch afterwards. + server.Close() + cancel() + + require.Eventually(t, func() bool { + return len(rec.snapshotRemoved()) == 1 + }, 2*time.Second, 10*time.Millisecond, "SubscriberRemoved must fire despite the panic") + + close(repo.gate) + select { + case <-repo.done: + case <-time.After(2 * time.Second): + require.Fail(t, "the replay producer was stranded: the teardown must drain before its callbacks") + } +} + +// slowReplayRepository streams a large batch one event at a time through an +// unbuffered channel, so a test can end the connection while the batch is +// still draining. +type slowReplayRepository struct { + count int + data string +} + +func (r slowReplayRepository) Replay(channel, id string) chan Event { + out := make(chan Event) + go func() { + defer close(out) + for i := 0; i < r.count; i++ { + out <- &publication{id: fmt.Sprintf("id-%d", i), data: r.data} + time.Sleep(100 * time.Microsecond) + } + }() + return out +} + +func TestServerTraceReplayAborted(t *testing.T) { + rec := &traceRecorder{} + server := NewServer() + server.Trace = rec.trace() + server.ReplayAll = true + // Large enough that the cancel below lands mid-batch, small enough that + // the post-test background drain finishes promptly. + repo := slowReplayRepository{count: 2000, data: "0123456789"} + server.Register("test", repo) + defer server.Close() + + w := &traceTestWriter{} + cancel, done := startTraceHandler(server, "test", w, nil) + + // Wait until the batch is genuinely mid-drain: the replay has started and + // some replayed events have been encoded to the connection. + require.Eventually(t, func() bool { + return len(rec.snapshotReplayStarted()) == 1 && w.bytesWritten() > 0 + }, 5*time.Second, time.Millisecond) + + cancel() // the client hangs up mid-batch + waitClosed(t, done) + + finished := rec.snapshotReplayFinished() + require.Len(t, finished, 1, "an aborted batch must still report ReplayFinished") + assert.True(t, finished[0].Aborted) + assert.Equal(t, "test", finished[0].Channel) + assert.Greater(t, finished[0].EventCount, 0) + assert.Less(t, finished[0].EventCount, repo.count) + assert.Equal(t, int64(finished[0].EventCount*len(repo.data)), finished[0].TotalDataSize) + assert.Empty(t, rec.snapshotEventsSent()) +} + +// countingDataEvent counts Data() invocations, so a test can pin how many +// times the server reads a caller-supplied payload. +type countingDataEvent struct { + data string + calls *atomic.Int64 +} + +func (e *countingDataEvent) Id() string { return "1" } //nolint:revive // required by Event +func (e *countingDataEvent) Event() string { return "" } +func (e *countingDataEvent) Data() string { + e.calls.Add(1) + return e.data +} + +type countingReplayRepository struct { + calls *atomic.Int64 + count int + data string +} + +func (r countingReplayRepository) Replay(channel, id string) chan Event { + out := make(chan Event, r.count) + for i := 0; i < r.count; i++ { + out <- &countingDataEvent{data: r.data, calls: r.calls} + } + close(out) + return out +} + +// The replay byte accounting reads Event.Data() a second time per event (the +// encoder's own read is the first). Data() is caller-supplied code with no +// cost contract, so that second read must happen only when something consumes +// the sum: the ReplayFinished callback or a Logger. +func TestServerTraceReplayDataCallGating(t *testing.T) { + run := func(t *testing.T, configure func(*Server)) int64 { + var calls atomic.Int64 + server := NewServer() + server.ReplayAll = true + configure(server) + server.Register("test", countingReplayRepository{calls: &calls, count: 5, data: "0123456789"}) + defer server.Close() + + w := &traceTestWriter{} + cancel, done := startTraceHandler(server, "test", w, nil) + defer func() { + cancel() + waitClosed(t, done) + }() + + // Replay output appearing proves the subscription is registered and the + // batch enqueued, so the marker published next is ordered behind the + // batch. A live marker event is delivered only after the replay batch + // has been fully drained, so its appearance means every batch Data() + // call has happened. + require.Eventually(t, func() bool { + return w.contains("0123456789") + }, 2*time.Second, time.Millisecond) + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: "marker"}) + require.Eventually(t, func() bool { + return w.contains("marker") + }, 2*time.Second, time.Millisecond) + + return calls.Load() + } + + t.Run("no replay consumer reads Data once per event", func(t *testing.T) { + calls := run(t, func(s *Server) { + s.Trace = &ServerTrace{SubscriberAdded: func(context.Context, SubscriberAddedInfo) {}} + }) + assert.Equal(t, int64(5), calls, + "with neither ReplayFinished nor a Logger set, only the encoder should read Data()") + }) + + t.Run("ReplayFinished consumer reads Data twice per event", func(t *testing.T) { + calls := run(t, func(s *Server) { + s.Trace = &ServerTrace{ReplayFinished: func(context.Context, ReplayFinishedInfo) {}} + }) + assert.Equal(t, int64(10), calls) + }) +} + +func TestServerTraceNilCallbacksAreNoOps(t *testing.T) { + t.Run("nil trace pointer", func(t *testing.T) { + server := NewServer() + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: "x"}) + server.PublishComment([]string{"test"}, "c") + cancel() + waitClosed(t, done) + }) + + t.Run("all nil fields", func(t *testing.T) { + server := NewServer() + server.Trace = &ServerTrace{} + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: "x"}) + server.PublishComment([]string{"test"}, "c") + cancel() + waitClosed(t, done) + }) + + t.Run("only some fields set", func(t *testing.T) { + var addedCount int + var mu sync.Mutex + server := NewServer() + server.Trace = &ServerTrace{ + SubscriberAdded: func(context.Context, SubscriberAddedInfo) { + mu.Lock() + defer mu.Unlock() + addedCount++ + }, + } + defer server.Close() + + cancel, done := startTraceHandler(server, "test", &traceTestWriter{}, nil) + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return addedCount == 1 + }, 2*time.Second, 10*time.Millisecond) + + // SubscriberRemoved and the rest are nil; triggering them must not panic. + <-server.PublishWithAcknowledgment([]string{"test"}, &publication{data: "x"}) + cancel() + waitClosed(t, done) + }) +} + +func TestServerTraceLogging(t *testing.T) { + // The channel name stands in for a value that must never be logged; in + // ld-relay the channel is a raw SDK credential. + const sensitiveChannel = "sdk-key-do-not-log" + + t.Run("subscriber added and removed at debug", func(t *testing.T) { + logger := &captureLogger{} + server := NewServer() + server.Logger = logger + defer server.Close() + + cancel, done := startTraceHandler(server, sensitiveChannel, &traceTestWriter{}, nil) + require.Eventually(t, func() bool { + return logger.containing("[DEBUG] eventsource: subscriber added (id=1") + }, 2*time.Second, 10*time.Millisecond) + + cancel() + waitClosed(t, done) + + assert.True(t, logger.containing("[DEBUG] eventsource: subscriber removed (id=1")) + assert.False(t, logger.containing(sensitiveChannel), + "log lines must never include the channel name") + }) + + t.Run("buffer overflow at warn", func(t *testing.T) { + logger := &captureLogger{} + server := NewServer() + server.Logger = logger + server.BufferSize = 1 + defer server.Close() + + w := &traceTestWriter{gate: make(chan struct{})} + cancel, done := startTraceHandler(server, sensitiveChannel, w, nil) + defer cancel() + require.Eventually(t, func() bool { + return logger.containing("[DEBUG] eventsource: subscriber added") + }, 2*time.Second, 10*time.Millisecond) + + for _, data := range []string{"a", "b", "c"} { + <-server.PublishWithAcknowledgment([]string{sensitiveChannel}, &publication{data: data}) + } + + require.Eventually(t, func() bool { + return logger.containing("[WARN] eventsource: dropped subscriber (id=1") + }, 2*time.Second, 10*time.Millisecond) + + close(w.gate) + waitClosed(t, done) + + assert.False(t, logger.containing(sensitiveChannel), + "log lines must never include the channel name") + }) + + t.Run("replay drained at debug", func(t *testing.T) { + logger := &captureLogger{} + server := NewServer() + server.Logger = logger + server.ReplayAll = true + server.Register(sensitiveChannel, &testServerRepository{}) + defer server.Close() + + cancel, done := startTraceHandler(server, sensitiveChannel, &traceTestWriter{}, nil) + defer func() { + cancel() + waitClosed(t, done) + }() + + require.Eventually(t, func() bool { + return logger.containing("[DEBUG] eventsource: replay drained (id=1") + }, 2*time.Second, 10*time.Millisecond) + + assert.False(t, logger.containing(sensitiveChannel), + "log lines must never include the channel name") + }) +}