qkc/slave 06: add PeerConn and peer routing

qkc/cluster/slave/master_conn.go

@@ -16,6 +16,17 @@ import (
 	"github.com/ethereum/go-ethereum/qkc/types"
 )
 
+// PeerResolver resolves the virtual PeerConn a forwarded peer frame is
+// addressed to. It is implemented by the composition layer that owns the peer
+// registry and injected into MasterConn, which uses it only for frame routing
+// — never for request delegation (that is MasterHandler's job).
+type PeerResolver interface {
+	// LookupPeer returns the PeerConn for (clusterPeerID, branch), or nil when
+	// this slave has none; the frame is then dropped (py: slave.py:131-146
+	// NULL_CONNECTION).
+	LookupPeer(clusterPeerID uint64, branch uint32) *PeerConn
+}
+
 // MasterHandler handles master requests delegated by MasterConn.
 // It is implemented by the composition layer and injected into MasterConn.
 type MasterHandler interface {
@@ -71,8 +82,7 @@ type MasterHandler interface {
 	GetTotalBalance(req *wire.GetTotalBalanceRequest) (*wire.GetTotalBalanceResponse, error)
 }
 
-// MasterConnConfig configures a MasterConn. Conn and Handler are required;
-// Logger defaults to log.Root().
+// MasterConnConfig configures a MasterConn.
 type MasterConnConfig struct {
 	// Conn is the accepted TCP connection from the master. The slave never
 	// dials the master (py: MasterServer connects, SlaveServer listens).
@@ -87,9 +97,19 @@ type MasterConnConfig struct {
 	LocalID              []byte
 	LocalFullShardIDList []uint32
 
+	// ClusterShardIDs is the cluster-wide configured full shard id set
+	// (py: env.quark_chain_config.get_full_shard_ids()). routeFrame uses it to
+	// reject frames from a master for a branch outside the global config, which
+	// is fatal for the connection (py: slave.py:123-129 close_with_error).
+	ClusterShardIDs []uint32
+
 	// Handler handles master requests delegated by MasterConn.
 	Handler MasterHandler
 
+	// PeerResolver resolves forwarded peer frames (cluster_peer_id != 0) to
+	// their virtual PeerConn. It is consulted by the routing forwarder only.
+	PeerResolver PeerResolver
+
 	// Logger defaults to log.Root() if nil.
 	Logger log.Logger
 }
@@ -101,8 +121,14 @@ type MasterConn struct {
 	*conn.BaseConn
 
 	handler              MasterHandler
+	peerResolver         PeerResolver
 	localID              []byte
 	localFullShardIDList []uint32
+
+	// clusterShardIDs is the cluster-wide configured full shard id set
+	// (py: env.quark_chain_config.get_full_shard_ids()); a frame for a branch
+	// outside it closes the connection (see routeFrame).
+	clusterShardIDs map[uint32]struct{}
 }
 
 // NewMasterConn wraps an accepted net.Conn from the master.
@@ -114,16 +140,35 @@ func NewMasterConn(cfg MasterConnConfig) (*MasterConn, error) {
 	if cfg.Handler == nil {
 		return nil, errors.New("master handler must not be nil")
 	}
+	if cfg.PeerResolver == nil {
+		return nil, errors.New("master peer resolver must not be nil")
+	}
+	if len(cfg.ClusterShardIDs) == 0 {
+		return nil, errors.New("cluster shard ids is required")
+	}
 	readFrame := func(r io.Reader) (*wire.Frame, error) {
 		return wire.ReadFrame(r, cfg.MaxPayloadSize)
 	}
 
+	clusterShardIDs := make(map[uint32]struct{}, len(cfg.ClusterShardIDs))
+	for _, id := range cfg.ClusterShardIDs {
+		clusterShardIDs[id] = struct{}{}
+	}
+
 	mc := &MasterConn{
 		handler:              cfg.Handler,
+		peerResolver:         cfg.PeerResolver,
 		localID:              append([]byte(nil), cfg.LocalID...),
 		localFullShardIDList: append([]uint32(nil), cfg.LocalFullShardIDList...),
+		clusterShardIDs:      clusterShardIDs,
 	}
 
+	// Forwarder: route cluster_peer_id != 0 frames to virtual PeerConns.
+	// routeFrame returns false for master-local traffic so MasterConn handles
+	// it normally. The forwarder runs on the reader goroutine; it enqueues
+	// frames without blocking (the PeerConn inbound queue is unbounded).
+	forwarder := mc.routeFrame
+
 	mc.BaseConn = conn.NewBaseConn(conn.Config{
 		Transport: conn.NewTCPTransport(cfg.Conn, readFrame, wire.WriteFrame),
 		Serializers: map[byte]*conn.OpSerializer{
@@ -209,11 +254,8 @@ func NewMasterConn(cfg MasterConnConfig) (*MasterConn, error) {
 		NonRPCOps: map[byte]struct{}{
 			byte(wire.ClusterOpDestroyClusterPeerConnectionCommand): {},
 		},
-		// Forwarder stays nil: routing peer traffic (cluster_peer_id != 0)
-		// to virtual PeerConns is PR6 (Dispatcher as the frame consumer).
-		// Until then, any peer frame (CommandOp opcode) is unregistered and
-		// closes the connection — MasterConn must not receive peer traffic.
-		Logger: cfg.Logger,
+		Forwarder: forwarder,
+		Logger:    cfg.Logger,
 	})
 	return mc, nil
 }
@@ -254,6 +296,45 @@ func (mc *MasterConn) SendAddMinorBlockHeaderList(ctx context.Context, req *wire
 	return r, nil
 }
 
+// ── Frame routing ───────────────────────────────────────────────────────
+
+// routeFrame handles frames addressed to virtual peer connections.
+// cluster_peer_id == 0 is master-local traffic and returns false so the
+// normal MasterConn dispatcher handles it. Peer traffic is validated and
+// forwarded to the corresponding PeerConn.
+// A branch outside the GLOBAL configured shard set is fatal for the
+// connection (py: slave.py:123-129 close_with_error); a branch that is
+// globally valid but not owned/created locally, or an unknown peer id,
+// follows Python's NULL_CONNECTION semantics (slave.py:131-146): the
+// frame is consumed and dropped without closing the connection.
+func (mc *MasterConn) routeFrame(frame *wire.Frame) bool {
+	if frame.Meta.ClusterPeerID == 0 {
+		return false
+	}
+
+	if _, ok := mc.clusterShardIDs[frame.Meta.Branch]; !ok {
+		mc.Logger().Error(
+			"incorrect forwarding branch",
+			"branch", fmt.Sprintf("0x%x", frame.Meta.Branch),
+		)
+		mc.Close()
+		return true
+	}
+
+	pc := mc.peerResolver.LookupPeer(frame.Meta.ClusterPeerID, frame.Meta.Branch)
+	if pc == nil {
+		// Covers both "shard valid globally but not created locally"
+		// (slave.py:131-134) and "peer not found" (slave.py:136-146): drop,
+		// keep the connection.
+		mc.Logger().Warn("dropping frame for unknown virtual peer connection",
+			"cluster_peer_id", frame.Meta.ClusterPeerID, "branch", frame.Meta.Branch)
+		return true
+	}
+
+	pc.HandleFrame(frame)
+	return true
+}
+
 // ── topology & shard activation handlers ───────────────────────────────────
 
 // handlePing handles the master's PING.

qkc/cluster/slave/master_conn_test.go

@@ -88,6 +88,12 @@ func (h *fakeMasterHandler) ConnectToSlaves(req *wire.ConnectToSlavesRequest) (*
 	return resp, nil
 }
 
+// LookupPeer implements PeerResolver. This fake models a slave with no
+// PeerConns at all: every lookup misses, so a peer frame follows MasterConn's
+// NULL_CONNECTION path (dropped, connection kept). Fakes that own a peer
+// registry (fakeSlaveService) shadow this with a real lookup.
+func (h *fakeMasterHandler) LookupPeer(uint64, uint32) *PeerConn { return nil }
+
 func (h *fakeMasterHandler) Mine(*wire.MineRequest) (*wire.MineResponse, error) {
 	return &wire.MineResponse{}, nil
 }
@@ -260,7 +266,9 @@ func newMasterConnWithPeer(t *testing.T, handler *fakeMasterHandler) (*MasterCon
 		Conn:                 slaveConn,
 		LocalID:              []byte("go-slave"),
 		LocalFullShardIDList: []uint32{0x00010001},
+		ClusterShardIDs:      []uint32{0x00010001},
 		Handler:              handler,
+		PeerResolver:         handler,
 		Logger:               log.New(),
 	})
 	if err != nil {
@@ -281,13 +289,26 @@ func newMasterConnWithPeer(t *testing.T, handler *fakeMasterHandler) (*MasterCon
 // ── construction ─────────────────────────────────────────────────────────────
 
 func TestMasterConn_ConfigValidation(t *testing.T) {
-	// Nil conn / nil handler must be rejected.
+	// Nil conn / nil handler / nil peer resolver must be rejected.
 	if _, err := NewMasterConn(MasterConnConfig{}); err == nil {
 		t.Fatal("expected error for nil conn")
 	}
 	if _, err := NewMasterConn(MasterConnConfig{Conn: &net.TCPConn{}}); err == nil {
 		t.Fatal("expected error for nil master handler")
 	}
+	if _, err := NewMasterConn(MasterConnConfig{
+		Conn:    &net.TCPConn{},
+		Handler: &fakeMasterHandler{},
+	}); err == nil {
+		t.Fatal("expected error for nil peer resolver")
+	}
+	if _, err := NewMasterConn(MasterConnConfig{
+		Conn:         &net.TCPConn{},
+		Handler:      &fakeMasterHandler{},
+		PeerResolver: &fakeMasterHandler{},
+	}); err == nil {
+		t.Fatal("expected error for empty cluster shard ids")
+	}
 }
 
 // TestMasterConn_IdentitySnapshot verifies that LocalID and
@@ -306,7 +327,9 @@ func TestMasterConn_IdentitySnapshot(t *testing.T) {
 		Conn:                 slaveConn,
 		LocalID:              localID,
 		LocalFullShardIDList: shardList,
+		ClusterShardIDs:      []uint32{0x00010001},
 		Handler:              &fakeMasterHandler{},
+		PeerResolver:         &fakeMasterHandler{},
 		Logger:               log.New(),
 	})
 	if err != nil {
@@ -366,8 +389,10 @@ func TestMasterConn_Ping(t *testing.T) {
 			t.Fatalf("serialize ping: %v", err)
 		}
 		// RPC IDs strictly increase across both pings.
+		// cluster_peer_id 0 keeps the frame on the master-local path; the
+		// branch is echoed untouched by the PONG.
 		if err := peer.send(&wire.Frame{
-			Meta:    wire.ClusterMetadata{},
+			Meta:    wire.ClusterMetadata{Branch: 0x00010001},
 			Opcode:  byte(wire.ClusterOpPing),
 			RPCID:   uint64(i + 1),
 			Payload: payload,
@@ -382,6 +407,12 @@ func TestMasterConn_Ping(t *testing.T) {
 		if resp.RPCID != uint64(i+1) {
 			t.Fatalf("pong rpc_id: got %d, want %d", resp.RPCID, uint64(i+1))
 		}
+		if resp.Meta.ClusterPeerID != 0 {
+			t.Fatalf("master-local response must keep cluster_peer_id 0, got %d", resp.Meta.ClusterPeerID)
+		}
+		if resp.Meta.Branch != 0x00010001 {
+			t.Fatalf("pong branch: got 0x%x, want 0x%x", resp.Meta.Branch, 0x00010001)
+		}
 		var pong wire.PongResponse
 		if err := serialize.Deserialize(serialize.NewByteBuffer(resp.Payload), &pong); err != nil {
 			t.Fatalf("deserialize pong: %v", err)
@@ -761,7 +792,9 @@ func TestMasterConn_SendAddMinorBlockHeader(t *testing.T) {
 		Conn:                 clientConn,
 		LocalID:              []byte("slave"),
 		LocalFullShardIDList: []uint32{0x00010001},
+		ClusterShardIDs:      []uint32{0x00010001},
 		Handler:              &fakeMasterHandler{},
+		PeerResolver:         &fakeMasterHandler{},
 		Logger:               log.New(),
 	})
 	if err != nil {
@@ -849,7 +882,9 @@ func TestMasterConn_SendAddMinorBlockHeaderList(t *testing.T) {
 		Conn:                 clientConn,
 		LocalID:              []byte("slave"),
 		LocalFullShardIDList: []uint32{0x00010001},
+		ClusterShardIDs:      []uint32{0x00010001},
 		Handler:              &fakeMasterHandler{},
+		PeerResolver:         &fakeMasterHandler{},
 		Logger:               log.New(),
 	})
 	if err != nil {