diff --git a/clients/go/mxgateway/client_session_test.go b/clients/go/mxgateway/client_session_test.go index bc5d597..fa1f85b 100644 --- a/clients/go/mxgateway/client_session_test.go +++ b/clients/go/mxgateway/client_session_test.go @@ -147,6 +147,59 @@ func TestEventsAfterCancelsStreamWhenCompatibilityChannelIsAbandoned(t *testing. } } +func TestEventsSlowConsumerYieldsErrSlowConsumerBeforeClose(t *testing.T) { + fake := &fakeGatewayServer{ + streamStarted: make(chan struct{}), + streamDone: make(chan struct{}), + streamEventCount: 64, + } + client, cleanup := newBufconnClient(t, fake) + defer cleanup() + session := NewSessionForID(client, "session-1") + + events, err := session.EventsAfter(context.Background(), 0) + if err != nil { + t.Fatalf("EventsAfter() error = %v", err) + } + <-fake.streamStarted + + // Do not drain the channel so the buffer overflows. The stream stops once + // the slow-consumer cancel fires on the producer side. + select { + case <-fake.streamDone: + case <-time.After(2 * time.Second): + t.Fatal("event stream did not stop after buffer overflow") + } + + var last EventResult + gotResult := false + for { + select { + case res, ok := <-events: + if !ok { + if !gotResult { + t.Fatal("events channel closed without yielding any result") + } + if !errors.Is(last.Err, ErrSlowConsumer) { + t.Fatalf("final event result err = %v, want one wrapping ErrSlowConsumer", last.Err) + } + var gwErr *GatewayError + if !errors.As(last.Err, &gwErr) { + t.Fatalf("final event result err is %T, want *GatewayError", last.Err) + } + if gwErr.Op != "stream events" { + t.Fatalf("final event result err Op = %q, want %q", gwErr.Op, "stream events") + } + return + } + last = res + gotResult = true + case <-time.After(2 * time.Second): + t.Fatal("events channel did not close after slow-consumer termination") + } + } +} + func TestSessionHelpersBuildCommandsAndExposeRawReply(t *testing.T) { fake := &fakeGatewayServer{ invokeReply: &pb.MxCommandReply{ diff --git a/clients/go/mxgateway/errors.go b/clients/go/mxgateway/errors.go index 92dd486..778e56e 100644 --- a/clients/go/mxgateway/errors.go +++ b/clients/go/mxgateway/errors.go @@ -1,11 +1,19 @@ package mxgateway import ( + "errors" "fmt" pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated" ) +// ErrSlowConsumer is the terminal error sent on the Events/EventsAfter +// (cancel-when-full) path when the buffered results channel overflows because +// the consumer fell behind. It is delivered as the final EventResult.Err before +// the channel closes, so overflow is always observable rather than silently +// dropping events. Match it with errors.Is. +var ErrSlowConsumer = errors.New("mxgateway: event consumer fell behind; stream terminated") + // GatewayError wraps transport-level gRPC failures. type GatewayError struct { // Op names the operation that failed (for example "dial" or "invoke"). diff --git a/clients/go/mxgateway/session.go b/clients/go/mxgateway/session.go index 00bcdab..e61cf46 100644 --- a/clients/go/mxgateway/session.go +++ b/clients/go/mxgateway/session.go @@ -18,6 +18,15 @@ import ( const maxBulkItems = 1000 +// eventBufferSize is the number of buffered event slots on the +// Events/EventsAfter (cancel-when-full) results channel. +const eventBufferSize = 16 + +// eventBufferReservedSlots is the extra capacity reserved beyond +// eventBufferSize so a terminal ErrSlowConsumer result can always be enqueued +// non-blockingly on overflow, even when all data slots are full. +const eventBufferReservedSlots = 1 + // EventResult carries either the next ordered event or a terminal stream error. type EventResult struct { // Event is the next event from the stream when Err is nil. @@ -674,11 +683,22 @@ func (s *Session) Write2Raw(ctx context.Context, serverHandle, itemHandle int32, // Events streams ordered session events until the server ends the stream, // context cancellation stops Recv, or a terminal error is sent. +// +// The returned channel is buffered. If the consumer falls behind and the buffer +// overflows, the stream is terminated and a final EventResult carrying a +// GatewayError that wraps ErrSlowConsumer is delivered before the channel +// closes. Callers must match it with errors.Is(res.Err, ErrSlowConsumer) to +// distinguish a slow-consumer drop from a graceful server end. Use +// SubscribeEvents for a blocking, backpressured stream that never drops. func (s *Session) Events(ctx context.Context) (<-chan EventResult, error) { return s.EventsAfter(ctx, 0) } // EventsAfter streams ordered session events after the given worker sequence. +// +// Like Events, the returned channel is buffered and terminates with a final +// EventResult wrapping ErrSlowConsumer (matchable via errors.Is) if the consumer +// falls behind and the buffer overflows, rather than silently closing. func (s *Session) EventsAfter(ctx context.Context, afterWorkerSequence uint64) (<-chan EventResult, error) { subscription, err := s.subscribeEventsAfter(ctx, afterWorkerSequence, true) if err != nil { @@ -708,7 +728,7 @@ func (s *Session) subscribeEventsAfter(ctx context.Context, afterWorkerSequence return nil, err } - results := make(chan EventResult, 16) + results := make(chan EventResult, eventBufferSize+eventBufferReservedSlots) done := make(chan struct{}) go func() { defer close(results) @@ -756,14 +776,30 @@ func sendEventResult( cancel context.CancelFunc, ) bool { if cancelWhenBufferFull { + // Treat the channel as full once the eventBufferSize data slots are + // occupied, keeping eventBufferReservedSlots free for the terminal + // error. This goroutine is the sole producer, so len(results) only + // grows by our own sends and shrinks as the consumer reads; the reserve + // therefore always survives to carry ErrSlowConsumer. A plain buffered + // send would instead consume the reserved slot as ordinary data. + if len(results) >= eventBufferSize { + // The consumer fell behind and the data slots are full. Cancel the + // stream, then use the reserved terminal slot to enqueue a loud + // ErrSlowConsumer result so overflow is always observable rather + // than a silently closed channel indistinguishable from a graceful + // server end. + cancel() + select { + case results <- EventResult{Err: &GatewayError{Op: "stream events", Err: ErrSlowConsumer}}: + default: + } + return false + } select { case results <- result: return true case <-ctx.Done(): return false - default: - cancel() - return false } }