grpcmesh

package
v0.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 31, 2026 License: GPL-3.0 Imports: 31 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrRemoteRoutingUnavailable = errors.New("cross-node routing unavailable: no cluster module configured")

ErrRemoteRoutingUnavailable is returned when a cross-node call is attempted but no cluster module has been configured for remote routing.

Functions

func AuthUnaryInterceptor added in v0.4.0 

func AuthUnaryInterceptor() grpc.UnaryServerInterceptor

AuthUnaryInterceptor returns a gRPC unary interceptor (deprecated shim). Use NewAuthInterceptor().UnaryInterceptor() for new code.

func GRPCTransportCredentials added in v0.4.0 

func GRPCTransportCredentials(certFile, keyFile, caCertFile string, mtlsEnabled bool) (credentials.TransportCredentials, error)

GRPCTransportCredentials creates transport credentials for the gRPC server from certificate and key files. If cert and key are both empty, returns (nil, nil) only when MUXCORE_INSECURE_DISABLE_TLS is set to true or 1.

Environment variables: 

MUXCORE_GRPC_TLS_CERT  — path to TLS certificate PEM file
MUXCORE_GRPC_TLS_KEY   — path to TLS private key PEM file
MUXCORE_GRPC_MTLS_ENABLED — if "true" or "1", enable mutual TLS

Types

type AuthInterceptor added in v0.4.0 

type AuthInterceptor struct {
	// contains filtered or unexported fields
}

AuthInterceptor provides a gRPC unary interceptor that extracts caller identity from gRPC metadata and, when an Authorizer is configured, enforces per-method access control.

Behavior:

  • Extracts x-caller-id from metadata and propagates it into the context.
  • When both IdentityProvider and Authorizer are set: extracts identity, checks authorization via Authorizer.Can(), and denies with PermissionDenied if not authorized.
  • When neither is set: logs debug-level warnings for anonymous calls but allows all traffic (backward compatible with dev/testing).

func NewAuthInterceptor added in v0.4.0 

func NewAuthInterceptor() *AuthInterceptor

NewAuthInterceptor creates an auth interceptor with no enforcement. Call SetAuthorizer and SetIdentityProvider to enable enforcement.

func (*AuthInterceptor) SetAuthorizer added in v0.4.0 

func (a *AuthInterceptor) SetAuthorizer(auth contracts.Authorizer)

SetAuthorizer sets the authorizer for permission checks. Pass nil to disable enforcement (dev/testing mode).

func (*AuthInterceptor) SetIdentityProvider added in v0.4.0 

func (a *AuthInterceptor) SetIdentityProvider(ip contracts.IdentityProvider)

SetIdentityProvider sets the identity provider for extracting caller identity.

func (*AuthInterceptor) UnaryInterceptor added in v0.4.0 

func (a *AuthInterceptor) UnaryInterceptor() grpc.UnaryServerInterceptor

UnaryInterceptor returns a gRPC unary server interceptor.

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client implements contracts.ModuleMeshClient. It routes calls: local modules get in-process dispatch; remote modules go over gRPC.

func NewClient 

func NewClient(srv *Server) *Client

NewClient creates a mesh client backed by the given server.

func (*Client) Call 

func (c *Client) Call(ctx context.Context, targetModule, method string, payload []byte) ([]byte, error)

Call dispatches a call to the target module. Local modules are called in-process. Remote modules go over gRPC.

func (*Client) RegisterHandler 

func (c *Client) RegisterHandler(moduleID string, handler contracts.MeshHandler)

RegisterHandler registers a local module to receive calls.

func (*Client) SetAuditLogger added in v0.4.0 

func (c *Client) SetAuditLogger(a contracts.AuditLogger)

SetAuditLogger attaches an audit logger for recording outbound mesh calls.

func (*Client) SetCallPolicy added in v0.4.0 

func (c *Client) SetCallPolicy(policy contracts.CallPolicyProvider)

SetCallPolicy attaches a call policy module for inter-module access control. When set, every Call() checks with the policy provider before dispatching.

func (*Client) SetCluster added in v0.4.0 

func (c *Client) SetCluster(cluster contracts.Cluster)

SetCluster attaches a cluster module for cross-node routing.

func (*Client) SetNodeID added in v0.4.0 

func (c *Client) SetNodeID(id string)

SetNodeID sets the node identifier for audit entries.

func (*Client) SetTransportCredentials added in v0.4.0 

func (c *Client) SetTransportCredentials(creds credentials.TransportCredentials)

SetTransportCredentials sets the TLS credentials used for cross-node gRPC connections. When nil, cross-node routing is disabled — the mesh will only route to local modules. This prevents accidental plaintext fallback in production.

