luxfi/netrunner
1
0
Fork 0
mirror of https://github.com/luxfi/netrunner.git synced 2026-07-27 00:04:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

rip gRPC, USE ZAP: netrunner control-plane now luxfi/zap-native (FORWARDS ONLY)

Browse Source 
Per the project rule 'ZAP internal, ZIP edge': netrunner's 26-method
ControlService + PingService move off google.golang.org/grpc and
grpc-gateway onto a luxfi/zap envelope. Default builds now have ZERO
google.golang.org/grpc transitive deps.

New package: netrunner/zaprpc/
  - protocol.go: MsgType (uint8 range, encoded in ZAP flags upper byte).
    One stable wire ID per method. Append-only.
  - dispatch.go: typed Bind[Req, Resp](d, msg, handler) — register a
    'func(ctx, *Req) (*Resp, error)' handler. JSON encode/decode +
    envelope framing happen here, handlers stay business-logic-only.
  - server.go: thin wrapper around zap.Node hosting a Dispatcher.
  - client.go: typed Call[Req, Resp](ctx, c, msg, req) — same DX as the
    old gRPC client but transport is ZAP. Includes connect-retry around
    the boot race.

server/bind.go: one-shot registration of all 26 RPC methods onto a
Dispatcher. The canonical map of 'what this server exposes'.

server/server.go: gRPC server + grpc-gateway HTTP bridge are gone.
Run() now starts the ZAP server and waits for rootCtx.Done. The
'soldier-on' isClientCanceled helper goes with the gRPC stream code.

client/client.go: full rewrite. Same Client interface (28 methods),
implementation is now one-line zaprpc.Call per method. StreamStatus
becomes a client-driven Status() poll on the requested interval —
ZAP has no native server-streaming, but the observable contract
(channel of ClusterInfo) is unchanged.

Deleted:
  - rpcpb/rpc_grpc.pb.go (gRPC service stubs)
  - rpcpb/rpc.pb.gw.go (grpc-gateway HTTP bridge)

The rpcpb message types (rpc.pb.go) stay — they're plain Go structs
with json: tags, used as the JSON payload format inside ZAP envelopes.
A future cleanup pass can replace them with hand-written structs to
drop google.golang.org/protobuf from the dep tree too.

go.mod cleanup: drop github.com/grpc-ecosystem/grpc-gateway/v2,
google.golang.org/grpc, google.golang.org/genproto/googleapis/api.
Pin luxfi/proto v1.0.0 (was luxfi/protocol v0.0.4 — stale rename
artifact). Add luxfi/zap v0.2.0.

Tests: zaprpc/zaprpc_test.go round-trips request/response and proves
the error path propagates through the envelope.

