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{

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").

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
 		}
 	}