type ConnPool added in v0.4.0 

type ConnPool struct {
	// contains filtered or unexported fields
}

ConnPool caches gRPC connections keyed by target address. Connections are evicted after connPoolIdleTimeout of inactivity. Thread-safe.

Usage: 

pool := NewConnPool(dialOpts...)
conn, err := pool.Get("node.example.com:9090")
pool.Close() // on shutdown

func NewConnPool added in v0.4.0 

func NewConnPool(dialOpts ...grpc.DialOption) *ConnPool

NewConnPool creates a connection pool with the given dial options (TLS credentials, etc.). A background goroutine periodically evicts idle or unhealthy connections.

Keepalive pings are injected automatically to detect dead peers within 20 seconds so that stale pool entries are evicted on the next cleanup tick rather than waiting the full idle timeout.

func (*ConnPool) Close added in v0.4.0 

func (p *ConnPool) Close()

Close drains the pool, closing all cached connections.

func (*ConnPool) Get added in v0.4.0 

func (p *ConnPool) Get(addr string) (*grpc.ClientConn, error)

Get returns a gRPC connection for the given address. If a cached connection exists, it is reused. Otherwise a new connection is dialed and cached. The caller should NOT close the returned connection — the pool manages its lifecycle.

func (*ConnPool) Size added in v0.4.0 

func (p *ConnPool) Size() int

Size returns the current number of cached connections.

type DiscoveryServer 

type DiscoveryServer struct {
	discoveryv1.UnimplementedDiscoveryServiceServer
	// contains filtered or unexported fields
}

DiscoveryServer implements the DiscoveryService gRPC service and manages cluster membership with term-based leader election.

Leader election is deterministic and rank-based: the member with the alphabetically lowest node ID among surviving members is always the leader. Each node independently computes the same leader from the shared member set.

Terms are monotonically increasing counters that prevent stale leadership claims. A node that detects a leader change increments its term and propagates the new term to peers via gRPC metadata on heartbeats and responses.

func NewDiscoveryServer 

func NewDiscoveryServer(nodeID, grpcAddr, httpAddr, joinToken string, moduleIDs func() []string) *DiscoveryServer

NewDiscoveryServer creates a discovery server. The leader is initially empty; it is set when the first node joins or when this node forms a cluster.

func (*DiscoveryServer) Close added in v0.4.0 

func (s *DiscoveryServer) Close()

Close cleanly stops the eviction goroutine.

func (*DiscoveryServer) FindByCapability added in v0.4.0 

func (s *DiscoveryServer) FindByCapability(ctx context.Context, req *discoveryv1.FindByCapabilityRequest) (*discoveryv1.FindByCapabilityResponse, error)

FindByCapability returns modules that advertise the given capability.

func (*DiscoveryServer) FindByRole added in v0.4.0 

func (s *DiscoveryServer) FindByRole(ctx context.Context, req *discoveryv1.FindByRoleRequest) (*discoveryv1.FindByRoleResponse, error)

FindByRole returns modules with the given role.

func (*DiscoveryServer) Heartbeat 

func (s *DiscoveryServer) Heartbeat(ctx context.Context, req *discoveryv1.HeartbeatRequest) (*discoveryv1.HeartbeatResponse, error)

Heartbeat processes a peer heartbeat. It updates lastSeen, syncs the module list, and reconciles the term. If the peer has a higher term, this node recognises the peer's leader. If the peer has a stale term, the response carries the current term so the peer can catch up.

func (*DiscoveryServer) IsLeader added in v0.4.0 

func (s *DiscoveryServer) IsLeader() bool

IsLeader reports whether this node is currently the cluster leader.

func (*DiscoveryServer) Join 

func (s *DiscoveryServer) Join(ctx context.Context, req *discoveryv1.JoinRequest) (*discoveryv1.JoinResponse, error)

Join handles a cluster join request. The first node to join becomes the initial leader. If the joining node has a lower ID than the current leader, leadership transfers to the new node (deterministic rank-based election).

func (*DiscoveryServer) LeaderID added in v0.4.0 

func (s *DiscoveryServer) LeaderID() string

LeaderID returns the current leader's node ID, or "" if none.

func (*DiscoveryServer) Leave 

func (s *DiscoveryServer) Leave(ctx context.Context, req *discoveryv1.LeaveRequest) (*discoveryv1.LeaveResponse, error)

Leave gracefully removes a node from the cluster and triggers leader re-election if the departing node was the leader.

func (*DiscoveryServer) LocalNode 

func (s *DiscoveryServer) LocalNode() *discoveryv1.NodeInfo

func (*DiscoveryServer) Members 

