feat(go-client): configurable event buffer, grpc.NewClient dial, typed diagnostics

- Options.EventBufferSize (default 1024) replaces the fixed 16-slot Events buffer - Dial/DialGalaxy use grpc.NewClient + explicit READY wait (replaces DialContext+WithBlock) - Session.Ping/GetSessionState/GetWorkerInfo/DrainEvents typed diagnostics
2026-10-22 06:08:31 -04:00
parent 488649ce99
commit a3a7bd2170
7 changed files with 389 additions and 14 deletions
@@ -121,7 +121,7 @@ func TestEventsAfterCancelsStreamWhenCompatibilityChannelIsAbandoned(t *testing.
 		streamDone:       make(chan struct{}),
 		streamEventCount: 64,
 	}
-	client, cleanup := newBufconnClient(t, fake)
+	client, cleanup := newBufconnClientWithOptions(t, fake, Options{EventBufferSize: eventBufferSize})
 	defer cleanup()
 	session := NewSessionForID(client, "session-1")

@@ -155,7 +155,7 @@ func TestEventsSlowConsumerYieldsErrSlowConsumerBeforeClose(t *testing.T) {
 		streamDone:       make(chan struct{}),
 		streamEventCount: 64,
 	}
-	client, cleanup := newBufconnClient(t, fake)
+	client, cleanup := newBufconnClientWithOptions(t, fake, Options{EventBufferSize: eventBufferSize})
 	defer cleanup()
 	session := NewSessionForID(client, "session-1")

@@ -209,7 +209,7 @@ func TestEventsFullBufferTerminalErrorKeepsRootCause(t *testing.T) {
 		streamEventCount:  eventBufferSize,
 		streamTerminalErr: status.Error(codes.Internal, "boom"),
 	}
-	client, cleanup := newBufconnClient(t, fake)
+	client, cleanup := newBufconnClientWithOptions(t, fake, Options{EventBufferSize: eventBufferSize})
 	defer cleanup()
 	session := NewSessionForID(client, "session-1")

@@ -280,7 +280,7 @@ func TestSubscribeEventsFullBufferDeliversTerminalError(t *testing.T) {
 		streamEventCount:  eventBufferSize + eventBufferReservedSlots,
 		streamTerminalErr: status.Error(codes.Internal, "boom"),
 	}
-	client, cleanup := newBufconnClient(t, fake)
+	client, cleanup := newBufconnClientWithOptions(t, fake, Options{EventBufferSize: eventBufferSize})
 	defer cleanup()
 	session := NewSessionForID(client, "session-1")

@@ -1047,3 +1047,240 @@ func TestWriteArrayElementsSendsWriteCommandWithSparseOneof(t *testing.T) {
 		t.Errorf("element float value = %v, want 1.5", elem.GetValue().GetFloatValue())
 	}
 }
