diff --git a/qkc/cluster/slave/dispatcher.go b/qkc/cluster/slave/dispatcher.go new file mode 100644 index 000000000000..d2bf05386661 --- /dev/null +++ b/qkc/cluster/slave/dispatcher.go @@ -0,0 +1,169 @@ +// Copyright 2026-2027, QuarkChain. + +package slave + +import ( + "fmt" + "sync" + + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/qkc/cluster/wire" +) + +// Dispatcher routes intra-cluster frames that carry a non-zero cluster_peer_id +// to the corresponding virtual PeerConn. Frames with cluster_peer_id == 0 are +// left for MasterConn to handle. +// +// It owns the registry of PeerConns as a two-layer map: +// +// cluster_peer_id -> branch -> *PeerConn +// +// This matches Python's MasterConnection.v_conn_map and shard.peers layout. +type Dispatcher struct { + mu sync.RWMutex + peers map[uint64]map[uint32]*PeerConn + log log.Logger +} + +// NewDispatcher creates an empty dispatcher. +func NewDispatcher(logger log.Logger) *Dispatcher { + if logger == nil { + logger = log.Root() + } + return &Dispatcher{ + peers: make(map[uint64]map[uint32]*PeerConn), + log: logger, + } +} + +// Register adds an already-created PeerConn to the registry. It returns an +// error if a PeerConn for the same cluster_peer_id and branch already exists. +func (d *Dispatcher) Register(pc *PeerConn) error { + d.mu.Lock() + defer d.mu.Unlock() + + branchMap, ok := d.peers[pc.ClusterPeerID()] + if !ok { + branchMap = make(map[uint32]*PeerConn) + d.peers[pc.ClusterPeerID()] = branchMap + } + if _, exists := branchMap[pc.Branch()]; exists { + return fmt.Errorf("peer connection already exists for cluster_peer_id %d branch %d", pc.ClusterPeerID(), pc.Branch()) + } + branchMap[pc.Branch()] = pc + return nil +} + +// Unregister removes a single PeerConn from the registry. It returns the +// removed PeerConn (if any) without closing it. +func (d *Dispatcher) Unregister(clusterPeerID uint64, branch uint32) *PeerConn { + d.mu.Lock() + defer d.mu.Unlock() + + branchMap, ok := d.peers[clusterPeerID] + if !ok { + return nil + } + pc := branchMap[branch] + delete(branchMap, branch) + if len(branchMap) == 0 { + delete(d.peers, clusterPeerID) + } + return pc +} + +// CreatePeerConns creates and starts one PeerConn per branch for the given +// cluster_peer_id, using masterConn as the transport. Existing branch entries +// are skipped (logged as duplicates), matching Python's behavior. +func (d *Dispatcher) CreatePeerConns(clusterPeerID uint64, branches []uint32, masterConn *MasterConn, logger log.Logger) { + if clusterPeerID == ReservedClusterPeerID { + d.log.Error("refusing to create peer connection with reserved cluster_peer_id", "cluster_peer_id", clusterPeerID) + return + } + + d.mu.Lock() + defer d.mu.Unlock() + + branchMap, ok := d.peers[clusterPeerID] + if !ok { + branchMap = make(map[uint32]*PeerConn) + d.peers[clusterPeerID] = branchMap + } + + for _, branch := range branches { + if _, exists := branchMap[branch]; exists { + d.log.Warn("duplicate create cluster peer connection", "cluster_peer_id", clusterPeerID, "branch", branch) + continue + } + pc := NewPeerConn(clusterPeerID, branch, masterConn, logger) + pc.Start() + branchMap[branch] = pc + } +} + +// DestroyPeerConns removes all PeerConns for clusterPeerID from the registry +// and closes them. Missing entries are silently ignored. +func (d *Dispatcher) DestroyPeerConns(clusterPeerID uint64) { + d.mu.Lock() + branchMap, ok := d.peers[clusterPeerID] + if ok { + delete(d.peers, clusterPeerID) + } + d.mu.Unlock() + + if !ok { + return + } + for _, pc := range branchMap { + pc.Close() + } +} + +// RouteFrame is the forwarder callback installed on MasterConn. It returns +// false for master-local traffic (cluster_peer_id == 0) so MasterConn handles +// the frame normally. For peer traffic it looks up the PeerConn, enqueues the +// frame if found, or drops it (matching Python's NULL_CONNECTION) and logs a +// warning if not found. +func (d *Dispatcher) RouteFrame(frame *wire.Frame) bool { + if frame.Meta.ClusterPeerID == 0 { + return false + } + + d.mu.RLock() + branchMap, ok := d.peers[frame.Meta.ClusterPeerID] + if !ok { + d.mu.RUnlock() + d.log.Warn("no peer connection for cluster_peer_id", "cluster_peer_id", frame.Meta.ClusterPeerID) + return true + } + pc, ok := branchMap[frame.Meta.Branch] + d.mu.RUnlock() + + if !ok { + d.log.Warn("no peer connection for branch", "cluster_peer_id", frame.Meta.ClusterPeerID, "branch", frame.Meta.Branch) + return true + } + + if err := pc.HandleFrame(frame); err != nil { + d.log.Warn("failed to deliver frame to peer connection", "cluster_peer_id", frame.Meta.ClusterPeerID, "branch", frame.Meta.Branch, "err", err) + } + return true +} + +// Close closes all registered PeerConns and clears the registry. +func (d *Dispatcher) Close() error { + d.mu.Lock() + all := make([]*PeerConn, 0) + for _, branchMap := range d.peers { + for _, pc := range branchMap { + all = append(all, pc) + } + } + d.peers = make(map[uint64]map[uint32]*PeerConn) + d.mu.Unlock() + + for _, pc := range all { + pc.Close() + } + return nil +} diff --git a/qkc/cluster/slave/master_conn.go b/qkc/cluster/slave/master_conn.go index 3cdfd1c35c52..c5bdb2fdd742 100644 --- a/qkc/cluster/slave/master_conn.go +++ b/qkc/cluster/slave/master_conn.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "net" + "slices" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/qkc/cluster/wire" @@ -29,6 +30,18 @@ type MasterConn struct { localID []byte localFullShardIDList []uint32 + + // dispatcher routes peer traffic (cluster_peer_id != 0) to virtual PeerConns. + // It is nil until wired by SetDispatcher. Once Start() is called, it is + // immutable — handlers and Close() read it without synchronization. + dispatcher *Dispatcher + + // peerRPCIDs tracks the most recent inbound RPC ID per cluster_peer_id. + // cluster_peer_id == 0 is the master itself; each non-zero peer has its + // own independent monotonic sequence so PeerConns sharing this MasterConn + // do not collide on rpc_id. + // Only accessed by readLoop; no lock needed. + peerRPCIDs map[uint64]int64 } // NewMasterConn dials the master at addr and returns a MasterConn. @@ -56,8 +69,11 @@ func newMasterConn(conn net.Conn, maxPayloadSize uint32, localID []byte, localFu baseConn: newBaseConnFromConn(conn, readFrame, wire.WriteFrame, logger), localID: append([]byte(nil), localID...), localFullShardIDList: append([]uint32(nil), localFullShardIDList...), + peerRPCIDs: make(map[uint64]int64), } + mc.baseConn.validateRPCID = mc.validatePeerRPCID + mc.registerOpSerializers() mc.registerHandlers() @@ -217,6 +233,21 @@ func emptyRawBytes() *wire.RawBytes { return rawBytes([]byte{}) } +// validatePeerRPCID validates inbound RPC IDs independently for each +// cluster_peer_id. This lets multiple PeerConns share one MasterConn without +// colliding on rpc_id. +func (mc *MasterConn) validatePeerRPCID(clusterPeerID uint64, rpcID uint64) bool { + last, ok := mc.peerRPCIDs[clusterPeerID] + if !ok { + last = -1 + } + if int64(rpcID) <= last { + return false + } + mc.peerRPCIDs[clusterPeerID] = int64(rpcID) + return true +} + // LocalID returns this slave's ID used in PONG responses. func (mc *MasterConn) LocalID() []byte { return append([]byte(nil), mc.localID...) @@ -245,16 +276,30 @@ func (mc *MasterConn) handlePing(req any) (any, error) { // handleCreateClusterPeerConnection creates virtual peer connections for all shards. // Python: returns CreateClusterPeerConnectionResponse(error_code=0) on success. func (mc *MasterConn) handleCreateClusterPeerConnection(req any) (any, error) { - _ = req.(*wire.CreateClusterPeerConnectionRequest) - // TODO: create PeerShardConnection instances and wire with the dispatcher (PR6). + r := req.(*wire.CreateClusterPeerConnectionRequest) + + // TODO: localFullShardIDList is a temporary stand-in for the real shard + // registry. Once the shard registry is available, replace with the actual + // per-shard branch list from the registry. + + d := mc.dispatcher + if d != nil { + d.CreatePeerConns(r.ClusterPeerID, mc.localFullShardIDList, mc, mc.baseConn.log) + } + return &wire.CreateClusterPeerConnectionResponse{ErrorCode: 0}, nil } // handleDestroyClusterPeerConnection is a fire-and-forget command to tear down // a virtual peer connection. No response is sent. func (mc *MasterConn) handleDestroyClusterPeerConnection(req any) (any, error) { - _ = req.(*wire.DestroyClusterPeerConnectionCommand) - // TODO: notify dispatcher / close peer shard connections (PR6). + r := req.(*wire.DestroyClusterPeerConnectionCommand) + + d := mc.dispatcher + if d != nil { + d.DestroyPeerConns(r.ClusterPeerID) + } + return nil, nil } @@ -540,6 +585,45 @@ func (mc *MasterConn) ForwardFrame(f *wire.Frame) error { return mc.baseConn.writeFrame(f) } +// SetDispatcher wires the dispatcher that routes peer traffic. It also installs +// a forwarder that validates the branch before routing. +// +// Must be called before Start(). dispatcher and forwarder are immutable after Start(). +// +// Python: MasterConnection.get_connection_to_forward() closes the connection if +// the branch is not in the configured full_shard_id_list. +func (mc *MasterConn) SetDispatcher(d *Dispatcher) { + mc.dispatcher = d + + // Create a wrapper forwarder that validates branch before routing. + // Python: MasterConnection.get_connection_to_forward() only validates + // branch when cluster_peer_id != 0 (i.e., on the forwarding path). + // Master-local RPCs (cluster_peer_id == 0) use branch=0 and must not + // be validated. + mc.SetForwarder(func(frame *wire.Frame) bool { + if frame.Meta.ClusterPeerID != 0 && !mc.isValidBranch(frame.Meta.Branch) { + mc.log.Error("incorrect forwarding branch", "branch", frame.Meta.Branch) + mc.Close() + return true + } + return d.RouteFrame(frame) + }) +} + +// isValidBranch checks if the given branch is in the local full shard ID list. +func (mc *MasterConn) isValidBranch(branch uint32) bool { + return slices.Contains(mc.localFullShardIDList, branch) +} + +// Close closes the master connection and all associated peer connections. +func (mc *MasterConn) Close() error { + d := mc.dispatcher + if d != nil { + d.Close() + } + return mc.baseConn.Close() +} + // SendRPCMeta sends a request with ClusterMetadata and waits for the response. func (mc *MasterConn) SendRPCMeta(ctx context.Context, opcode byte, payload []byte, meta wire.ClusterMetadata) (*wire.Frame, error) { return mc.baseConn.SendRPCMeta(ctx, opcode, payload, meta) diff --git a/qkc/cluster/slave/peer_conn.go b/qkc/cluster/slave/peer_conn.go new file mode 100644 index 000000000000..5dcd80302ebe --- /dev/null +++ b/qkc/cluster/slave/peer_conn.go @@ -0,0 +1,233 @@ +// Copyright 2026-2027, QuarkChain. + +package slave + +import ( + "fmt" + "sync" + + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/qkc/cluster/wire" +) + +// virtualTransport implements frameTransport for PeerConn. It has no TCP +// socket; inbound frames are pushed by the Dispatcher via receive(), and +// outbound frames are forwarded through the associated MasterConn. +type virtualTransport struct { + clusterPeerID uint64 + branch uint32 + masterConn *MasterConn + + inbound chan *wire.Frame + closedChan chan struct{} + closeOnce sync.Once + remoteAddr string +} + +func newVirtualTransport(clusterPeerID uint64, branch uint32, masterConn *MasterConn) *virtualTransport { + return &virtualTransport{ + clusterPeerID: clusterPeerID, + branch: branch, + masterConn: masterConn, + inbound: make(chan *wire.Frame, 64), + closedChan: make(chan struct{}), + remoteAddr: fmt.Sprintf("virtual://peer/%d/%d", clusterPeerID, branch), + } +} + +func (vt *virtualTransport) readFrame() (*wire.Frame, error) { + select { + case frame := <-vt.inbound: + return frame, nil + case <-vt.closedChan: + return nil, ErrConnectionClosed + } +} + +func (vt *virtualTransport) writeFrame(f *wire.Frame) error { + // PeerShardConnection in Python always writes with the shard branch and its + // own cluster_peer_id so the master can route the frame back to the peer. + f.Meta = wire.ClusterMetadata{ + Branch: vt.branch, + ClusterPeerID: vt.clusterPeerID, + } + return vt.masterConn.ForwardFrame(f) +} + +func (vt *virtualTransport) close() error { + vt.closeOnce.Do(func() { close(vt.closedChan) }) + return nil +} + +func (vt *virtualTransport) RemoteAddr() string { + return vt.remoteAddr +} + +// receive pushes a frame into the inbound queue. It returns false if the +// transport is already closed. +func (vt *virtualTransport) receive(frame *wire.Frame) bool { + select { + case vt.inbound <- frame: + return true + case <-vt.closedChan: + return false + } +} + +// PeerConn is a virtual RPC channel representing the slave-side endpoint of a +// forwarded external peer connection. It does not own a TCP socket; all wire +// traffic is tunneled through the slave's MasterConn. +// +// It corresponds to Python's PeerShardConnection and shares the same +// responsibilities: independent RPC ID namespace, CommandOp handler dispatch, +// and lifecycle tied to master commands. +type PeerConn struct { + *baseConn + + clusterPeerID uint64 + branch uint32 + vt *virtualTransport +} + +// NewPeerConn creates a virtual peer connection for the given cluster_peer_id +// and branch, tunneling outbound frames through masterConn. +func NewPeerConn(clusterPeerID uint64, branch uint32, masterConn *MasterConn, logger log.Logger) *PeerConn { + vt := newVirtualTransport(clusterPeerID, branch, masterConn) + pc := &PeerConn{ + baseConn: newBaseConn(vt, logger), + clusterPeerID: clusterPeerID, + branch: branch, + vt: vt, + } + pc.registerOpSerializers() + pc.registerHandlers() + return pc +} + +// ReservedClusterPeerID is the reserved cluster_peer_id used by the master for +// its own control traffic. PeerConn must not use this value. +const ReservedClusterPeerID = 0 + +// registerOpSerializers registers serializers for every CommandOp so that both +// inbound requests and outbound responses can be (de)serialized. +func (pc *PeerConn) registerOpSerializers() { + pc.baseConn.RegisterOpSerializers(map[byte]*OpSerializer{ + // §1 Hello / master-only + byte(wire.CommandOpHello): OpSerializerFor[wire.HelloCommand, wire.HelloCommand](), + byte(wire.CommandOpNewMinorBlockHeaderList): OpSerializerFor[wire.NewMinorBlockHeaderListCommand, wire.NewMinorBlockHeaderListCommand](), + byte(wire.CommandOpNewTransactionList): OpSerializerFor[wire.NewTransactionListCommand, wire.NewTransactionListCommand](), + byte(wire.CommandOpGetPeerListRequest): OpSerializerFor[wire.GetPeerListRequest, wire.GetPeerListResponse](), + byte(wire.CommandOpGetPeerListResponse): OpSerializerFor[wire.GetPeerListResponse, wire.GetPeerListRequest](), + byte(wire.CommandOpGetRootBlockHeaderListRequest): OpSerializerFor[wire.GetRootBlockHeaderListRequest, wire.GetRootBlockHeaderListResponse](), + byte(wire.CommandOpGetRootBlockHeaderListResponse): OpSerializerFor[wire.GetRootBlockHeaderListResponse, wire.GetRootBlockHeaderListRequest](), + byte(wire.CommandOpGetRootBlockListRequest): OpSerializerFor[wire.GetRootBlockListRequest, wire.GetRootBlockListResponse](), + byte(wire.CommandOpGetRootBlockListResponse): OpSerializerFor[wire.GetRootBlockListResponse, wire.GetRootBlockListRequest](), + + // §2 Slave RPC request/response pairs + byte(wire.CommandOpGetMinorBlockListRequest): OpSerializerFor[wire.GetMinorBlockListRequest, wire.GetMinorBlockListResponse](), + byte(wire.CommandOpGetMinorBlockListResponse): OpSerializerFor[wire.GetMinorBlockListResponse, wire.GetMinorBlockListRequest](), + byte(wire.CommandOpGetMinorBlockHeaderListRequest): OpSerializerFor[wire.GetMinorBlockHeaderListRequest, wire.GetMinorBlockHeaderListResponse](), + byte(wire.CommandOpGetMinorBlockHeaderListResponse): OpSerializerFor[wire.GetMinorBlockHeaderListResponse, wire.GetMinorBlockHeaderListRequest](), + + // §3 More master-only / root-chain peer opcodes + byte(wire.CommandOpNewBlockMinor): OpSerializerFor[wire.NewBlockMinorCommand, wire.NewBlockMinorCommand](), + byte(wire.CommandOpPing): OpSerializerFor[wire.PingPongCommand, wire.PingPongCommand](), + byte(wire.CommandOpPong): OpSerializerFor[wire.PingPongCommand, wire.PingPongCommand](), + byte(wire.CommandOpGetRootBlockHeaderListWithSkipRequest): OpSerializerFor[wire.GetRootBlockHeaderListWithSkipRequest, wire.GetRootBlockHeaderListResponse](), + byte(wire.CommandOpGetRootBlockHeaderListWithSkipResponse): OpSerializerFor[wire.GetRootBlockHeaderListResponse, wire.GetRootBlockHeaderListWithSkipRequest](), + byte(wire.CommandOpNewRootBlock): OpSerializerFor[wire.NewRootBlockCommand, wire.NewRootBlockCommand](), + byte(wire.CommandOpGetMinorBlockHeaderListWithSkipRequest): OpSerializerFor[wire.GetMinorBlockHeaderListWithSkipRequest, wire.GetMinorBlockHeaderListResponse](), + byte(wire.CommandOpGetMinorBlockHeaderListWithSkipResponse): OpSerializerFor[wire.GetMinorBlockHeaderListResponse, wire.GetMinorBlockHeaderListWithSkipRequest](), + }) +} + +// registerHandlers registers the shard-level peer handlers. These are stubs; +// real implementations require the shard runtime to be ported. +func (pc *PeerConn) registerHandlers() { + pc.baseConn.RegisterTypedHandlers(map[byte]TypedHandler{ + // ── Migration stubs ───────────────────────────────────────────── + // These handlers exist only to preserve protocol compatibility. + // Real implementations must be added outside the connection layer. + // After migration, remove these stub registrations and handlers. + + // Non-RPC commands (fire-and-forget). + byte(wire.CommandOpNewMinorBlockHeaderList): pc.handleNewMinorBlockHeaderList, + byte(wire.CommandOpNewTransactionList): pc.handleNewTransactionList, + byte(wire.CommandOpNewBlockMinor): pc.handleNewBlockMinor, + + // RPC requests; responses use opcode+1. + byte(wire.CommandOpGetMinorBlockListRequest): pc.handleGetMinorBlockListRequest, + byte(wire.CommandOpGetMinorBlockHeaderListRequest): pc.handleGetMinorBlockHeaderListRequest, + byte(wire.CommandOpGetMinorBlockHeaderListWithSkipRequest): pc.handleGetMinorBlockHeaderListWithSkipRequest, + }) + + pc.baseConn.RegisterNonRPCOps([]byte{ + byte(wire.CommandOpNewMinorBlockHeaderList), + byte(wire.CommandOpNewTransactionList), + byte(wire.CommandOpNewBlockMinor), + }) +} + +// HandleFrame receives a frame routed by the Dispatcher. It enqueues the frame +// for the PeerConn read loop. Frames received after close are dropped. +func (pc *PeerConn) HandleFrame(frame *wire.Frame) error { + if pc.Closed() { + return ErrConnectionClosed + } + if !pc.vt.receive(frame) { + return ErrConnectionClosed + } + return nil +} + +// ClusterPeerID returns the peer's cluster-scoped identifier. +func (pc *PeerConn) ClusterPeerID() uint64 { return pc.clusterPeerID } + +// Branch returns the shard branch this virtual connection serves. +func (pc *PeerConn) Branch() uint32 { return pc.branch } + +// ── stub handlers ──────────────────────────────────────────────────────────── + +func (pc *PeerConn) handleNewMinorBlockHeaderList(req any) (any, error) { + _ = req.(*wire.NewMinorBlockHeaderListCommand) + // TODO: delegate to shard synchronizer once Shard Runtime is ported. + return nil, nil +} + +func (pc *PeerConn) handleNewTransactionList(req any) (any, error) { + _ = req.(*wire.NewTransactionListCommand) + // TODO: delegate to shard tx pool once Shard Runtime is ported. + return nil, nil +} + +func (pc *PeerConn) handleNewBlockMinor(req any) (any, error) { + _ = req.(*wire.NewBlockMinorCommand) + // TODO: delegate to shard block processing once Shard Runtime is ported. + return nil, nil +} + +func (pc *PeerConn) handleGetMinorBlockListRequest(req any) (any, error) { + _ = req.(*wire.GetMinorBlockListRequest) + // TODO: fetch blocks from shard state db once Shard Runtime is ported. + return &wire.GetMinorBlockListResponse{MinorBlockList: []*wire.RawBytes{}}, nil +} + +func (pc *PeerConn) handleGetMinorBlockHeaderListRequest(req any) (any, error) { + _ = req.(*wire.GetMinorBlockHeaderListRequest) + // TODO: fetch headers from shard state db once Shard Runtime is ported. + return &wire.GetMinorBlockHeaderListResponse{ + RootTip: nil, + ShardTip: nil, + BlockHeaderList: []*wire.RawBytes{}, + }, nil +} + +func (pc *PeerConn) handleGetMinorBlockHeaderListWithSkipRequest(req any) (any, error) { + _ = req.(*wire.GetMinorBlockHeaderListWithSkipRequest) + // TODO: fetch headers from shard state db once Shard Runtime is ported. + return &wire.GetMinorBlockHeaderListResponse{ + RootTip: nil, + ShardTip: nil, + BlockHeaderList: []*wire.RawBytes{}, + }, nil +} diff --git a/qkc/cluster/slave/peer_conn_test.go b/qkc/cluster/slave/peer_conn_test.go new file mode 100644 index 000000000000..980584321cb6 --- /dev/null +++ b/qkc/cluster/slave/peer_conn_test.go @@ -0,0 +1,675 @@ +// Copyright 2026-2027, QuarkChain. + +package slave + +import ( + "context" + "net" + "testing" + "time" + + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/qkc/cluster/wire" + "github.com/ethereum/go-ethereum/qkc/serialize" +) + +// newMasterConnWithDispatcher creates a MasterConn over a local TCP pair and +// wires a Dispatcher. The caller gets the raw server-side net.Conn so it can +// act as the fake master, plus the client MasterConn and cleanup. +func newMasterConnWithDispatcher(t *testing.T) (client *MasterConn, serverConn net.Conn, cleanup func()) { + t.Helper() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + + var srvConn net.Conn + var acceptErr error + accepted := make(chan struct{}) + go func() { + defer close(accepted) + srvConn, acceptErr = ln.Accept() + ln.Close() + }() + + clientConn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + <-accepted + if acceptErr != nil { + t.Fatalf("accept: %v", acceptErr) + } + + logger := log.New() + client = NewMasterConnFromConn(clientConn, 0, []byte("go-slave"), []uint32{0x00010001, 0x00020001}, logger) + dispatcher := NewDispatcher(logger) + client.SetDispatcher(dispatcher) + client.Start() + serverConn = srvConn + + cleanup = func() { + client.Close() + if serverConn != nil { + serverConn.Close() + } + } + return +} + +// readMasterFrame reads a 12-byte metadata frame from the fake master side. +func readMasterFrame(t *testing.T, conn net.Conn) *wire.Frame { + t.Helper() + if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatalf("set read deadline: %v", err) + } + frame, err := wire.ReadFrame(conn, 0) + if err != nil { + t.Fatalf("read frame: %v", err) + } + return frame +} + +// writeMasterFrame writes a 12-byte metadata frame from the fake master side. +func writeMasterFrame(t *testing.T, conn net.Conn, frame *wire.Frame) { + t.Helper() + if err := conn.SetWriteDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatalf("set write deadline: %v", err) + } + if err := wire.WriteFrame(conn, frame); err != nil { + t.Fatalf("write frame: %v", err) + } +} + +// TestDispatcher_RouteToMasterConn verifies that frames with cluster_peer_id == 0 +// are handled by MasterConn itself (PING -> PONG). +func TestDispatcher_RouteToMasterConn(t *testing.T) { + client, serverConn, cleanup := newMasterConnWithDispatcher(t) + defer cleanup() + + pingPayload, err := serialize.SerializeToBytes(&wire.PingRequest{ + ID: []byte("master"), + FullShardIDList: []uint32{0x00010001}, + }) + if err != nil { + t.Fatalf("serialize ping: %v", err) + } + + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001}, + Opcode: byte(wire.ClusterOpPing), + RPCID: 1, + Payload: pingPayload, + }) + + resp := readMasterFrame(t, serverConn) + if resp.Opcode != byte(wire.ClusterOpPong) { + t.Fatalf("expected PONG opcode 0x%x, got 0x%x", wire.ClusterOpPong, resp.Opcode) + } + if resp.RPCID != 1 { + t.Fatalf("expected rpc_id 1, got %d", resp.RPCID) + } + if resp.Meta.ClusterPeerID != 0 { + t.Fatalf("expected cluster_peer_id 0 for master-local response, got %d", resp.Meta.ClusterPeerID) + } + + var pong wire.PongResponse + if err := serialize.Deserialize(serialize.NewByteBuffer(resp.Payload), &pong); err != nil { + t.Fatalf("deserialize pong: %v", err) + } + if string(pong.ID) != "go-slave" { + t.Fatalf("pong id mismatch: got %q", pong.ID) + } + + _ = client +} + +// TestDispatcher_RouteToPeerConn verifies that frames with cluster_peer_id != 0 +// are forwarded to the matching virtual PeerConn and the stub response is sent +// back through MasterConn. +func TestDispatcher_RouteToPeerConn(t *testing.T) { + client, serverConn, cleanup := newMasterConnWithDispatcher(t) + defer cleanup() + + const clusterPeerID uint64 = 7 + const branch uint32 = 0x00010001 + + client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New()) + + reqPayload, err := serialize.SerializeToBytes(&wire.GetMinorBlockListRequest{ + MinorBlockHashList: [][wire.HashLength]byte{}, + }) + if err != nil { + t.Fatalf("serialize request: %v", err) + } + + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: branch, ClusterPeerID: clusterPeerID}, + Opcode: byte(wire.CommandOpGetMinorBlockListRequest), + RPCID: 3, + Payload: reqPayload, + }) + + resp := readMasterFrame(t, serverConn) + if resp.Opcode != byte(wire.CommandOpGetMinorBlockListResponse) { + t.Fatalf("expected response opcode 0x%x, got 0x%x", wire.CommandOpGetMinorBlockListResponse, resp.Opcode) + } + if resp.RPCID != 3 { + t.Fatalf("expected rpc_id 3, got %d", resp.RPCID) + } + if resp.Meta.Branch != branch || resp.Meta.ClusterPeerID != clusterPeerID { + t.Fatalf("metadata mismatch: got %+v, want branch=%d cluster_peer_id=%d", resp.Meta, branch, clusterPeerID) + } + + var listResp wire.GetMinorBlockListResponse + if err := serialize.Deserialize(serialize.NewByteBuffer(resp.Payload), &listResp); err != nil { + t.Fatalf("deserialize response: %v", err) + } +} + +// TestDispatcher_UnknownPeerDropped verifies that frames for an unregistered +// cluster_peer_id are dropped and do not close MasterConn. +func TestDispatcher_UnknownPeerDropped(t *testing.T) { + _, serverConn, cleanup := newMasterConnWithDispatcher(t) + defer cleanup() + + reqPayload, err := serialize.SerializeToBytes(&wire.GetMinorBlockListRequest{ + MinorBlockHashList: [][wire.HashLength]byte{}, + }) + if err != nil { + t.Fatalf("serialize request: %v", err) + } + + // Unknown peer frame should be consumed and dropped. + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001, ClusterPeerID: 999}, + Opcode: byte(wire.CommandOpGetMinorBlockListRequest), + RPCID: 1, + Payload: reqPayload, + }) + + // The fake master should see no response for the dropped frame, but a + // subsequent master-local PING must still work. + if err := serverConn.SetReadDeadline(time.Now().Add(200 * time.Millisecond)); err != nil { + t.Fatalf("set read deadline: %v", err) + } + if _, err := wire.ReadFrame(serverConn, 0); err == nil { + t.Fatal("expected no response for unknown peer frame") + } + + pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{ + ID: []byte("master"), + FullShardIDList: []uint32{0x00010001}, + }) + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001}, + Opcode: byte(wire.ClusterOpPing), + RPCID: 2, + Payload: pingPayload, + }) + + resp := readMasterFrame(t, serverConn) + if resp.Opcode != byte(wire.ClusterOpPong) { + t.Fatalf("expected PONG after unknown peer drop, got opcode 0x%x", resp.Opcode) + } + if resp.RPCID != 2 { + t.Fatalf("expected rpc_id 2 in pong, got %d", resp.RPCID) + } +} + +// TestPeerConn_RPCIDIsolation verifies that two PeerConns sharing a MasterConn +// can use the same RPC ID without collision; responses are routed back to the +// correct peer via metadata. +func TestPeerConn_RPCIDIsolation(t *testing.T) { + client, serverConn, cleanup := newMasterConnWithDispatcher(t) + defer cleanup() + + client.dispatcher.CreatePeerConns(7, []uint32{0x00010001}, client, log.New()) + client.dispatcher.CreatePeerConns(9, []uint32{0x00020001}, client, log.New()) + + reqPayload, err := serialize.SerializeToBytes(&wire.GetMinorBlockListRequest{ + MinorBlockHashList: [][wire.HashLength]byte{}, + }) + if err != nil { + t.Fatalf("serialize request: %v", err) + } + + // Both peers use rpc_id=5. + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001, ClusterPeerID: 7}, + Opcode: byte(wire.CommandOpGetMinorBlockListRequest), + RPCID: 5, + Payload: reqPayload, + }) + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00020001, ClusterPeerID: 9}, + Opcode: byte(wire.CommandOpGetMinorBlockListRequest), + RPCID: 5, + Payload: reqPayload, + }) + + resp1 := readMasterFrame(t, serverConn) + resp2 := readMasterFrame(t, serverConn) + + if resp1.RPCID != 5 || resp2.RPCID != 5 { + t.Fatalf("expected both responses to have rpc_id 5, got %d and %d", resp1.RPCID, resp2.RPCID) + } + + // Each response must belong to a distinct peer/branch pair. + peers := map[uint64]uint32{ + resp1.Meta.ClusterPeerID: resp1.Meta.Branch, + resp2.Meta.ClusterPeerID: resp2.Meta.Branch, + } + if len(peers) != 2 { + t.Fatalf("responses were not routed to distinct peers: %+v", peers) + } + if peers[7] != 0x00010001 { + t.Fatalf("peer 7 response routed to wrong branch: got 0x%x", peers[7]) + } + if peers[9] != 0x00020001 { + t.Fatalf("peer 9 response routed to wrong branch: got 0x%x", peers[9]) + } + + for _, resp := range []*wire.Frame{resp1, resp2} { + if resp.Opcode != byte(wire.CommandOpGetMinorBlockListResponse) { + t.Fatalf("expected GetMinorBlockListResponse, got opcode 0x%x", resp.Opcode) + } + var listResp wire.GetMinorBlockListResponse + if err := serialize.Deserialize(serialize.NewByteBuffer(resp.Payload), &listResp); err != nil { + t.Fatalf("deserialize response: %v", err) + } + } +} + +// TestPeerConn_PeerHandlerRPCRoundTrip sends a CommandOp RPC through a virtual +// PeerConn and verifies the stub response deserializes correctly. +func TestPeerConn_PeerHandlerRPCRoundTrip(t *testing.T) { + client, serverConn, cleanup := newMasterConnWithDispatcher(t) + defer cleanup() + + const clusterPeerID uint64 = 11 + const branch uint32 = 0x00010001 + + client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New()) + + reqPayload, err := serialize.SerializeToBytes(&wire.GetMinorBlockHeaderListRequest{ + Branch: branch, + BlockHash: [wire.HashLength]byte{}, + Limit: 10, + Direction: 0, + }) + if err != nil { + t.Fatalf("serialize request: %v", err) + } + + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: branch, ClusterPeerID: clusterPeerID}, + Opcode: byte(wire.CommandOpGetMinorBlockHeaderListRequest), + RPCID: 1, + Payload: reqPayload, + }) + + resp := readMasterFrame(t, serverConn) + if resp.Opcode != byte(wire.CommandOpGetMinorBlockHeaderListResponse) { + t.Fatalf("expected response opcode 0x%x, got 0x%x", wire.CommandOpGetMinorBlockHeaderListResponse, resp.Opcode) + } + if resp.RPCID != 1 { + t.Fatalf("expected rpc_id 1, got %d", resp.RPCID) + } + + var headerResp wire.GetMinorBlockHeaderListResponse + if err := serialize.Deserialize(serialize.NewByteBuffer(resp.Payload), &headerResp); err != nil { + t.Fatalf("deserialize response: %v", err) + } +} + +// TestMasterConn_CreateDestroyPeerConnection verifies that the master commands +// create and destroy virtual peer connections through the dispatcher. +func TestMasterConn_CreateDestroyPeerConnection(t *testing.T) { + client, serverConn, cleanup := newMasterConnWithDispatcher(t) + defer cleanup() + + const clusterPeerID uint64 = 21 + + createPayload, err := serialize.SerializeToBytes(&wire.CreateClusterPeerConnectionRequest{ClusterPeerID: clusterPeerID}) + if err != nil { + t.Fatalf("serialize create request: %v", err) + } + + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001}, + Opcode: byte(wire.ClusterOpCreateClusterPeerConnectionRequest), + RPCID: 1, + Payload: createPayload, + }) + + resp := readMasterFrame(t, serverConn) + if resp.Opcode != byte(wire.ClusterOpCreateClusterPeerConnectionResponse) { + t.Fatalf("expected create response opcode 0x%x, got 0x%x", wire.ClusterOpCreateClusterPeerConnectionResponse, resp.Opcode) + } + var createResp wire.CreateClusterPeerConnectionResponse + if err := serialize.Deserialize(serialize.NewByteBuffer(resp.Payload), &createResp); err != nil { + t.Fatalf("deserialize create response: %v", err) + } + if createResp.ErrorCode != 0 { + t.Fatalf("expected error_code 0, got %d", createResp.ErrorCode) + } + + // Capture PeerConn pointers before destroy to avoid racing with dispatcher's + // internal map mutation during DestroyPeerConns. + branchMap := client.dispatcher.peers[clusterPeerID] + if len(branchMap) != len(client.localFullShardIDList) { + t.Fatalf("expected %d peer conns, got %d", len(client.localFullShardIDList), len(branchMap)) + } + peerConns := make([]*PeerConn, 0, len(client.localFullShardIDList)) + for _, branch := range client.localFullShardIDList { + pc := branchMap[branch] + if pc == nil { + t.Fatalf("missing peer conn for branch 0x%x", branch) + } + if pc.IsClosed() { + t.Fatalf("peer conn for branch 0x%x is already closed", branch) + } + peerConns = append(peerConns, pc) + } + + // Destroy the peer connections. + destroyPayload, err := serialize.SerializeToBytes(&wire.DestroyClusterPeerConnectionCommand{ClusterPeerID: clusterPeerID}) + if err != nil { + t.Fatalf("serialize destroy command: %v", err) + } + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001}, + Opcode: byte(wire.ClusterOpDestroyClusterPeerConnectionCommand), + RPCID: 0, // non-RPC + Payload: destroyPayload, + }) + + // Wait for peer connections to be closed (async since handler runs in goroutine). + // Check captured pointers instead of reading dispatcher.peers to avoid data race. + waitForCondition(t, 2*time.Second, func() bool { + for _, pc := range peerConns { + if !pc.IsClosed() { + return false + } + } + return true + }) + + // MasterConn must still be alive for a follow-up PING. + pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{ + ID: []byte("master"), + FullShardIDList: []uint32{0x00010001}, + }) + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001}, + Opcode: byte(wire.ClusterOpPing), + RPCID: 2, + Payload: pingPayload, + }) + + pingResp := readMasterFrame(t, serverConn) + if pingResp.Opcode != byte(wire.ClusterOpPong) { + t.Fatalf("expected PONG after destroy, got opcode 0x%x", pingResp.Opcode) + } +} + +// TestMasterConn_CloseClosesPeerConns verifies that closing MasterConn closes +// all associated PeerConns and clears the dispatcher registry. +func TestMasterConn_CloseClosesPeerConns(t *testing.T) { + client, _, cleanup := newMasterConnWithDispatcher(t) + defer cleanup() + + client.dispatcher.CreatePeerConns(7, []uint32{0x00010001, 0x00020001}, client, log.New()) + client.dispatcher.CreatePeerConns(9, []uint32{0x00010001}, client, log.New()) + + // Keep references before Close clears the map. + var peerConns []*PeerConn + for _, branchMap := range client.dispatcher.peers { + for _, pc := range branchMap { + peerConns = append(peerConns, pc) + } + } + if len(peerConns) != 3 { + t.Fatalf("expected 3 peer conns, got %d", len(peerConns)) + } + + client.Close() + + for _, pc := range peerConns { + if !pc.IsClosed() { + t.Fatalf("peer conn %d/%d was not closed by MasterConn.Close", pc.ClusterPeerID(), pc.Branch()) + } + } + + if len(client.dispatcher.peers) != 0 { + t.Fatalf("dispatcher registry not cleared: got %d cluster_peer_id entries", len(client.dispatcher.peers)) + } +} + +// TestPeerConn_OutboundRPCThroughMasterConn verifies that a PeerConn can issue +// an outbound RPC and the request is written to the underlying MasterConn with +// the correct cluster_peer_id metadata. +func TestPeerConn_OutboundRPCThroughMasterConn(t *testing.T) { + client, serverConn, cleanup := newMasterConnWithDispatcher(t) + defer cleanup() + + const clusterPeerID uint64 = 31 + const branch uint32 = 0x00010001 + + client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New()) + pc := client.dispatcher.peers[clusterPeerID][branch] + + req := &wire.GetMinorBlockListRequest{MinorBlockHashList: [][wire.HashLength]byte{}} + reqPayload, err := serialize.SerializeToBytes(req) + if err != nil { + t.Fatalf("serialize request: %v", err) + } + + // Echo the request back as a response from the fake master. + go func() { + frame := readMasterFrame(t, serverConn) + if frame.Meta.ClusterPeerID != clusterPeerID { + t.Errorf("outbound request cluster_peer_id mismatch: got %d, want %d", frame.Meta.ClusterPeerID, clusterPeerID) + } + if frame.Meta.Branch != branch { + t.Errorf("outbound request branch mismatch: got 0x%x, want 0x%x", frame.Meta.Branch, branch) + } + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: frame.Meta, + Opcode: frame.Opcode + 1, + RPCID: frame.RPCID, + Payload: frame.Payload, + }) + }() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + resp, err := pc.SendRPCMeta(ctx, byte(wire.CommandOpGetMinorBlockListRequest), reqPayload, wire.ClusterMetadata{}) + if err != nil { + t.Fatalf("peer conn SendRPCMeta: %v", err) + } + if resp.Opcode != byte(wire.CommandOpGetMinorBlockListResponse) { + t.Fatalf("expected response opcode 0x%x, got 0x%x", wire.CommandOpGetMinorBlockListResponse, resp.Opcode) + } +} + +// ── Additional tests ───────────────────────────────────────────────────────── + +// TestDispatcher_DuplicateCreatePeerConn verifies that a duplicate create +// request for the same cluster_peer_id and branch does not replace the existing +// PeerConn. This matches Python's behavior of logging an error and skipping. +// Python: slave.py#L335-L341 +func TestDispatcher_DuplicateCreatePeerConn(t *testing.T) { + client, serverConn, cleanup := newMasterConnWithDispatcher(t) + defer cleanup() + + const clusterPeerID uint64 = 41 + const branch uint32 = 0x00010001 + + // First create. + client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New()) + original := client.dispatcher.peers[clusterPeerID][branch] + if original == nil { + t.Fatal("expected peer conn after first create") + } + + // Duplicate create — should not replace the existing PeerConn. + client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New()) + afterDup := client.dispatcher.peers[clusterPeerID][branch] + if afterDup != original { + t.Fatal("duplicate create replaced the existing PeerConn") + } + + // The branch map should still have exactly one entry. + if len(client.dispatcher.peers[clusterPeerID]) != 1 { + t.Fatalf("expected 1 branch entry, got %d", len(client.dispatcher.peers[clusterPeerID])) + } + + // MasterConn must still be alive. + pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{ + ID: []byte("master"), + FullShardIDList: []uint32{0x00010001}, + }) + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001}, + Opcode: byte(wire.ClusterOpPing), + RPCID: 1, + Payload: pingPayload, + }) + resp := readMasterFrame(t, serverConn) + if resp.Opcode != byte(wire.ClusterOpPong) { + t.Fatalf("expected PONG, got opcode 0x%x", resp.Opcode) + } +} + +// TestDispatcher_NonRPCCommandRouted verifies that fire-and-forget (non-RPC) +// commands are routed through the Dispatcher to the correct PeerConn. +// Python: shard.py OP_NONRPC_MAP (L275-L279) +func TestDispatcher_NonRPCCommandRouted(t *testing.T) { + client, serverConn, cleanup := newMasterConnWithDispatcher(t) + defer cleanup() + + const clusterPeerID uint64 = 42 + const branch uint32 = 0x00010001 + + client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New()) + + cmdPayload, err := serialize.SerializeToBytes(&wire.NewMinorBlockHeaderListCommand{ + RootBlockHeader: nil, + MinorBlockHeaderList: nil, + }) + if err != nil { + t.Fatalf("serialize command: %v", err) + } + + // Send a non-RPC command (rpc_id must be 0). + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: branch, ClusterPeerID: clusterPeerID}, + Opcode: byte(wire.CommandOpNewMinorBlockHeaderList), + RPCID: 0, + Payload: cmdPayload, + }) + + // Non-RPC commands produce no response. Verify no frame is sent back. + if err := serverConn.SetReadDeadline(time.Now().Add(200 * time.Millisecond)); err != nil { + t.Fatalf("set read deadline: %v", err) + } + if _, err := wire.ReadFrame(serverConn, 0); err == nil { + t.Fatal("expected no response for non-RPC command") + } + + // MasterConn must still be alive — a follow-up PING should work. + pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{ + ID: []byte("master"), + FullShardIDList: []uint32{0x00010001}, + }) + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001}, + Opcode: byte(wire.ClusterOpPing), + RPCID: 1, + Payload: pingPayload, + }) + resp := readMasterFrame(t, serverConn) + if resp.Opcode != byte(wire.ClusterOpPong) { + t.Fatalf("expected PONG after non-RPC command, got opcode 0x%x", resp.Opcode) + } +} + +// TestPeerConn_CloseStopsReadLoop verifies that closing a PeerConn causes its +// read loop to exit (no goroutine leak). After Close(), the read loop's +// deferred Close should be a no-op and the closed channel should be signaled. +func TestPeerConn_CloseStopsReadLoop(t *testing.T) { + client, _, cleanup := newMasterConnWithDispatcher(t) + defer cleanup() + + const clusterPeerID uint64 = 43 + const branch uint32 = 0x00010001 + + client.dispatcher.CreatePeerConns(clusterPeerID, []uint32{branch}, client, log.New()) + pc := client.dispatcher.peers[clusterPeerID][branch] + + // Verify the PeerConn is active and its read loop is running. + if !pc.IsActive() { + t.Fatal("expected PeerConn to be active after Start") + } + + // Close the PeerConn. + pc.Close() + + // The closed channel should be signaled promptly. + select { + case <-pc.WaitUntilClosed(): + // OK + case <-time.After(2 * time.Second): + t.Fatal("PeerConn read loop did not exit after Close") + } + + // HandleFrame after close should return an error. + if err := pc.HandleFrame(&wire.Frame{}); err == nil { + t.Fatal("expected error from HandleFrame after Close") + } +} + +// TestDispatcher_DestroyNonexistentPeer verifies that destroying a non-existent +// cluster_peer_id is a no-op and does not affect MasterConn. +// Python: slave.py#L321-L327 (pop with default None) +func TestDispatcher_DestroyNonexistentPeer(t *testing.T) { + client, serverConn, cleanup := newMasterConnWithDispatcher(t) + defer cleanup() + + // Destroy a cluster_peer_id that was never created. + client.dispatcher.DestroyPeerConns(9999) + + // MasterConn must still be alive. + pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{ + ID: []byte("master"), + FullShardIDList: []uint32{0x00010001}, + }) + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001}, + Opcode: byte(wire.ClusterOpPing), + RPCID: 1, + Payload: pingPayload, + }) + resp := readMasterFrame(t, serverConn) + if resp.Opcode != byte(wire.ClusterOpPong) { + t.Fatalf("expected PONG after destroying nonexistent peer, got opcode 0x%x", resp.Opcode) + } + + // Destroy the same ID again — should still be idempotent. + client.dispatcher.DestroyPeerConns(9999) + + writeMasterFrame(t, serverConn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001}, + Opcode: byte(wire.ClusterOpPing), + RPCID: 2, + Payload: pingPayload, + }) + resp = readMasterFrame(t, serverConn) + if resp.Opcode != byte(wire.ClusterOpPong) { + t.Fatalf("expected PONG after second destroy, got opcode 0x%x", resp.Opcode) + } +}