func (s *DiscoveryServer) Members(ctx context.Context, req *discoveryv1.MembersRequest) (*discoveryv1.MembersResponse, error)

func (*DiscoveryServer) MembersSnapshot added in v0.4.0 

func (s *DiscoveryServer) MembersSnapshot() []*discoveryv1.NodeInfo

MembersSnapshot returns a copy of the current member list. Safe for external consumers that need a stable view of the cluster.

func (*DiscoveryServer) RegisterWithGRPC 

func (s *DiscoveryServer) RegisterWithGRPC(srv *grpc.Server)

func (*DiscoveryServer) Resolve added in v0.4.0 

func (s *DiscoveryServer) Resolve(ctx context.Context, req *discoveryv1.ResolveRequest) (*discoveryv1.ResolveResponse, error)

Resolve looks up a single module by ID.

func (*DiscoveryServer) SetConnPool added in v0.4.0 

func (s *DiscoveryServer) SetConnPool(p *ConnPool)

SetConnPool attaches a connection pool for heartbeat and cross-node routing.

func (*DiscoveryServer) SetRegistry added in v0.4.0 

func (s *DiscoveryServer) SetRegistry(reg *registry.Registry)

SetRegistry attaches the module registry for discovery queries.

func (*DiscoveryServer) StartHeartbeatLoop added in v0.4.0 

func (s *DiscoveryServer) StartHeartbeatLoop(ctx context.Context, dialOpts []grpc.DialOption)

StartHeartbeatLoop begins sending periodic heartbeats to all known cluster members. Heartbeats carry the current module list and term so peers learn about module registrations and leadership changes.

dialOpts are the gRPC dial options for connecting to peers (TLS config, etc.). The loop stops when ctx is cancelled.

func (*DiscoveryServer) Term added in v0.4.0 

func (s *DiscoveryServer) Term() uint64

Term returns the current leader term.

func (*DiscoveryServer) Watch 

func (s *DiscoveryServer) Watch(req *discoveryv1.MembersRequest, stream discoveryv1.DiscoveryService_WatchServer) error

type EventServer 

type EventServer struct {
	eventsv1.UnimplementedEventServiceServer
	// contains filtered or unexported fields
}

EventServer implements the EventService gRPC service. It relays events between nodes when NATS is not in use.

func NewEventServer 

func NewEventServer(bus contracts.EventBus) *EventServer

NewEventServer creates an event relay gRPC server.

func (*EventServer) Publish 

func (s *EventServer) Publish(ctx context.Context, req *eventsv1.PublishRequest) (*eventsv1.PublishResponse, error)

Publish receives an event from a remote node and publishes it locally.

func (*EventServer) RegisterWithGRPC 

func (s *EventServer) RegisterWithGRPC(srv *grpc.Server)

RegisterWithGRPC registers this server with a gRPC server.

func (*EventServer) Request 

func (s *EventServer) Request(ctx context.Context, req *eventsv1.RequestEvent) (*eventsv1.Event, error)

Request sends a request event and waits for a single reply.

func (*EventServer) Subscribe 

func (s *EventServer) Subscribe(req *eventsv1.SubscribeRequest, stream eventsv1.EventService_SubscribeServer) error

Subscribe streams events matching the given types from the remote node.

type HealthServer 

type HealthServer struct {
	healthv1.UnimplementedHealthServiceServer
	// contains filtered or unexported fields
}

HealthServer implements the HealthService gRPC service. It exposes module and node health over gRPC, using the core registry.

func NewHealthServer 

func NewHealthServer(reg contracts.Registry) *HealthServer

NewHealthServer creates a health gRPC server.

func (*HealthServer) Check 

func (s *HealthServer) Check(ctx context.Context, req *healthv1.HealthCheckRequest) (*healthv1.HealthCheckResponse, error)

Check returns health status for a node or specific module.

func (*HealthServer) RegisterWithGRPC 

func (s *HealthServer) RegisterWithGRPC(srv *grpc.Server)

RegisterWithGRPC registers this server with a gRPC server.

func (*HealthServer) Watch 

func (s *HealthServer) Watch(req *healthv1.HealthWatchRequest, stream healthv1.HealthService_WatchServer) error

Watch streams health status changes for a set of modules.

type RateLimitInterceptor added in v0.4.0 

type RateLimitInterceptor struct {
	// contains filtered or unexported fields
}

RateLimitInterceptor enforces per-IP request rate limiting on gRPC calls. The limit is configured via MUXCORE_GRPC_RATE_LIMIT (requests per minute). When 0 or unset, the default (1000 req/min) is used.

func NewRateLimitInterceptor added in v0.4.0 

func NewRateLimitInterceptor() *RateLimitInterceptor

NewRateLimitInterceptor creates a rate limiter. Call Close() on shutdown.