+
+// newBufconnClientWithOptions mirrors newBufconnClient but lets a test tune Options.
+func newBufconnClientWithOptions(t *testing.T, fake *fakeGatewayServer, opts Options) (*Client, func()) {
+	t.Helper()
+	listener := bufconn.Listen(bufSize)
+	server := grpc.NewServer()
+	pb.RegisterMxAccessGatewayServer(server, fake)
+	go func() { _ = server.Serve(listener) }()
+	opts.Endpoint = "bufnet"
+	opts.APIKey = "test-api-key"
+	opts.Plaintext = true
+	opts.DialOptions = append(opts.DialOptions, grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) {
+		return listener.DialContext(ctx)
+	}))
+	client, err := Dial(context.Background(), opts)
+	if err != nil {
+		t.Fatalf("Dial() error = %v", err)
+	}
+	return client, func() {
+		client.Close()
+		server.Stop()
+		listener.Close()
+	}
+}
+
+func TestEventsHonorsConfiguredEventBufferSize(t *testing.T) {
+	// 100 events exceed the old fixed 16-slot buffer; a 256-slot buffer must
+	// hold them all without a slow-consumer termination.
+	fake := &fakeGatewayServer{
+		streamStarted:     make(chan struct{}),
+		streamDone:        make(chan struct{}),
+		streamEventCount:  100,
+		streamTerminalErr: status.Error(codes.Internal, "boom"),
+	}
+	client, cleanup := newBufconnClientWithOptions(t, fake, Options{EventBufferSize: 256})
+	defer cleanup()
+	session := NewSessionForID(client, "session-1")
+
+	events, err := session.Events(context.Background())
+	if err != nil {
+		t.Fatalf("Events() error = %v", err)
+	}
+	<-fake.streamDone
+	time.Sleep(250 * time.Millisecond)
+
+	count := 0
+	var last EventResult
+	for res := range events {
+		if res.Event != nil {
+			count++
+		}
+		last = res
+	}
+	if count != 100 {
+		t.Fatalf("delivered events = %d, want 100", count)
+	}
+	if errors.Is(last.Err, ErrSlowConsumer) || status.Code(last.Err) != codes.Internal {
+		t.Fatalf("terminal err = %v, want the genuine Internal error", last.Err)
+	}
+}
+
+func TestEventsOverflowDeliversBufferedEventsBeforeSlowConsumerError(t *testing.T) {
+	fake := &fakeGatewayServer{
+		streamStarted:    make(chan struct{}),
+		streamDone:       make(chan struct{}),
+		streamEventCount: 64,
+	}
+	client, cleanup := newBufconnClientWithOptions(t, fake, Options{EventBufferSize: 4})
+	defer cleanup()
+	session := NewSessionForID(client, "session-1")
+
+	events, err := session.Events(context.Background())
+	if err != nil {
+		t.Fatalf("Events() error = %v", err)
+	}
+	<-fake.streamStarted
+	time.Sleep(250 * time.Millisecond)
+
+	buffered := 0
+	var last EventResult
+	for res := range events {
+		if res.Event != nil {
+			buffered++
+			if got := res.Event.GetWorkerSequence(); got != uint64(buffered) {
+				t.Fatalf("event %d has sequence %d, want in-order", buffered, got)
+			}
+		}
+		last = res
+	}
+	if buffered != 4 {
+		t.Fatalf("buffered events delivered = %d, want 4 (all data slots)", buffered)
+	}
+	if !errors.Is(last.Err, ErrSlowConsumer) {
+		t.Fatalf("final err = %v, want ErrSlowConsumer", last.Err)
+	}
+}
+
+func TestEventBufferSizeDefault(t *testing.T) {
+	if got := (Options{}).eventBufferSize(); got != DefaultEventBufferSize {
+		t.Fatalf("default = %d, want %d", got, DefaultEventBufferSize)
+	}
+	if got := (Options{EventBufferSize: 7}).eventBufferSize(); got != 7 {
+		t.Fatalf("explicit = %d, want 7", got)
+	}
+}
+
+func TestDialFailsWhenEndpointUnreachable(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	_, err := Dial(ctx, Options{
+		Endpoint:    "bufnet",
+		Plaintext:   true,
+		DialTimeout: 300 * time.Millisecond,
+		DialOptions: []grpc.DialOption{grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+			return nil, errors.New("refused")
+		})},
+	})
+	if err == nil {
+		t.Fatal("Dial() error = nil, want connect failure")
+	}
+	var gwErr *GatewayError
+	if !errors.As(err, &gwErr) || gwErr.Op != "dial" {
+		t.Fatalf("Dial() error = %v, want *GatewayError Op=dial", err)
+	}
+}
+
+func okStatus() *pb.ProtocolStatus {
+	return &pb.ProtocolStatus{Code: pb.ProtocolStatusCode_PROTOCOL_STATUS_CODE_OK}
+}
+
+func TestSessionPingReturnsEchoedMessage(t *testing.T) {
+	fake := &fakeGatewayServer{invokeReply: &pb.MxCommandReply{
+		Kind:              pb.MxCommandKind_MX_COMMAND_KIND_PING,
+		ProtocolStatus:    okStatus(),
+		DiagnosticMessage: "pong:hi",
+	}}
+	client, cleanup := newBufconnClient(t, fake)
+	defer cleanup()
+	session := NewSessionForID(client, "session-1")
+
+	got, err := session.Ping(context.Background(), "hi")
+	if err != nil {
+		t.Fatalf("Ping() error = %v", err)
+	}
+	if got != "pong:hi" {
+		t.Fatalf("Ping() = %q, want pong:hi", got)
+	}
+	if fake.invokeRequest.GetCommand().GetPing().GetMessage() != "hi" {
+		t.Fatalf("ping payload not sent: %v", fake.invokeRequest)
+	}
+}
+
+func TestSessionGetSessionState(t *testing.T) {
+	fake := &fakeGatewayServer{invokeReply: &pb.MxCommandReply{
+		Kind:           pb.MxCommandKind_MX_COMMAND_KIND_GET_SESSION_STATE,
+		ProtocolStatus: okStatus(),
+		Payload: &pb.MxCommandReply_SessionState{SessionState: &pb.SessionStateReply{
+			State: pb.SessionState_SESSION_STATE_CREATING,
+		}},
+	}}
+	client, cleanup := newBufconnClient(t, fake)
+	defer cleanup()
+	session := NewSessionForID(client, "session-1")
+
+	got, err := session.GetSessionState(context.Background())
+	if err != nil {
+		t.Fatalf("GetSessionState() error = %v", err)
+	}
+	if got != pb.SessionState_SESSION_STATE_CREATING {
+		t.Fatalf("state = %v", got)
+	}
+	if fake.invokeRequest.GetCommand().GetKind() != pb.MxCommandKind_MX_COMMAND_KIND_GET_SESSION_STATE {
+		t.Fatalf("kind = %v", fake.invokeRequest.GetCommand().GetKind())
+	}
+}
+
+func TestSessionGetWorkerInfo(t *testing.T) {
+	fake := &fakeGatewayServer{invokeReply: &pb.MxCommandReply{
+		Kind:           pb.MxCommandKind_MX_COMMAND_KIND_GET_WORKER_INFO,
+		ProtocolStatus: okStatus(),
+		Payload: &pb.MxCommandReply_WorkerInfo{WorkerInfo: &pb.WorkerInfoReply{
+			WorkerProcessId: 42, WorkerVersion: "1.2.3", MxaccessProgid: "prog", MxaccessClsid: "clsid",
+		}},
+	}}
+	client, cleanup := newBufconnClient(t, fake)
+	defer cleanup()
+	session := NewSessionForID(client, "session-1")
+
+	info, err := session.GetWorkerInfo(context.Background())
+	if err != nil {
+		t.Fatalf("GetWorkerInfo() error = %v", err)
+	}
+	if info.GetWorkerProcessId() != 42 || info.GetWorkerVersion() != "1.2.3" {
+		t.Fatalf("info = %v", info)
+	}
+}
+
+func TestSessionDrainEvents(t *testing.T) {
+	fake := &fakeGatewayServer{invokeReply: &pb.MxCommandReply{
+		Kind:           pb.MxCommandKind_MX_COMMAND_KIND_DRAIN_EVENTS,
+		ProtocolStatus: okStatus(),
+		Payload: &pb.MxCommandReply_DrainEvents{DrainEvents: &pb.DrainEventsReply{
+			Events: []*pb.MxEvent{{WorkerSequence: 1}, {WorkerSequence: 2}},
+		}},
+	}}
+	client, cleanup := newBufconnClient(t, fake)
+	defer cleanup()
+	session := NewSessionForID(client, "session-1")
+
+	events, err := session.DrainEvents(context.Background(), 10)
+	if err != nil {
+		t.Fatalf("DrainEvents() error = %v", err)
+	}
+	if len(events) != 2 || events[1].GetWorkerSequence() != 2 {
+		t.Fatalf("events = %v", events)
+	}
+	if fake.invokeRequest.GetCommand().GetDrainEvents().GetMaxEvents() != 10 {
+		t.Fatalf("max_events not sent: %v", fake.invokeRequest)
+	}
+}
+
+func TestSessionDiagnosticsErrorOnMissingPayload(t *testing.T) {
+	fake := &fakeGatewayServer{}
+	client, cleanup := newBufconnClient(t, fake)
+	defer cleanup()
+	session := NewSessionForID(client, "session-1")
+
+	if _, err := session.GetWorkerInfo(context.Background()); err == nil {
+		t.Fatal("GetWorkerInfo() with empty reply: want error")
+	}
+	if _, err := session.GetSessionState(context.Background()); err == nil {
+		t.Fatal("GetSessionState() with empty reply: want error")
+	}
+	if _, err := session.DrainEvents(context.Background(), 1); err == nil {
+		t.Fatal("DrainEvents() with empty reply: want error")
+	}
+}