From 873d6df67009ff3d92dd08ba2a51bdab2dd6d461 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Thu, 6 Aug 2026 12:46:21 -0400 Subject: [PATCH 1/3] feat: Add the ServerTrace API surface and its reporting helpers ServerTrace is a struct of optional callbacks modeled on net/http/httptrace.ClientTrace, exposed as Server.Trace: fields can be added without breaking implementers, info structs (not positional arguments) carry the payloads, and a nil struct or nil field disables that hook. The API is marked EXPERIMENTAL and for LaunchDarkly libraries only. This commit adds the contract and the Server-side reporting helpers; the next commit wires them through the handler and dispatch paths. Design points: - Callbacks fire synchronously on internal goroutines and must return promptly and never panic; SubscriberDropped fires on the dispatch goroutine, where an unrecovered panic would take down the process rather than one connection, so that helper contains panics and reports them through the Logger (rendering only the panic value's type, never its contents). - Handler-goroutine callbacks receive the subscriber's request context for telemetry correlation only. - The same instrumentation points feed Logger: DEBUG on subscriber add/remove and replay drain, WARN on a slow-subscriber drop. Log lines identify connections by an opaque per-Server subscriber id and never include the channel name, because channel names may contain credentials. - Measurement is gated on the consuming callback being set (shouldMeasureWrite, writeTraced, beginSubscription/sinceOrZero), so a Server with neither Trace nor Logger reads no clock and allocates nothing new on the write path. - Encode errors render unexpected values with %T, not %v, so payload contents cannot leak into error strings that reach WriteError consumers and logs. --- encoder.go | 5 +- server.go | 26 ++- server_trace.go | 502 ++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 526 insertions(+), 7 deletions(-) create mode 100644 server_trace.go 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/server.go b/server.go index 56a25e5..4a8c9d7 100644 --- a/server.go +++ b/server.go @@ -5,6 +5,7 @@ import ( "net/http" "strings" "sync" + "sync/atomic" "time" ) @@ -20,6 +21,7 @@ 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 } type eventOrComment interface{} @@ -51,12 +53,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 +83,7 @@ type Server struct { isClosed bool isClosedMutex sync.RWMutex jitter time.Duration + subCounter atomic.Uint64 } // NewServer creates a new Server instance. 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) + } +} From 596bfdc4b355ab9cc88a46eb80e6f6d17548779c Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Thu, 6 Aug 2026 12:46:44 -0400 Subject: [PATCH 2/3] feat: Wire ServerTrace through the handler and dispatch paths Calls the hooks at every lifecycle point and makes what they report true under the teardown races the server already had: - Exit reasons: each read-loop exit records its reason; for Server-initiated closes, run() records closeReason before closing the subscriber's channel and the handler reads it only after observing that close (the channel close is the happens-before edge). When a Server close races the handler's own exit, reasons rank: buffer_overflow above everything (SubscriberDropped already promised a matching removal), write_error above the remaining Server reasons, any Server reason above the speculative client_closed and max_conn_time. - Teardown: the handler tears down in two defers. reportExit runs first and holds the consumer-reachable reporting; cleanup registers earlier so it still resolves the reason, unsubscribes, and emits SubscriberRemoved while a panic from a reporting callback is unwinding. reportedAdded is set before SubscriberAdded so even a panic inside that callback produces the balancing removal. SubscriberAdded fires before the subscription is registered, so no other callback can precede it. - Replay: batches report at batch level (count, summed payload bytes, DrainDuration through the end-of-batch flush, measured only when consumed). Every ReplayStarted gets exactly one ReplayFinished; a batch abandoned mid-drain reports Aborted, with a handler-local exit probe that keeps the common disconnect-races-sentinel case reported as completed. Aborted is documented best-effort: a Server shutdown draining an abandoned batch concurrently can make it read as completed. Concurrent drains themselves are safe -- receivers split discarded events until the channel closes. - Accounting: a jitter-parked event whose connection ends is reported discarded (connection_ended), so every event is accounted sent or discarded; the parking slot clears before the write so a panicking EventSent cannot double-report it. - Drops: a replay-batch enqueue that overflows the buffer reports SubscriberDropped exactly as trySend does; a handler racing Server.Close escapes the registration send instead of parking forever. --- interface.go | 4 +- server.go | 407 ++++++++++++++++++++++++++++++++++++++++----------- 2 files changed, 323 insertions(+), 88 deletions(-) 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 4a8c9d7..dcc4530 100644 --- a/server.go +++ b/server.go @@ -22,6 +22,11 @@ type subscription struct { // 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{} @@ -140,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 @@ -162,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 @@ -221,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. @@ -236,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 @@ -252,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 } @@ -279,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 @@ -304,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 - } } } @@ -352,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) } @@ -360,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 } } } @@ -450,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. @@ -459,6 +705,7 @@ func (srv *Server) run() { if !sub.send(ec) { sub.close() delete(subs[sub.channel], sub) + srv.traceSubscriberDropped(sub) } } for { @@ -471,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 @@ -483,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 } @@ -511,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 @@ -598,6 +830,7 @@ func (s *subscription) send(e eventOrComment) bool { case s.out <- e: return true default: + s.closeReason = ReasonBufferOverflow s.close() return false } From 1b907e81e60326d898e9f489d8f164527130fc87 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Thu, 6 Aug 2026 12:47:00 -0400 Subject: [PATCH 3/3] test: Cover the ServerTrace contract Covers every callback, all six removal reasons and their precedence under races, context propagation, replay batch accounting (count, bytes, durations, best-effort Aborted), the panic-teardown invariants (a panicking callback never leaks the subscription, strands a producer, or breaks the 1:1 Added/Removed pairing), zero-overhead gating, and a concurrency stress test of the pairing invariant. Includes the encoder %T error-format pin. --- encoder_test.go | 10 + server_trace_test.go | 1786 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 1796 insertions(+) create mode 100644 server_trace_test.go 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/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") + }) +}