README
¶
Modulex gRPC Adapter
This package provides an optional gRPC topology adapter for Modulex: a
Modulex-managed server lifecycle, OpenTelemetry context propagation
interceptors, a consistent domain-error-to-status mapping layer, and a
health integration backed by a modulex.Manager's real registered
health/readiness checks.
This is gRPC only — Connect (connectrpc.com/connect) support is out of
scope for this package. See the package doc comment (doc.go) and
docs/planning/grpc-adapter-guide.md
for the full scoping rationale and a worked example.
Usage
import (
googlegrpc "google.golang.org/grpc"
healthpb "google.golang.org/grpc/health/grpc_health_v1"
"github.com/mediusfy/modulex"
modulexgrpc "github.com/mediusfy/modulex/grpc"
)
// Composition root, typically inside a module's Init:
grpcServer := googlegrpc.NewServer(modulexgrpc.ServerOptions(myErrorMapping)...)
myservicepb.RegisterMyServiceServer(grpcServer, myServiceImpl)
healthpb.RegisterHealthServer(grpcServer, modulexgrpc.NewHealthServer(mgr))
listener, err := net.Listen("tcp", ":50051")
if err != nil {
return err
}
server, err := modulexgrpc.NewServer(grpcServer, listener)
if err != nil {
return err
}
// server implements modulex.Starter and modulex.Stopper — register it (or a
// module that delegates Start/Stop to it) with the Manager so
// StartModules/StopModules own the gRPC server's lifecycle.
Behavior
Serveradapts a*grpc.Serverintomodulex.Starter/modulex.Stopper:Startserves in the background;Stopperforms a bounded graceful shutdown (GracefulStop, falling back to a hardStopif it doesn't complete before the configured timeout or ctx's deadline).TraceUnaryClientInterceptor/TraceUnaryServerInterceptorand their streaming counterparts propagate the active OpenTelemetry trace context via gRPC metadata, usingotel.GetTextMapPropagator()— the same mechanism thenatsandrabbitmqadapters use for message headers. The composition root must callotel.SetTextMapPropagator(...)at startup (e.g.propagation.TraceContext{}); the global default is a no-op.UnaryServerErrorInterceptormaps domain errors tostatus.Errorusing a caller-suppliedErrorMapping;TranslateErrorconverts a client-received status error back into one of this package's sentinel errors (ErrNotFound,ErrInvalidInput, etc.) for use witherrors.Is.HealthServerimplementsgrpc_health_v1.HealthServerby evaluating aHealthChecker's (e.g. a*modulex.Manager's) actual registered health/readiness checks on every call.
Testing
The adapter's tests use google.golang.org/grpc/test/bufconn for in-memory
client/server round trips — no network or external service is required.
go test ./grpc/...
Documentation
¶
Overview ¶
Package grpc provides an optional gRPC topology adapter for Modulex: a Modulex-managed server lifecycle, OpenTelemetry context propagation interceptors, a consistent domain-error-to-status mapping layer, and a health integration that reports a modulex.Manager's real registered health/readiness checks over the standard gRPC health-checking protocol.
Scoping: gRPC only, not Connect ¶
ADR-0031's roadmap item 3 calls for "gRPC and/or Connect" topology adapters. This package implements gRPC only. connectrpc.com/connect is not a dependency of this module today, and adding it would be a genuinely new dependency; google.golang.org/grpc and google.golang.org/protobuf, by contrast, are already indirect dependencies of this module (pulled in transitively by the OTLP gRPC trace exporter), so depending on them directly from this package does not add a new dependency to the module's dependency graph — it only promotes an existing one. A future package (e.g. modulex/connect) can add Connect support without changing anything here.
Core module boundary ¶
This package is a sibling of chi, nats, rabbitmq, watermill, and otel: an optional integration package that the core github.com/mediusfy/modulex package does not import. A consumer that imports only the core package never pulls in google.golang.org/grpc or google.golang.org/protobuf; see scripts/check-consumer-boundary.sh.
What this package provides ¶
- Server: adapts a *grpc.Server into modulex.Starter and modulex.Stopper so a modulex.Manager owns starting it and gracefully stopping it, mirroring how the httpx package's Serve function manages a *http.Server — except Server implements the lifecycle interfaces directly, since a gRPC server's shutdown (GracefulStop with a bounded fallback to Stop) is a self-contained concern independent of task supervision.
- Trace context propagation: TraceUnaryClientInterceptor / TraceUnaryServerInterceptor inject and extract the active OpenTelemetry trace context via gRPC metadata, using the same otel.GetTextMapPropagator() pattern the nats and rabbitmq adapters use for message headers. TraceStreamClientInterceptor / TraceStreamServerInterceptor do the same for streaming RPCs.
- Consistent error mapping: ErrorMapping lets a service map its domain errors to gRPC status codes; UnaryServerErrorInterceptor applies that mapping to every unary RPC's returned error. TranslateError converts a status error received by a client back into one of this package's sentinel errors, so callers can use errors.Is instead of inspecting codes.Code directly. There is no streaming error-mapping interceptor: a streaming RPC's error can surface from any Send/Recv call across the life of the stream rather than from one return value, so a single "wrap the terminal error" interceptor would not have a consistent place to run. Callers of a streaming client should call TranslateError explicitly on the error returned from Recv/CloseSend.
- Health integration: HealthServer implements grpc_health_v1.HealthServer by evaluating a modulex.Manager's actual registered health and readiness checks on every call, instead of reporting a hardcoded SERVING status.
What this package does not provide ¶
Client registration is inherently tied to a specific generated service stub, so there is no generic "bind a port to a gRPC client" helper here. DialOptions bundles the reusable pieces (trace propagation and error translation) that any client dial should apply; the client adapter itself — dialing a *grpc.ClientConn and calling the generated stub — belongs next to the domain port it implements. See examples/deployment/notification/adapters/grpc_client.go for a worked example, and docs/planning/grpc-adapter-guide.md for the full writeup.
Index ¶
- Constants
- Variables
- func DefaultErrorMapping(err error) codes.Code
- func DialOptions() []googlegrpc.DialOption
- func ServerOptions(mapping ErrorMapping) []googlegrpc.ServerOption
- func TraceStreamClientInterceptor() googlegrpc.StreamClientInterceptor
- func TraceStreamServerInterceptor() googlegrpc.StreamServerInterceptor
- func TraceUnaryClientInterceptor() googlegrpc.UnaryClientInterceptor
- func TraceUnaryServerInterceptor() googlegrpc.UnaryServerInterceptor
- func TranslateError(err error) error
- func UnaryClientErrorInterceptor() googlegrpc.UnaryClientInterceptor
- func UnaryServerErrorInterceptor(mapping ErrorMapping) googlegrpc.UnaryServerInterceptor
- type ErrorMapping
- type HealthChecker
- type HealthServer
- type HealthServerOption
- type Server
- type ServerOption
Constants ¶
const DefaultShutdownTimeout = 10 * time.Second
DefaultShutdownTimeout bounds how long Stop waits for (*google.golang.org/grpc.Server).GracefulStop to finish before forcing an immediate (*google.golang.org/grpc.Server).Stop. It is used by NewServer when no WithShutdownTimeout option is given.
const ReadinessService = "readiness"
ReadinessService is the health-check "service" name (per the standard gRPC health-checking protocol's HealthCheckRequest.Service field) that reports a HealthChecker's readiness checks instead of its liveness checks. The empty string reports liveness, matching the protocol's convention that an empty service name means "the server's overall health."
Variables ¶
var ( // ErrInvalidInput corresponds to codes.InvalidArgument. ErrInvalidInput = errors.New("grpc: invalid input") // ErrNotFound corresponds to codes.NotFound. ErrNotFound = errors.New("grpc: not found") // ErrAlreadyExists corresponds to codes.AlreadyExists. ErrAlreadyExists = errors.New("grpc: already exists") // ErrPermissionDenied corresponds to codes.PermissionDenied. ErrPermissionDenied = errors.New("grpc: permission denied") // ErrUnauthenticated corresponds to codes.Unauthenticated. ErrUnauthenticated = errors.New("grpc: unauthenticated") ErrUnavailable = errors.New("grpc: unavailable") // ErrDeadlineExceeded corresponds to codes.DeadlineExceeded. ErrDeadlineExceeded = errors.New("grpc: deadline exceeded") // ErrCanceled corresponds to codes.Canceled. ErrCanceled = errors.New("grpc: canceled") // ErrInternal is returned for any other non-OK code, including // codes.Internal itself and codes.Unknown. ErrInternal = errors.New("grpc: internal error") )
Sentinel errors a client can compare against with errors.Is after calling TranslateError on an error returned by a gRPC call. Each corresponds to a commonly used gRPC status code; see TranslateError's doc comment for the full code table.
Functions ¶
func DefaultErrorMapping ¶
DefaultErrorMapping is a conservative, domain-agnostic fallback: it maps context cancellation and deadline errors to their gRPC equivalents, and everything else to codes.Internal. This package cannot see a specific service's domain errors (e.g. "not found", "invalid input"), so a service should supply its own ErrorMapping that checks its own sentinel errors first and falls back to DefaultErrorMapping — see the notification example's grpcErrorMapping in examples/deployment/notification/grpc_module.go.
func DialOptions ¶
func DialOptions() []googlegrpc.DialOption
DialOptions returns the reusable grpc.DialOption set every client dial created for a Modulex-composed service should apply: trace-context propagation (unary and streaming) and consistent error translation (unary). Callers append their own transport credentials and any service-specific options — DialOptions never sets credentials, since that is a security-sensitive choice the caller must make explicitly (e.g. insecure.NewCredentials() for a private network, or a real TLS configuration otherwise):
conn, err := grpc.NewClient(target, append(
[]grpc.DialOption{grpc.WithTransportCredentials(insecure.NewCredentials())},
modulexgrpc.DialOptions()...,
)...)
func ServerOptions ¶
func ServerOptions(mapping ErrorMapping) []googlegrpc.ServerOption
ServerOptions returns the reusable grpc.ServerOption set every server hosting a Modulex-composed service should apply: trace-context extraction (unary and streaming) and consistent error mapping (unary), using mapping to convert domain errors to status codes. If mapping is nil, DefaultErrorMapping is used.
grpcServer := grpc.NewServer(modulexgrpc.ServerOptions(myErrorMapping)...)
func TraceStreamClientInterceptor ¶
func TraceStreamClientInterceptor() googlegrpc.StreamClientInterceptor
TraceStreamClientInterceptor injects the active OpenTelemetry trace context into the outgoing metadata used to open the stream. Context propagation for streaming RPCs is scoped to the stream's opening context only: unlike a unary call, a stream has no single point after which new trace context could be attached, so (as with the standard gRPC and OpenTelemetry gRPC instrumentation conventions) the trace context reflects the caller's context at Stream-open time.
func TraceStreamServerInterceptor ¶
func TraceStreamServerInterceptor() googlegrpc.StreamServerInterceptor
TraceStreamServerInterceptor extracts a trace context carried in the metadata that opened the stream and makes it available from the wrapped ServerStream's Context, so a handler that starts a span from ss.Context() continues the client's trace.
func TraceUnaryClientInterceptor ¶
func TraceUnaryClientInterceptor() googlegrpc.UnaryClientInterceptor
TraceUnaryClientInterceptor injects the active OpenTelemetry trace context from ctx into the outgoing gRPC metadata of every unary call, using otel.GetTextMapPropagator() — the same propagator the nats and rabbitmq adapters use for message headers.
func TraceUnaryServerInterceptor ¶
func TraceUnaryServerInterceptor() googlegrpc.UnaryServerInterceptor
TraceUnaryServerInterceptor extracts a trace context carried in the incoming gRPC metadata (as injected by TraceUnaryClientInterceptor, or any other W3C-trace-context-compatible client) and merges it into the handler's context, so a server-side span started from that context continues the client's trace.
func TranslateError ¶
TranslateError converts an error returned by a gRPC client call into one of this package's sentinel errors, so a caller can write errors.Is(err, grpc.ErrNotFound) instead of inspecting status.Code directly. The original status message is preserved via %w-wrapping.
Code table:
codes.InvalidArgument -> ErrInvalidInput codes.NotFound -> ErrNotFound codes.AlreadyExists -> ErrAlreadyExists codes.PermissionDenied -> ErrPermissionDenied codes.Unauthenticated -> ErrUnauthenticated codes.Unavailable -> ErrUnavailable codes.DeadlineExceeded -> ErrDeadlineExceeded codes.Canceled -> ErrCanceled anything else -> ErrInternal
TranslateError returns nil for a nil error or a status with codes.OK, and returns err unchanged if it does not carry a gRPC status at all (e.g. a local dial error).
func UnaryClientErrorInterceptor ¶
func UnaryClientErrorInterceptor() googlegrpc.UnaryClientInterceptor
UnaryClientErrorInterceptor applies TranslateError to the error returned by every unary RPC, so a caller using this interceptor gets sentinel-error translation automatically instead of having to call TranslateError at every call site.
func UnaryServerErrorInterceptor ¶
func UnaryServerErrorInterceptor(mapping ErrorMapping) googlegrpc.UnaryServerInterceptor
UnaryServerErrorInterceptor converts a non-nil error returned by a unary handler into a status.Error using mapping, so every RPC on the server returns the same consistent error shape instead of a bespoke internal error. If mapping is nil, DefaultErrorMapping is used.
An error that already carries a gRPC status (for example, one returned by a nested gRPC client call the handler made and passed straight through) is left unchanged rather than re-wrapped, so its original code is preserved.
Types ¶
type ErrorMapping ¶
ErrorMapping maps a domain error to the gRPC status code that best describes it. A caller implements this once per service, checking its own domain sentinel errors with errors.Is/As, and passes it to UnaryServerErrorInterceptor or ServerOptions.
See docs/planning/grpc-adapter-guide.md for the mapping table used by the notification example (service.ErrEmptyMessage -> codes.InvalidArgument, falling back to DefaultErrorMapping for everything else).
type HealthChecker ¶
type HealthChecker interface {
// HealthChecks returns the currently registered liveness checks.
HealthChecks() map[string]func(context.Context) error
// ReadinessChecks returns the currently registered readiness checks.
ReadinessChecks() map[string]func(context.Context) error
}
HealthChecker is the subset of modulex.Registry (and modulex.Manager, which implements the full Registry) that HealthServer needs to answer health-check requests from real, currently-registered checks rather than a hardcoded status.
type HealthServer ¶
type HealthServer struct {
healthpb.UnimplementedHealthServer
// contains filtered or unexported fields
}
HealthServer implements grpc_health_v1.HealthServer by evaluating a HealthChecker's registered health and readiness checks on every call, instead of reporting a hardcoded SERVING status. Register it with:
healthpb.RegisterHealthServer(grpcServer, grpcadapter.NewHealthServer(mgr))
where mgr is a *modulex.Manager (or anything else implementing HealthChecker).
Service name convention ¶
- "" (empty) or any name not recognized below evaluates the registered liveness checks (modulex.Registry.RegisterHealthCheck).
- ReadinessService ("readiness") evaluates the registered readiness checks (modulex.Registry.RegisterReadinessCheck).
Unlike the standard library's google.golang.org/grpc/health.Server (which requires a caller to push status updates via SetServingStatus), HealthServer never goes stale: every Check and Watch tick re-runs the checker's actual check functions, so the reported status always reflects the Manager's real state at the moment of the call.
func NewHealthServer ¶
func NewHealthServer(checker HealthChecker, opts ...HealthServerOption) *HealthServer
NewHealthServer creates a HealthServer backed by checker.
func (*HealthServer) Check ¶
func (h *HealthServer) Check(ctx context.Context, req *healthpb.HealthCheckRequest) (*healthpb.HealthCheckResponse, error)
Check implements grpc_health_v1.HealthServer.
func (*HealthServer) Watch ¶
func (h *HealthServer) Watch(req *healthpb.HealthCheckRequest, stream healthpb.Health_WatchServer) error
Watch implements grpc_health_v1.HealthServer. It sends the current status immediately, then re-evaluates on an interval (see WithWatchInterval), sending a new message only when the status changes, until the stream's context is done.
type HealthServerOption ¶
type HealthServerOption func(*HealthServer)
HealthServerOption configures a HealthServer during construction.
func WithWatchInterval ¶
func WithWatchInterval(d time.Duration) HealthServerOption
WithWatchInterval overrides how often Watch re-evaluates checks while a client is streaming. The default is 5 seconds.
type Server ¶
type Server struct {
// contains filtered or unexported fields
}
Server adapts a *googlegrpc.Server into modulex.Starter and modulex.Stopper so a modulex.Manager's lifecycle owns starting it and gracefully stopping it — the gRPC analogue of what the httpx package's Serve function does for a *http.Server.
Ownership ¶
Server owns exactly the serve loop and its shutdown. It does not own:
- The *googlegrpc.Server's construction: the caller builds it (with whatever ServerOptions, interceptors, and credentials it needs — see ServerOptions for a reusable bundle) and registers services on it before passing it to NewServer. Server never calls a registration function itself.
- The net.Listener's creation: the caller supplies an already-bound listener (typically via net.Listen), so bind failures surface to the caller before NewServer is ever called, not from Start.
Lifecycle ¶
Start begins Serve in a background goroutine and returns immediately (matching how Manager.StartModules expects Starter.Start to behave: it starts background work, it does not block for the server's lifetime). Stop performs a graceful shutdown bounded by the configured (or ctx's) deadline: it calls GracefulStop and waits for in-flight RPCs to finish. If that does not complete before the bound is reached, Stop calls Stop on the underlying *googlegrpc.Server (which unblocks GracefulStop by closing the listener and aborting all pending RPCs) and returns an error explaining that the shutdown was forced. Stop always waits for the Serve goroutine to fully exit before returning, so a caller can rely on the listener being closed and all resources released once Stop returns, regardless of which path was taken.
Stop is idempotent: calling it more than once returns the result of the first call without repeating the shutdown. Calling Stop before Start is also safe and returns immediately, after closing the listener passed to NewServer so it is never leaked even if Start is never called.
The forced fallback's real bound ¶
Closing the listener and transport (what the forced Stop call does) cancels the context of every in-flight RPC. That bounds shutdown for any handler that behaves correctly — i.e. one that returns once its context is canceled, per the standard gRPC/Go handler contract (the same assumption net/http's Shutdown makes about handlers checking r.Context().Done()). It does not, and cannot, bound a handler that blocks on something unrelated to its own context and never checks it: Go has no supported way to force a running goroutine to stop from the outside. A handler that ignores context cancellation entirely can still block Stop indefinitely — that is a bug in the handler, not a gap this package (or grpc-go itself) can close for it.
func NewServer ¶
func NewServer(grpcServer *googlegrpc.Server, listener net.Listener, opts ...ServerOption) (*Server, error)
NewServer creates a Modulex-managed lifecycle wrapper around grpcServer, serving on listener once Start is called.
grpcServer and listener must not be nil. Register all services (and the health service, if desired — see NewHealthServer) on grpcServer before calling NewServer, since Server never registers services itself.
func (*Server) Start ¶
Start implements modulex.Starter. It begins serving in a background goroutine and returns immediately without blocking module initialization. A Serve error (other than the expected error returned after Stop closes the server) is captured and surfaced from Stop's return value.
Start must be called at most once; subsequent calls return an error without launching another serving goroutine.
type ServerOption ¶
type ServerOption func(*Server)
ServerOption configures a Server during construction.
func WithServerLogger ¶
func WithServerLogger(logger *slog.Logger) ServerOption
WithServerLogger sets the logger used to report a forced shutdown or a Serve error. If not provided, or if nil, slog.Default() is used.
func WithShutdownTimeout ¶
func WithShutdownTimeout(d time.Duration) ServerOption
WithShutdownTimeout overrides DefaultShutdownTimeout for the bound Stop waits on GracefulStop before forcing an immediate Stop. A non-positive duration disables the internal bound entirely, so Stop then waits only on the ctx passed to it (if that ctx has no deadline either, Stop can block until GracefulStop completes on its own).
Source Files