diff --git a/clients/go/README.md b/clients/go/README.md index 5e5fa0f..ad36447 100644 --- a/clients/go/README.md +++ b/clients/go/README.md @@ -100,6 +100,29 @@ judges each `MXSTATUS_PROXY` entry by its category — an entry fails when `Category` is not `MX_STATUS_CATEGORY_OK`, and the raw `Success` member is a diagnostic that never decides the verdict. A nil entry is success. +`Dial` is connect-or-fail: it creates the connection with `grpc.NewClient` +(the endpoint resolves with the `passthrough` scheme unless it names one) and +then waits for the channel to become READY, bounded by `DialTimeout` (default +10 s) or the context deadline. A failure is returned as a `GatewayError` with +`Op == "dial"`. Passing a connection to `NewClient`/`NewGalaxyClient` stays lazy. + +### Event buffering + +The channel returned by `Session.Events`/`EventsAfter` buffers +`Options.EventBufferSize` data results (default `DefaultEventBufferSize`, 1024). +If the consumer falls behind and the buffer fills, the stream is cancelled; the +results already buffered are still delivered in order, followed by a final +result whose error matches `errors.Is(err, mxgateway.ErrSlowConsumer)`. Raise +`EventBufferSize` for bursty feeds, or use `SubscribeEvents`/`SubscribeEventsAfter`, +which block instead of dropping and are the non-dropping alternative. + +### Diagnostics + +`Session.Ping`, `GetSessionState`, `GetWorkerInfo`, and `DrainEvents` are typed +wrappers over the diagnostic commands (`PingRaw` remains for the raw reply). +`DrainEvents(ctx, maxEvents)` returns fewer events than requested when the +server caps the reply; call it until it returns an empty slice. + ### Reconnect-replay gap Each `EventResult` carries exactly one of `Event`, `ReplayGap`, or `Err`. When diff --git a/clients/go/mxgateway/client.go b/clients/go/mxgateway/client.go index 9bf2370..5b647cf 100644 --- a/clients/go/mxgateway/client.go +++ b/clients/go/mxgateway/client.go @@ -15,10 +15,12 @@ import ( "context" "crypto/tls" "errors" + "strings" "time" pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated" "google.golang.org/grpc" + "google.golang.org/grpc/connectivity" "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" "google.golang.org/protobuf/types/known/durationpb" @@ -61,11 +63,10 @@ func Dial(ctx context.Context, opts Options) (*Client, error) { grpc.WithTransportCredentials(transportCredentials), grpc.WithUnaryInterceptor(unaryAuthInterceptor(opts.APIKey)), grpc.WithStreamInterceptor(streamAuthInterceptor(opts.APIKey)), - grpc.WithBlock(), } dialOptions = append(dialOptions, opts.DialOptions...) - conn, err := grpc.DialContext(dialCtx, opts.Endpoint, dialOptions...) + conn, err := dialConnected(dialCtx, opts.Endpoint, dialOptions) if err != nil { return nil, &GatewayError{Op: "dial", Err: err} } @@ -265,3 +266,34 @@ func (o OpenSessionOptions) Request() *OpenSessionRequest { } return req } + +// dialConnected creates a connection with grpc.NewClient and then waits for it +// to become READY, bounded by ctx. This preserves the connect-or-fail contract +// the removed grpc.DialContext + grpc.WithBlock combination gave Dial callers. +// The endpoint is resolved with the passthrough scheme unless it already names a +// scheme, matching DialContext's historical default (NewClient defaults to dns). +func dialConnected(ctx context.Context, endpoint string, dialOptions []grpc.DialOption) (*grpc.ClientConn, error) { + target := endpoint + if !strings.Contains(target, "://") { + target = "passthrough:///" + target + } + conn, err := grpc.NewClient(target, dialOptions...) + if err != nil { + return nil, err + } + conn.Connect() + for { + state := conn.GetState() + if state == connectivity.Ready { + return conn, nil + } + if state == connectivity.Shutdown { + _ = conn.Close() + return nil, errors.New("connection shut down while connecting") + } + if !conn.WaitForStateChange(ctx, state) { + _ = conn.Close() + return nil, ctx.Err() + } + } +} diff --git a/clients/go/mxgateway/client_session_test.go b/clients/go/mxgateway/client_session_test.go index 7088294..4b17227 100644 --- a/clients/go/mxgateway/client_session_test.go +++ b/clients/go/mxgateway/client_session_test.go @@ -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") + } +} diff --git a/clients/go/mxgateway/galaxy.go b/clients/go/mxgateway/galaxy.go index 3a3a339..8d09056 100644 --- a/clients/go/mxgateway/galaxy.go +++ b/clients/go/mxgateway/galaxy.go @@ -99,11 +99,10 @@ func DialGalaxy(ctx context.Context, opts Options) (*GalaxyClient, error) { grpc.WithTransportCredentials(transportCredentials), grpc.WithUnaryInterceptor(unaryAuthInterceptor(opts.APIKey)), grpc.WithStreamInterceptor(streamAuthInterceptor(opts.APIKey)), - grpc.WithBlock(), } dialOptions = append(dialOptions, opts.DialOptions...) - conn, err := grpc.DialContext(dialCtx, opts.Endpoint, dialOptions...) + conn, err := dialConnected(dialCtx, opts.Endpoint, dialOptions) if err != nil { return nil, &GatewayError{Op: "dial", Err: err} } diff --git a/clients/go/mxgateway/options.go b/clients/go/mxgateway/options.go index 732c662..173e406 100644 --- a/clients/go/mxgateway/options.go +++ b/clients/go/mxgateway/options.go @@ -38,6 +38,21 @@ type Options struct { // no CACertFile is pinned. Default false: the gateway's self-signed cert is // accepted without verification (internal-tool posture). RequireCertificateValidation bool + // EventBufferSize is the number of data results the channel returned by + // Session.Events/EventsAfter buffers before a slow consumer is cancelled with + // ErrSlowConsumer. Zero or negative applies DefaultEventBufferSize. It does not + // affect SubscribeEvents, which blocks instead of dropping. + EventBufferSize int +} + +// DefaultEventBufferSize is the default Options.EventBufferSize. +const DefaultEventBufferSize = 1024 + +func (o Options) eventBufferSize() int { + if o.EventBufferSize <= 0 { + return DefaultEventBufferSize + } + return o.EventBufferSize } // BrowseChildrenOptions configures lazy Galaxy hierarchy walks performed by diff --git a/clients/go/mxgateway/session.go b/clients/go/mxgateway/session.go index d20caaf..f2233f3 100644 --- a/clients/go/mxgateway/session.go +++ b/clients/go/mxgateway/session.go @@ -18,8 +18,9 @@ import ( const maxBulkItems = 1000 -// eventBufferSize is the number of buffered event slots on the -// Events/EventsAfter (cancel-when-full) results channel. +// eventBufferSize is the data-slot count the unit tests use for the legacy +// fixed-size assertions; the runtime value comes from Options.EventBufferSize +// (default DefaultEventBufferSize). const eventBufferSize = 16 // eventBufferReservedSlots is the extra capacity reserved beyond @@ -666,6 +667,68 @@ func buildSparseArrayValue(elementDataType MxDataType, totalLength uint32, eleme } } +// Ping sends a diagnostic PING command and returns the message the gateway +// echoed back in the reply's DiagnosticMessage field. +func (s *Session) Ping(ctx context.Context, message string) (string, error) { + reply, err := s.PingRaw(ctx, message) + if err != nil { + return "", err + } + return reply.GetDiagnosticMessage(), nil +} + +// GetSessionState returns the gateway's current lifecycle state for the session. +func (s *Session) GetSessionState(ctx context.Context) (SessionState, error) { + reply, err := s.invokeCommand(ctx, &pb.MxCommand{ + Kind: pb.MxCommandKind_MX_COMMAND_KIND_GET_SESSION_STATE, + Payload: &pb.MxCommand_GetSessionState{GetSessionState: &pb.GetSessionStateCommand{}}, + }) + if err != nil { + return 0, err + } + payload := reply.GetSessionState() + if payload == nil { + return 0, errors.New("mxgateway: get session state reply carried no session_state payload") + } + return payload.GetState(), nil +} + +// GetWorkerInfo returns diagnostic information about the session's worker +// process (process id, version, MXAccess ProgID and CLSID). +func (s *Session) GetWorkerInfo(ctx context.Context) (*WorkerInfoReply, error) { + reply, err := s.invokeCommand(ctx, &pb.MxCommand{ + Kind: pb.MxCommandKind_MX_COMMAND_KIND_GET_WORKER_INFO, + Payload: &pb.MxCommand_GetWorkerInfo{GetWorkerInfo: &pb.GetWorkerInfoCommand{}}, + }) + if err != nil { + return nil, err + } + payload := reply.GetWorkerInfo() + if payload == nil { + return nil, errors.New("mxgateway: get worker info reply carried no worker_info payload") + } + return payload, nil +} + +// DrainEvents removes and returns up to maxEvents queued events through the +// diagnostic DRAIN_EVENTS command. The reply is bounded server-side, so it can +// carry fewer events than requested or queued; call repeatedly until it returns +// an empty slice. It is a diagnostic tool, not a replacement for Events. +func (s *Session) DrainEvents(ctx context.Context, maxEvents uint32) ([]*MxEvent, error) { + reply, err := s.invokeCommand(ctx, &pb.MxCommand{ + Kind: pb.MxCommandKind_MX_COMMAND_KIND_DRAIN_EVENTS, + Payload: &pb.MxCommand_DrainEvents{DrainEvents: &pb.DrainEventsCommand{MaxEvents: maxEvents}}, + }) + if err != nil { + return nil, err + } + payload := reply.GetDrainEvents() + if payload == nil { + return nil, errors.New("mxgateway: drain events reply carried no drain_events payload") + } + return payload.GetEvents(), nil +} + // PingRaw sends a diagnostic PING command and returns the raw reply. // The message is echoed back by the gateway in the reply's DiagnosticMessage field. func (s *Session) PingRaw(ctx context.Context, message string) (*MxCommandReply, error) { @@ -1001,7 +1064,9 @@ func stringSecrets(values ...*MxValue) []string { // The returned channel is buffered. If the consumer falls behind and the buffer // overflows with data, the stream is terminated and a final EventResult carrying // a GatewayError that wraps ErrSlowConsumer is delivered before the channel -// closes; match it with errors.Is(res.Err, ErrSlowConsumer) to distinguish a +// closes. Results already buffered are delivered first, in order, ahead of that +// terminal result. The buffer holds Options.EventBufferSize data results +// (default DefaultEventBufferSize); match it with errors.Is(res.Err, ErrSlowConsumer) to distinguish a // slow-consumer drop from a graceful server end. A genuine stream error is // reported as itself even under overflow — it is never relabeled as // ErrSlowConsumer, so the underlying gRPC status stays inspectable. Use @@ -1046,7 +1111,8 @@ func (s *Session) subscribeEventsAfter(ctx context.Context, afterWorkerSequence return nil, err } - results := make(chan EventResult, eventBufferSize+eventBufferReservedSlots) + dataSlots := s.client.opts.eventBufferSize() + results := make(chan EventResult, dataSlots+eventBufferReservedSlots) done := make(chan struct{}) go func() { defer close(results) @@ -1062,7 +1128,7 @@ func (s *Session) subscribeEventsAfter(ctx context.Context, afterWorkerSequence if gap := event.GetReplayGap(); gap != nil { result = EventResult{ReplayGap: gap} } - if !sendEventResult(streamCtx, results, result, cancelWhenResultBufferFull, cancel) { + if !sendEventResult(streamCtx, results, result, dataSlots, cancelWhenResultBufferFull, cancel) { return } continue @@ -1126,6 +1192,7 @@ func sendEventResult( ctx context.Context, results chan<- EventResult, result EventResult, + dataSlots int, cancelWhenBufferFull bool, cancel context.CancelFunc, ) bool { @@ -1136,7 +1203,7 @@ func sendEventResult( // 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 { + if len(results) >= dataSlots { // 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 diff --git a/clients/go/mxgateway/types.go b/clients/go/mxgateway/types.go index e5a6f1a..b085c99 100644 --- a/clients/go/mxgateway/types.go +++ b/clients/go/mxgateway/types.go @@ -28,6 +28,8 @@ type ( MxCommandReply = pb.MxCommandReply // MxCommand is the discriminated union of MXAccess command payloads. MxCommand = pb.MxCommand + // WorkerInfoReply is the GetWorkerInfo diagnostic reply payload. + WorkerInfoReply = pb.WorkerInfoReply // MxEvent is one ordered event delivered on a session event stream. MxEvent = pb.MxEvent // ReplayGap is the gateway sentinel payload signalling that a resumed event