Merge stream H (Go client: event buffer option, grpc.NewClient, typed diagnostics)

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

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()
+		}
+	}
+}

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")
+	}
+}

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

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

clients/go/mxgateway/session.go

@@ -18,8 +18,9 @@ import (
 
 const maxBulkItems = 1000
 
-// eventBufferSize is the number of buffered event slots on the
+// eventBufferSize is the data-slot count the unit tests use for the legacy
-// Events/EventsAfter (cancel-when-full) results channel.
+// 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

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