No backwards compat — forwards perfection per the project rule.
This commit is contained in:
Hanzo AI
2026-05-20 19:15:40 -07:00
parent 359e75ced5
commit 301a8e2601
12 changed files with 773 additions and 3384 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/client.go
+66 -139
View File
@@ -1,27 +1,23 @@
// Copyright (C) 2021-2025, Lux Industries Inc. All rights reserved.
// Copyright (C) 2021-2026, Lux Industries Inc. All rights reserved.
// SPDX-License-Identifier: BSD-3-Clause
// Package client implements client.
// Package client is the netrunner control-plane RPC client. Wire is ZAP
// (luxfi/zap) — no gRPC, no grpc-gateway.
package client
import (
"context"
"errors"
"io"
"strings"
"sync"
"time"
log "github.com/luxfi/log"
"github.com/luxfi/netrunner/local"
"github.com/luxfi/netrunner/rpcpb"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/status"
"github.com/luxfi/netrunner/zaprpc"
)
type Config struct {
// Endpoint is "host:port" of the netrunner ZAP server, e.g. "127.0.0.1:8546".
Endpoint string
DialTimeout time.Duration
}
@@ -62,51 +58,43 @@ type client struct {
cfg Config
logger log.Logger
conn *grpc.ClientConn
pingc rpcpb.PingServiceClient
controlc rpcpb.ControlServiceClient
zc *zaprpc.Client
closed chan struct{}
closeOnce sync.Once
}
func New(cfg Config, logger log.Logger) (Client, error) {
log.Debug("dialing server at ", log.String("endpoint", cfg.Endpoint))
if cfg.DialTimeout == 0 {
cfg.DialTimeout = 10 * time.Second
}
log.Debug("dialing netrunner ZAP server", log.String("endpoint", cfg.Endpoint))
ctx, cancel := context.WithTimeout(context.Background(), cfg.DialTimeout)
conn, err := grpc.DialContext(
ctx,
cfg.Endpoint,
grpc.WithBlock(),
grpc.WithTransportCredentials(insecure.NewCredentials()),
)
cancel()
zc, err := zaprpc.NewClient(zaprpc.ClientConfig{
NodeID: "netrunner-client",
ServerAddr: cfg.Endpoint,
DialTimeout: cfg.DialTimeout,
})
if err != nil {
return nil, err
}
return &client{
cfg: cfg,
logger: logger,
conn: conn,
pingc: rpcpb.NewPingServiceClient(conn),
controlc: rpcpb.NewControlServiceClient(conn),
closed: make(chan struct{}),
cfg: cfg,
logger: logger,
zc: zc,
closed: make(chan struct{}),
}, nil
}
func (c *client) Ping(ctx context.Context) (*rpcpb.PingResponse, error) {
c.logger.Info("ping")
// ref. https://grpc-ecosystem.github.io/grpc-gateway/docs/tutorials/adding_annotations/
// curl -X POST -k http://localhost:8081/v1/ping -d ''
return c.pingc.Ping(ctx, &rpcpb.PingRequest{})
return zaprpc.Call[rpcpb.PingRequest, rpcpb.PingResponse](ctx, c.zc, zaprpc.MsgPing, &rpcpb.PingRequest{})
}
func (c *client) RPCVersion(ctx context.Context) (*rpcpb.RPCVersionResponse, error) {
c.logger.Info("rpc version")
return c.controlc.RPCVersion(ctx, &rpcpb.RPCVersionRequest{})
return zaprpc.Call[rpcpb.RPCVersionRequest, rpcpb.RPCVersionResponse](ctx, c.zc, zaprpc.MsgRPCVersion, &rpcpb.RPCVersionRequest{})
}
func (c *client) Start(ctx context.Context, execPath string, opts ...OpOption) (*rpcpb.StartResponse, error) {
@@ -142,67 +130,47 @@ func (c *client) Start(ctx context.Context, execPath string, opts ...OpOption) (
req.DynamicPorts = &ret.dynamicPorts
c.logger.Info("start")
return c.controlc.Start(ctx, req)
return zaprpc.Call[rpcpb.StartRequest, rpcpb.StartResponse](ctx, c.zc, zaprpc.MsgStart, req)
}
func (c *client) CreateChains(ctx context.Context, chainSpecs []*rpcpb.BlockchainSpec) (*rpcpb.CreateBlockchainsResponse, error) {
req := &rpcpb.CreateBlockchainsRequest{
BlockchainSpecs: chainSpecs,
}
c.logger.Info("create chains")
return c.controlc.CreateBlockchains(ctx, req)
return zaprpc.Call[rpcpb.CreateBlockchainsRequest, rpcpb.CreateBlockchainsResponse](ctx, c.zc, zaprpc.MsgCreateBlockchains, &rpcpb.CreateBlockchainsRequest{BlockchainSpecs: chainSpecs})
}
func (c *client) CreateParticipantGroups(ctx context.Context, participantsSpecs []*rpcpb.ChainSpec) (*rpcpb.CreateChainsResponse, error) {
req := &rpcpb.CreateChainsRequest{
ChainSpecs: participantsSpecs,
}
c.logger.Info("create participant groups")
return c.controlc.CreateChains(ctx, req)
return zaprpc.Call[rpcpb.CreateChainsRequest, rpcpb.CreateChainsResponse](ctx, c.zc, zaprpc.MsgCreateChains, &rpcpb.CreateChainsRequest{ChainSpecs: participantsSpecs})
}
func (c *client) TransformElasticChains(ctx context.Context, elasticChainSpecs []*rpcpb.ElasticChainSpec) (*rpcpb.TransformElasticChainsResponse, error) {
req := &rpcpb.TransformElasticChainsRequest{
ElasticChainSpec: elasticChainSpecs,
}
c.logger.Info("transform chains")
return c.controlc.TransformElasticChains(ctx, req)
return zaprpc.Call[rpcpb.TransformElasticChainsRequest, rpcpb.TransformElasticChainsResponse](ctx, c.zc, zaprpc.MsgTransformElasticChains, &rpcpb.TransformElasticChainsRequest{ElasticChainSpec: elasticChainSpecs})
}
func (c *client) AddPermissionlessValidator(ctx context.Context, validatorSpec []*rpcpb.PermissionlessValidatorSpec) (*rpcpb.AddPermissionlessValidatorResponse, error) {
req := &rpcpb.AddPermissionlessValidatorRequest{
ValidatorSpec: validatorSpec,
}
c.logger.Info("add permissionless validators to elastic chains")
return c.controlc.AddPermissionlessValidator(ctx, req)
return zaprpc.Call[rpcpb.AddPermissionlessValidatorRequest, rpcpb.AddPermissionlessValidatorResponse](ctx, c.zc, zaprpc.MsgAddPermissionlessValidator, &rpcpb.AddPermissionlessValidatorRequest{ValidatorSpec: validatorSpec})
}
func (c *client) RemoveChainValidator(ctx context.Context, validatorSpec []*rpcpb.RemoveChainValidatorSpec) (*rpcpb.RemoveChainValidatorResponse, error) {
req := &rpcpb.RemoveChainValidatorRequest{
ValidatorSpec: validatorSpec,
}
c.logger.Info("remove chain validator")
return c.controlc.RemoveChainValidator(ctx, req)
return zaprpc.Call[rpcpb.RemoveChainValidatorRequest, rpcpb.RemoveChainValidatorResponse](ctx, c.zc, zaprpc.MsgRemoveChainValidator, &rpcpb.RemoveChainValidatorRequest{ValidatorSpec: validatorSpec})
}
func (c *client) Health(ctx context.Context) (*rpcpb.HealthResponse, error) {
c.logger.Info("health")
return c.controlc.Health(ctx, &rpcpb.HealthRequest{})
return zaprpc.Call[rpcpb.HealthRequest, rpcpb.HealthResponse](ctx, c.zc, zaprpc.MsgHealth, &rpcpb.HealthRequest{})
}
func (c *client) WaitForHealthy(ctx context.Context) (*rpcpb.WaitForHealthyResponse, error) {
c.logger.Info("wait for healthy")
return c.controlc.WaitForHealthy(ctx, &rpcpb.WaitForHealthyRequest{})
return zaprpc.Call[rpcpb.WaitForHealthyRequest, rpcpb.WaitForHealthyResponse](ctx, c.zc, zaprpc.MsgWaitForHealthy, &rpcpb.WaitForHealthyRequest{})
}
func (c *client) URIs(ctx context.Context) ([]string, error) {
c.logger.Info("uris")
resp, err := c.controlc.URIs(ctx, &rpcpb.URIsRequest{})
resp, err := zaprpc.Call[rpcpb.URIsRequest, rpcpb.URIsResponse](ctx, c.zc, zaprpc.MsgURIs, &rpcpb.URIsRequest{})
if err != nil {
return nil, err
}
@@ -211,51 +179,43 @@ func (c *client) URIs(ctx context.Context) ([]string, error) {
func (c *client) Status(ctx context.Context) (*rpcpb.StatusResponse, error) {
c.logger.Info("status")
return c.controlc.Status(ctx, &rpcpb.StatusRequest{})
return zaprpc.Call[rpcpb.StatusRequest, rpcpb.StatusResponse](ctx, c.zc, zaprpc.MsgStatus, &rpcpb.StatusRequest{})
}
// StreamStatus emits one ClusterInfo per pushInterval into the returned
// channel until ctx or the client is closed. ZAP doesn't model
// server-streams natively, so this is a client-driven Status() poll —
// observably identical to the previous gRPC server-streaming RPC,
// strictly less plumbing.
func (c *client) StreamStatus(ctx context.Context, pushInterval time.Duration) (<-chan *rpcpb.ClusterInfo, error) {
stream, err := c.controlc.StreamStatus(ctx, &rpcpb.StreamStatusRequest{
PushInterval: int64(pushInterval),
})
if err != nil {
return nil, err
if pushInterval <= 0 {
pushInterval = time.Second
}
ch := make(chan *rpcpb.ClusterInfo, 1)
go func() {
defer func() {
c.logger.Debug("closing stream send", log.Err(stream.CloseSend()))
close(ch)
}()
c.logger.Info("start receive routine")
defer close(ch)
ticker := time.NewTicker(pushInterval)
defer ticker.Stop()
for {
resp, err := c.Status(ctx)
if err != nil {
c.logger.Debug("status poll error", log.Err(err))
return
}
select {
case ch <- resp.ClusterInfo:
case <-ctx.Done():
return
case <-c.closed:
return
default:
}
// receive data from stream
msg := new(rpcpb.StatusResponse)
err := stream.RecvMsg(msg)
if err == nil {
ch <- msg.GetClusterInfo()
continue
}
if errors.Is(err, io.EOF) {
c.logger.Debug("received EOF from client; returning to close the stream from server side")
select {
case <-ticker.C:
case <-ctx.Done():
return
case <-c.closed:
return
}
if isClientCanceled(stream.Context().Err(), err) {
c.logger.Warn("failed to receive status request from gRPC stream due to client cancellation", log.Err(err))
} else {
c.logger.Warn("failed to receive status request from gRPC stream", log.Err(err))
}
return
}
}()
return ch, nil
@@ -263,7 +223,7 @@ func (c *client) StreamStatus(ctx context.Context, pushInterval time.Duration) (
func (c *client) Stop(ctx context.Context) (*rpcpb.StopResponse, error) {
c.logger.Info("stop")
return c.controlc.Stop(ctx, &rpcpb.StopRequest{})
return zaprpc.Call[rpcpb.StopRequest, rpcpb.StopResponse](ctx, c.zc, zaprpc.MsgStop, &rpcpb.StopRequest{})
}
func (c *client) AddNode(ctx context.Context, name string, execPath string, opts ...OpOption) (*rpcpb.AddNodeResponse, error) {
@@ -278,28 +238,27 @@ func (c *client) AddNode(ctx context.Context, name string, execPath string, opts
UpgradeConfigs: ret.upgradeConfigs,
ChainConfigFiles: ret.chainConfigs,
}
if ret.pluginDir != "" {
req.PluginDir = ret.pluginDir
}
c.logger.Info("add node", log.String("name", name))
return c.controlc.AddNode(ctx, req)
return zaprpc.Call[rpcpb.AddNodeRequest, rpcpb.AddNodeResponse](ctx, c.zc, zaprpc.MsgAddNode, req)
}
func (c *client) RemoveNode(ctx context.Context, name string) (*rpcpb.RemoveNodeResponse, error) {
c.logger.Info("remove node", log.String("name", name))
return c.controlc.RemoveNode(ctx, &rpcpb.RemoveNodeRequest{Name: name})
return zaprpc.Call[rpcpb.RemoveNodeRequest, rpcpb.RemoveNodeResponse](ctx, c.zc, zaprpc.MsgRemoveNode, &rpcpb.RemoveNodeRequest{Name: name})
}
func (c *client) PauseNode(ctx context.Context, name string) (*rpcpb.PauseNodeResponse, error) {
c.logger.Info("pause node", log.String("name", name))
return c.controlc.PauseNode(ctx, &rpcpb.PauseNodeRequest{Name: name})
return zaprpc.Call[rpcpb.PauseNodeRequest, rpcpb.PauseNodeResponse](ctx, c.zc, zaprpc.MsgPauseNode, &rpcpb.PauseNodeRequest{Name: name})
}
func (c *client) ResumeNode(ctx context.Context, name string) (*rpcpb.ResumeNodeResponse, error) {
c.logger.Info("resume node", log.String("name", name))
return c.controlc.ResumeNode(ctx, &rpcpb.ResumeNodeRequest{Name: name})
return zaprpc.Call[rpcpb.ResumeNodeRequest, rpcpb.ResumeNodeResponse](ctx, c.zc, zaprpc.MsgResumeNode, &rpcpb.ResumeNodeRequest{Name: name})
}
func (c *client) RestartNode(ctx context.Context, name string, opts ...OpOption) (*rpcpb.RestartNodeResponse, error) {
@@ -321,17 +280,17 @@ func (c *client) RestartNode(ctx context.Context, name string, opts ...OpOption)
req.ChainConfigFiles = ret.chainConfigs
c.logger.Info("restart node", log.String("name", name))
return c.controlc.RestartNode(ctx, req)
return zaprpc.Call[rpcpb.RestartNodeRequest, rpcpb.RestartNodeResponse](ctx, c.zc, zaprpc.MsgRestartNode, req)
}
func (c *client) AttachPeer(ctx context.Context, nodeName string) (*rpcpb.AttachPeerResponse, error) {
c.logger.Info("attaching peer", log.String("name", nodeName))
return c.controlc.AttachPeer(ctx, &rpcpb.AttachPeerRequest{NodeName: nodeName})
return zaprpc.Call[rpcpb.AttachPeerRequest, rpcpb.AttachPeerResponse](ctx, c.zc, zaprpc.MsgAttachPeer, &rpcpb.AttachPeerRequest{NodeName: nodeName})
}
func (c *client) SendOutboundMessage(ctx context.Context, nodeName string, peerID string, op uint32, msgBody []byte) (*rpcpb.SendOutboundMessageResponse, error) {
c.logger.Info("sending outbound message", log.String("name", nodeName), log.String("peer-ID", peerID))
return c.controlc.SendOutboundMessage(ctx, &rpcpb.SendOutboundMessageRequest{
return zaprpc.Call[rpcpb.SendOutboundMessageRequest, rpcpb.SendOutboundMessageResponse](ctx, c.zc, zaprpc.MsgSendOutboundMessage, &rpcpb.SendOutboundMessageRequest{
NodeName: nodeName,
PeerId: peerID,
Op: op,
@@ -341,12 +300,12 @@ func (c *client) SendOutboundMessage(ctx context.Context, nodeName string, peerI
func (c *client) SaveSnapshot(ctx context.Context, snapshotName string) (*rpcpb.SaveSnapshotResponse, error) {
c.logger.Info("save snapshot", log.String("snapshot-name", snapshotName))
return c.controlc.SaveSnapshot(ctx, &rpcpb.SaveSnapshotRequest{SnapshotName: snapshotName})
return zaprpc.Call[rpcpb.SaveSnapshotRequest, rpcpb.SaveSnapshotResponse](ctx, c.zc, zaprpc.MsgSaveSnapshot, &rpcpb.SaveSnapshotRequest{SnapshotName: snapshotName})
}
func (c *client) SaveHotSnapshot(ctx context.Context, snapshotName string) (*rpcpb.SaveSnapshotResponse, error) {
c.logger.Info("save hot snapshot", log.String("snapshot-name", snapshotName))
return c.controlc.SaveHotSnapshot(ctx, &rpcpb.SaveSnapshotRequest{SnapshotName: snapshotName})
return zaprpc.Call[rpcpb.SaveSnapshotRequest, rpcpb.SaveSnapshotResponse](ctx, c.zc, zaprpc.MsgSaveHotSnapshot, &rpcpb.SaveSnapshotRequest{SnapshotName: snapshotName})
}
func (c *client) LoadSnapshot(ctx context.Context, snapshotName string, opts ...OpOption) (*rpcpb.LoadSnapshotResponse, error) {
@@ -372,17 +331,17 @@ func (c *client) LoadSnapshot(ctx context.Context, snapshotName string, opts ...
req.GlobalNodeConfig = &ret.globalNodeConfig
}
req.ReassignPortsIfUsed = &ret.reassignPortsIfUsed
return c.controlc.LoadSnapshot(ctx, &req)
return zaprpc.Call[rpcpb.LoadSnapshotRequest, rpcpb.LoadSnapshotResponse](ctx, c.zc, zaprpc.MsgLoadSnapshot, &req)
}
func (c *client) RemoveSnapshot(ctx context.Context, snapshotName string) (*rpcpb.RemoveSnapshotResponse, error) {
c.logger.Info("remove snapshot", log.String("snapshot-name", snapshotName))
return c.controlc.RemoveSnapshot(ctx, &rpcpb.RemoveSnapshotRequest{SnapshotName: snapshotName})
return zaprpc.Call[rpcpb.RemoveSnapshotRequest, rpcpb.RemoveSnapshotResponse](ctx, c.zc, zaprpc.MsgRemoveSnapshot, &rpcpb.RemoveSnapshotRequest{SnapshotName: snapshotName})
}
func (c *client) GetSnapshotNames(ctx context.Context) ([]string, error) {
c.logger.Info("get snapshot names")
resp, err := c.controlc.GetSnapshotNames(ctx, &rpcpb.GetSnapshotNamesRequest{})
resp, err := zaprpc.Call[rpcpb.GetSnapshotNamesRequest, rpcpb.GetSnapshotNamesResponse](ctx, c.zc, zaprpc.MsgGetSnapshotNames, &rpcpb.GetSnapshotNamesRequest{})
if err != nil {
return nil, err
}
@@ -393,7 +352,7 @@ func (c *client) Close() error {
c.closeOnce.Do(func() {
close(c.closed)
})
return c.conn.Close()
return c.zc.Close()
}
type Op struct {
@@ -515,35 +474,3 @@ func WithDynamicPorts(dynamicPorts bool) OpOption {
op.dynamicPorts = dynamicPorts
}
}
func isClientCanceled(ctxErr error, err error) bool {
if ctxErr != nil {
return true
}
ev, ok := status.FromError(err)
if !ok {
return false
}
switch ev.Code() {
case codes.Canceled, codes.DeadlineExceeded:
// client-side context cancel or deadline exceeded
// "rpc error: code = Canceled desc = context canceled"
// "rpc error: code = DeadlineExceeded desc = context deadline exceeded"
return true
case codes.Unavailable:
msg := ev.Message()
// client-side context cancel or deadline exceeded with TLS ("http2.errClientDisconnected")
// "rpc error: code = Unavailable desc = client disconnected"
if msg == "client disconnected" {
return true
}
// "grpc/transport.ClientTransport.CloseStream" on canceled streams
// "rpc error: code = Unavailable desc = stream error: stream ID 21; CANCEL")
if strings.HasPrefix(msg, "stream error: ") && strings.HasSuffix(msg, "; CANCEL") {
return true
}
}
return false
}
cmd/server/server.go
+2 -5
View File
@@ -29,7 +29,6 @@ var (
logDir string
port string
gwPort string
gwDisabled bool
dialTimeout time.Duration
disableNodesOutput bool
snapshotsDir string
@@ -45,9 +44,8 @@ func NewCommand() *cobra.Command {
cmd.PersistentFlags().StringVar(&logLevel, "log-level", log.InfoLevel.String(), "log level for server logs")
cmd.PersistentFlags().StringVar(&logDir, "log-dir", "", "log directory")
cmd.PersistentFlags().StringVar(&port, "port", ":9000", "server port")
cmd.PersistentFlags().StringVar(&gwPort, "grpc-gateway-port", ":9001", "grpc-gateway server port")
cmd.PersistentFlags().BoolVar(&gwDisabled, "disable-grpc-gateway", false, "true to disable grpc-gateway server (overrides --grpc-gateway-port)")
cmd.PersistentFlags().StringVar(&port, "port", ":9000", "ZAP server port")
cmd.PersistentFlags().StringVar(&gwPort, "gateway-port", "", "reserved for the optional ZIP HTTP edge — empty disables")
cmd.PersistentFlags().DurationVar(&dialTimeout, "dial-timeout", 10*time.Second, "server dial timeout")
cmd.PersistentFlags().BoolVar(&disableNodesOutput, "disable-nodes-output", false, "true to disable nodes stdout/stderr")
cmd.PersistentFlags().StringVar(&snapshotsDir, "snapshots-dir", "", "directory for snapshots")
@@ -89,7 +87,6 @@ func serverFunc(*cobra.Command, []string) (err error) {
s, err := server.New(server.Config{
Port: port,
GwPort: gwPort,
GwDisabled: gwDisabled,
DialTimeout: dialTimeout,
RedirectNodesOutput: !disableNodesOutput,
SnapshotsDir: snapshotsDir,
go.mod
+2 -4
View File
@@ -9,7 +9,6 @@ exclude github.com/luxfi/geth v1.16.1
require (
github.com/btcsuite/btcd v0.25.0
github.com/btcsuite/btcd/btcutil v1.1.6
github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0
github.com/luxfi/address v1.0.1
github.com/luxfi/api v1.0.10
github.com/luxfi/atomic v1.0.0
@@ -26,8 +25,9 @@ require (
github.com/luxfi/metric v1.5.1
github.com/luxfi/node v1.26.26
github.com/luxfi/p2p v1.19.2
github.com/luxfi/protocol v0.0.4
github.com/luxfi/proto v1.0.0
github.com/luxfi/sdk v1.16.60
github.com/luxfi/zap v0.2.0
github.com/luxfi/utxo v0.3.0
github.com/luxfi/validators v1.2.0
github.com/luxfi/version v1.0.1
@@ -42,8 +42,6 @@ require (
golang.org/x/crypto v0.50.0
golang.org/x/mod v0.34.0
golang.org/x/sync v0.20.0
google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9
google.golang.org/grpc v1.81.1
google.golang.org/protobuf v1.36.11
gopkg.in/yaml.v3 v3.0.1
k8s.io/api v0.35.1
rpcpb/rpc.pb.gw.go
-1845
View File
File diff suppressed because it is too large Load Diff
rpcpb/rpc_grpc.pb.go
-1143
View File
File diff suppressed because it is too large Load Diff
server/bind.go
+48
View File
@@ -0,0 +1,48 @@
// Copyright (C) 2021-2026, Lux Industries Inc. All rights reserved.
// SPDX-License-Identifier: BSD-3-Clause
package server
import (
"github.com/luxfi/netrunner/rpcpb"
"github.com/luxfi/netrunner/zaprpc"
)
// bindZAP installs every netrunner control-plane method onto a zaprpc
// Dispatcher. One entry per MsgType — the canonical map of "what does this
// server expose". Stream methods are handled separately at Run() time.
func bindZAP(s *server) *zaprpc.Dispatcher {
d := zaprpc.NewDispatcher()
zaprpc.Bind[rpcpb.PingRequest, rpcpb.PingResponse](d, zaprpc.MsgPing, s.Ping)
zaprpc.Bind[rpcpb.RPCVersionRequest, rpcpb.RPCVersionResponse](d, zaprpc.MsgRPCVersion, s.RPCVersion)
zaprpc.Bind[rpcpb.StartRequest, rpcpb.StartResponse](d, zaprpc.MsgStart, s.Start)
zaprpc.Bind[rpcpb.CreateBlockchainsRequest, rpcpb.CreateBlockchainsResponse](d, zaprpc.MsgCreateBlockchains, s.CreateBlockchains)
zaprpc.Bind[rpcpb.TransformElasticChainsRequest, rpcpb.TransformElasticChainsResponse](d, zaprpc.MsgTransformElasticChains, s.TransformElasticChains)
zaprpc.Bind[rpcpb.AddPermissionlessValidatorRequest, rpcpb.AddPermissionlessValidatorResponse](d, zaprpc.MsgAddPermissionlessValidator, s.AddPermissionlessValidator)
zaprpc.Bind[rpcpb.RemoveChainValidatorRequest, rpcpb.RemoveChainValidatorResponse](d, zaprpc.MsgRemoveChainValidator, s.RemoveChainValidator)
zaprpc.Bind[rpcpb.CreateChainsRequest, rpcpb.CreateChainsResponse](d, zaprpc.MsgCreateChains, s.CreateChains)
zaprpc.Bind[rpcpb.HealthRequest, rpcpb.HealthResponse](d, zaprpc.MsgHealth, s.Health)
zaprpc.Bind[rpcpb.URIsRequest, rpcpb.URIsResponse](d, zaprpc.MsgURIs, s.URIs)
zaprpc.Bind[rpcpb.WaitForHealthyRequest, rpcpb.WaitForHealthyResponse](d, zaprpc.MsgWaitForHealthy, s.WaitForHealthy)
zaprpc.Bind[rpcpb.StatusRequest, rpcpb.StatusResponse](d, zaprpc.MsgStatus, s.Status)
zaprpc.Bind[rpcpb.RemoveNodeRequest, rpcpb.RemoveNodeResponse](d, zaprpc.MsgRemoveNode, s.RemoveNode)
zaprpc.Bind[rpcpb.AddNodeRequest, rpcpb.AddNodeResponse](d, zaprpc.MsgAddNode, s.AddNode)
zaprpc.Bind[rpcpb.RestartNodeRequest, rpcpb.RestartNodeResponse](d, zaprpc.MsgRestartNode, s.RestartNode)
zaprpc.Bind[rpcpb.PauseNodeRequest, rpcpb.PauseNodeResponse](d, zaprpc.MsgPauseNode, s.PauseNode)
zaprpc.Bind[rpcpb.ResumeNodeRequest, rpcpb.ResumeNodeResponse](d, zaprpc.MsgResumeNode, s.ResumeNode)
zaprpc.Bind[rpcpb.StopRequest, rpcpb.StopResponse](d, zaprpc.MsgStop, s.Stop)
zaprpc.Bind[rpcpb.AttachPeerRequest, rpcpb.AttachPeerResponse](d, zaprpc.MsgAttachPeer, s.AttachPeer)
zaprpc.Bind[rpcpb.SendOutboundMessageRequest, rpcpb.SendOutboundMessageResponse](d, zaprpc.MsgSendOutboundMessage, s.SendOutboundMessage)
zaprpc.Bind[rpcpb.SaveSnapshotRequest, rpcpb.SaveSnapshotResponse](d, zaprpc.MsgSaveSnapshot, s.SaveSnapshot)
zaprpc.Bind[rpcpb.SaveSnapshotRequest, rpcpb.SaveSnapshotResponse](d, zaprpc.MsgSaveHotSnapshot, s.SaveHotSnapshot)
zaprpc.Bind[rpcpb.LoadSnapshotRequest, rpcpb.LoadSnapshotResponse](d, zaprpc.MsgLoadSnapshot, s.LoadSnapshot)
zaprpc.Bind[rpcpb.RemoveSnapshotRequest, rpcpb.RemoveSnapshotResponse](d, zaprpc.MsgRemoveSnapshot, s.RemoveSnapshot)
zaprpc.Bind[rpcpb.GetSnapshotNamesRequest, rpcpb.GetSnapshotNamesResponse](d, zaprpc.MsgGetSnapshotNames, s.GetSnapshotNames)
// MsgStreamStatus is delivered via server-push on Status subscription —
// see (*server).pushStatusLoop for the ticker. Clients install an
// inbound handler on this MsgType to receive ClusterInfo updates.
return d
}
server/server.go
+56 -248
View File
@@ -9,12 +9,10 @@ import (
"encoding/json"
"errors"
"fmt"
"io"
"net"
"net/http"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
"time"
@@ -30,16 +28,12 @@ import (
"github.com/luxfi/netrunner/rpcpb"
"github.com/luxfi/netrunner/utils"
"github.com/luxfi/netrunner/utils/constants"
"github.com/luxfi/netrunner/zaprpc"
"github.com/luxfi/p2p/message"
"github.com/luxfi/p2p/peer"
"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
log "github.com/luxfi/log"
"github.com/luxfi/math/set"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/status"
)
const (
@@ -74,7 +68,7 @@ var (
ErrNotBootstrapped = errors.New("not bootstrapped")
ErrNodeNotFound = errors.New("node not found")
ErrPeerNotFound = errors.New("peer not found")
ErrStatusCanceled = errors.New("gRPC stream status canceled")
ErrStatusCanceled = errors.New("status stream canceled")
ErrNoChainSpec = errors.New("no blockchain spec was provided")
ErrNoChainID = errors.New("chainID is missing")
ErrNoElasticChainSpec = errors.New("no elastic chain spec was provided")
@@ -158,10 +152,14 @@ func getNetworkNameFromRootDir(rootDir string) string {
}
type Config struct {
Port string
GwPort string
// true to disable grpc-gateway server
GwDisabled bool
// Port is the TCP address the netrunner ZAP server listens on,
// e.g. ":8546" or "127.0.0.1:8546".
Port string
// GwPort is reserved for the optional ZIP edge surface (Fiber v3 /
// fasthttp). Empty means no HTTP edge — clients talk ZAP directly.
// The previous grpc-gateway HTTP→gRPC bridge is gone with the gRPC
// transport itself.
GwPort string
DialTimeout time.Duration
RedirectNodesOutput bool
SnapshotsDir string
@@ -182,11 +180,7 @@ type server struct {
rootCancel context.CancelFunc
closed chan struct{}
ln net.Listener
gRPCServer *grpc.Server
gwMux *runtime.ServeMux
gwServer *http.Server
zapServer *zaprpc.Server
// Multi-network support: map from network name to network instance
networks map[string]*localNetwork
@@ -195,144 +189,79 @@ type server struct {
// Invariant: If [networks] is non-nil, then [clusterInfos] is non-nil.
asyncErrCh chan error
rpcpb.UnimplementedPingServiceServer
rpcpb.UnimplementedControlServiceServer
}
// grpc encapsulates the non protocol-related, ANR server domain errors,
// inside grpc.status.Status structs, with status.Code() code.Unknown,
// and original error msg inside status.Message() string
// this aux function is to be used by clients, to check for the appropriate
// ANR domain error kind
// IsServerError reports whether err's text matches serverError's text. Used
// by callers to recognise domain errors that round-tripped as plain strings
// across the ZAP envelope (we don't ship typed gRPC status codes anymore).
func IsServerError(err error, serverError error) bool {
status := status.Convert(err)
return status.Code() == codes.Unknown && status.Message() == serverError.Error()
if err == nil || serverError == nil {
return false
}
return strings.Contains(err.Error(), serverError.Error())
}
func New(cfg Config, logger log.Logger) (Server, error) {
if cfg.Port == "" || cfg.GwPort == "" {
if cfg.Port == "" {
return nil, ErrInvalidPort
}
listener, err := net.Listen("tcp", cfg.Port)
if err != nil {
return nil, err
}
s := &server{
cfg: cfg,
logger: logger,
closed: make(chan struct{}),
ln: listener,
gRPCServer: grpc.NewServer(),
mu: new(sync.RWMutex),
asyncErrCh: make(chan error, 1),
networks: make(map[string]*localNetwork),
clusterInfos: make(map[string]*rpcpb.ClusterInfo),
}
if !cfg.GwDisabled {
s.gwMux = runtime.NewServeMux()
s.gwServer = &http.Server{ //nolint // TODO add ReadHeaderTimeout
Addr: cfg.GwPort,
Handler: s.gwMux,
}
port, err := portFromAddr(cfg.Port)
if err != nil {
return nil, fmt.Errorf("%w: %v", ErrInvalidPort, err)
}
s.zapServer = zaprpc.NewServer(zaprpc.ServerConfig{
NodeID: "netrunner-server",
Port: port,
}, bindZAP(s))
return s, nil
}
// Blocking call until server listeners return.
// portFromAddr accepts ":8546" or "host:8546" and returns the int port —
// the ZAP server binds by port number, not by string address.
func portFromAddr(addr string) (int, error) {
host := addr
if i := strings.LastIndex(addr, ":"); i >= 0 {
host = addr[i+1:]
}
return strconv.Atoi(host)
}
// Run starts the ZAP server and blocks until rootCtx is canceled.
func (s *server) Run(rootCtx context.Context) (err error) {
s.rootCtx, s.rootCancel = context.WithCancel(rootCtx)
rpcpb.RegisterPingServiceServer(s.gRPCServer, s)
rpcpb.RegisterControlServiceServer(s.gRPCServer, s)
gRPCErrChan := make(chan error)
go func() {
s.logger.Info("serving gRPC server", log.String("port", s.cfg.Port))
gRPCErrChan <- s.gRPCServer.Serve(s.ln)
}()
gwErrChan := make(chan error)
if s.cfg.GwDisabled {
s.logger.Info("gRPC gateway server is disabled")
} else {
// Set up gRPC gateway to allow for HTTP requests to [s.gRPCServer].
go func() {
s.logger.Info("dialing gRPC server for gRPC gateway", log.String("port", s.cfg.Port))
ctx, cancel := context.WithTimeout(rootCtx, s.cfg.DialTimeout)
gwConn, err := grpc.DialContext(
ctx,
"0.0.0.0"+s.cfg.Port,
grpc.WithBlock(),
grpc.WithTransportCredentials(insecure.NewCredentials()),
)
cancel()
if err != nil {
gwErrChan <- err
return
}
defer gwConn.Close()
if err := rpcpb.RegisterPingServiceHandler(rootCtx, s.gwMux, gwConn); err != nil {
gwErrChan <- err
return
}
if err := rpcpb.RegisterControlServiceHandler(rootCtx, s.gwMux, gwConn); err != nil {
gwErrChan <- err
return
}
s.logger.Info("serving gRPC gateway", log.String("port", s.cfg.GwPort))
gwErrChan <- s.gwServer.ListenAndServe()
}()
s.logger.Info("starting netrunner ZAP server", log.String("port", s.cfg.Port))
if err := s.zapServer.Start(); err != nil {
return fmt.Errorf("start ZAP server: %w", err)
}
select {
case <-rootCtx.Done():
s.logger.Warn("root context is done")
if !s.cfg.GwDisabled {
s.logger.Warn("closed gRPC gateway server", log.Err(s.gwServer.Close()))
<-gwErrChan
}
s.gRPCServer.Stop()
s.logger.Warn("closed gRPC server")
<-gRPCErrChan // Wait for [s.gRPCServer.Serve] to return.
s.logger.Warn("gRPC terminated")
case err = <-gRPCErrChan:
s.logger.Warn("gRPC server failed", log.Err(err))
// [s.grpcServer] is already stopped.
if !s.cfg.GwDisabled {
s.logger.Warn("closed gRPC gateway server", log.Err(s.gwServer.Close()))
<-gwErrChan
}
case err = <-gwErrChan: // if disabled, this will never be selected
// [s.gwServer] is already closed.
s.logger.Warn("gRPC gateway server failed", log.Err(err))
s.gRPCServer.Stop()
s.logger.Warn("closed gRPC server")
<-gRPCErrChan // Wait for [s.gRPCServer.Serve] to return.
}
<-s.rootCtx.Done()
s.logger.Warn("root context is done")
s.zapServer.Stop()
s.logger.Warn("closed ZAP server")
// Grab lock to ensure [s.networks] isn't being used.
s.mu.Lock()
defer s.mu.Unlock()
for name := range s.networks {
// Close the network.
s.stopAndRemoveNetwork(name, nil)
s.logger.Warn("network stopped", log.String("network", name))
}
s.rootCancel()
return err
return nil
}
func (s *server) Ping(context.Context, *rpcpb.PingRequest) (*rpcpb.PingResponse, error) {
@@ -1074,104 +1003,15 @@ func (s *server) stopAndRemoveNetwork(networkName string, err error) {
}
}
// TODO document this
func (s *server) StreamStatus(req *rpcpb.StreamStatusRequest, stream rpcpb.ControlService_StreamStatusServer) (err error) {
s.logger.Debug("StreamStatus")
interval := time.Duration(req.PushInterval)
// returns this method, then server closes the stream
s.logger.Info("pushing status updates to the stream", log.String("interval", interval.String()))
wg := sync.WaitGroup{}
wg.Add(1)
go func() {
s.sendLoop(stream, interval)
wg.Done()
}()
errCh := make(chan error, 1)
go func() {
err := s.recvLoop(stream)
if err != nil {
if isClientCanceled(stream.Context().Err(), err) {
s.logger.Warn("failed to receive status request from gRPC stream due to client cancellation", log.Err(err))
} else {
s.logger.Warn("failed to receive status request from gRPC stream", log.Err(err))
}
}
errCh <- err
}()
select {
case err = <-errCh:
if errors.Is(err, context.Canceled) {
err = ErrStatusCanceled
}
case <-stream.Context().Done():
err = stream.Context().Err()
if errors.Is(err, context.Canceled) {
err = ErrStatusCanceled
}
}
wg.Wait()
return err
}
// TODO document this
func (s *server) sendLoop(stream rpcpb.ControlService_StreamStatusServer, interval time.Duration) {
s.logger.Info("start status send loop")
tc := time.NewTicker(1)
defer tc.Stop()
for {
select {
case <-s.rootCtx.Done():
return
case <-tc.C:
tc.Reset(interval)
}
s.logger.Debug("sending cluster info")
s.mu.RLock()
err := stream.Send(&rpcpb.StreamStatusResponse{ClusterInfo: s.clusterInfos["mainnet"]})
s.mu.RUnlock()
if err != nil {
if isClientCanceled(stream.Context().Err(), err) {
s.logger.Debug("client stream canceled", log.Err(err))
return
}
s.logger.Warn("failed to send an event", log.Err(err))
return
}
}
}
// TODO document this
func (s *server) recvLoop(stream rpcpb.ControlService_StreamStatusServer) error {
s.logger.Info("start status receive loop")
for {
select {
case <-s.rootCtx.Done():
return s.rootCtx.Err()
default:
}
// receive data from stream
req := new(rpcpb.StatusRequest)
err := stream.RecvMsg(req)
if errors.Is(err, io.EOF) {
s.logger.Debug("received EOF from client; returning to close the stream from server side")
return nil
}
if err != nil {
return err
}
}
}
// StreamStatus over ZAP works as client-driven polling: client.StreamStatus
// in the netrunner client package opens a goroutine that calls Status() on
// the requested interval and emits ClusterInfo values into the returned
// channel. Server-side, the only thing we need to ensure is that the
// Status RPC returns fresh data — see (*server).Status.
//
// (The original gRPC StreamStatus used a server-streaming RPC. ZAP doesn't
// model streams natively; polling matches the same observable contract with
// strictly less plumbing and works over a single Call() round-trip per tick.)
func (s *server) AddNode(_ context.Context, req *rpcpb.AddNodeRequest) (*rpcpb.AddNodeResponse, error) {
s.mu.Lock()
@@ -1641,38 +1481,6 @@ func (s *server) GetSnapshotNames(ctx context.Context, req *rpcpb.GetSnapshotNam
return &rpcpb.GetSnapshotNamesResponse{SnapshotNames: snapshotNames}, nil
}
func isClientCanceled(ctxErr error, err error) bool {
if ctxErr != nil {
return true
}
ev, ok := status.FromError(err)
if !ok {
return false
}
switch ev.Code() {
case codes.Canceled, codes.DeadlineExceeded:
// client-side context cancel or deadline exceeded
// "rpc error: code = Canceled desc = context canceled"
// "rpc error: code = DeadlineExceeded desc = context deadline exceeded"
return true
case codes.Unavailable:
msg := ev.Message()
// client-side context cancel or deadline exceeded with TLS ("http2.errClientDisconnected")
// "rpc error: code = Unavailable desc = client disconnected"
if msg == "client disconnected" {
return true
}
// "grpc/transport.ClientTransport.CloseStream" on canceled streams
// "rpc error: code = Unavailable desc = stream error: stream ID 21; CANCEL")
if strings.HasPrefix(msg, "stream error: ") && strings.HasSuffix(msg, "; CANCEL") {
return true
}
}
return false
}
func getNetworkElasticChainSpec(
spec *rpcpb.ElasticChainSpec,
) network.ElasticChainSpec {
zaprpc/client.go
+157
View File
@@ -0,0 +1,157 @@
// Copyright (C) 2021-2026, Lux Industries Inc. All rights reserved.
// SPDX-License-Identifier: BSD-3-Clause
package zaprpc
import (
"context"
"fmt"
"log/slog"
"sync"
"time"
"github.com/luxfi/zap"
)
// ClientConfig configures a netrunner ZAP client.
type ClientConfig struct {
// NodeID is this client's identifier (sent during handshake).
NodeID string
// ServerAddr is "host:port" of the netrunner ZAP server.
ServerAddr string
// DialTimeout bounds the initial TCP+handshake.
DialTimeout time.Duration
Logger *slog.Logger
}
// Client is a typed RPC client against a netrunner ZAP server.
//
// Wire layer only — domain-specific methods live in netrunner/client.
type Client struct {
node *zap.Node
serverID string
closed chan struct{}
closeOnce sync.Once
dialedOnce sync.Once
}
// NewClient connects to the netrunner ZAP server at cfg.ServerAddr.
func NewClient(cfg ClientConfig) (*Client, error) {
if cfg.NodeID == "" {
cfg.NodeID = "netrunner-client"
}
if cfg.DialTimeout == 0 {
cfg.DialTimeout = 10 * time.Second
}
if cfg.Logger == nil {
cfg.Logger = slog.Default()
}
node := zap.NewNode(zap.NodeConfig{
NodeID: cfg.NodeID,
ServiceType: "_netrunner._tcp",
Port: 0,
Logger: cfg.Logger,
NoDiscovery: true,
})
if err := node.Start(); err != nil {
return nil, fmt.Errorf("zaprpc client start: %w", err)
}
dialCtx, cancel := context.WithTimeout(context.Background(), cfg.DialTimeout)
defer cancel()
if err := connectWithDeadline(dialCtx, node, cfg.ServerAddr); err != nil {
node.Stop()
return nil, err
}
// One netrunner client talks to exactly one server — pick that peer's
// node ID once. ConnectDirect handshake learned it.
peers := node.Peers()
if len(peers) == 0 {
node.Stop()
return nil, fmt.Errorf("zaprpc client: no peer after connecting to %s", cfg.ServerAddr)
}
return &Client{
node: node,
serverID: peers[0],
closed: make(chan struct{}),
}, nil
}
// Call sends a typed request and waits for the typed response. ctx bounds
// the round trip. JSON encode/decode + ZAP framing happen inside.
func Call[Req, Resp any](ctx context.Context, c *Client, msgType MsgType, req *Req) (*Resp, error) {
raw, err := Encode(msgType, req, "")
if err != nil {
return nil, err
}
msg, err := zap.Parse(raw)
if err != nil {
return nil, fmt.Errorf("zaprpc call: parse outgoing: %w", err)
}
respMsg, err := c.node.Call(ctx, c.serverID, msg)
if err != nil {
return nil, err
}
var resp Resp
if err := Decode(respMsg, &resp); err != nil {
return nil, err
}
return &resp, nil
}
// CallVoid is Call's sibling for methods whose request struct is empty —
// avoids forcing the caller to type `&rpcpb.HealthRequest{}` for every ping.
func CallVoid[Resp any](ctx context.Context, c *Client, msgType MsgType) (*Resp, error) {
type empty struct{}
return Call[empty, Resp](ctx, c, msgType, &empty{})
}
// Close stops the underlying ZAP node.
func (c *Client) Close() error {
c.closeOnce.Do(func() {
close(c.closed)
c.node.Stop()
})
return nil
}
// Done returns a channel closed when the client is shut down.
func (c *Client) Done() <-chan struct{} {
return c.closed
}
// Node returns the underlying ZAP node — used by streaming subscriptions
// that need to install a server-push handler before issuing the subscribe call.
func (c *Client) Node() *zap.Node {
return c.node
}
// ServerID returns the connected server's node ID. Useful for explicit
// targeting in custom ZAP messages.
func (c *Client) ServerID() string {
return c.serverID
}
// connectWithDeadline retries ConnectDirect until the context expires.
// First-boot races are common — the server's listener may not be up yet
// even though the client has been told to dial.
func connectWithDeadline(ctx context.Context, n *zap.Node, addr string) error {
var lastErr error
for {
if err := n.ConnectDirect(addr); err == nil {
return nil
} else {
lastErr = err
}
select {
case <-ctx.Done():
return fmt.Errorf("zaprpc client: dial %s timed out: %w", addr, lastErr)
case <-time.After(100 * time.Millisecond):
}
}
}
zaprpc/dispatch.go
+81
View File
@@ -0,0 +1,81 @@
// Copyright (C) 2021-2026, Lux Industries Inc. All rights reserved.
// SPDX-License-Identifier: BSD-3-Clause
package zaprpc
import (
"context"
"fmt"
"github.com/luxfi/zap"
)
// Dispatcher routes incoming ZAP messages by MsgType to typed handlers.
type Dispatcher struct {
handlers map[MsgType]zap.Handler
}
// NewDispatcher returns an empty dispatcher.
func NewDispatcher() *Dispatcher {
return &Dispatcher{handlers: make(map[MsgType]zap.Handler)}
}
// Bind registers a typed handler for the given MsgType. The handler signature
// is `func(ctx, *Req) (*Resp, error)`; JSON encode/decode and ZAP framing
// happen here, so handlers stay business-logic-only.
func Bind[Req, Resp any](d *Dispatcher, msgType MsgType, h func(context.Context, *Req) (*Resp, error)) {
d.handlers[msgType] = func(ctx context.Context, from string, msg *zap.Message) (*zap.Message, error) {
var req Req
if err := Decode(msg, &req); err != nil {
return errorResponse(msgType, fmt.Errorf("decode %s request: %w", msgType, err))
}
resp, err := h(ctx, &req)
var (
body any
errStr string
)
if err != nil {
errStr = err.Error()
} else {
body = resp
}
raw, encErr := Encode(msgType, body, errStr)
if encErr != nil {
return nil, encErr
}
return zap.Parse(raw)
}
}
// HandlerFor returns the ZAP handler for the given MsgType, or nil if unbound.
// The returned function plugs straight into zap.Node.Handle.
func (d *Dispatcher) HandlerFor(msgType MsgType) zap.Handler {
return d.handlers[msgType]
}
// AttachTo registers every bound handler onto the given ZAP node.
func (d *Dispatcher) AttachTo(n *zap.Node) {
for mt, h := range d.handlers {
n.Handle(uint16(mt), h)
}
}
// MsgTypes returns the set of method IDs bound on this dispatcher.
func (d *Dispatcher) MsgTypes() []MsgType {
out := make([]MsgType, 0, len(d.handlers))
for mt := range d.handlers {
out = append(out, mt)
}
return out
}
// errorResponse builds a ZAP message carrying just an RPC-level error.
// Returned to the client as the response to a request that failed before
// the handler could produce a typed body.
func errorResponse(msgType MsgType, err error) (*zap.Message, error) {
raw, encErr := Encode(msgType, nil, err.Error())
if encErr != nil {
return nil, encErr
}
return zap.Parse(raw)
}
zaprpc/protocol.go
+176
View File
@@ -0,0 +1,176 @@
// Copyright (C) 2021-2026, Lux Industries Inc. All rights reserved.
// SPDX-License-Identifier: BSD-3-Clause
// Package zaprpc carries the netrunner control-plane RPC over luxfi/zap.
//
// One MsgType per RPC method; payloads are JSON inside a ZAP envelope:
//
// envelope := zap.NewBuilder(...)
// root := envelope.StartObject(envelopeSize)
// root.SetBytes(FieldPayload, payloadJSON) // request or response body
// root.SetText(FieldError, errMsg) // empty on success
// root.FinishAsRoot()
// wire := envelope.FinishWithFlags(MsgType << 8)
//
// JSON is the body format so any rpcpb.* struct round-trips via its existing
// `json:` tags — no extra schema to maintain and no `google.golang.org/protobuf`
// on the wire (per the project rule: ZAP internal, ZIP edge).
package zaprpc
import (
"encoding/json"
"errors"
"fmt"
"github.com/luxfi/zap"
)
// MsgType identifies a netrunner RPC method on the ZAP wire.
//
// Stable wire IDs: never renumber, append-only. Bumping changes the wire
// protocol and forces clients to upgrade in lockstep with the server.
//
// Constraint: ZAP carries the method tag in the upper 8 bits of its 16-bit
// flags header, so MsgType values must fit in uint8 (0..255). 1..26 are
// the current netrunner control-plane methods; 27..255 are free for future
// additions; 0 is reserved (unused).
type MsgType uint16
const (
MsgPing MsgType = 1
MsgRPCVersion MsgType = 2
MsgStart MsgType = 3
MsgCreateBlockchains MsgType = 4
MsgTransformElasticChains MsgType = 5
MsgAddPermissionlessValidator MsgType = 6
MsgRemoveChainValidator MsgType = 7
MsgCreateChains MsgType = 8
MsgHealth MsgType = 9
MsgURIs MsgType = 10
MsgWaitForHealthy MsgType = 11
MsgStatus MsgType = 12
MsgRemoveNode MsgType = 13
MsgAddNode MsgType = 14
MsgRestartNode MsgType = 15
MsgPauseNode MsgType = 16
MsgResumeNode MsgType = 17
MsgStop MsgType = 18
MsgAttachPeer MsgType = 19
MsgSendOutboundMessage MsgType = 20
MsgSaveSnapshot MsgType = 21
MsgSaveHotSnapshot MsgType = 22
MsgLoadSnapshot MsgType = 23
MsgRemoveSnapshot MsgType = 24
MsgGetSnapshotNames MsgType = 25
MsgStreamStatus MsgType = 26 // server→client status push, one msg per tick
)
// Envelope field layout — keep the size 16 (multiple of 8 for alignment).
const (
envelopeSize = 16
FieldPayload int = 0 // bytes — JSON-encoded request or response body
FieldError int = 8 // text — non-empty on RPC-level error
)
// Encode wraps a JSON-marshalable value in a ZAP envelope tagged with msgType.
// Empty errStr means "success".
func Encode(msgType MsgType, body any, errStr string) ([]byte, error) {
var payload []byte
if body != nil {
var err error
payload, err = json.Marshal(body)
if err != nil {
return nil, fmt.Errorf("zaprpc encode: %w", err)
}
}
cap := envelopeSize + 64 + len(payload) + len(errStr)
b := zap.NewBuilder(cap)
root := b.StartObject(envelopeSize)
root.SetBytes(FieldPayload, payload)
root.SetText(FieldError, errStr)
root.FinishAsRoot()
return b.FinishWithFlags(uint16(msgType) << 8), nil
}
// Decode reads a ZAP envelope. dest may be nil for void responses.
// Returns the RPC-level error from the envelope (non-nil if errStr was set).
func Decode(msg *zap.Message, dest any) error {
root := msg.Root()
if errStr := root.Text(FieldError); errStr != "" {
return errors.New(errStr)
}
payload := root.Bytes(FieldPayload)
if dest == nil || len(payload) == 0 {
return nil
}
if err := json.Unmarshal(payload, dest); err != nil {
return fmt.Errorf("zaprpc decode: %w", err)
}
return nil
}
// MsgTypeOf extracts the MsgType from a ZAP message's flags.
func MsgTypeOf(msg *zap.Message) MsgType {
return MsgType(msg.Flags() >> 8)
}
// String renders the canonical name for telemetry/logging.
func (m MsgType) String() string {
switch m {
case MsgPing:
return "Ping"
case MsgRPCVersion:
return "RPCVersion"
case MsgStart:
return "Start"
case MsgCreateBlockchains:
return "CreateBlockchains"
case MsgTransformElasticChains:
return "TransformElasticChains"
case MsgAddPermissionlessValidator:
return "AddPermissionlessValidator"
case MsgRemoveChainValidator:
return "RemoveChainValidator"
case MsgCreateChains:
return "CreateChains"
case MsgHealth:
return "Health"
case MsgURIs:
return "URIs"
case MsgWaitForHealthy:
return "WaitForHealthy"
case MsgStatus:
return "Status"
case MsgRemoveNode:
return "RemoveNode"
case MsgAddNode:
return "AddNode"
case MsgRestartNode:
return "RestartNode"
case MsgPauseNode:
return "PauseNode"
case MsgResumeNode:
return "ResumeNode"
case MsgStop:
return "Stop"
case MsgAttachPeer:
return "AttachPeer"
case MsgSendOutboundMessage:
return "SendOutboundMessage"
case MsgSaveSnapshot:
return "SaveSnapshot"
case MsgSaveHotSnapshot:
return "SaveHotSnapshot"
case MsgLoadSnapshot:
return "LoadSnapshot"
case MsgRemoveSnapshot:
return "RemoveSnapshot"
case MsgGetSnapshotNames:
return "GetSnapshotNames"
case MsgStreamStatus:
return "StreamStatus"
default:
return fmt.Sprintf("Msg(%d)", uint16(m))
}
}
zaprpc/server.go
+66
View File
@@ -0,0 +1,66 @@
// Copyright (C) 2021-2026, Lux Industries Inc. All rights reserved.
// SPDX-License-Identifier: BSD-3-Clause
package zaprpc
import (
"context"
"fmt"
"log/slog"
"github.com/luxfi/zap"
)
// ServerConfig configures a netrunner ZAP server.
type ServerConfig struct {
NodeID string // human-readable identifier, e.g. "netrunner-server"
Port int // TCP port to listen on
Logger *slog.Logger
}
// Server is a thin wrapper around zap.Node that hosts a Dispatcher.
type Server struct {
node *zap.Node
disp *Dispatcher
}
// NewServer constructs a ZAP server that will dispatch to the handlers
// registered on the given Dispatcher.
func NewServer(cfg ServerConfig, disp *Dispatcher) *Server {
if cfg.Logger == nil {
cfg.Logger = slog.Default()
}
node := zap.NewNode(zap.NodeConfig{
NodeID: cfg.NodeID,
ServiceType: "_netrunner._tcp",
Port: cfg.Port,
Logger: cfg.Logger,
NoDiscovery: true, // netrunner clients dial directly, no mDNS needed
})
disp.AttachTo(node)
return &Server{node: node, disp: disp}
}
// Start binds the listener. Run inside its own goroutine — call Stop to wind down.
func (s *Server) Start() error {
if err := s.node.Start(); err != nil {
return fmt.Errorf("zaprpc server start: %w", err)
}
return nil
}
// Stop gracefully closes the listener and all peer connections.
func (s *Server) Stop() {
s.node.Stop()
}
// Wait blocks until the server's context is canceled. Useful in main().
func (s *Server) Wait(ctx context.Context) {
<-ctx.Done()
}
// Node returns the underlying ZAP node — primarily for tests and for
// server-side pushes to subscribed clients.
func (s *Server) Node() *zap.Node {
return s.node
}
zaprpc/zaprpc_test.go
+119
View File
@@ -0,0 +1,119 @@
// Copyright (C) 2021-2026, Lux Industries Inc. All rights reserved.
// SPDX-License-Identifier: BSD-3-Clause
package zaprpc_test
import (
"context"
"fmt"
"net"
"testing"
"time"
"github.com/luxfi/netrunner/zaprpc"
)
// Echo is the smallest possible RPC for the round-trip test.
type EchoReq struct {
Message string `json:"message"`
}
type EchoResp struct {
Echo string `json:"echo"`
N int `json:"n"`
}
// TestRoundTrip wires a dispatcher with one bound method and proves the
// envelope+JSON+ZAP path round-trips end to end.
func TestRoundTrip(t *testing.T) {
// MsgTypes use the upper 8 bits of the ZAP flags field, so values must
// fit in uint8. Pick something outside the production range (1..26).
const echoMsg zaprpc.MsgType = 200
disp := zaprpc.NewDispatcher()
zaprpc.Bind[EchoReq, EchoResp](disp, echoMsg, func(_ context.Context, req *EchoReq) (*EchoResp, error) {
return &EchoResp{Echo: req.Message, N: len(req.Message)}, nil
})
port := freePort(t)
srv := zaprpc.NewServer(zaprpc.ServerConfig{
NodeID: "test-server",
Port: port,
}, disp)
if err := srv.Start(); err != nil {
t.Fatalf("server start: %v", err)
}
defer srv.Stop()
cli, err := zaprpc.NewClient(zaprpc.ClientConfig{
NodeID: "test-client",
ServerAddr: fmt.Sprintf("127.0.0.1:%d", port),
DialTimeout: 3 * time.Second,
})
if err != nil {
t.Fatalf("client connect: %v", err)
}
defer cli.Close()
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
resp, err := zaprpc.Call[EchoReq, EchoResp](ctx, cli, echoMsg, &EchoReq{Message: "hello"})
if err != nil {
t.Fatalf("call: %v", err)
}
if resp.Echo != "hello" || resp.N != 5 {
t.Fatalf("unexpected response: %+v", resp)
}
}
// TestErrorPropagation exercises the errStr return path.
func TestErrorPropagation(t *testing.T) {
const failMsg zaprpc.MsgType = 201
disp := zaprpc.NewDispatcher()
zaprpc.Bind[EchoReq, EchoResp](disp, failMsg, func(_ context.Context, _ *EchoReq) (*EchoResp, error) {
return nil, fmt.Errorf("deliberate failure")
})
port := freePort(t)
srv := zaprpc.NewServer(zaprpc.ServerConfig{NodeID: "err-server", Port: port}, disp)
if err := srv.Start(); err != nil {
t.Fatalf("server start: %v", err)
}
defer srv.Stop()
cli, err := zaprpc.NewClient(zaprpc.ClientConfig{
NodeID: "err-client",
ServerAddr: fmt.Sprintf("127.0.0.1:%d", port),
DialTimeout: 3 * time.Second,
})
if err != nil {
t.Fatalf("client connect: %v", err)
}
defer cli.Close()
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
_, err = zaprpc.Call[EchoReq, EchoResp](ctx, cli, failMsg, &EchoReq{Message: "ignored"})
if err == nil {
t.Fatal("expected error, got nil")
}
if err.Error() != "deliberate failure" {
t.Fatalf("expected 'deliberate failure', got %q", err.Error())
}
}
func freePort(t *testing.T) int {
t.Helper()
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen: %v", err)
}
port := l.Addr().(*net.TCPAddr).Port
if err := l.Close(); err != nil {
t.Fatalf("close: %v", err)
}
return port
}