qkc/slave 06: add PeerConn, Dispatcher, and peer routing

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
+}

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)