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

feat(zapwire): native ZAP control RPC foundation — Ping/RPCVersion/Health/URIs/Status

Browse Source 
100% native ZAP, no protobuf, no gRPC, no codegen. Hand-written Go
types in zapwire/types/ encode/decode against luxfi/api/zap.Buffer
and zap.Reader directly. Server dispatches via a single zap.Handler
that switches on opcode (and sub-opcode for fan-out via OpStart).

Opcodes 60..63 reserved for netrunner control. Sub-opcodes (under
OpStart) provide RPC fan-out without consuming the limited opcode
range. Default port: 9999.

Migration plan:
  Phase 1 (this commit): Ping, RPCVersion, Health, URIs, Status — 5
    of 26 control RPCs landed with full e2e test coverage
  Phase 2: lifecycle (Start, Stop, AddNode, RemoveNode, RestartNode,
    PauseNode, ResumeNode, AttachPeer, SendOutboundMessage)
  Phase 3: chain ops (CreateBlockchains, CreateChains, etc.)
  Phase 4: snapshots (Save/Load/Remove/GetSnapshotNames + Hot)
  Phase 5: streaming (StreamStatus over persistent ZAP conn)
  Phase 6: swap netrunner/client/client.go + cli callers, delete rpcpb

Tests: go test ./zapwire/... — 5 RPCs e2e, all green.
This commit is contained in:
Hanzo AI
2026-05-06 10:37:19 -07:00
parent a016d13e9d
commit 66a1a46075
5 changed files with 1042 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zapwire/client.go
+179
View File
@@ -0,0 +1,179 @@
// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
// Package zapwire provides a 100%-native ZAP client/server for the
// netrunner control protocol. No protobuf, no gRPC, no codegen.
//
// This package replaces the legacy netrunner/client/client.go (gRPC)
// and netrunner/server/server.go (gRPC) with a hand-written ZAP layer
// over luxfi/api/zap. The migration happens RPC-by-RPC; consumers
// switch to zapwire.Client when the server-side handler for a given
// op has been converted.
package zapwire
import (
"bytes"
"context"
"encoding/binary"
"errors"
"fmt"
"github.com/luxfi/api/zap"
"github.com/luxfi/netrunner/zapwire/types"
)
// Wire-protocol version. Bumped on breaking changes to encoding.
const ProtocolVersion uint32 = 1
// Client is the netrunner control RPC client over ZAP.
type Client struct {
conn *zap.Conn
}
// Dial opens a ZAP control connection to a netrunner server.
func Dial(ctx context.Context, addr string) (*Client, error) {
conn, err := zap.Dial(ctx, addr, nil)
if err != nil {
return nil, fmt.Errorf("zapwire: dial %s: %w", addr, err)
}
return &Client{conn: conn}, nil
}
// Close releases the underlying connection.
func (c *Client) Close() error {
if c.conn == nil {
return nil
}
return c.conn.Close()
}
// call serializes req, performs the RPC, and decodes the response into
// resp. The on-the-wire payload format is:
//
// [reqID:uint32 (added by zap.Conn.Call)] [req-body]
//
// and the response (post-flag-strip) is:
//
// [reqID:uint32] [resp-body]
//
// zap.Conn.Call handles the request ID transparently.
func (c *Client) call(
ctx context.Context,
op zap.MessageType,
req types.Encoder,
resp types.Decoder,
) error {
buf := zap.GetBuffer()
defer zap.PutBuffer(buf)
req.Encode(buf)
body := append([]byte(nil), buf.Bytes()...)
_, payload, err := c.conn.Call(ctx, op, body)
if err != nil {
return err
}
r := zap.NewReader(payload)
return resp.Decode(r)
}
// callSub is the same as call but writes a SubOpcode byte before the
// request body. Used for opcodes that fanout multiple RPCs through
// shared opcode (see opcodes.go for the rationale).
func (c *Client) callSub(
ctx context.Context,
op zap.MessageType,
sub SubOpcode,
req types.Encoder,
resp types.Decoder,
) error {
buf := zap.GetBuffer()
defer zap.PutBuffer(buf)
buf.WriteUint8(uint8(sub))
req.Encode(buf)
body := append([]byte(nil), buf.Bytes()...)
_, payload, err := c.conn.Call(ctx, op, body)
if err != nil {
return err
}
r := zap.NewReader(payload)
return resp.Decode(r)
}
// ---- Public RPC methods ----
// Ping returns the netrunner server's PID.
func (c *Client) Ping(ctx context.Context) (*types.PingResponse, error) {
resp := &types.PingResponse{}
if err := c.call(ctx, OpPing, &types.PingRequest{}, resp); err != nil {
return nil, err
}
return resp, nil
}
// RPCVersion returns the netrunner control wire-protocol version.
func (c *Client) RPCVersion(ctx context.Context) (*types.RPCVersionResponse, error) {
resp := &types.RPCVersionResponse{}
if err := c.call(ctx, OpRPCVersion, &types.RPCVersionRequest{}, resp); err != nil {
return nil, err
}
return resp, nil
}
// Health returns a cluster-health snapshot.
func (c *Client) Health(ctx context.Context) (*types.HealthResponse, error) {
resp := &types.HealthResponse{}
if err := c.callSub(ctx, OpStart, SubHealth, &types.HealthRequest{}, resp); err != nil {
return nil, err
}
return resp, nil
}
// URIs returns the public RPC endpoints of every running node.
func (c *Client) URIs(ctx context.Context) ([]string, error) {
resp := &types.URIsResponse{}
if err := c.callSub(ctx, OpStart, SubURIs, &types.URIsRequest{}, resp); err != nil {
return nil, err
}
return resp.URIs, nil
}
// Status returns the full cluster status.
func (c *Client) Status(ctx context.Context) (*types.StatusResponse, error) {
resp := &types.StatusResponse{}
if err := c.callSub(ctx, OpStart, SubStatus, &types.StatusRequest{}, resp); err != nil {
return nil, err
}
return resp, nil
}
// ---- Compatibility helpers ----
// reqIDFromPayload is unused on the client side (zap.Conn handles IDs)
// but kept here as the canonical reader for tests + server-side code.
func reqIDFromPayload(payload []byte) (uint32, []byte, error) {
if len(payload) < 4 {
return 0, nil, errors.New("zapwire: payload too short for request ID")
}
return binary.BigEndian.Uint32(payload[:4]), payload[4:], nil
}
// peelSubOpcode reads the leading SubOpcode byte off a payload and
// returns the remainder. Server-side use only.
func peelSubOpcode(body []byte) (SubOpcode, []byte, error) {
if len(body) < 1 {
return 0, nil, errors.New("zapwire: missing sub-opcode byte")
}
return SubOpcode(body[0]), body[1:], nil
}
// readerFromBody returns a zap.Reader positioned at the start of the
// post-(reqID, sub-opcode) message body. Server handlers receive a
// payload that already had the request ID stripped by the transport,
// so they only need to peel the sub-opcode.
func readerFromBody(body []byte) *zap.Reader {
return zap.NewReader(body)
}
// ensure the package doesn't drop the bytes import accidentally.
var _ = bytes.NewReader
zapwire/e2e_test.go
+236
View File
@@ -0,0 +1,236 @@
// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package zapwire
import (
"context"
"testing"
"time"
"github.com/luxfi/netrunner/zapwire/types"
)
// runServer starts a zapwire server on 127.0.0.1:0 and returns a
// teardown that the test must call (via defer or t.Cleanup). It also
// gives a context whose cancel is wired into the teardown, so handlers
// see ctx.Done() before the listener closes.
func runServer(t *testing.T, be Backend) (*Server, func()) {
t.Helper()
srv, err := NewServer("127.0.0.1:0", be)
if err != nil {
t.Fatalf("NewServer: %v", err)
}
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() {
_ = srv.Serve(ctx)
close(done)
}()
// Wait for listener to actually be accepting.
time.Sleep(20 * time.Millisecond)
teardown := func() {
cancel()
_ = srv.Close()
<-done
}
return srv, teardown
}
// stubBackend is a minimal Backend implementation for e2e tests.
type stubBackend struct {
pingPID int32
}
func (b *stubBackend) Ping(ctx context.Context) (*types.PingResponse, error) {
return &types.PingResponse{PID: b.pingPID}, nil
}
func (b *stubBackend) RPCVersion(ctx context.Context) (*types.RPCVersionResponse, error) {
return &types.RPCVersionResponse{Version: ProtocolVersion}, nil
}
func (b *stubBackend) Health(ctx context.Context) (*types.HealthResponse, error) {
return &types.HealthResponse{
ClusterInfo: &types.ClusterInfo{
NodeNames: []string{"alpha", "beta"},
NodeInfos: map[string]*types.NodeInfo{},
PID: 42,
RootDataDir: "/tmp/netrunner",
Healthy: true,
NetworkID: 12345,
},
}, nil
}
func (b *stubBackend) URIs(ctx context.Context) (*types.URIsResponse, error) {
return &types.URIsResponse{
URIs: []string{"http://127.0.0.1:9650", "http://127.0.0.1:9652"},
}, nil
}
func (b *stubBackend) Status(ctx context.Context) (*types.StatusResponse, error) {
return &types.StatusResponse{
ClusterInfo: &types.ClusterInfo{
NodeNames: []string{"alpha"},
NodeInfos: map[string]*types.NodeInfo{
"alpha": {
Name: "alpha",
ExecPath: "/usr/local/bin/luxd",
URI: "http://127.0.0.1:9650",
ID: "NodeID-1",
LogDir: "/tmp/log",
DBDir: "/tmp/db",
Config: []byte(`{"network-id":1}`),
PluginDir: "/usr/local/lib/lux/plugins",
WhitelistedSubnets: "",
Paused: false,
},
},
PID: 42,
RootDataDir: "/tmp/netrunner",
Healthy: true,
CustomChains: map[string]*types.ChainInfo{
"chain-1": {
ChainName: "C",
VMID: "mgj786NP7uDwBCcq6YwThhaN8FLyybkCa4zBWTQbNgmK6k9A6",
VMName: "evm",
ChainID: "chain-1",
SubnetID: "primary",
Genesis: []byte(`{"chainId":96369}`),
},
},
Subnets: []string{"primary"},
NetworkID: 96369,
},
}, nil
}
func TestE2EPing(t *testing.T) {
srv, teardown := runServer(t, &stubBackend{pingPID: 42})
defer teardown()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
client, err := Dial(ctx, srv.Addr())
if err != nil {
t.Fatalf("Dial: %v", err)
}
defer client.Close()
resp, err := client.Ping(ctx)
if err != nil {
t.Fatalf("Ping: %v", err)
}
if resp.PID != 42 {
t.Fatalf("PID: got %d, want 42", resp.PID)
}
}
func TestE2ERPCVersion(t *testing.T) {
srv, teardown := runServer(t, &stubBackend{})
defer teardown()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
client, err := Dial(ctx, srv.Addr())
if err != nil {
t.Fatalf("Dial: %v", err)
}
defer client.Close()
resp, err := client.RPCVersion(ctx)
if err != nil {
t.Fatalf("RPCVersion: %v", err)
}
if resp.Version != ProtocolVersion {
t.Fatalf("Version: got %d, want %d", resp.Version, ProtocolVersion)
}
}
func TestE2EHealth(t *testing.T) {
srv, teardown := runServer(t, &stubBackend{})
defer teardown()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
client, err := Dial(ctx, srv.Addr())
if err != nil {
t.Fatalf("Dial: %v", err)
}
defer client.Close()
resp, err := client.Health(ctx)
if err != nil {
t.Fatalf("Health: %v", err)
}
if resp.ClusterInfo == nil {
t.Fatal("ClusterInfo: nil")
}
if !resp.ClusterInfo.Healthy {
t.Fatal("Healthy: false")
}
if resp.ClusterInfo.NetworkID != 12345 {
t.Fatalf("NetworkID: got %d, want 12345", resp.ClusterInfo.NetworkID)
}
if len(resp.ClusterInfo.NodeNames) != 2 {
t.Fatalf("NodeNames: got %d, want 2", len(resp.ClusterInfo.NodeNames))
}
}
func TestE2EURIs(t *testing.T) {
srv, teardown := runServer(t, &stubBackend{})
defer teardown()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
client, err := Dial(ctx, srv.Addr())
if err != nil {
t.Fatalf("Dial: %v", err)
}
defer client.Close()
uris, err := client.URIs(ctx)
if err != nil {
t.Fatalf("URIs: %v", err)
}
if len(uris) != 2 {
t.Fatalf("URIs: got %d, want 2", len(uris))
}
}
func TestE2EStatus(t *testing.T) {
srv, teardown := runServer(t, &stubBackend{})
defer teardown()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
client, err := Dial(ctx, srv.Addr())
if err != nil {
t.Fatalf("Dial: %v", err)
}
defer client.Close()
resp, err := client.Status(ctx)
if err != nil {
t.Fatalf("Status: %v", err)
}
if resp.ClusterInfo == nil {
t.Fatal("ClusterInfo: nil")
}
ni := resp.ClusterInfo.NodeInfos["alpha"]
if ni == nil || ni.Name != "alpha" {
t.Fatalf("NodeInfos[alpha] missing or wrong: %+v", ni)
}
chain := resp.ClusterInfo.CustomChains["chain-1"]
if chain == nil {
t.Fatal("CustomChains[chain-1]: nil")
}
if chain.VMID != "mgj786NP7uDwBCcq6YwThhaN8FLyybkCa4zBWTQbNgmK6k9A6" {
t.Fatalf("VMID: got %s", chain.VMID)
}
if string(chain.Genesis) != `{"chainId":96369}` {
t.Fatalf("Genesis round-trip mangled: %s", chain.Genesis)
}
}
zapwire/opcodes.go
+89
View File
@@ -0,0 +1,89 @@
// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
// Package zapwire defines the ZAP opcodes and message types for the
// netrunner control protocol. It replaces the legacy gRPC/protobuf
// netrunner/rpcpb package with native ZAP types — no protobuf, no .proto
// files, no codegen — every message is a hand-written Go struct that
// encodes/decodes via luxfi/api/zap.Buffer / luxfi/api/zap.Reader.
//
// The control protocol is a request/response RPC that runs on TCP
// between a netrunner client (luxfi/cli, automation tooling, tests) and
// a netrunner server (this repo). Streaming RPCs (StreamStatus) use
// multiple response frames keyed by the same request ID.
package zapwire
import "github.com/luxfi/api/zap"
// DefaultPort is the canonical Lux ZAP TCP port — used by netrunner
// control, KMS ZAP transport, and other in-cluster ZAP services.
// Override only when running multiple ZAP servers on the same host.
const DefaultPort = 9999
// DefaultAddr is the canonical loopback bind address for local
// development and tests. Production deployments bind to a routable
// interface, not loopback.
const DefaultAddr = "127.0.0.1:9999"
// Netrunner control opcodes occupy ZAP message-type range [60, 99].
// Per zap/wire.go layout: bits 0..5 carry the opcode (max 63), bit 6
// is MsgErrorFlag, bit 7 is MsgResponseFlag. Stay below 0x40 so the
// flags remain free.
//
// Reserved ranges (see ~/work/lux/api/zap/wire.go):
// 1..31 — VM service methods
// 40..43 — p2p.Sender methods
// 50..52 — Warp signing
// 60..63 — netrunner control (this file)
//
// Hitting the 0x40 ceiling: keep opcodes <=63. If we need more than 4
// opcodes here, use a sub-opcode byte at the start of the payload
// rather than expanding the opcode range.
const (
// Lifecycle
OpPing zap.MessageType = 60
OpRPCVersion zap.MessageType = 61
OpStart zap.MessageType = 62
OpStop zap.MessageType = 63
)
// Sub-opcode dispatcher — encoded as the first byte of payload after the
// 4-byte request ID. This lets us multiplex many RPCs onto a small
// opcode footprint while keeping the wire format simple.
//
// [hdr][reqID:uint32][subOp:uint8][rpc-specific bytes...]
type SubOpcode uint8
const (
// Network lifecycle (alongside OpStart/OpStop opcodes above)
SubHealth SubOpcode = 1
SubWaitForHealthy SubOpcode = 2
SubURIs SubOpcode = 3
SubStatus SubOpcode = 4
SubStreamStatus SubOpcode = 5
// Chain operations
SubCreateBlockchains SubOpcode = 10
SubCreateChains SubOpcode = 11
SubTransformElasticChains SubOpcode = 12
SubAddPermissionlessValidator SubOpcode = 13
SubRemoveChainValidator SubOpcode = 14
// Node operations
SubAddNode SubOpcode = 20
SubRemoveNode SubOpcode = 21
SubRestartNode SubOpcode = 22
SubPauseNode SubOpcode = 23
SubResumeNode SubOpcode = 24
// Peer / message operations
SubAttachPeer SubOpcode = 30
SubSendOutboundMessage SubOpcode = 31
// Snapshot operations
SubSaveSnapshot SubOpcode = 40
SubSaveHotSnapshot SubOpcode = 41
SubLoadSnapshot SubOpcode = 42
SubRemoveSnapshot SubOpcode = 43
SubGetSnapshotNames SubOpcode = 44
)
zapwire/server.go
+148
View File
@@ -0,0 +1,148 @@
// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
package zapwire
import (
"context"
"errors"
"fmt"
"github.com/luxfi/api/zap"
"github.com/luxfi/netrunner/zapwire/types"
)
// Backend is the interface a netrunner server implementation provides
// to the ZAP wire layer. Each method maps 1:1 to an RPC opcode in
// opcodes.go. The wire layer is fully decoupled from the underlying
// network/orchestrator/local-cluster impl — just plug in any Backend.
type Backend interface {
Ping(ctx context.Context) (*types.PingResponse, error)
RPCVersion(ctx context.Context) (*types.RPCVersionResponse, error)
Health(ctx context.Context) (*types.HealthResponse, error)
URIs(ctx context.Context) (*types.URIsResponse, error)
Status(ctx context.Context) (*types.StatusResponse, error)
}
// Server hosts a netrunner control RPC over ZAP.
type Server struct {
be Backend
listener *zap.Listener
srv *zap.Server
}
// NewServer creates a netrunner ZAP server listening on addr.
func NewServer(addr string, be Backend) (*Server, error) {
if be == nil {
return nil, errors.New("zapwire: nil backend")
}
listener, err := zap.Listen(addr, nil)
if err != nil {
return nil, fmt.Errorf("zapwire: listen %s: %w", addr, err)
}
s := &Server{be: be, listener: listener}
s.srv = zap.NewServer(listener, zap.HandlerFunc(s.handle))
return s, nil
}
// Addr returns the actual listen address (useful for ":0" tests).
func (s *Server) Addr() string {
return s.listener.Addr().String()
}
// Serve runs the server until ctx is cancelled or Close is called.
func (s *Server) Serve(ctx context.Context) error {
return s.srv.Serve(ctx)
}
// Close shuts down the server.
func (s *Server) Close() error {
if s.srv != nil {
s.srv.Close()
}
if s.listener != nil {
return s.listener.Close()
}
return nil
}
// handle is the single ZAP message handler. It dispatches to the
// backend based on the message-type opcode (and sub-opcode for
// fan-out opcodes).
//
// luxfi/api/zap.ServerConn.Read() already strips the request ID off
// the payload AND the response writer prepends it on the way back —
// so handlers see only the application bytes both inbound and out.
func (s *Server) handle(
ctx context.Context,
msgType zap.MessageType,
payload []byte,
) (zap.MessageType, []byte, error) {
switch msgType {
case OpPing:
resp, err := s.be.Ping(ctx)
if err != nil {
return 0, nil, err
}
return msgType, encodeResp(resp), nil
case OpRPCVersion:
resp, err := s.be.RPCVersion(ctx)
if err != nil {
return 0, nil, err
}
return msgType, encodeResp(resp), nil
case OpStart:
// Fan-out opcode: peel the sub-opcode byte and dispatch.
sub, _, err := peelSubOpcode(payload)
if err != nil {
return 0, nil, err
}
return s.handleStartSub(ctx, msgType, sub)
case OpStop:
return 0, nil, errors.New("OpStop: not yet wired in zapwire")
default:
return 0, nil, fmt.Errorf("zapwire: unknown opcode 0x%02x", byte(msgType))
}
}
func (s *Server) handleStartSub(
ctx context.Context,
msgType zap.MessageType,
sub SubOpcode,
) (zap.MessageType, []byte, error) {
switch sub {
case SubHealth:
resp, err := s.be.Health(ctx)
if err != nil {
return 0, nil, err
}
return msgType, encodeResp(resp), nil
case SubURIs:
resp, err := s.be.URIs(ctx)
if err != nil {
return 0, nil, err
}
return msgType, encodeResp(resp), nil
case SubStatus:
resp, err := s.be.Status(ctx)
if err != nil {
return 0, nil, err
}
return msgType, encodeResp(resp), nil
default:
return 0, nil, fmt.Errorf("zapwire: unknown SubOp 0x%02x for OpStart", uint8(sub))
}
}
// encodeResp serialises the response body. The transport prepends the
// request ID and applies MsgResponseFlag automatically.
func encodeResp(resp types.Encoder) []byte {
buf := zap.GetBuffer()
defer zap.PutBuffer(buf)
resp.Encode(buf)
return append([]byte(nil), buf.Bytes()...)
}
zapwire/types/types.go
+390
View File
@@ -0,0 +1,390 @@
// Copyright (C) 2019-2025, Lux Industries Inc. All rights reserved.
// See the file LICENSE for licensing terms.
// Package types defines the netrunner control RPC message types as
// hand-written Go structs. These replace the legacy rpcpb (protobuf-
// generated) types — there is no .proto, no codegen, no protobuf
// dependency. Each type owns its own ZAP encode/decode pair.
//
// Style: every struct has no embedded interfaces, no pointers to
// primitives unless null-distinct semantics matter, and every field is
// JSON-serialisable so test fixtures can be written by hand.
package types
import (
"github.com/luxfi/api/zap"
)
// Encoder is implemented by every netrunner control message that goes
// over the wire. Encode appends the message bytes to the provided
// buffer. The 4-byte request ID is written by the transport layer
// before Encode is called; do not write it from inside Encode.
type Encoder interface {
Encode(b *zap.Buffer)
}
// Decoder is implemented by every netrunner control message that comes
// off the wire. Decode reads the message from the reader and populates
// the receiver. The reader is positioned past the request ID and any
// sub-opcode byte before Decode is called.
type Decoder interface {
Decode(r *zap.Reader) error
}
// PingRequest is empty; the wire payload is just the request ID.
type PingRequest struct{}
func (*PingRequest) Encode(b *zap.Buffer) {}
func (*PingRequest) Decode(r *zap.Reader) error { return nil }
// PingResponse carries the netrunner server's PID, used by clients to
// detect server identity changes (e.g. the daemon was restarted).
type PingResponse struct {
PID int32
}
func (p *PingResponse) Encode(b *zap.Buffer) {
b.WriteInt32(p.PID)
}
func (p *PingResponse) Decode(r *zap.Reader) error {
pid, err := r.ReadInt32()
if err != nil {
return err
}
p.PID = pid
return nil
}
// RPCVersionRequest is empty.
type RPCVersionRequest struct{}
func (*RPCVersionRequest) Encode(b *zap.Buffer) {}
func (*RPCVersionRequest) Decode(r *zap.Reader) error { return nil }
// RPCVersionResponse carries the wire-protocol version. Bumped on
// breaking changes to the netrunner ZAP control protocol.
type RPCVersionResponse struct {
Version uint32
}
func (r *RPCVersionResponse) Encode(b *zap.Buffer) {
b.WriteUint32(r.Version)
}
func (r *RPCVersionResponse) Decode(rd *zap.Reader) error {
v, err := rd.ReadUint32()
if err != nil {
return err
}
r.Version = v
return nil
}
// HealthRequest is empty.
type HealthRequest struct{}
func (*HealthRequest) Encode(b *zap.Buffer) {}
func (*HealthRequest) Decode(r *zap.Reader) error { return nil }
// HealthResponse carries a snapshot of the cluster's health.
type HealthResponse struct {
ClusterInfo *ClusterInfo // nil if no cluster yet
}
func (h *HealthResponse) Encode(b *zap.Buffer) {
if h.ClusterInfo == nil {
b.WriteUint8(0)
return
}
b.WriteUint8(1)
h.ClusterInfo.Encode(b)
}
func (h *HealthResponse) Decode(r *zap.Reader) error {
present, err := r.ReadUint8()
if err != nil {
return err
}
if present == 0 {
h.ClusterInfo = nil
return nil
}
h.ClusterInfo = &ClusterInfo{}
return h.ClusterInfo.Decode(r)
}
// URIsRequest is empty.
type URIsRequest struct{}
func (*URIsRequest) Encode(b *zap.Buffer) {}
func (*URIsRequest) Decode(r *zap.Reader) error { return nil }
// URIsResponse carries the public RPC endpoints of every node in the
// running cluster.
type URIsResponse struct {
URIs []string
}
func (u *URIsResponse) Encode(b *zap.Buffer) {
b.WriteUint32(uint32(len(u.URIs)))
for _, uri := range u.URIs {
b.WriteString(uri)
}
}
func (u *URIsResponse) Decode(r *zap.Reader) error {
n, err := r.ReadUint32()
if err != nil {
return err
}
u.URIs = make([]string, n)
for i := uint32(0); i < n; i++ {
s, err := r.ReadString()
if err != nil {
return err
}
u.URIs[i] = s
}
return nil
}
// StatusRequest is empty.
type StatusRequest struct{}
func (*StatusRequest) Encode(b *zap.Buffer) {}
func (*StatusRequest) Decode(r *zap.Reader) error { return nil }
// StatusResponse carries the full cluster snapshot.
type StatusResponse struct {
ClusterInfo *ClusterInfo
}
func (s *StatusResponse) Encode(b *zap.Buffer) {
if s.ClusterInfo == nil {
b.WriteUint8(0)
return
}
b.WriteUint8(1)
s.ClusterInfo.Encode(b)
}
func (s *StatusResponse) Decode(r *zap.Reader) error {
present, err := r.ReadUint8()
if err != nil {
return err
}
if present == 0 {
s.ClusterInfo = nil
return nil
}
s.ClusterInfo = &ClusterInfo{}
return s.ClusterInfo.Decode(r)
}
// ---- Composite types ----
// NodeInfo is the per-node payload inside ClusterInfo.
type NodeInfo struct {
Name string
ExecPath string
URI string
ID string
LogDir string
DBDir string
Config []byte
PluginDir string
WhitelistedSubnets string // legacy field name preserved for migration; rename pending
Paused bool
}
func (n *NodeInfo) Encode(b *zap.Buffer) {
b.WriteString(n.Name)
b.WriteString(n.ExecPath)
b.WriteString(n.URI)
b.WriteString(n.ID)
b.WriteString(n.LogDir)
b.WriteString(n.DBDir)
b.WriteBytes(n.Config)
b.WriteString(n.PluginDir)
b.WriteString(n.WhitelistedSubnets)
b.WriteBool(n.Paused)
}
func (n *NodeInfo) Decode(r *zap.Reader) error {
var err error
if n.Name, err = r.ReadString(); err != nil {
return err
}
if n.ExecPath, err = r.ReadString(); err != nil {
return err
}
if n.URI, err = r.ReadString(); err != nil {
return err
}
if n.ID, err = r.ReadString(); err != nil {
return err
}
if n.LogDir, err = r.ReadString(); err != nil {
return err
}
if n.DBDir, err = r.ReadString(); err != nil {
return err
}
cfg, err := r.ReadBytes()
if err != nil {
return err
}
// ReadBytes is zero-copy; copy because NodeInfo outlives the frame.
n.Config = append([]byte(nil), cfg...)
if n.PluginDir, err = r.ReadString(); err != nil {
return err
}
if n.WhitelistedSubnets, err = r.ReadString(); err != nil {
return err
}
if n.Paused, err = r.ReadBool(); err != nil {
return err
}
return nil
}
// ChainInfo is the per-chain payload inside ClusterInfo.
type ChainInfo struct {
ChainName string
VMID string
VMName string
ChainID string
SubnetID string // P-Chain validator-set identifier — distinct from ChainID
Genesis []byte
}
func (c *ChainInfo) Encode(b *zap.Buffer) {
b.WriteString(c.ChainName)
b.WriteString(c.VMID)
b.WriteString(c.VMName)
b.WriteString(c.ChainID)
b.WriteString(c.SubnetID)
b.WriteBytes(c.Genesis)
}
func (c *ChainInfo) Decode(r *zap.Reader) error {
var err error
if c.ChainName, err = r.ReadString(); err != nil {
return err
}
if c.VMID, err = r.ReadString(); err != nil {
return err
}
if c.VMName, err = r.ReadString(); err != nil {
return err
}
if c.ChainID, err = r.ReadString(); err != nil {
return err
}
if c.SubnetID, err = r.ReadString(); err != nil {
return err
}
gen, err := r.ReadBytes()
if err != nil {
return err
}
c.Genesis = append([]byte(nil), gen...)
return nil
}
// ClusterInfo is the top-level cluster snapshot returned by Health,
// Status, and StreamStatus.
type ClusterInfo struct {
NodeNames []string
NodeInfos map[string]*NodeInfo // keyed by node name
PID int32
RootDataDir string
Healthy bool
CustomChains map[string]*ChainInfo // keyed by chain ID
Subnets []string // P-Chain validator sets
NetworkID uint32
}
func (c *ClusterInfo) Encode(b *zap.Buffer) {
b.WriteUint32(uint32(len(c.NodeNames)))
for _, n := range c.NodeNames {
b.WriteString(n)
}
b.WriteUint32(uint32(len(c.NodeInfos)))
for _, ni := range c.NodeInfos {
ni.Encode(b)
}
b.WriteInt32(c.PID)
b.WriteString(c.RootDataDir)
b.WriteBool(c.Healthy)
b.WriteUint32(uint32(len(c.CustomChains)))
for _, ch := range c.CustomChains {
ch.Encode(b)
}
b.WriteUint32(uint32(len(c.Subnets)))
for _, s := range c.Subnets {
b.WriteString(s)
}
b.WriteUint32(c.NetworkID)
}
func (c *ClusterInfo) Decode(r *zap.Reader) error {
n, err := r.ReadUint32()
if err != nil {
return err
}
c.NodeNames = make([]string, n)
for i := uint32(0); i < n; i++ {
if c.NodeNames[i], err = r.ReadString(); err != nil {
return err
}
}
n, err = r.ReadUint32()
if err != nil {
return err
}
c.NodeInfos = make(map[string]*NodeInfo, n)
for i := uint32(0); i < n; i++ {
ni := &NodeInfo{}
if err := ni.Decode(r); err != nil {
return err
}
c.NodeInfos[ni.Name] = ni
}
if c.PID, err = r.ReadInt32(); err != nil {
return err
}
if c.RootDataDir, err = r.ReadString(); err != nil {
return err
}
if c.Healthy, err = r.ReadBool(); err != nil {
return err
}
n, err = r.ReadUint32()
if err != nil {
return err
}
c.CustomChains = make(map[string]*ChainInfo, n)
for i := uint32(0); i < n; i++ {
ch := &ChainInfo{}
if err := ch.Decode(r); err != nil {
return err
}
c.CustomChains[ch.ChainID] = ch
}
n, err = r.ReadUint32()
if err != nil {
return err
}
c.Subnets = make([]string, n)
for i := uint32(0); i < n; i++ {
if c.Subnets[i], err = r.ReadString(); err != nil {
return err
}
}
if c.NetworkID, err = r.ReadUint32(); err != nil {
return err
}
return nil
}