func (*RateLimitInterceptor) Close added in v0.4.0 

func (rl *RateLimitInterceptor) Close()

Close stops the cleanup goroutine.

func (*RateLimitInterceptor) StreamInterceptor added in v0.4.0 

func (rl *RateLimitInterceptor) StreamInterceptor() grpc.StreamServerInterceptor

StreamInterceptor returns a gRPC streaming server interceptor.

func (*RateLimitInterceptor) UnaryInterceptor added in v0.4.0 

func (rl *RateLimitInterceptor) UnaryInterceptor() grpc.UnaryServerInterceptor

UnaryInterceptor returns a gRPC unary server interceptor that enforces the rate limit.

type Server 

type Server struct {
	meshv1.UnimplementedModuleMeshServer
	// contains filtered or unexported fields
}

Server implements the ModuleMesh gRPC service and routes calls to local modules.

func NewServer 

func NewServer() *Server

NewServer creates a new mesh server.

func (*Server) Call 

func (s *Server) Call(ctx context.Context, req *meshv1.CallRequest) (*meshv1.CallResponse, error)

Call handles an incoming gRPC Call request.

func (*Server) RegisterHandler 

func (s *Server) RegisterHandler(moduleID string, handler contracts.MeshHandler)

RegisterHandler registers a local module to receive Call requests.

func (*Server) RegisterWithGRPC 

func (s *Server) RegisterWithGRPC(srv *grpc.Server)

RegisterWithGRPC registers this server with a gRPC server.

func (*Server) SetAuditLogger added in v0.4.0 

func (s *Server) SetAuditLogger(a contracts.AuditLogger)

SetAuditLogger attaches an audit logger for recording gRPC calls.

func (*Server) SetClient 

func (s *Server) SetClient(c *Client)

SetClient sets the mesh client for remote call routing.

func (*Server) SetNodeID added in v0.4.0 

func (s *Server) SetNodeID(id string)

SetNodeID sets the node identifier for audit entries.

func (*Server) StreamCall 

func (s *Server) StreamCall(stream meshv1.ModuleMesh_StreamCallServer) error

StreamCall handles bidirectional streaming of module calls. Each incoming CallRequest is dispatched to the target module's handler, and the result is streamed back as a CallResponse. This allows long-running or multi-message interactions over a single gRPC stream.

type StorageServer added in v0.4.0 

type StorageServer struct {
	storagev1.UnimplementedStorageServiceServer
	// contains filtered or unexported fields
}

StorageServer implements the StorageService gRPC service, wrapping core's StorageOrchestrator so sidecar modules can access storage without in-process Fabric access.

When a CallPolicyProvider is configured via SetCallPolicy(), every storage operation checks whether the caller module has the \"storage\" capability. Without a policy, all access is denied (deny-by-default).

func NewStorageServer added in v0.4.0 

func NewStorageServer(store contracts.StorageOrchestrator) *StorageServer

NewStorageServer creates a StorageServer backed by the given orchestrator.

func (*StorageServer) Capabilities added in v0.4.0 

func (s *StorageServer) Capabilities(ctx context.Context, req *storagev1.CapabilitiesRequest) (*storagev1.CapabilitiesResponse, error)

func (*StorageServer) Delete added in v0.4.0 

func (s *StorageServer) Delete(ctx context.Context, req *storagev1.DeleteRequest) (*storagev1.DeleteResponse, error)

func (*StorageServer) Get added in v0.4.0 

func (s *StorageServer) Get(req *storagev1.GetRequest, stream storagev1.StorageService_GetServer) error

Get streams an object from storage to the client.

func (*StorageServer) List added in v0.4.0 

func (s *StorageServer) List(ctx context.Context, req *storagev1.ListRequest) (*storagev1.ListResponse, error)

func (*StorageServer) Put added in v0.4.0 

func (s *StorageServer) Put(stream storagev1.StorageService_PutServer) error

Put receives a client-streamed object and stores it.

func (*StorageServer) RegisterWithGRPC added in v0.4.0 

func (s *StorageServer) RegisterWithGRPC(srv *grpc.Server)

RegisterWithGRPC registers this server with a gRPC server.

func (*StorageServer) SetCallPolicy added in v0.4.0 

func (s *StorageServer) SetCallPolicy(cp contracts.CallPolicyProvider)

SetCallPolicy attaches a call policy provider for capability enforcement. When set, every storage operation checks the caller's capabilities before accessing data. Pass nil to disable (dev/testing only).

func (*StorageServer) Stat added in v0.4.0 

func (s *StorageServer) Stat(ctx context.Context, req *storagev1.StatRequest) (*storagev1.StatResponse, error)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.