diff --git a/qkc/cluster/slave/interop_harness.go b/qkc/cluster/slave/interop_harness.go new file mode 100644 index 00000000000..dbf494c3acb --- /dev/null +++ b/qkc/cluster/slave/interop_harness.go @@ -0,0 +1,627 @@ +// Copyright 2026-2027, QuarkChain. + +//go:build interop + +package slave + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/internal/testlog" + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/qkc/types" +) + +// ============================================================================= +// Environment guards +// ============================================================================= + +// requirePyquarkchain returns the pyquarkchain root path. +// Skips the test if PYQUARKCHAIN is not set or the directory doesn't exist. +func requirePyquarkchain(t *testing.T) string { + t.Helper() + root := os.Getenv("PYQUARKCHAIN") + if root == "" { + t.Skip("PYQUARKCHAIN not set") + } + if _, err := os.Stat(root); err != nil { + t.Fatalf("PYQUARKCHAIN directory not found: %v", err) + } + return root +} + +// requirePython3 skips the test if python3 is not available. +func requirePython3(t *testing.T) { + t.Helper() + if _, err := exec.LookPath("python3"); err != nil { + t.Skip("python3 not available") + } +} + +// safeBuffer is a goroutine-safe bytes.Buffer. +type safeBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *safeBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *safeBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + +// freePort returns an unused TCP port on 127.0.0.1. +func freePort(t *testing.T) int { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + port := ln.Addr().(*net.TCPAddr).Port + ln.Close() + return port +} + +// ============================================================================= +// interopBackend — communication-only slave backend for Python interop tests +// ============================================================================= + +// interopBackend serves every business RPC with a trivially-successful response +// and records the lifecycle events the tests assert on (shard creation from a +// PING root tip). It satisfies SlaveConfig.Master, .Peer and .Xshard. Business +// methods are inherited from fakeMasterHandler, stubPeerHandler and +// testXshardHandler; only orchestration-relevant methods are overridden here. +type interopBackend struct { + *fakeMasterHandler + fullShardIDList []uint32 + + mu sync.Mutex + // createShardsCalls counts CreateShards invocations (shard creation only, + // driven by ShardCreator below, not the business backend). + createShardsCalls int + // shardsCreated is closed once at least one PING root tip initialized this + // slave's shards. It is the readiness gate WaitBootstrap polls. + shardsCreated chan struct{} +} + +func newInteropBackend(fullShardIDList []uint32) *interopBackend { + return &interopBackend{ + fakeMasterHandler: &fakeMasterHandler{}, + fullShardIDList: append([]uint32(nil), fullShardIDList...), + shardsCreated: make(chan struct{}), + } +} + +// ShardCreator implements MasterBackend: it records the invocation and reports +// every configured shard as created, closing shardsCreated on the first call. +// The root tip itself carries no meaning for the communication-only backend. +func (b *interopBackend) ShardCreator(_ *types.RootBlock) ([]uint32, error) { + b.mu.Lock() + b.createShardsCalls++ + b.mu.Unlock() + select { + case <-b.shardsCreated: + default: + close(b.shardsCreated) + } + return append([]uint32(nil), b.fullShardIDList...), nil +} + +func (b *interopBackend) CreateShardsCalls() int { + b.mu.Lock() + defer b.mu.Unlock() + return b.createShardsCalls +} + +func (b *interopBackend) ShardsCreated() <-chan struct{} { + return b.shardsCreated +} + +// ============================================================================= +// Slave startup +// ============================================================================= + +// startTestSlave starts a SlaveComm with a communication-only backend. fullShards +// are this slave's shards and must equal the config's FULL_SHARD_ID_LIST for the +// master's PONG validation; clusterShards are the cluster-wide shard set. +// peerOver optionally overrides the shared PeerHandler (defaults to stubPeerHandler). +func startTestSlave(t *testing.T, id string, fullShards, clusterShards []uint32, peerOver ...PeerHandler) (*SlaveComm, *interopBackend, int) { + t.Helper() + backend := newInteropBackend(fullShards) + + var peerHandler PeerHandler = stubPeerHandler{} + if len(peerOver) > 0 { + peerHandler = peerOver[0] + } + + for attempt := 0; ; attempt++ { + port := freePort(t) + cfg := SlaveConfig{ + ID: []byte(id), + FullShardIDList: append([]uint32(nil), fullShards...), + ClusterFullShardIDList: append([]uint32(nil), clusterShards...), + Port: port, + MaxPayloadSize: 0, + Logger: testlog.Logger(t, log.LvlInfo), + Master: backend, + Peer: peerHandler, + Xshard: testXshardHandler{}, + } + srv, err := NewSlaveComm(cfg) + if err != nil { + t.Fatalf("new slave server: %v", err) + } + if err := srv.Start(); err != nil { + if attempt < 5 && strings.Contains(err.Error(), "address already in use") { + time.Sleep(10 * time.Millisecond) + continue + } + t.Fatalf("start slave server: %v", err) + } + t.Cleanup(func() { srv.Stop() }) + return srv, backend, port + } +} + +// clusterShardSet returns the union of all slaves' shard sets. +func clusterShardSet(shardLists [][]uint32) []uint32 { + seen := make(map[uint32]struct{}) + var out []uint32 + for _, list := range shardLists { + for _, s := range list { + if _, ok := seen[s]; ok { + continue + } + seen[s] = struct{}{} + out = append(out, s) + } + } + return out +} + +// ============================================================================= +// Python master process driver +// ============================================================================= + +// scenarioMasterProc runs a testdata/master_harness.py scenario in the +// background, capturing its combined stdout/stderr. Scenarios run to completion +// (exit 0 on success) but some need to be observed mid-flight, so the process +// is started eagerly and its output polled line by line. +type scenarioMasterProc struct { + t *testing.T + cmd *exec.Cmd + cancel context.CancelFunc + out *safeBuffer + done chan struct{} // closed once the process exits + err error // set before done is closed +} + +// startScenarioMaster drives a master_harness.py sub-command in the background. +// subcommand is one of "rpc", "peer", "disconnect" (and optionally "bootstrap"). +// The caller must Close() it to release the process. +func startScenarioMaster(t *testing.T, subcommand string, args ...string) *scenarioMasterProc { + t.Helper() + requirePython3(t) + pyRoot := requirePyquarkchain(t) + + script := masterScript(t) + pyArgs := append([]string{"-u", script, subcommand}, args...) + + ctx, cancel := context.WithCancel(context.Background()) + cmd := exec.CommandContext(ctx, "python3", pyArgs...) + cmd.Env = append(os.Environ(), "PYQUARKCHAIN="+pyRoot) + + out := &safeBuffer{} + cmd.Stdout = out + cmd.Stderr = out + + if err := cmd.Start(); err != nil { + cancel() + t.Fatalf("start master_harness: %v", err) + } + + p := &scenarioMasterProc{t: t, cmd: cmd, cancel: cancel, out: out, done: make(chan struct{})} + go func() { + perr := cmd.Wait() + if perr != nil { + p.cancel() + } + p.err = perr + close(p.done) + }() + + t.Cleanup(p.Close) + return p +} + +// output returns all output so far. +func (p *scenarioMasterProc) output() string { + return p.out.String() +} + +// WaitLine polls the output until substr appears or the process exits or +// timeout elapses. Returns true when the line appeared in time. +func (p *scenarioMasterProc) WaitLine(substr string, timeout time.Duration) bool { + p.t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if strings.Contains(p.output(), substr) { + return true + } + select { + case <-p.done: + return strings.Contains(p.output(), substr) + case <-time.After(50 * time.Millisecond): + } + } + return false +} + +// WaitExit waits for the process to exit within timeout and returns its error. +func (p *scenarioMasterProc) WaitExit(timeout time.Duration) error { + p.t.Helper() + select { + case <-p.done: + case <-time.After(timeout): + return context.DeadlineExceeded + } + return p.err +} + +// Close cancels the process and waits for it to exit. Idempotent. +func (p *scenarioMasterProc) Close() { + p.cancel() + select { + case <-p.done: + case <-time.After(5 * time.Second): + } +} + +// masterScript returns the path to testdata/master_harness.py next to this file. +func masterScript(t *testing.T) string { + t.Helper() + _, thisFile, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("cannot determine source file location") + } + return filepath.Join(filepath.Dir(thisFile), "testdata", "master_harness.py") +} + +// ============================================================================= +// InteropCluster — full Python Master + N Go Slaves bootstrap harness +// ============================================================================= +// +// InteropCluster starts N Go slaves, writes a cluster_config.json, and launches +// the real Python Master via master_harness.py bootstrap → master.main(). The +// master performs the full bootstrap by itself: connects to every slave, +// PING/PONGs, instructs the slaves to dial each other (CONNECT_TO_SLAVES → +// xshard), and initializes shards with a root-tip PING. WaitBootstrap reports +// when all of that has happened. InteropCluster implements no protocol and no +// business logic; it only owns processes, ports and config. + +type InteropCluster struct { + slaves []*SlaveComm + backends []*interopBackend + shardLists [][]uint32 + ports []int + p2pPort int + configPath string + + masterCmd *exec.Cmd + masterOut *safeBuffer + cancel context.CancelFunc +} + +// startInteropCluster starts the Go slaves, generates cluster_config.json and +// launches the full real Python Master. The caller must call Stop() to tear +// down the master process. shardLists[i] is slave i's full shard id list. +func startInteropCluster(t *testing.T, shardLists [][]uint32) *InteropCluster { + t.Helper() + + requirePyquarkchain(t) + requirePython3(t) + + n := len(shardLists) + if n == 0 { + t.Fatal("need at least 1 slave") + } + + clusterShards := clusterShardSet(shardLists) + + // 1. Start Go slaves. + slaves := make([]*SlaveComm, n) + backends := make([]*interopBackend, n) + ports := make([]int, n) + for i := range n { + id := fmt.Sprintf("S%d", i) + slaves[i], backends[i], ports[i] = startTestSlave(t, id, shardLists[i], clusterShards) + } + + // 2. Reserve P2P port. + p2pPort := freePort(t) + + // 3. Generate cluster_config.json. + configPath := filepath.Join(t.TempDir(), "cluster_config.json") + writeClusterConfig(t, configPath, ports, p2pPort, shardLists) + + // 4. Launch the full real Python Master. + ctx, cancel := context.WithCancel(context.Background()) + cmd := exec.CommandContext(ctx, "python3", "-u", masterScript(t), "bootstrap", "--cluster_config", configPath) + cmd.Env = append(os.Environ(), "PYQUARKCHAIN="+requirePyquarkchain(t)) + var out safeBuffer + cmd.Stdout = &out + cmd.Stderr = &out + + if err := cmd.Start(); err != nil { + cancel() + t.Fatalf("start master: %v", err) + } + t.Cleanup(func() { cancel(); cmd.Wait() }) + + return &InteropCluster{ + slaves: slaves, + backends: backends, + shardLists: shardLists, + ports: ports, + p2pPort: p2pPort, + configPath: configPath, + masterCmd: cmd, + masterOut: &out, + cancel: cancel, + } +} + +// Stop cancels the Python master process. Go slaves are stopped by t.Cleanup +// registered in startTestSlave. +func (c *InteropCluster) Stop() { + c.cancel() +} + +// WaitBootstrap returns true when every slave is fully bootstrapped: shards +// initialized by a root-tip PING and, for multi-slave clusters, at least one +// registered xshard connection. +func (c *InteropCluster) WaitBootstrap(timeout time.Duration) bool { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if c.bootstrapReady() { + return true + } + time.Sleep(100 * time.Millisecond) + } + return false +} + +// bootstrapReady reports whether all slaves reached the fully-initialized state. +func (c *InteropCluster) bootstrapReady() bool { + n := len(c.slaves) + for i := range n { + select { + case <-c.backends[i].ShardsCreated(): + default: + return false + } + if n > 1 && !hasXshard(c.slaves[i], c.shardLists[i], clusterShardSet(c.shardLists)) { + return false + } + } + return true +} + +// hasXshard reports whether the slave registered an xshard connection to a +// shard it does not itself own (i.e. a real peer slave, not a self-link). +func hasXshard(s *SlaveComm, ownShards, clusterShards []uint32) bool { + own := make(map[uint32]struct{}, len(ownShards)) + for _, shard := range ownShards { + own[shard] = struct{}{} + } + for _, shard := range clusterShards { + if _, ok := own[shard]; ok { + continue + } + if len(s.xshardPool.Lookup(shard)) > 0 { + return true + } + } + return false +} + +// Slave returns the i-th SlaveComm. +func (c *InteropCluster) Slave(i int) *SlaveComm { return c.slaves[i] } + +// SlaveCount returns the number of Go slaves in the cluster. +func (c *InteropCluster) SlaveCount() int { return len(c.slaves) } + +// Backend returns the i-th communication-only backend. +func (c *InteropCluster) Backend(i int) *interopBackend { return c.backends[i] } + +// MasterOutput returns the combined stdout/stderr of the Python master. +func (c *InteropCluster) MasterOutput() string { return c.masterOut.String() } + +// ============================================================================= +// Cluster config generation +// ============================================================================= + +func writeClusterConfig(t *testing.T, path string, ports []int, p2pPort int, shardLists [][]uint32) { + t.Helper() + config := buildClusterConfig(ports, p2pPort, shardLists) + data, err := json.MarshalIndent(config, "", " ") + if err != nil { + t.Fatalf("marshal cluster config: %v", err) + } + if err := os.WriteFile(path, data, 0644); err != nil { + t.Fatalf("write cluster config: %v", err) + } +} + +func buildClusterConfig(ports []int, p2pPort int, shardLists [][]uint32) map[string]any { + n := len(ports) + + formatShardList := func(shards []uint32) []string { + list := make([]string, len(shards)) + for i, s := range shards { + list[i] = fmt.Sprintf("0x%08x", s) + } + return list + } + + slaveList := make([]any, n) + for i := range n { + slaveList[i] = map[string]any{ + "HOST": "127.0.0.1", + "PORT": ports[i], + "ID": fmt.Sprintf("S%d", i), + "FULL_SHARD_ID_LIST": formatShardList(shardLists[i]), + } + } + chains := make([]any, n) + for i := range n { + chains[i] = chainConfig(i) + } + + return map[string]any{ + "P2P_PORT": p2pPort, + "JSON_RPC_PORT": 0, + "PRIVATE_JSON_RPC_PORT": 0, + "ENABLE_TRANSACTION_HISTORY": false, + "DB_PATH_ROOT": "", + "LOG_LEVEL": "info", + "START_SIMULATED_MINING": false, + "CLEAN": false, + "GENESIS_DIR": nil, + + "QUARKCHAIN": map[string]any{ + "CHAIN_SIZE": n, + "BASE_ETH_CHAIN_ID": 110000, + "MAX_NEIGHBORS": 32, + "NETWORK_ID": 255, + "TRANSACTION_QUEUE_SIZE_LIMIT_PER_SHARD": 10000, + "BLOCK_EXTRA_DATA_SIZE_LIMIT": 1024, + "GUARDIAN_PUBLIC_KEY": "ab856abd0983a82972021e454fcf66ed5940ed595b0898bcd75cbe2d0a51a00f5358b566df22395a2a8bf6c022c1d51a2c3defe654e91a8d244947783029694d", + "ROOT_SIGNER_PRIVATE_KEY": nil, + "P2P_PROTOCOL_VERSION": 0, + "P2P_COMMAND_SIZE_LIMIT": 134217728, + "SKIP_ROOT_DIFFICULTY_CHECK": false, + "SKIP_MINOR_DIFFICULTY_CHECK": false, + "GENESIS_TOKEN": "QKC", + "ROOT": rootConfig(), + "CHAINS": chains, + "REWARD_TAX_RATE": 0.5, + "BLOCK_REWARD_DECAY_FACTOR": 0.88, + "ROOT_CHAIN_POSW_CONTRACT_BYTECODE_HASH": "0000000000000000000000000000000000000000000000000000000000000000", + }, + + "MASTER": map[string]any{ + "MASTER_TO_SLAVE_CONNECT_RETRY_DELAY": 1.0, + }, + + "SLAVE_LIST": slaveList, + + "P2P": map[string]any{ + "BOOT_NODES": "", + "PRIV_KEY": "", + "MAX_PEERS": 25, + "UPNP": false, + "ALLOW_DIAL_IN_RATIO": 1.0, + "PREFERRED_NODES": "", + "DISCOVERY_ONLY": false, + "CRAWLING_ROUTING_TABLE_FILE_PATH": nil, + }, + + "MONITORING": map[string]any{ + "NETWORK_NAME": "", + "CLUSTER_ID": "127.0.0.1", + "KAFKA_REST_ADDRESS": "", + "MINER_TOPIC": "qkc_miner", + "PROPAGATION_TOPIC": "block_propagation", + "ERRORS": "error", + }, + } +} + +func rootConfig() map[string]any { + return map[string]any{ + "MAX_STALE_ROOT_BLOCK_HEIGHT_DIFF": 22500, + "CONSENSUS_TYPE": "POW_SIMULATE", + "CONSENSUS_CONFIG": map[string]any{ + "TARGET_BLOCK_TIME": 10, + "REMOTE_MINE": false, + }, + "GENESIS": map[string]any{ + "VERSION": 0, + "HEIGHT": 0, + "HASH_PREV_BLOCK": "0000000000000000000000000000000000000000000000000000000000000000", + "HASH_MERKLE_ROOT": "0000000000000000000000000000000000000000000000000000000000000000", + "TIMESTAMP": 1556639999, + "DIFFICULTY": 100000, + "NONCE": 0, + }, + "COINBASE_ADDRESS": "000000000000000000000000000000000000000000000000", + "COINBASE_AMOUNT": json.Number("156000000000000000000"), + "DIFFICULTY_ADJUSTMENT_CUTOFF_TIME": 40, + "DIFFICULTY_ADJUSTMENT_FACTOR": 1024, + "EPOCH_INTERVAL": 525600, + "POSW_CONFIG": map[string]any{ + "ENABLED": false, + "ENABLE_TIMESTAMP": 0, + "DIFF_DIVIDER": 100, + "WINDOW_SIZE": 256, + "TOTAL_STAKE_PER_BLOCK": 0, + }, + } +} + +func chainConfig(chainID int) map[string]any { + return map[string]any{ + "CHAIN_ID": chainID, + "SHARD_SIZE": 1, + "DEFAULT_CHAIN_TOKEN": "QKC", + "CONSENSUS_TYPE": "POW_SIMULATE", + "CONSENSUS_CONFIG": map[string]any{ + "TARGET_BLOCK_TIME": 10, + "REMOTE_MINE": false, + }, + "GENESIS": map[string]any{ + "ROOT_HEIGHT": 0, + "VERSION": 0, + "HEIGHT": 0, + "HASH_PREV_MINOR_BLOCK": "0000000000000000000000000000000000000000000000000000000000000000", + "HASH_MERKLE_ROOT": "0000000000000000000000000000000000000000000000000000000000000000", + "EXTRA_DATA": "497420776173207468652062657374206f662074696d6573", + "TIMESTAMP": 1556639999, + "DIFFICULTY": 10000, + "GAS_LIMIT": 12000000, + "NONCE": 0, + "ALLOC": map[string]any{}, + }, + "COINBASE_ADDRESS": "000000000000000000000000000000000000000000000000", + "COINBASE_AMOUNT": json.Number("6500000000000000000"), + "DIFFICULTY_ADJUSTMENT_CUTOFF_TIME": 7, + "DIFFICULTY_ADJUSTMENT_FACTOR": 512, + "EXTRA_SHARD_BLOCKS_IN_ROOT_BLOCK": 3, + "POSW_CONFIG": map[string]any{ + "ENABLED": false, + "DIFF_DIVIDER": 20, + "WINDOW_SIZE": 256, + "TOTAL_STAKE_PER_BLOCK": 0, + }, + "EPOCH_INTERVAL": 3153600, + } +} diff --git a/qkc/cluster/slave/interop_test.go b/qkc/cluster/slave/interop_test.go new file mode 100644 index 00000000000..58d018ba82b --- /dev/null +++ b/qkc/cluster/slave/interop_test.go @@ -0,0 +1,327 @@ +// Copyright 2026-2027, QuarkChain. + +//go:build interop + +package slave + +import ( + "context" + "fmt" + "math/big" + "strconv" + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/qkc/account" + "github.com/ethereum/go-ethereum/qkc/cluster/wire" + qkcCommon "github.com/ethereum/go-ethereum/qkc/common" + "github.com/ethereum/go-ethereum/qkc/serialize" + "github.com/ethereum/go-ethereum/qkc/types" +) + +// ============================================================================= +// Interop tests: real Python Master ↔ real Go Slave +// +// These verify wire / opcode / serializer / bootstrap compatibility against the +// real pyquarkchain wire stack, driven by testdata/master_harness.py. All +// business behavior is served by the communication-only interopBackend. +// ============================================================================= + +// TestInteropBootstrap runs the full Python Master (master.main()) against two +// Go slaves and verifies the complete bootstrap: PING/PONG, root-tip shard +// initialization, and CONNECT_TO_SLAVES establishing xshard connections. +func TestInteropBootstrap(t *testing.T) { + cluster := startInteropCluster(t, [][]uint32{ + {0x00000001}, + {0x00010001}, + }) + defer cluster.Stop() + + if !cluster.WaitBootstrap(30 * time.Second) { + t.Fatalf("bootstrap timed out — master did not fully bootstrap all slaves\nmaster output:\n%s", cluster.MasterOutput()) + } + + for i := 0; i < cluster.SlaveCount(); i++ { + s := cluster.Slave(i) + if s.master.Load() == nil { + t.Errorf("slave %d has no established MasterConn", i) + } + if n := numXshardConns(s.xshardPool); n == 0 { + t.Errorf("slave %d registered no xshard connections", i) + } + // Bootstrap must not create any cluster peer connections. + if n := len(s.peers); n != 0 { + t.Errorf("slave %d has %d unexpected peer connections", i, n) + } + t.Logf("slave %d: master=%v createShards=%d xshard=%d", + i, s.master.Load() != nil, cluster.Backend(i).CreateShardsCalls(), numXshardConns(s.xshardPool)) + } +} + +// TestInteropMasterRpcRoundTrip drives a GET_ACCOUNT_DATA round-trip over the +// real wire and confirms the slave answers with error_code 0. +func TestInteropMasterRpcRoundTrip(t *testing.T) { + _, _, port := startTestSlave(t, "S0", []uint32{0x00000001}, []uint32{0x00000001}) + + zeroAddress := "0000000000000000000000000000000000000000" + p := startScenarioMaster(t, "rpc", "127.0.0.1", strconv.Itoa(port), "0x00000001", zeroAddress) + + if !p.WaitLine("RPC_OK error_code=0", 15*time.Second) { + t.Fatalf("rpc round-trip not confirmed\n%s", p.output()) + } + if err := p.WaitExit(15 * time.Second); err != nil { + t.Fatalf("scenario master failed: %v\n%s", err, p.output()) + } +} + +// TestInteropPeerCreateOrDestroy drives two CREATE_CLUSTER_PEER_CONNECTION and a +// single DESTROY over the real wire, confirming both peers appear, and that +// destroying peer 1 leaves peer 2 untouched. +func TestInteropPeerCreateOrDestroy(t *testing.T) { + slave, _, port := startTestSlave(t, "S0", []uint32{0x00000001}, []uint32{0x00000001}) + + // Pre-activate the local branch so created virtual peers get a PeerConn. + if err := slave.createShards(nil); err != nil { + t.Fatalf("create shards: %v", err) + } + + const peer1, peer2 = 1, 2 + p := startScenarioMaster(t, "peer", "127.0.0.1", strconv.Itoa(port), + "0x00000001", strconv.Itoa(peer1), strconv.Itoa(peer2), "--hold", "3") + + if !p.WaitLine("PEER_CREATED 1", 15*time.Second) { + t.Fatalf("peer 1 create not confirmed\n%s", p.output()) + } + if !waitForPeer(t, slave, peer1, true, 10*time.Second) { + t.Fatalf("peer 1 not present in slave registry\n%s", p.output()) + } + + if !p.WaitLine("PEER_CREATED 2", 15*time.Second) { + t.Fatalf("peer 2 create not confirmed\n%s", p.output()) + } + if !waitForPeer(t, slave, peer2, true, 10*time.Second) { + t.Fatalf("peer 2 not present in slave registry\n%s", p.output()) + } + + if !p.WaitLine("PEER_DESTROYED", 15*time.Second) { + t.Fatalf("peer destroy not confirmed\n%s", p.output()) + } + if !waitForPeer(t, slave, peer1, false, 10*time.Second) { + t.Fatalf("peer 1 not removed from slave registry") + } + if !waitForPeer(t, slave, peer2, true, 10*time.Second) { + t.Fatalf("peer 2 was removed with peer 1: destroy(peer1) must not affect peer2") + } + + if err := p.WaitExit(15 * time.Second); err != nil { + t.Fatalf("scenario master failed: %v\n%s", err, p.output()) + } +} + +// routingRecorder is the shared PeerHandler for TestInteropPeerMessageRouting. It +// signals over gotNewTxList when a NEW_TRANSACTION_LIST command is delivered. +// PeerHandler is shared across all of a slave's PeerConns and its methods carry no +// route identity, so routing-correctness for (cluster_peer_id, branch) is asserted +// by combining this delivery signal with the peer (cluster_peer_id, branch) registry. +type routingRecorder struct { + mu sync.Mutex + calls int + gotNewTxList chan struct{} +} + +func newRoutingRecorder() *routingRecorder { + return &routingRecorder{gotNewTxList: make(chan struct{})} +} + +func (r *routingRecorder) NewTransactionList(*wire.NewTransactionListCommand) error { + r.mu.Lock() + r.calls++ + r.mu.Unlock() + select { + case <-r.gotNewTxList: + default: + close(r.gotNewTxList) + } + return nil +} + +func (r *routingRecorder) NewMinorBlockHeaderList(*wire.NewMinorBlockHeaderListCommand) error { + return nil +} +func (r *routingRecorder) NewBlockMinor(*wire.NewBlockMinorCommand) error { return nil } + +func (r *routingRecorder) GetMinorBlockHeaderList(*wire.GetMinorBlockHeaderListRequest) (*wire.GetMinorBlockHeaderListResponse, error) { + return &wire.GetMinorBlockHeaderListResponse{}, nil +} + +func (r *routingRecorder) GetMinorBlockList(*wire.GetMinorBlockListRequest) (*wire.GetMinorBlockListResponse, error) { + return &wire.GetMinorBlockListResponse{}, nil +} + +func (r *routingRecorder) GetMinorBlockHeaderListWithSkip(*wire.GetMinorBlockHeaderListWithSkipRequest) (*wire.GetMinorBlockHeaderListResponse, error) { + return &wire.GetMinorBlockHeaderListResponse{}, nil +} + +// TestInteropPeerMessageRouting drives a P2P NEW_TRANSACTION_LIST command from the +// Python master through the real frame path: +// +// Python Master (ClusterMetadata{branch, cluster_peer_id}) +// → Go MasterConn.routeFrame +// → PeerConn.HandleFrame +// → PeerHandler.NewTransactionList +// +// It asserts the unique target peer (cluster_peer_id, branch) is registered and +// that the command actually reaches the PeerHandler. +func TestInteropPeerMessageRouting(t *testing.T) { + recorder := newRoutingRecorder() + slave, _, port := startTestSlave(t, "S0", []uint32{0x00000001}, []uint32{0x00000001}, recorder) + + // Pre-activate the local branch so the created virtual peer gets a PeerConn. + if err := slave.createShards(nil); err != nil { + t.Fatalf("create shards: %v", err) + } + + const clusterPeerID = 1 + const branch = 0x00000001 + p := startScenarioMaster(t, "peermsg", "127.0.0.1", strconv.Itoa(port), + "0x00000001", strconv.Itoa(clusterPeerID), fmt.Sprintf("%d", branch)) + + // The peer must be created and registered on the exact (cluster_peer_id, branch). + if !waitForPeer(t, slave, clusterPeerID, true, 10*time.Second) { + t.Fatalf("peer %d not created\n%s", clusterPeerID, p.output()) + } + slave.peersMu.RLock() + _, hasBranch := slave.peers[clusterPeerID][branch] + slave.peersMu.RUnlock() + if !hasBranch { + t.Fatalf("peer %d has no PeerConn for branch 0x%x\n%s", clusterPeerID, branch, p.output()) + } + + // Python now sends a NEW_TRANSACTION_LIST addressed to (cluster_peer_id, branch). + // It must be routed through the real PeerConn to the recording PeerHandler. + select { + case <-recorder.gotNewTxList: + t.Log("NEW_TRANSACTION_LIST reached PeerHandler via routeFrame→PeerConn") + case <-time.After(15 * time.Second): + t.Fatalf("peer command did not reach PeerHandler\n%s", p.output()) + } + + if err := p.WaitExit(15 * time.Second); err != nil { + t.Fatalf("scenario master failed: %v\n%s", err, p.output()) + } +} + +// TestInteropMasterDisconnectShutdown confirms that losing the master +// connection (per pyquarkchain semantics) shuts the slave down. +func TestInteropMasterDisconnectShutdown(t *testing.T) { + slave, _, port := startTestSlave(t, "S0", []uint32{0x00000001}, []uint32{0x00000001}) + + p := startScenarioMaster(t, "disconnect", "127.0.0.1", strconv.Itoa(port), "0x00000001") + + if !p.WaitLine("CONNECTED", 15*time.Second) { + t.Fatalf("master did not connect to slave\n%s", p.output()) + } + if !p.WaitLine("DISCONNECT_SENT", 15*time.Second) { + t.Fatalf("master did not close the connection\n%s", p.output()) + } + + select { + case <-slave.WaitStopped(): + t.Log("slave shut down after master disconnect") + case <-time.After(15 * time.Second): + t.Fatalf("slave did not shut down after master disconnect\n%s", p.output()) + } + + if err := p.WaitExit(15 * time.Second); err != nil { + t.Fatalf("scenario master failed: %v\n%s", err, p.output()) + } +} + +// waitForPeer polls slave.peers until the given clusterPeerID exists (true) or +// no longer exists (false). Returns false on timeout. +func waitForPeer(t *testing.T, srv *SlaveComm, clusterPeerID uint64, expectExists bool, timeout time.Duration) bool { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + srv.peersMu.RLock() + _, exists := srv.peers[clusterPeerID] + srv.peersMu.RUnlock() + if exists == expectExists { + return true + } + time.Sleep(100 * time.Millisecond) + } + return false +} + +// TestInteropAddMinorBlockHeaderToMaster exercises the only outbound protocol +// direction: Go Slave → Python Master. The slave serializes a real +// AddMinorBlockHeaderRequest, the Python harness decodes it with the real +// request serializer, replies with a real AddMinorBlockHeaderResponse, and the +// slave decodes it. This verifies Go wire serialization, Python wire +// deserialization, and the RPC request/response framing in the slave→master +// direction (py: SlaveServer.send_minor_block_header_to_master). +func TestInteropAddMinorBlockHeaderToMaster(t *testing.T) { + slave, _, port := startTestSlave(t, "S0", []uint32{0x00000001}, []uint32{0x00000001}) + + // The Python harness connects, blocks on PING→PONG, then prints READY: at + // that point the slave's MasterConn is established and active. + p := startScenarioMaster(t, "addminorblock", "127.0.0.1", strconv.Itoa(port), "0x00000001") + if !p.WaitLine("READY", 15*time.Second) { + t.Fatalf("python master did not complete handshake\n%s", p.output()) + } + + // Go encodes the request and awaits the Python-decoded response. + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + resp, err := slave.SendMinorBlockHeaderToMaster(ctx, newInteropMinorBlockHeaderRequest()) + if err != nil { + t.Fatalf("SendMinorBlockHeaderToMaster: %v\n%s", err, p.output()) + } + if resp.ErrorCode != 0 { + t.Fatalf("AddMinorBlockHeader error_code=%d, want 0\n%s", resp.ErrorCode, p.output()) + } + + if !p.WaitLine("ADD_MINOR_BLOCK_OK error_code=0", 15*time.Second) { + t.Fatalf("add minor block round-trip not confirmed\n%s", p.output()) + } + if err := p.WaitExit(15 * time.Second); err != nil { + t.Fatalf("scenario master failed: %v\n%s", err, p.output()) + } +} + +// newInteropMinorBlockHeaderRequest builds a minimal-but-legal +// AddMinorBlockHeaderRequest through the formal wire types. The field layout of +// MinorBlockHeader mirrors pyquarkchain core.MinorBlockHeader (core.py) so that +// the Python harness's real deserializer can consume the encoded bytes. +func newInteropMinorBlockHeaderRequest() *wire.AddMinorBlockHeaderRequest { + return &wire.AddMinorBlockHeaderRequest{ + MinorBlockHeader: newInteropMinorBlockHeader(), + TxCount: 0, + XShardTxCount: 0, + CoinbaseAmountMap: qkcCommon.NewEmptyTokenBalances(), + ShardStats: wire.ShardStats{Branch: 0x00000001}, + } +} + +func newInteropMinorBlockHeader() *types.MinorBlockHeader { + return &types.MinorBlockHeader{ + Version: 0, + Branch: account.Branch{Value: 0x00000001}, + Number: 0, + Coinbase: account.Address{}, + CoinbaseAmount: qkcCommon.NewEmptyTokenBalances(), + ParentHash: common.Hash{}, + PrevRootBlockHash: common.Hash{}, + GasLimit: &serialize.Uint256{}, + MetaHash: common.Hash{}, + Time: 0, + Difficulty: new(big.Int), + Nonce: 0, + Bloom: types.Bloom{}, + Extra: nil, + MixDigest: common.Hash{}, + } +} diff --git a/qkc/cluster/slave/slave.go b/qkc/cluster/slave/slave.go new file mode 100644 index 00000000000..35a7cb89ca3 --- /dev/null +++ b/qkc/cluster/slave/slave.go @@ -0,0 +1,673 @@ +// Copyright 2026-2027, QuarkChain. + +package slave + +import ( + "context" + "errors" + "fmt" + "net" + "strconv" + "sync" + "sync/atomic" + + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/qkc/cluster/conn" + "github.com/ethereum/go-ethereum/qkc/cluster/wire" + "github.com/ethereum/go-ethereum/qkc/types" +) + +// MasterBackend defines the business operations used by SlaveComm. +// It is implemented by the external slave runtime; SlaveComm consumes +// these operations and performs the communication-side orchestration. +type MasterBackend interface { + // ShardCreator creates the business runtime's shards for a root tip + // and returns the newly-created branches. + ShardCreator(rootTip *types.RootBlock) ([]uint32, error) + + // ── business RPCs ── + Mine(req *wire.MineRequest) (*wire.MineResponse, error) + GenTx(req *wire.GenTxRequest) (*wire.GenTxResponse, error) + AddRootBlock(req *wire.AddRootBlockRequest) (*wire.AddRootBlockResponse, error) + GetEcoInfoList(req *wire.GetEcoInfoListRequest) (*wire.GetEcoInfoListResponse, error) + GetNextBlockToMine(req *wire.GetNextBlockToMineRequest) (*wire.GetNextBlockToMineResponse, error) + AddMinorBlock(req *wire.AddMinorBlockRequest) (*wire.AddMinorBlockResponse, error) + GetUnconfirmedHeaders(req *wire.GetUnconfirmedHeadersRequest) (*wire.GetUnconfirmedHeadersResponse, error) + GetAccountData(req *wire.GetAccountDataRequest) (*wire.GetAccountDataResponse, error) + AddTransaction(req *wire.AddTransactionRequest) (*wire.AddTransactionResponse, error) + GetMinorBlock(req *wire.GetMinorBlockRequest) (*wire.GetMinorBlockResponse, error) + GetTransaction(req *wire.GetTransactionRequest) (*wire.GetTransactionResponse, error) + SyncMinorBlockList(req *wire.SyncMinorBlockListRequest) (*wire.SyncMinorBlockListResponse, error) + ExecuteTransaction(req *wire.ExecuteTransactionRequest) (*wire.ExecuteTransactionResponse, error) + GetTransactionReceipt(req *wire.GetTransactionReceiptRequest) (*wire.GetTransactionReceiptResponse, error) + GetTransactionListByAddress(req *wire.GetTransactionListByAddressRequest) (*wire.GetTransactionListByAddressResponse, error) + GetLogs(req *wire.GetLogRequest) (*wire.GetLogResponse, error) + EstimateGas(req *wire.EstimateGasRequest) (*wire.EstimateGasResponse, error) + GetStorageAt(req *wire.GetStorageRequest) (*wire.GetStorageResponse, error) + GetCode(req *wire.GetCodeRequest) (*wire.GetCodeResponse, error) + GasPrice(req *wire.GasPriceRequest) (*wire.GasPriceResponse, error) + GetWork(req *wire.GetWorkRequest) (*wire.GetWorkResponse, error) + SubmitWork(req *wire.SubmitWorkRequest) (*wire.SubmitWorkResponse, error) + CheckMinorBlock(req *wire.CheckMinorBlockRequest) (*wire.CheckMinorBlockResponse, error) + GetAllTransactions(req *wire.GetAllTransactionsRequest) (*wire.GetAllTransactionsResponse, error) + GetRootChainStakes(req *wire.GetRootChainStakesRequest) (*wire.GetRootChainStakesResponse, error) + GetTotalBalance(req *wire.GetTotalBalanceRequest) (*wire.GetTotalBalanceResponse, error) +} + +// ── masterHandler facade ───────────────────────────────────────────────────── +// +// masterHandler is the MasterHandler SlaveComm installs into MasterConn. It +// combines SlaveComm's communication orchestration with the external business +// backend: the topology & shard-activation commands below are served by +// SlaveComm itself; the business RPCs are embedded MasterBackend methods. +type masterHandler struct { + MasterBackend + comm *SlaveComm +} + +var _ MasterHandler = (*masterHandler)(nil) + +// newMasterHandler builds the MasterHandler MasterConn is configured with. +func (s *SlaveComm) newMasterHandler() MasterHandler { + return &masterHandler{ + MasterBackend: s.cfg.Master, + comm: s, + } +} + +// ── topology & shard activation: served by SlaveComm ── + +// CreateShards handles a PING root tip: it creates the local shards and equips +// them with PeerConns. +func (h *masterHandler) CreateShards(rootTip *types.RootBlock) error { + return h.comm.createShards(rootTip) +} + +// ConnectToSlaves dials the advertised slaves into the xshard pool. Per-entry +// failures are reported in the response's result list and the master +// connection stays up; the bootstrap shutdown decision belongs to the master +// side (see SlaveComm.connectToSlaves for the Python evidence chain). +func (h *masterHandler) ConnectToSlaves(req *wire.ConnectToSlavesRequest) (*wire.ConnectToSlavesResponse, error) { + return h.comm.connectToSlaves(req) +} + +// CreateClusterPeerConnection registers a new cluster peer and creates a +// PeerConn for it on every local branch. +func (h *masterHandler) CreateClusterPeerConnection(req *wire.CreateClusterPeerConnectionRequest) (*wire.CreateClusterPeerConnectionResponse, error) { + return h.comm.createClusterPeerConnection(req) +} + +// DestroyClusterPeerConnection removes a cluster peer and closes its PeerConns. +func (h *masterHandler) DestroyClusterPeerConnection(req *wire.DestroyClusterPeerConnectionCommand) error { + return h.comm.destroyClusterPeerConnection(req) +} + +// SlaveConfig holds the runtime configuration of a SlaveComm together with the +// handlers it delegates protocol work to. The slave runtime implements the +// handlers; SlaveComm only wires and owns the communication resources. +type SlaveConfig struct { + // ID is this slave's unique identifier (e.g., []byte("S0")). + ID []byte + // FullShardIDList contains the shards managed by this slave. + FullShardIDList []uint32 + // Port is the TCP port on which the slave listens for cluster connections. + Port int + // ClusterFullShardIDList is the cluster-wide shard id set (py: + // get_full_shard_ids()); feeds the xshard pool route filter and the + // MasterConn branch validator. + ClusterFullShardIDList []uint32 + // MaxPayloadSize limits incoming frame payload size; 0 disables the limit. + MaxPayloadSize uint32 + + // Master handles business RPCs routed through MasterConn. + Master MasterBackend + // Peer builds and serves slave-to-slave PeerConns for virtual cluster peers. + Peer PeerHandler + // Xshard serves requests received through XshardConns. + Xshard XshardHandler + + // Logger defaults to log.Root() if nil. + Logger log.Logger +} + +// Validate returns an error if the configuration is unusable. +func (cfg *SlaveConfig) Validate() error { + if len(cfg.ID) == 0 { + return errors.New("slave id is required") + } + if len(cfg.FullShardIDList) == 0 { + return errors.New("full shard id list is required") + } + if cfg.Port <= 0 { + return errors.New("slave port must be positive") + } + + if len(cfg.ClusterFullShardIDList) == 0 { + return errors.New("cluster full shard id list is required") + } + + if cfg.Master == nil { + return errors.New("master handler must not be nil") + } + if cfg.Peer == nil { + return errors.New("peer handler must not be nil") + } + if cfg.Xshard == nil { + return errors.New("xshard handler must not be nil") + } + return nil +} + +// SlaveComm owns the slave's communication resources: the listener, +// MasterConn, XshardPool, and virtual cluster-peer topology. +// +// Lifecycle: New → Start → Stop. Start is called once by the owner. +// Stop is idempotent and returns without waiting for goroutines to exit. +type SlaveComm struct { + cfg SlaveConfig + logger log.Logger + + listener net.Listener + // master is the established MasterConn (py: slave_server.master), published + // atomically by runMasterConn for the first inbound and never replaced. The + // Send*ToMaster APIs may run on any goroutine, so publication is atomic: Load + // is race-free against the single Store. nil means not established + // (ErrNotActive); open/closed is the delegate's own state. This records + // establishment, not classification — which inbound is the master is + // acceptLoop's loop-local control flow. + master atomic.Pointer[MasterConn] + + xshardPool *XshardPool + + // Peer topology, guarded by peersMu. Invariant: + // peers[p][b] exists ⇒ p ∈ clusterPeerIDs ∧ b ∈ localBranches. + peersMu sync.RWMutex + // localBranches is the set of branches currently served by this slave. + localBranches map[uint32]struct{} + // clusterPeerIDs is the set of announced virtual cluster peers (py: + // SlaveServer.cluster_peer_ids). + clusterPeerIDs map[uint64]struct{} + // peers is the (cluster_peer_id, branch) → PeerConn registry (py: shard.peers). + peers map[uint64]map[uint32]*PeerConn + + // shutdownOnce guards only the shutdown notification (py: shutdown_future.done()): + // close(stopped) must happen exactly once across Stop's concurrent triggers (owner, + // master loss, startup failure). The resource closes below are NOT once-guarded — + // each is individually idempotent and repeats on every Stop call, as in py. + shutdownOnce sync.Once + // stopped is the shutdown notification (py: SlaveServer.shutdown_future), closed + // once every close request has been issued, without waiting for the goroutines + // those closes unblock. Consumers (process main, tests) read it to learn shutdown + // was triggered. + stopped chan struct{} +} + +var _ PeerResolver = (*SlaveComm)(nil) + +// NewSlaveComm constructs a fully-initialized but unstarted SlaveComm. An error +// here means the object is unusable and discarded. +func NewSlaveComm(cfg SlaveConfig) (*SlaveComm, error) { + if err := cfg.Validate(); err != nil { + return nil, fmt.Errorf("invalid slave config: %w", err) + } + if cfg.Logger == nil { + cfg.Logger = log.Root() + } + pool, err := NewXshardPool(cfg.ID, cfg.FullShardIDList, cfg.ClusterFullShardIDList, cfg.MaxPayloadSize, cfg.Xshard, cfg.Logger) + if err != nil { + return nil, fmt.Errorf("new xshard pool: %w", err) + } + return &SlaveComm{ + cfg: cfg, + localBranches: make(map[uint32]struct{}), + clusterPeerIDs: make(map[uint64]struct{}), + peers: make(map[uint64]map[uint32]*PeerConn), + xshardPool: pool, + stopped: make(chan struct{}), + logger: cfg.Logger, + }, nil +} + +// ── Lifecycle ──────────────────────────────────────────────────────────────── + +// Start binds the listener and dispatches the event loop. Owner contract: called +// exactly once, before Stop; restart is unsupported. The only failure is binding +// the listener, after which the object must be discarded. +func (s *SlaveComm) Start() error { + addr := net.JoinHostPort("0.0.0.0", strconv.Itoa(s.cfg.Port)) + ln, err := net.Listen("tcp", addr) + if err != nil { + return fmt.Errorf("listen on %s: %w", addr, err) + } + + // Install before any goroutine exists that could read these fields. + s.listener = ln + + go s.acceptLoop() + + s.logger.Info("slave server started", "addr", ln.Addr().String(), "id", string(s.cfg.ID)) + return nil +} + +// Stop initiates shutdown and returns without waiting for goroutines to exit. +// Resource closes are intentionally not once-guarded (each is individually +// idempotent); only the shutdown notification is once-guarded. +// +// stopped is closed before loading master so runMasterConn can detect a MasterConn +// published after Stop has already observed master as nil, and close it in its own +// post-publication compensation. +func (s *SlaveComm) Stop() { + s.shutdownOnce.Do(func() { + close(s.stopped) + }) + if s.listener != nil { + s.listener.Close() + } + s.xshardPool.Close() + s.closeAllPeers() + if mc := s.master.Load(); mc != nil { + mc.Close() + } + s.logger.Info("slave server stopped") +} + +// WaitStopped returns the shutdown notification channel (py: get_shutdown_future), +// closed once Stop has issued every close request, without waiting for the +// goroutines those closes unblock. +func (s *SlaveComm) WaitStopped() <-chan struct{} { + return s.stopped +} + +// ── Business outbound: master sends ────────────────────────────────────────── + +// SendMinorBlockHeaderToMaster reports a new minor block header (py: +// send_minor_block_header_to_master). Returns ErrNotActive before the master +// connection exists and ErrConnectionClosed after it closes. +func (s *SlaveComm) SendMinorBlockHeaderToMaster(ctx context.Context, req *wire.AddMinorBlockHeaderRequest) (*wire.AddMinorBlockHeaderResponse, error) { + mc := s.master.Load() + if mc == nil { + return nil, conn.ErrNotActive + } + return mc.SendAddMinorBlockHeader(ctx, req) +} + +// SendMinorBlockHeaderListToMaster reports a list of new minor block headers +// to the master (py: SlaveServer.send_minor_block_header_list_to_master). +// Before the master connection exists it returns ErrNotActive; after it +// closes the delegate returns ErrConnectionClosed. +func (s *SlaveComm) SendMinorBlockHeaderListToMaster(ctx context.Context, req *wire.AddMinorBlockHeaderListRequest) (*wire.AddMinorBlockHeaderListResponse, error) { + mc := s.master.Load() + if mc == nil { + return nil, conn.ErrNotActive + } + return mc.SendAddMinorBlockHeaderList(ctx, req) +} + +// ── Business outbound: xshard broadcasts ───────────────────────────────────── + +// broadcastToBranch concurrently sends to every xshard connection serving +// branch (py: broadcast_xshard_tx_list / batch_broadcast_xshard_tx_list). +// +// The delivery set is the full snapshot: all connections are attempted even +// if some sends fail. The call waits for all sends to complete and returns +// nil only if all sends succeed; otherwise, all errors are aggregated. +// An empty connection set is a no-op, matching Python's gather([]). +func (s *SlaveComm) broadcastToBranch(branch uint32, send func(*XshardConn) error) error { + conns := s.xshardPool.Lookup(branch) + errs := make([]error, len(conns)) + + var wg sync.WaitGroup + for i, c := range conns { + wg.Add(1) + go func(i int, c *XshardConn) { + defer wg.Done() + + if err := send(c); err != nil { + errs[i] = fmt.Errorf("xshard conn#%d (remote=%x): %w", i, c.RemoteID(), err) + } + }(i, c) + } + + wg.Wait() + return errors.Join(errs...) +} + +// SendXshardTxList broadcasts an AddXshardTxListRequest to every slave connection +// serving branch (py: broadcast_xshard_tx_list, remote leg); local delivery is the +// caller's. Delivery is attempt-all — a failing connection never suppresses the +// others — and the result is binary: nil iff every connection acknowledged. An +// empty connection set is a no-op, matching py's gather([]). +func (s *SlaveComm) SendXshardTxList(ctx context.Context, branch uint32, req *wire.AddXshardTxListRequest) error { + return s.broadcastToBranch(branch, func(c *XshardConn) error { + return c.SendAddXshardTxList(ctx, req) + }) +} + +// SendBatchXshardTxList broadcasts a BatchAddXshardTxListRequest to every +// slave connection serving branch, with the same attempt-all and binary-result +// semantics (py: batch_broadcast_xshard_tx_list). +func (s *SlaveComm) SendBatchXshardTxList(ctx context.Context, branch uint32, req *wire.BatchAddXshardTxListRequest) error { + return s.broadcastToBranch(branch, func(c *XshardConn) error { + return c.SendBatchAddXshardTxList(ctx, req) + }) +} + +// ── Business outbound: peer sends ──────────────────────────────────────────── +// +// The business layer never holds a *PeerConn; these veneers resolve it by +// (clusterPeerID, branch) inside SlaveComm. + +// SendPeerNewBlock sends a minor block to the peer's (clusterPeerID, branch) +// connection (py: PeerShardConnection.send_new_block). +func (s *SlaveComm) SendPeerNewBlock(clusterPeerID uint64, branch uint32, cmd *wire.NewBlockMinorCommand) error { + pc, err := s.requirePeer(clusterPeerID, branch) + if err != nil { + return err + } + return pc.SendNewBlock(cmd) +} + +// SendPeerNewMinorBlockHeaderList sends a new-tip header list to the peer's +// (clusterPeerID, branch) connection (py: PeerShardConnection.broadcast_new_tip). +func (s *SlaveComm) SendPeerNewMinorBlockHeaderList(clusterPeerID uint64, branch uint32, cmd *wire.NewMinorBlockHeaderListCommand) error { + pc, err := s.requirePeer(clusterPeerID, branch) + if err != nil { + return err + } + return pc.SendNewMinorBlockHeaderList(cmd) +} + +// SendPeerTransactionList sends a transaction list to the peer's +// (clusterPeerID, branch) connection (py: PeerShardConnection.broadcast_tx_list). +func (s *SlaveComm) SendPeerTransactionList(clusterPeerID uint64, branch uint32, cmd *wire.NewTransactionListCommand) error { + pc, err := s.requirePeer(clusterPeerID, branch) + if err != nil { + return err + } + return pc.SendTransactionList(cmd) +} + +// GetPeerMinorBlockList issues an active RPC to the peer's (clusterPeerID, +// branch) connection (py: write_rpc_request(GET_MINOR_BLOCK_LIST_REQUEST)). +func (s *SlaveComm) GetPeerMinorBlockList(ctx context.Context, clusterPeerID uint64, branch uint32, req *wire.GetMinorBlockListRequest) (*wire.GetMinorBlockListResponse, error) { + pc, err := s.requirePeer(clusterPeerID, branch) + if err != nil { + return nil, err + } + return pc.GetMinorBlockList(ctx, req) +} + +// GetPeerMinorBlockHeaderList issues an active RPC to the peer's +// (clusterPeerID, branch) connection (py: SyncTask.__download_block_headers). +func (s *SlaveComm) GetPeerMinorBlockHeaderList(ctx context.Context, clusterPeerID uint64, branch uint32, req *wire.GetMinorBlockHeaderListRequest) (*wire.GetMinorBlockHeaderListResponse, error) { + pc, err := s.requirePeer(clusterPeerID, branch) + if err != nil { + return nil, err + } + return pc.GetMinorBlockHeaderList(ctx, req) +} + +// GetPeerMinorBlockHeaderListWithSkip issues an active RPC to the peer's +// (clusterPeerID, branch) connection, skipping headers already known locally. +func (s *SlaveComm) GetPeerMinorBlockHeaderListWithSkip(ctx context.Context, clusterPeerID uint64, branch uint32, req *wire.GetMinorBlockHeaderListWithSkipRequest) (*wire.GetMinorBlockHeaderListResponse, error) { + pc, err := s.requirePeer(clusterPeerID, branch) + if err != nil { + return nil, err + } + return pc.GetMinorBlockHeaderListWithSkip(ctx, req) +} + +// LookupPeer routes virtual peer frames from the master to the PeerConn serving +// (cluster_peer_id, branch), or nil when there is none (py: NULL_CONNECTION). +func (s *SlaveComm) LookupPeer(clusterPeerID uint64, branch uint32) *PeerConn { + s.peersMu.RLock() + defer s.peersMu.RUnlock() + if bm, ok := s.peers[clusterPeerID]; ok { + return bm[branch] + } + return nil +} + +// ── Internals ──────────────────────────────────────────────────────────────── + +// connectToSlaves dials every advertised slave into the xshard pool, reporting +// per-entry failures in the response result list so the master connection +// stays up. +func (s *SlaveComm) connectToSlaves(req *wire.ConnectToSlavesRequest) (*wire.ConnectToSlavesResponse, error) { + resultList := make([]wire.PrependedSizeBytes4, len(req.SlaveInfoList)) + for i := range req.SlaveInfoList { + info := req.SlaveInfoList[i] + if err := s.xshardPool.DialToSlave(context.Background(), info); err != nil { + // Wire semantics stay per-entry (see above); this log only aids + // ops, mirroring py:866's per-entry log. + s.logger.Warn("connect to slave failed", "remote_id", string(info.ID), "err", err) + resultList[i] = wire.PrependedSizeBytes4([]byte(err.Error())) + } + } + return &wire.ConnectToSlavesResponse{ResultList: resultList}, nil +} + +// createShards records newly-created branches in localBranches and equips +// each with a PeerConn for every announced cluster peer. Existing branches +// are skipped. +func (s *SlaveComm) createShards(rootTip *types.RootBlock) error { + // A business failure fails the PING before any topology change. + createdBranches, err := s.cfg.Master.ShardCreator(rootTip) + if err != nil { + return err + } + if len(createdBranches) == 0 { + return nil + } + + s.peersMu.Lock() + newBranches := make([]uint32, 0, len(createdBranches)) + for _, branch := range createdBranches { + if _, exists := s.localBranches[branch]; exists { + continue + } + s.localBranches[branch] = struct{}{} + newBranches = append(newBranches, branch) + } + // Snapshot the announced cluster peers to equip each new branch with. + peers := make([]uint64, 0, len(s.clusterPeerIDs)) + for id := range s.clusterPeerIDs { + peers = append(peers, id) + } + s.peersMu.Unlock() + + for _, branch := range newBranches { + for _, id := range peers { + if _, err := s.addPeerConnection(id, branch); err != nil { + s.logger.Error("equip peer connection failed", "cluster_peer_id", id, "branch", branch, "err", err) + } + } + } + return nil +} + +// createClusterPeerConnection registers a new cluster peer and creates a +// PeerConn for it on every currently-created local branch. It always succeeds +// from the master's point of view (error_code 0); duplicates are logged and +// skipped. Branches not created yet are equipped later by createShards. +func (s *SlaveComm) createClusterPeerConnection(req *wire.CreateClusterPeerConnectionRequest) (*wire.CreateClusterPeerConnectionResponse, error) { + id := req.ClusterPeerID + + s.peersMu.Lock() + s.clusterPeerIDs[id] = struct{}{} + branches := make([]uint32, 0, len(s.localBranches)) + for branch := range s.localBranches { + branches = append(branches, branch) + } + s.peersMu.Unlock() + + for _, branch := range branches { + created, err := s.addPeerConnection(id, branch) + if err != nil { + s.logger.Error("create peer connection failed", "cluster_peer_id", id, "branch", branch, "err", err) + continue + } + if !created { + s.logger.Error("duplicated create cluster peer connection", "cluster_peer_id", id, "branch", branch) + } + } + return &wire.CreateClusterPeerConnectionResponse{}, nil +} + +// destroyClusterPeerConnection deregisters the cluster peer and closes every +// PeerConn of it. Fire-and-forget; destroying an unknown id is a no-op. +// Connections are closed outside peersMu. +func (s *SlaveComm) destroyClusterPeerConnection(req *wire.DestroyClusterPeerConnectionCommand) error { + id := req.ClusterPeerID + + s.peersMu.Lock() + delete(s.clusterPeerIDs, id) + bm, ok := s.peers[id] + if ok { + delete(s.peers, id) + } + s.peersMu.Unlock() + if !ok { + return nil + } + for _, pc := range bm { + pc.Close() + } + return nil +} + +// acceptLoop accepts inbound connections and is the single owner of +// classification, encoding py's "if not self.master" as its own serialized +// control flow: the first accepted connection is always the master, every later +// one is xshard. The claim is a loop-local flag — no other goroutine classifies +// connections — so no lock or atomic is needed. Stop closes the listener, which +// makes Accept return and this loop exit. +func (s *SlaveComm) acceptLoop() { + masterClaimed := false + for { + conn, err := s.listener.Accept() + if err != nil { + // Listener closed by Stop; other errors are transient and retried. + if errors.Is(err, net.ErrClosed) { + return + } + s.logger.Warn("accept failed", "err", err) + continue + } + + if !masterClaimed { + masterClaimed = true + go s.runMasterConn(conn) + continue + } + go s.runXshardConn(conn) + } +} + +// runMasterConn runs the first inbound connection as the MasterConn and blocks +// until that connection is gone. The pointer is stored before Start so any frame +// the readLoop processes sees an established master. Master loss triggers Stop; an +// external Stop reaches here by closing the established MasterConn directly. Not +// joined or waited on by Stop: it holds nothing Stop waits for, and Stop is +// non-blocking. +func (s *SlaveComm) runMasterConn(conn net.Conn) { + mc, err := NewMasterConn(MasterConnConfig{ + Conn: conn, + MaxPayloadSize: s.cfg.MaxPayloadSize, + LocalID: s.cfg.ID, + LocalFullShardIDList: s.cfg.FullShardIDList, + ClusterShardIDs: s.cfg.ClusterFullShardIDList, + Handler: s.newMasterHandler(), + PeerResolver: s, + Logger: s.logger, + }) + if err != nil { + conn.Close() + s.logger.Error("failed to create master connection", "err", err) + s.Stop() + return + } + + s.master.Store(mc) + // If Stop resolved while this connection was being established, the master + // pointer was published too late for Stop to close it (its Load saw nil). + // Compensate here: close the just-published MasterConn and never Start it, + // so no net.Conn / readLoop is owned beyond a resolved shutdown. + select { + case <-s.stopped: + mc.Close() + return + default: + } + + mc.Start() + s.logger.Info("master connection established", "remote", conn.RemoteAddr()) + + <-mc.WaitUntilClosed() + s.logger.Info("master connection closed") + s.Stop() +} + +// runXshardConn hands a subsequent inbound connection to the pool, which owns +// the full inbound lifecycle: handshake, indexing and cleanup on close. Stop +// closes the pool, which closes the tracked connections and unblocks this +// handshake, so it exits on its own. +func (s *SlaveComm) runXshardConn(conn net.Conn) { + s.logger.Info("accepted xshard connection", "remote", conn.RemoteAddr()) + s.xshardPool.HandleInbound(conn) +} + +// requirePeer resolves the (clusterPeerID, branch) connection or reports the +// NULL_CONNECTION case: no sendable connection exists. +func (s *SlaveComm) requirePeer(clusterPeerID uint64, branch uint32) (*PeerConn, error) { + if pc := s.LookupPeer(clusterPeerID, branch); pc != nil { + return pc, nil + } + return nil, fmt.Errorf("no peer connection for cluster_peer_id %d branch 0x%x", clusterPeerID, branch) +} + +// addPeerConnection is the single construction path for every PeerConn: it builds, +// starts and registers (clusterPeerID, branch), reporting created=false on a +// duplicate. Ownership stays with SlaveComm; callers are master-command +// dispatchers, so the master connection is always published here. It does not gate +// on Stop: an in-flight handler may complete one registration after the registry +// drains — the terminal window py accepts. +func (s *SlaveComm) addPeerConnection(clusterPeerID uint64, branch uint32) (created bool, err error) { + s.peersMu.Lock() + defer s.peersMu.Unlock() + bm, ok := s.peers[clusterPeerID] + if !ok { + bm = make(map[uint32]*PeerConn) + } + if _, exists := bm[branch]; exists { + return false, nil + } + pc, err := NewPeerConn(clusterPeerID, branch, s.master.Load(), s.cfg.Peer, s.logger) + if err != nil { + return false, err + } + pc.Start() + bm[branch] = pc + s.peers[clusterPeerID] = bm + return true, nil +} + +// closeAllPeers removes and closes every registered PeerConn and clears the known +// peer set (py: MasterConnection.close, the master-loss leg). PeerConns recorded +// later by an in-flight handler are a terminal best-effort residue the process is +// about to exit with (py semantics). Close happens outside peersMu. +func (s *SlaveComm) closeAllPeers() { + s.peersMu.Lock() + var all []*PeerConn + for _, bm := range s.peers { + for _, pc := range bm { + all = append(all, pc) + } + } + s.peers = make(map[uint64]map[uint32]*PeerConn) + s.clusterPeerIDs = make(map[uint64]struct{}) + s.peersMu.Unlock() + for _, pc := range all { + pc.Close() + } +} diff --git a/qkc/cluster/slave/slave_test.go b/qkc/cluster/slave/slave_test.go new file mode 100644 index 00000000000..58231330357 --- /dev/null +++ b/qkc/cluster/slave/slave_test.go @@ -0,0 +1,972 @@ +// Copyright 2026-2027, QuarkChain. + +package slave + +import ( + "bufio" + "bytes" + "context" + "errors" + "net" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/qkc/cluster/conn" + "github.com/ethereum/go-ethereum/qkc/cluster/wire" + "github.com/ethereum/go-ethereum/qkc/serialize" + "github.com/ethereum/go-ethereum/qkc/types" +) + +// ── helpers ────────────────────────────────────────────────────────────────── + +var ( + testSlaveID = []byte("S0") + testSlaveShards = []uint32{0x00010001, 0x00020001} +) + +// commTestHandler is the test business backend: it serves the business RPCs +// with the established fakeMasterHandler test double. The master's +// communication-topology commands (CONNECT_TO_SLAVES, +// CREATE/DESTROY_CLUSTER_PEER_CONNECTION, PING CreateShards) are served by the +// SlaveComm under test via the masterHandler facade, not forwarded here. +type commTestHandler struct { + *fakeMasterHandler +} + +// ShardCreator arms the created-branch report after delegating count/error +// injection to the embedded double. Under Go's GENESIS.ROOT_HEIGHT 0 +// simplification the business runtime always reports every configured shard as +// created, unless it fails (errCreateShards). +func (h *commTestHandler) ShardCreator(rootTip *types.RootBlock) ([]uint32, error) { + if err := h.fakeMasterHandler.CreateShards(rootTip); err != nil { + return nil, err + } + return append([]uint32(nil), testSlaveShards...), nil +} + +// testRootTip returns a minimal RootBlock payload. The communication layer never +// decodes it — the business handler owns the RootTip semantics — so tests only +// need a non-nil payload to trigger the PING orchestration. +func testRootTip() *types.RootBlock { + return types.NewRootBlockWithHeader(&types.RootBlockHeader{Number: 1}) +} + +// startTestSlaveComm starts a SlaveComm on a free loopback port with both +// local shards already created (as if an earlier PING had reported them) and +// returns it plus its dial address. +func startTestSlaveComm(t *testing.T) (*SlaveComm, string) { + t.Helper() + return startTestSlaveCommWithBranches(t, testSlaveShards) +} + +// startTestSlaveCommWithBranches starts a SlaveComm whose already-created +// branch set is preCreated, standing in for the branches an earlier +// CreateShards had reported. Passing nil starts with nothing created. +func startTestSlaveCommWithBranches(t *testing.T, preCreated []uint32) (*SlaveComm, string) { + t.Helper() + // Reserve a port, release it, then let Start bind it. Back-to-back runs + // (go test -count) can briefly reuse a port whose accepted connection is + // still closing asynchronously (runMasterConn is deliberately not joined + // by Stop), so retry with a fresh reservation on EADDRINUSE. + for attempt := 0; ; attempt++ { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("reserve port: %v", err) + } + addr := ln.Addr().String() + port := ln.Addr().(*net.TCPAddr).Port + ln.Close() + + handler := &commTestHandler{fakeMasterHandler: &fakeMasterHandler{}} + comm, err := NewSlaveComm(SlaveConfig{ + ID: append([]byte(nil), testSlaveID...), + FullShardIDList: append([]uint32(nil), testSlaveShards...), + ClusterFullShardIDList: append([]uint32(nil), testSlaveShards...), + Port: port, + Logger: log.New(), + Master: handler, + Peer: stubPeerHandler{}, + Xshard: testXshardHandler{}, + }) + if err != nil { + t.Fatalf("new slave comm: %v", err) + } + if err := comm.Start(); err != nil { + if attempt < 5 && strings.Contains(err.Error(), "address already in use") { + time.Sleep(10 * time.Millisecond) + continue + } + t.Fatalf("start slave comm: %v", err) + } + comm.peersMu.Lock() + for _, branch := range preCreated { + comm.localBranches[branch] = struct{}{} + } + comm.peersMu.Unlock() + t.Cleanup(comm.Stop) + return comm, addr + } +} + +// dialComm dials the comm's listener and returns the raw connection. +func dialComm(t *testing.T, addr string) net.Conn { + t.Helper() + conn, err := net.Dial("tcp", addr) + if err != nil { + t.Fatalf("dial comm: %v", err) + } + t.Cleanup(func() { conn.Close() }) + return conn +} + +// sendFrame writes one frame to conn. +func sendFrame(t *testing.T, conn net.Conn, f *wire.Frame) { + t.Helper() + if err := wire.WriteFrame(conn, f); err != nil { + t.Fatalf("write frame: %v", err) + } +} + +// readFrame reads one frame from conn with a bounded wait. +func readFrame(t *testing.T, conn net.Conn) *wire.Frame { + t.Helper() + // Reading is bounded by the test's overall timeout via SetReadDeadline. + if err := conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatalf("set read deadline: %v", err) + } + f, err := wire.ReadFrame(bufio.NewReader(conn), 0) + if err != nil { + t.Fatalf("read frame: %v", err) + } + return f +} + +// waitFor polls until cond returns true. Fails the test on timeout. +func waitFor(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", what) +} + +// sendCreatePeer sends CREATE_CLUSTER_PEER_CONNECTION and returns the +// response's error code. +func sendCreatePeer(t *testing.T, conn net.Conn, rpcID uint64, clusterPeerID uint64) uint32 { + t.Helper() + payload, err := serialize.SerializeToBytes(&wire.CreateClusterPeerConnectionRequest{ClusterPeerID: clusterPeerID}) + if err != nil { + t.Fatalf("serialize create request: %v", err) + } + sendFrame(t, conn, &wire.Frame{ + Meta: wire.ClusterMetadata{}, + Opcode: byte(wire.ClusterOpCreateClusterPeerConnectionRequest), + RPCID: rpcID, + Payload: payload, + }) + + resp := readFrame(t, conn) + if resp.Opcode != byte(wire.ClusterOpCreateClusterPeerConnectionResponse) || resp.RPCID != rpcID { + t.Fatalf("unexpected create response: opcode=0x%x rpc_id=%d", resp.Opcode, resp.RPCID) + } + var out wire.CreateClusterPeerConnectionResponse + if err := serialize.DeserializeFromBytes(resp.Payload, &out); err != nil { + t.Fatalf("deserialize create response: %v", err) + } + return out.ErrorCode +} + +// sendDestroyPeer sends DESTROY_CLUSTER_PEER_CONNECTION (fire-and-forget). +func sendDestroyPeer(t *testing.T, conn net.Conn, clusterPeerID uint64) { + t.Helper() + payload, err := serialize.SerializeToBytes(&wire.DestroyClusterPeerConnectionCommand{ClusterPeerID: clusterPeerID}) + if err != nil { + t.Fatalf("serialize destroy command: %v", err) + } + sendFrame(t, conn, &wire.Frame{ + Meta: wire.ClusterMetadata{}, + Opcode: byte(wire.ClusterOpDestroyClusterPeerConnectionCommand), + RPCID: 0, + Payload: payload, + }) +} + +// sendPingRootTip sends a master PING carrying rootTip with the given rpcID +// (RPC ids must strictly increase per connection) and waits for the PONG, +// which is written only after the CreateShards orchestration has completed. +func sendPingRootTip(t *testing.T, conn net.Conn, rpcID uint64, rootTip *types.RootBlock) { + t.Helper() + payload, err := serialize.SerializeToBytes(&wire.PingRequest{ + ID: append([]byte(nil), testSlaveID...), + FullShardIDList: append([]uint32(nil), testSlaveShards...), + RootTip: rootTip, + }) + if err != nil { + t.Fatalf("serialize ping: %v", err) + } + sendFrame(t, conn, &wire.Frame{ + Meta: wire.ClusterMetadata{}, + Opcode: byte(wire.ClusterOpPing), + RPCID: rpcID, + Payload: payload, + }) + resp := readFrame(t, conn) + if resp.Opcode != byte(wire.ClusterOpPong) || resp.RPCID != rpcID { + t.Fatalf("unexpected ping response: opcode=0x%x rpc_id=%d", resp.Opcode, resp.RPCID) + } +} + +// sendPingRootTipNoResponse sends a master PING carrying rootTip without +// reading a response. Used when the orchestration is expected to fail and the +// connection to close before any PONG is written. +func sendPingRootTipNoResponse(t *testing.T, conn net.Conn, rpcID uint64, rootTip *types.RootBlock) { + t.Helper() + payload, err := serialize.SerializeToBytes(&wire.PingRequest{ + ID: append([]byte(nil), testSlaveID...), + FullShardIDList: append([]uint32(nil), testSlaveShards...), + RootTip: rootTip, + }) + if err != nil { + t.Fatalf("serialize ping: %v", err) + } + sendFrame(t, conn, &wire.Frame{ + Meta: wire.ClusterMetadata{}, + Opcode: byte(wire.ClusterOpPing), + RPCID: rpcID, + Payload: payload, + }) +} + +// numXshardConns returns the number of connections tracked by the pool. It is +// a white-box test helper: the pool intentionally exposes no connection +// counter (Python has no such public API). +func numXshardConns(p *XshardPool) int { + p.mu.RLock() + defer p.mu.RUnlock() + return len(p.connections) +} + +// establishXshardInbound dials the comm's listener as a second connection and +// completes the xshard handshake as the remote outbound client, so the comm +// indexes it in its pool. +func establishXshardInbound(t *testing.T, comm *SlaveComm, addr string) { + t.Helper() + conn := dialComm(t, addr) + client, err := newXshardConn(conn, 0, []byte("S1"), []uint32{0x00010001}, append([]byte(nil), testSlaveID...), append([]uint32(nil), testSlaveShards...), testXshardHandler{}, log.New()) + if err != nil { + t.Fatalf("new client xshard conn: %v", err) + } + client.Start() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + id, shards, err := client.sendPing(ctx) + if err != nil { + t.Fatalf("xshard handshake: %v", err) + } + if !bytes.Equal(id, testSlaveID) || !slicesEqualUint32(shards, testSlaveShards) { + t.Fatalf("unexpected pong identity: id=%q shards=%v", id, shards) + } + waitFor(t, "xshard inbound indexing", func() bool { + return comm.xshardPool.hasSlaveID([]byte("S1")) + }) +} + +func slicesEqualUint32(a, b []uint32) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +// White-box introspection helpers into SlaveComm's peer registry (this is a +// same-package test; the production surface is LookupPeer/peerCount). + +func (s *SlaveComm) lookupPeer(clusterPeerID uint64, branch uint32) *PeerConn { + return s.LookupPeer(clusterPeerID, branch) +} + +func (s *SlaveComm) peerCountFor(clusterPeerID uint64) int { + s.peersMu.RLock() + defer s.peersMu.RUnlock() + return len(s.peers[clusterPeerID]) +} + +func (s *SlaveComm) peerCount() int { + s.peersMu.RLock() + defer s.peersMu.RUnlock() + return len(s.peers) +} + +func (s *SlaveComm) localBranchCount() int { + s.peersMu.RLock() + defer s.peersMu.RUnlock() + return len(s.localBranches) +} + +// ── tests ──────────────────────────────────────────────────────────────────── + +// TestSlaveComm_FirstInboundIsMasterAndRestAreXshard verifies the accept-loop +// dispatch: the first inbound connection becomes the MasterConn, and a second +// inbound connection goes through the xshard handshake into the pool +// (py: __handle_new_connection). +func TestSlaveComm_FirstInboundIsMasterAndRestAreXshard(t *testing.T) { + comm, addr := startTestSlaveComm(t) + + masterConn := dialComm(t, addr) + _ = masterConn + + // A second connection is not claimed as master; it must complete the + // xshard handshake and land in the pool. + establishXshardInbound(t, comm, addr) +} + +// TestSlaveComm_CreateAndDestroyPeerConn verifies CREATE creates a PeerConn on +// every local branch, a duplicate CREATE is a no-op, and DESTROY removes and +// closes them (py: slave.py:329-370, 321-327). +func TestSlaveComm_CreateAndDestroyPeerConn(t *testing.T) { + comm, addr := startTestSlaveComm(t) + masterConn := dialComm(t, addr) + + const cid = 7 + if code := sendCreatePeer(t, masterConn, 1, cid); code != 0 { + t.Fatalf("create returned error_code=%d", code) + } + for _, branch := range testSlaveShards { + if pc := comm.lookupPeer(cid, branch); pc == nil { + t.Fatalf("no PeerConn for branch 0x%x after CREATE", branch) + } + } + if got := comm.peerCountFor(cid); got != len(testSlaveShards) { + t.Fatalf("registered %d PeerConns, want %d", got, len(testSlaveShards)) + } + + // Duplicate CREATE: existing branches are skipped, response stays success. + if code := sendCreatePeer(t, masterConn, 2, cid); code != 0 { + t.Fatalf("duplicate create returned error_code=%d", code) + } + if got := comm.peerCountFor(cid); got != len(testSlaveShards) { + t.Fatalf("duplicate create changed registry: %d PeerConns", got) + } + + // DESTROY removes and closes every PeerConn of the id. + sendDestroyPeer(t, masterConn, cid) + waitFor(t, "peer destruction", func() bool { + return comm.peerCountFor(cid) == 0 + }) + if pc := comm.lookupPeer(cid, testSlaveShards[0]); pc != nil && !pc.IsClosed() { + t.Fatal("destroyed PeerConn is not closed") + } +} + +// TestSlaveComm_PingDelegatesCreateShardsAndBackfills verifies the PING +// orchestration chain: MasterConn → masterHandler.CreateShards → +// SlaveComm.createShards → business MasterHandler.CreateShards, followed by an +// idempotent peer-registry convergence (every known cluster peer × every local +// branch), and that multiple peers coexist without cross-talk. +func TestSlaveComm_PingDelegatesCreateShardsAndBackfills(t *testing.T) { + comm, addr := startTestSlaveComm(t) + masterConn := dialComm(t, addr) + handler := comm.cfg.Master.(*commTestHandler) + + // PING with a RootTip before any peer exists: business CreateShards runs, + // the registry stays empty. + sendPingRootTip(t, masterConn, 1, testRootTip()) + if got := handler.createShardsCalls.Load(); got != 1 { + t.Fatalf("business CreateShards calls: got %d, want 1", got) + } + if got := comm.peerCount(); got != 0 { + t.Fatalf("registry holds %d peers before any CREATE", got) + } + + // CREATE two peers: each gets one PeerConn per local branch. + const cidA, cidB = 13, 14 + for i, cid := range []uint64{cidA, cidB} { + if code := sendCreatePeer(t, masterConn, uint64(i+2), cid); code != 0 { + t.Fatalf("create %d returned error_code=%d", cid, code) + } + } + if got := comm.peerCount(); got != 2 { + t.Fatalf("registered %d peers, want 2", got) + } + for _, cid := range []uint64{cidA, cidB} { + if got := comm.peerCountFor(cid); got != len(testSlaveShards) { + t.Fatalf("cluster_peer_id %d has %d PeerConns, want %d", cid, got, len(testSlaveShards)) + } + } + + // A second PING converges the registry idempotently: no new conns, no + // duplicate logs, and the business handler is invoked again. + sendPingRootTip(t, masterConn, 4, testRootTip()) + if got := handler.createShardsCalls.Load(); got != 2 { + t.Fatalf("business CreateShards calls: got %d, want 2", got) + } + if got := comm.peerCount(); got != 2 { + t.Fatalf("second PING changed peer count to %d", got) + } + for _, cid := range []uint64{cidA, cidB} { + if got := comm.peerCountFor(cid); got != len(testSlaveShards) { + t.Fatalf("second PING changed cluster_peer_id %d to %d PeerConns", cid, got) + } + } +} + +// TestSlaveComm_PingAfterDestroyKeepsPeerGone verifies that a later PING +// (backfill) does not resurrect a destroyed cluster peer id: it has been +// removed from the known-peer set, so convergence has nothing to create. +func TestSlaveComm_PingAfterDestroyKeepsPeerGone(t *testing.T) { + comm, addr := startTestSlaveComm(t) + masterConn := dialComm(t, addr) + handler := comm.cfg.Master.(*commTestHandler) + + const cid = 15 + if code := sendCreatePeer(t, masterConn, 1, cid); code != 0 { + t.Fatalf("create returned error_code=%d", code) + } + pc := comm.lookupPeer(cid, testSlaveShards[0]) + + sendDestroyPeer(t, masterConn, cid) + waitFor(t, "peer destruction", func() bool { + return comm.peerCountFor(cid) == 0 + }) + if !pc.IsClosed() { + t.Fatal("PeerConn closed by DESTROY was not closed") + } + + // The first PING after the DESTROY creates every configured shard, but the + // peer id is gone from the known set, so nothing is resurrected. + sendPingRootTip(t, masterConn, 2, testRootTip()) + if got := handler.createShardsCalls.Load(); got != 1 { + t.Fatalf("business CreateShards calls: got %d, want 1", got) + } + if got := comm.peerCount(); got != 0 { + t.Fatalf("backfill after DESTROY resurrected %d peer entries", got) + } +} + +// TestSlaveComm_CreateShardsEquipsEveryConfiguredBranch verifies the shard/peer +// model under Go's GENESIS.ROOT_HEIGHT 0 simplification: a PING's CreateShards +// creates every configured shard at once, and each announced cluster peer is +// equipped with a PeerConn on every local branch (py: +// Shard.create_peer_shard_connections per shard). A peer announced before any +// PING gets no PeerConn until the first PING equips it; a peer announced +// afterwards lands on every already-created branch. +func TestSlaveComm_CreateShardsEquipsEveryConfiguredBranch(t *testing.T) { + comm, addr := startTestSlaveCommWithBranches(t, nil) + handler := comm.cfg.Master.(*commTestHandler) + masterConn := dialComm(t, addr) + + const cid = 21 + // CREATE before any PING: no branch exists yet, so the peer is registered + // but not connected even though both branches are in FullShardIDList. + if code := sendCreatePeer(t, masterConn, 1, cid); code != 0 { + t.Fatalf("create returned error_code=%d", code) + } + if got := comm.peerCountFor(cid); got != 0 { + t.Fatalf("cluster_peer_id %d has %d PeerConns before any PING, want 0", cid, got) + } + + // The first PING creates every configured shard and backfills the peer. + sendPingRootTip(t, masterConn, 2, testRootTip()) + if got := handler.createShardsCalls.Load(); got != 1 { + t.Fatalf("business CreateShards calls: got %d, want 1", got) + } + for _, branch := range testSlaveShards { + if pc := comm.lookupPeer(cid, branch); pc == nil { + t.Fatalf("no PeerConn on branch 0x%x after PING", branch) + } + } + if got := comm.peerCountFor(cid); got != len(testSlaveShards) { + t.Fatalf("cluster_peer_id %d has %d PeerConns, want %d", cid, got, len(testSlaveShards)) + } + + // A second peer announced afterwards lands on every created branch. + const cid2 = 22 + if code := sendCreatePeer(t, masterConn, 3, cid2); code != 0 { + t.Fatalf("create returned error_code=%d", code) + } + if got := comm.peerCountFor(cid2); got != len(testSlaveShards) { + t.Fatalf("cluster_peer_id %d has %d PeerConns, want %d", cid2, got, len(testSlaveShards)) + } + if got := comm.peerCount(); got != 2 { + t.Fatalf("registry holds %d peers, want 2", got) + } +} + +// TestSlaveComm_CreateShardsFailureLeavesTopology verifies that a business +// CreateShards failure aborts before any branch is recorded or any PeerConn is +// created: the topology (localBranches, peer registry) is unchanged. The PING +// handler fails and the connection closes, so the business method runs exactly +// once (py: the create_shards exception propagates through handle_ping). +func TestSlaveComm_CreateShardsFailureLeavesTopology(t *testing.T) { + comm, addr := startTestSlaveCommWithBranches(t, nil) + handler := comm.cfg.Master.(*commTestHandler) + masterConn := dialComm(t, addr) + + const cid = 31 + if code := sendCreatePeer(t, masterConn, 1, cid); code != 0 { + t.Fatalf("create returned error_code=%d", code) + } + + handler.errCreateShards = errors.New("boom") + sendPingRootTipNoResponse(t, masterConn, 2, testRootTip()) + select { + case <-comm.WaitStopped(): + case <-time.After(10 * time.Second): + t.Fatal("CreateShards error did not trigger shutdown") + } + if got := handler.createShardsCalls.Load(); got != 1 { + t.Fatalf("business CreateShards calls: got %d, want 1", got) + } + if got := comm.localBranchCount(); got != 0 { + t.Fatalf("created-branch set holds %d branches, want 0", got) + } + if got := comm.peerCountFor(cid); got != 0 { + t.Fatalf("cluster_peer_id %d has %d PeerConns, want 0", cid, got) + } +} + +// TestSlaveComm_PingCreatesEveryReportedBranch covers the "root height past +// every genesis height" case: one CreateShards reporting several branches +// equips the announced peer on all of them, and repeating the same report +// changes nothing. +func TestSlaveComm_PingCreatesEveryReportedBranch(t *testing.T) { + comm, addr := startTestSlaveCommWithBranches(t, nil) + handler := comm.cfg.Master.(*commTestHandler) + masterConn := dialComm(t, addr) + + const cid = 32 + if code := sendCreatePeer(t, masterConn, 1, cid); code != 0 { + t.Fatalf("create returned error_code=%d", code) + } + + sendPingRootTip(t, masterConn, 2, testRootTip()) + if got := handler.createShardsCalls.Load(); got != 1 { + t.Fatalf("business CreateShards calls: got %d, want 1", got) + } + + if got := comm.localBranchCount(); got != len(testSlaveShards) { + t.Fatalf("created-branch set holds %d branches, want %d", got, len(testSlaveShards)) + } + for _, branch := range testSlaveShards { + if pc := comm.lookupPeer(cid, branch); pc == nil { + t.Fatalf("no PeerConn for branch 0x%x after it was created", branch) + } + } + first := comm.lookupPeer(cid, testSlaveShards[0]) + + // A repeated report is a no-op: the branch is already created, so the + // existing PeerConn is kept rather than rebuilt. + sendPingRootTip(t, masterConn, 3, testRootTip()) + if got := handler.createShardsCalls.Load(); got != 2 { + t.Fatalf("business CreateShards calls: got %d, want 2", got) + } + if got := comm.peerCountFor(cid); got != len(testSlaveShards) { + t.Fatalf("repeated report changed cluster_peer_id %d to %d PeerConns", cid, got) + } + if comm.lookupPeer(cid, testSlaveShards[0]) != first { + t.Fatal("repeated report replaced an existing PeerConn") + } +} + +// TestSlaveComm_RepeatDestroyIsNoop verifies a repeated DESTROY for the same +// cluster peer id leaves the registry consistent and does not double-close. +func TestSlaveComm_RepeatDestroyIsNoop(t *testing.T) { + comm, addr := startTestSlaveComm(t) + masterConn := dialComm(t, addr) + + const cid = 16 + if code := sendCreatePeer(t, masterConn, 1, cid); code != 0 { + t.Fatalf("create returned error_code=%d", code) + } + pc := comm.lookupPeer(cid, testSlaveShards[0]) + + sendDestroyPeer(t, masterConn, cid) + sendDestroyPeer(t, masterConn, cid) + waitFor(t, "peer destruction", func() bool { + return comm.peerCount() == 0 + }) + if !pc.IsClosed() { + t.Fatal("PeerConn is not closed after DESTROY") + } +} + +// TestSlaveComm_DestroyUnknownPeerIsNoop verifies DESTROY for a cluster peer +// id that was never created is a no-op and leaves the comm usable. +func TestSlaveComm_DestroyUnknownPeerIsNoop(t *testing.T) { + comm, addr := startTestSlaveComm(t) + masterConn := dialComm(t, addr) + + sendDestroyPeer(t, masterConn, 99) + if got := comm.peerCount(); got != 0 { + t.Fatalf("registry holds %d entries after destroying an unknown id", got) + } + + const cid = 17 + if code := sendCreatePeer(t, masterConn, 1, cid); code != 0 { + t.Fatalf("create returned error_code=%d", code) + } + if got := comm.peerCountFor(cid); got != len(testSlaveShards) { + t.Fatalf("registered %d PeerConns, want %d", got, len(testSlaveShards)) + } +} + +// TestSlaveComm_ConnectToSlaves verifies the comm dials the advertised slaves +// into its xshard pool, skips itself, and reports per-entry failures. +func TestSlaveComm_ConnectToSlaves(t *testing.T) { + comm, addr := startTestSlaveComm(t) + masterConn := dialComm(t, addr) + + rs := startRemoteSlave(t, []byte("S1"), []uint32{0x00010001}) + defer rs.close() + + req := &wire.ConnectToSlavesRequest{ + SlaveInfoList: []wire.SlaveInfo{ + // Self: must be skipped without dialing. + {ID: append([]byte(nil), testSlaveID...), Host: []byte("127.0.0.1"), Port: 1, FullShardIDList: append([]uint32(nil), testSlaveShards...)}, + rs.slaveInfo([]byte("S1"), []uint32{0x00010001}), + }, + } + payload, err := serialize.SerializeToBytes(req) + if err != nil { + t.Fatalf("serialize connect request: %v", err) + } + sendFrame(t, masterConn, &wire.Frame{ + Meta: wire.ClusterMetadata{}, + Opcode: byte(wire.ClusterOpConnectToSlavesRequest), + RPCID: 1, + Payload: payload, + }) + + resp := readFrame(t, masterConn) + if resp.Opcode != byte(wire.ClusterOpConnectToSlavesResponse) { + t.Fatalf("unexpected response opcode 0x%x", resp.Opcode) + } + var out wire.ConnectToSlavesResponse + if err := serialize.DeserializeFromBytes(resp.Payload, &out); err != nil { + t.Fatalf("deserialize connect response: %v", err) + } + if len(out.ResultList) != 2 { + t.Fatalf("result list has %d entries, want 2", len(out.ResultList)) + } + if len(out.ResultList[0]) != 0 { + t.Fatalf("self entry reported failure: %q", out.ResultList[0]) + } + if len(out.ResultList[1]) != 0 { + t.Fatalf("remote entry reported failure: %q", out.ResultList[1]) + } + waitFor(t, "xshard pool registration of S1", func() bool { + return comm.xshardPool.hasSlaveID([]byte("S1")) + }) +} + +// TestSlaveComm_MasterCloseCascade verifies the Python close cascade: losing +// the master closes all PeerConns, the xshard pool, and the listener. +func TestSlaveComm_MasterCloseCascade(t *testing.T) { + comm, addr := startTestSlaveComm(t) + masterConn := dialComm(t, addr) + + establishXshardInbound(t, comm, addr) + + const cid = 9 + if code := sendCreatePeer(t, masterConn, 1, cid); code != 0 { + t.Fatalf("create returned error_code=%d", code) + } + + // The remote master side drops the TCP connection. + masterConn.Close() + + // The master-loss shutdown cascade: runMasterConn notices the close and + // calls Stop. WaitStopped resolves once Stop has issued every close, and + // since those closes clear the peer registry and pool synchronously, the + // drained state below is deterministic. + select { + case <-comm.WaitStopped(): + case <-time.After(10 * time.Second): + t.Fatal("master close did not trigger shutdown") + } + + // closeAllPeers runs synchronously after WaitStopped resolves; wait for the + // registry to actually drain before asserting the cascade closed every + // PeerConn. + waitFor(t, "peer registry drain after master close", func() bool { + return comm.peerCount() == 0 + }) + if pc := comm.lookupPeer(cid, testSlaveShards[0]); pc != nil { + t.Fatal("PeerConn survived the master close cascade") + } + if got := comm.peerCount(); got != 0 { + t.Fatalf("peer registry still holds %d entries", got) + } + if n := numXshardConns(comm.xshardPool); n != 0 { + t.Fatalf("xshard pool still holds %d connections", n) + } + if c, err := net.DialTimeout("tcp", addr, time.Second); err == nil { + c.Close() + t.Fatal("listener still accepts connections after shutdown") + } +} + +// TestSlaveComm_StopIdempotent verifies Stop is safe to call repeatedly. +func TestSlaveComm_StopIdempotent(t *testing.T) { + comm, addr := startTestSlaveComm(t) + dialComm(t, addr) + + done := make(chan struct{}) + go func() { + comm.Stop() + comm.Stop() + close(done) + }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("repeated Stop did not return (deadlock?)") + } +} + +// TestSlaveComm_ConcurrentShutdown verifies a master-initiated stop and an +// explicit Stop racing each other terminate cleanly without deadlock. +func TestSlaveComm_ConcurrentShutdown(t *testing.T) { + comm, addr := startTestSlaveComm(t) + masterConn := dialComm(t, addr) + + // Close the master TCP side while Stop runs concurrently. + go masterConn.Close() + done := make(chan struct{}) + go func() { + comm.Stop() + close(done) + }() + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("concurrent shutdown deadlocked") + } + // Stop's body runs synchronously, so shutdown is fully initiated once it + // returns; both racing paths converge on the same once-only sequence. + select { + case <-comm.WaitStopped(): + default: + t.Fatal("WaitStopped not closed after Stop returned") + } +} + +// TestSlaveComm_LookupPeerUnknown verifies that LookupPeer returns nil for an +// unknown cluster peer id; peer lookup for a created peer is covered by +// TestSlaveComm_CreateAndDestroyPeerConn. +func TestSlaveComm_LookupPeerUnknown(t *testing.T) { + comm, _ := startTestSlaveComm(t) + + if pc := comm.LookupPeer(11, testSlaveShards[0]); pc != nil { + t.Fatal("LookupPeer returned a peer for an unknown id") + } +} + +// TestSlaveComm_SendToMasterNotReady verifies the nil gate: a master-dependent +// send before the master connection is established returns ErrNotActive +// without panicking (py: the equivalent call crashes with AttributeError; +// ErrNotActive is the diagnostic equivalent). +func TestSlaveComm_SendToMasterNotReady(t *testing.T) { + comm, _ := startTestSlaveComm(t) + + if _, err := comm.SendMinorBlockHeaderToMaster(context.Background(), &wire.AddMinorBlockHeaderRequest{}); !errors.Is(err, conn.ErrNotActive) { + t.Fatalf("SendMinorBlockHeaderToMaster before establishment: err=%v, want ErrNotActive", err) + } + if _, err := comm.SendMinorBlockHeaderListToMaster(context.Background(), &wire.AddMinorBlockHeaderListRequest{}); !errors.Is(err, conn.ErrNotActive) { + t.Fatalf("SendMinorBlockHeaderListToMaster before establishment: err=%v, want ErrNotActive", err) + } +} + +// TestSlaveComm_SendToMasterEstablishedButClosed verifies that once the master +// connection has been published and then closed, a send reaches the delegate +// (loading a non-nil pointer) and surfaces the delegate's closed error rather +// than the nil-gate ErrNotActive. +func TestSlaveComm_SendToMasterEstablishedButClosed(t *testing.T) { + comm, addr := startTestSlaveComm(t) + masterConn := dialComm(t, addr) + defer masterConn.Close() + + // Establishing the master publishes the pointer. + sendPingRootTip(t, masterConn, 1, testRootTip()) + if comm.master.Load() == nil { + t.Fatal("master not published after establishment") + } + + // Closing the socket makes the slave-side MasterConn reach Closed. + masterConn.Close() + waitFor(t, "master connection closed", func() bool { + mc := comm.master.Load() + return mc != nil && mc.IsClosed() + }) + + if _, err := comm.SendMinorBlockHeaderToMaster(context.Background(), &wire.AddMinorBlockHeaderRequest{}); !errors.Is(err, conn.ErrConnectionClosed) { + t.Fatalf("send with closed delegate: err=%v, want ErrConnectionClosed", err) + } +} + +// TestSlaveComm_MasterAtomicPublication verifies that many goroutines loading +// the published master pointer all observe the same fully-initialized +// MasterConn. The publication word is atomic: a Store by runMasterConn and the +// concurrent Loads below are race-free under go test -race. +func TestSlaveComm_MasterAtomicPublication(t *testing.T) { + comm, addr := startTestSlaveComm(t) + masterConn := dialComm(t, addr) + defer masterConn.Close() + + sendPingRootTip(t, masterConn, 1, testRootTip()) + want := comm.master.Load() + if want == nil { + t.Fatal("master not published after establishment") + } + + var wg sync.WaitGroup + for i := 0; i < 64; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if got := comm.master.Load(); got != want { + t.Errorf("master.Load() = %v, want %v", got, want) + } + }() + } + wg.Wait() +} + +// TestSlaveComm_DrainPeersOnStop verifies Stop drains every registered PeerConn +// (py: MasterConnection.close → close all peer forwarding connections): after a +// peer connection was created, Stop leaves the registry empty. +func TestSlaveComm_DrainPeersOnStop(t *testing.T) { + comm, addr := startTestSlaveComm(t) + masterConn := dialComm(t, addr) + defer masterConn.Close() + + // Establish the master and create a peer so Stop has something to drain. + sendPingRootTip(t, masterConn, 1, testRootTip()) + if code := sendCreatePeer(t, masterConn, 2, 1); code != 0 { + t.Fatalf("create returned error_code=%d", code) + } + + comm.Stop() + + if n := comm.peerCount(); n != 0 { + t.Fatalf("peerCount after Stop = %d, want 0", n) + } +} + +type countingXshardHandler struct { + adds atomic.Int32 +} + +func (h *countingXshardHandler) AddXshardTxList( + *wire.AddXshardTxListRequest, +) (*wire.AddXshardTxListResponse, error) { + h.adds.Add(1) + return &wire.AddXshardTxListResponse{}, nil +} + +func (*countingXshardHandler) BatchAddXshardTxList( + *wire.BatchAddXshardTxListRequest, +) (*wire.BatchAddXshardTxListResponse, error) { + panic("unexpected call") +} + +func TestSlaveComm_SendXshardTxListAttemptsAllOnFailure(t *testing.T) { + comm, addr := startTestSlaveComm(t) + + // first inbound always claims master slot + dialComm(t, addr) + + newInbound := func(id string, h *countingXshardHandler) *XshardConn { + raw := dialComm(t, addr) + + c, err := newXshardConn( + raw, + 0, + []byte(id), + []uint32{0x00010001}, + append([]byte(nil), testSlaveID...), + append([]uint32(nil), testSlaveShards...), + h, + log.New(), + ) + if err != nil { + t.Fatalf("newXshardConn(%s): %v", id, err) + } + + c.Start() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + if _, _, err := c.sendPing(ctx); err != nil { + t.Fatalf("sendPing(%s): %v", id, err) + } + + waitFor(t, "indexed "+id, func() bool { + return comm.xshardPool.hasSlaveID([]byte(id)) + }) + + return c + } + + h1 := &countingXshardHandler{} + h2 := &countingXshardHandler{} + h3 := &countingXshardHandler{} + + newInbound("S1", h1) + c2 := newInbound("S2", h2) + newInbound("S3", h3) + + conns := comm.xshardPool.Lookup(0x00010001) + if len(conns) != 3 { + t.Fatalf("pool holds %d conns, want 3", len(conns)) + } + + c2.Close() + + waitFor(t, "conn closed", func() bool { + return conns[1].IsClosed() + }) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err := comm.SendXshardTxList( + ctx, + 0x00010001, + &wire.AddXshardTxListRequest{ + Branch: 1, + TxList: &types.CrossShardTransactionList{}, + }, + ) + + if err == nil { + t.Fatal("expected error") + } + + if got := h1.adds.Load(); got != 1 { + t.Fatalf("S1 calls = %d, want 1", got) + } + + if got := h2.adds.Load(); got != 0 { + t.Fatalf("S2 calls = %d, want 0", got) + } + + if got := h3.adds.Load(); got != 1 { + t.Fatalf("S3 calls = %d, want 1", got) + } +} diff --git a/qkc/cluster/slave/testdata/README_INTEROP.md b/qkc/cluster/slave/testdata/README_INTEROP.md new file mode 100644 index 00000000000..4bb9e5376c8 --- /dev/null +++ b/qkc/cluster/slave/testdata/README_INTEROP.md @@ -0,0 +1,144 @@ +# Python Interop Tests + +These tests verify wire / opcode / serializer / bootstrap compatibility between +the real Go Slave and the real Python Master (pyquarkchain). They are driven by +a single Python harness (`master_harness.py`) that reuses the *real* pyquarkchain +wire stack (`ClusterConnection`, the cluster OP serializer map, and the real RPC +request classes) to drive concrete interactions against Go Slaves. + +## Prerequisites + +1. Python 3.8+ **with pyquarkchain's Python dependencies installed** (see the + `requirements.txt` in the pyquarkchain checkout, e.g. `ecdsa`, `aiohttp`). + A dedicated virtualenv is recommended; the harness launches whatever + `python3` resolves to on `PATH`. +2. A checkout of pyquarkchain: + ```bash + git clone https://github.com/QuarkChain/pyquarkchain.git + ``` +3. Set the PYQUARKCHAIN environment variable: + ```bash + export PYQUARKCHAIN=/path/to/pyquarkchain + ``` + +## Running + +All interop tests: +```bash +PYQUARKCHAIN=/path/to/pyquarkchain go test -tags interop ./qkc/cluster/slave/ +``` + +Specific tests: +```bash +PYQUARKCHAIN=/path/to/pyquarkchain go test -tags interop -run TestInteropBootstrap ./qkc/cluster/slave/ +PYQUARKCHAIN=/path/to/pyquarkchain go test -tags interop -run TestInteropMasterRpcRoundTrip ./qkc/cluster/slave/ +PYQUARKCHAIN=/path/to/pyquarkchain go test -tags interop -run TestInteropPeerCreateOrDestroy ./qkc/cluster/slave/ +PYQUARKCHAIN=/path/to/pyquarkchain go test -tags interop -run TestInteropMasterDisconnectShutdown ./qkc/cluster/slave/ +PYQUARKCHAIN=/path/to/pyquarkchain go test -tags interop -run TestInteropPeerMessageRouting ./qkc/cluster/slave/ +PYQUARKCHAIN=/path/to/pyquarkchain go test -tags interop -run TestInteropAddMinorBlockHeaderToMaster ./qkc/cluster/slave/ +``` + +With race detector: +```bash +PYQUARKCHAIN=/path/to/pyquarkchain go test -race -tags interop ./qkc/cluster/slave/ +``` + +Distinguish three environment states: + +- `PYQUARKCHAIN` unset → tests **skip**. +- `python3` exists on `PATH` but lacks pyquarkchain's dependencies (e.g. + `ecdsa`, `aiohttp`) → the Python harness fails at startup/import and the + tests **fail** — they do *not* skip. This is a test-environment problem, not + a Go regression. +- `python3` + pyquarkchain + dependencies complete → tests run normally. + +## What is tested + +- **Bootstrap handshake** (`TestInteropBootstrap`) — `master_harness.py bootstrap` + runs the full `master.main()` against multiple Go Slaves: PING/PONG, root-tip + shard initialization, and `CONNECT_TO_SLAVES_REQUEST` establishing + slave-to-slave (xshard) connections. +- **Master→slave RPC round-trip** (`TestInteropMasterRpcRoundTrip`) — + `master_harness.py rpc` issues `GET_ACCOUNT_DATA_REQUEST` and confirms a + successful response. +- **Peer lifecycle isolation** (`TestInteropPeerCreateOrDestroy`) — + `master_harness.py peer ` drives two + `CREATE_CLUSTER_PEER_CONNECTION_REQUEST`s then destroys only the first via + `DESTROY_CLUSTER_PEER_CONNECTION_COMMAND`. Confirms both peers appear in the + slave's registry, peer 1 is removed on destroy, and peer 2 survives untouched — + i.e. `destroy(peer1)` does not affect peer 2. +- **Peer message routing** (`TestInteropPeerMessageRouting`) — + `master_harness.py peermsg` creates a cluster peer connection, then sends a P2P + `NEW_TRANSACTION_LIST` command addressed to a target `ClusterMetadata{branch, + cluster_peer_id}`. Confirms the frame travels the real path + `MasterConn.routeFrame → PeerConn.HandleFrame → PeerHandler` and that the command + reaches the handler for the exact `(cluster_peer_id, branch)` target. +- **Master disconnect → slave shutdown** (`TestInteropMasterDisconnectShutdown`) — + `master_harness.py disconnect` closes its connection and confirms the slave shuts down. +- **Slave → Master AddMinorBlockHeader** (`TestInteropAddMinorBlockHeaderToMaster`) — + `master_harness.py addminorblock` awaits the Go slave's AddMinorBlockHeader request, + replies with a real AddMinorBlockHeaderResponse. This verifies the only Go → Python + outbound RPC direction: Go wire serialization → Python deserialization → Python + response encoding → Go response decoding. + +## Scope + +Interop verifies only the real Python Master ↔ real Go Slave boundary: +wire, opcode, serializer and bootstrap compatibility. It deliberately does not +verify lock ordering, races, timeouts, pending RPCs or internal state — those are +covered by the Go unit tests in `conn/base_test.go`, `master_conn_test.go`, +`peer_conn_test.go`, `xshard_test.go` and `slave_test.go`. + +## Architecture + +``` +slave/ +├── interop_harness.go # Single Go harness: env guards, backend, slave/process harness, cluster config +├── interop_test.go # All interop tests (Bootstrap / RPC / Peer / Routing / Disconnect / AddMinor) +└── testdata/ + ├── master_harness.py # Single Python driver: bootstrap / rpc / peer / disconnect / addminorblock + └── README_INTEROP.md # This file +``` + +### Go harness + +- `interopBackend` serves every business handler with a trivial-success response + while recording lifecycle events (shard creation), so tests exercise the real + communication layer without a business runtime. +- `startTestSlave` starts a single Go Slave with a communication-only backend. +- `startScenarioMaster` runs a `master_harness.py` sub-command in the background + and lets tests poll its output (`WaitLine`) and exit status (`WaitExit`). +- `InteropCluster` starts N Go Slaves, writes a `cluster_config.json`, launches + the full `master.main()` via `master_harness.py bootstrap`, and `WaitBootstrap` + polls the slaves' `ShardsCreated` events and xshard registrations. + +### Python harness + +`master_harness.py` is a single entry point; it reuses the real `ClusterConnection` +wire stack and cluster OP serializer map and performs only test-scripted +orchestration on top. Sub-commands: + +- `bootstrap` — patches the environment (hostname, qkchash native stubs, SimpleNetwork) + then runs the real `master.main()`; this heavy path is loaded lazily. +- `rpc ` +- `peer [--hold]` — creates multiple + cluster peers, then destroys only the first. Used for `TestInteropPeerCreateOrDestroy`. +- `peermsg [--wait]` — creates + the cluster peer, waits for it to be registered by Go, then sends a P2P + `NEW_TRANSACTION_LIST` command addressed to the `(cluster_peer_id, branch)` via + `ClusterMetadata`. Used for `TestInteropPeerMessageRouting`. +- `disconnect ` +- `addminorblock [--wait]` — connects, PING/PONG, prints + `READY`, then answers the Go slave's AddMinorBlockHeader request and prints + `ADD_MINOR_BLOCK_OK error_code=0`. `--wait` bounds how long it waits for the request. + +It does *not* reimplement the wire protocol. The `addminorblock` handler uses the +real `AddMinorBlockHeaderRequest` serializer for decode, the real +`AddMinorBlockHeaderResponse` class for the reply, and the real `ClusterConnection` +RPC framing; it never touches MasterServer block-processing state. + +## Environment Variables + +| Variable | Required | Description | +|----------|----------|-------------| +| `PYQUARKCHAIN` | Yes | Path to pyquarkchain checkout | diff --git a/qkc/cluster/slave/testdata/master_harness.py b/qkc/cluster/slave/testdata/master_harness.py new file mode 100644 index 00000000000..4c5af0646fe --- /dev/null +++ b/qkc/cluster/slave/testdata/master_harness.py @@ -0,0 +1,429 @@ +#!/usr/bin/env python3 +# Copyright 2026-2027, QuarkChain. +"""Single Python driver for Go Slave interop tests. + +Unified entry point backed by the *real* pyquarkchain wire stack. It never +reimplements the wire protocol — it reuses ClusterConnection, the cluster OP +serializer map and the real RPC request classes, and only performs the +test-scripted orchestration on top. + +Sub-commands: + + bootstrap Run the full Python master (master.main()[mounted]). + rpc Connect to one Go slave, PING, then round-trip a business RPC. + peer Connect to one Go slave, drive CREATE/DESTROY cluster peer. + disconnect Connect to one Go slave, PING, then close the connection. + +The heavy bootstrap path (master.main + native qkchash) is loaded lazily only +when the "bootstrap" sub-command is used, so the lighter scenario paths do not +need the native mining libraries. +""" +import asyncio +import os +import sys + +pyquarkchain_path = os.environ.get("PYQUARKCHAIN") +if not pyquarkchain_path: + print("ERROR: PYQUARKCHAIN environment variable is not set", file=sys.stderr) + sys.exit(1) +sys.path.insert(0, pyquarkchain_path) + +from quarkchain.cluster.p2p_commands import ( + CommandOp, + NewTransactionListCommand, + OP_SERIALIZER_MAP as P2P_OP_SERIALIZER_MAP, +) +from quarkchain.cluster.protocol import ClusterConnection, ClusterMetadata +from quarkchain.cluster.rpc import ( + CLUSTER_OP_SERIALIZER_MAP, + ClusterOp, + AddMinorBlockHeaderResponse, + ArtificialTxConfig, + CreateClusterPeerConnectionRequest, + DestroyClusterPeerConnectionCommand, + GetAccountDataRequest, + Ping, +) +from quarkchain.core import Address, Branch +from quarkchain.protocol import ConnectionState + + +def _merge_op_maps(): + merged = dict(CLUSTER_OP_SERIALIZER_MAP) + merged.update(P2P_OP_SERIALIZER_MAP) + return merged + + +OP_SER_MAP = _merge_op_maps() + + +class DummyEnv: + """Minimal environment object required by quarkchain.protocol.Connection.""" + + class cluster_config: + @staticmethod + def get_slave_command_size_limit(): + return None + + +class MasterToSlaveConnection(ClusterConnection): + """Master-side connection to a Go Slave. + + Reuses the real ClusterConnection wire stack with the real cluster OP + serializer map. It does not forward slave->peer traffic because the + scenario tests only exercise master->slave orchestration. + + ADD_MINOR_BLOCK_HEADER_REQUEST is answered here (not forwarded to a real + MasterServer) because this harness tests Go↔Python wire compatibility, not + Python block-processing. The real request serializer performs the decode + and the real AddMinorBlockHeaderResponse class performs the reply encode. + """ + + def __init__(self, reader, writer, name=None): + super().__init__( + DummyEnv, + reader, + writer, + OP_SER_MAP, + {}, # op_non_rpc_map + { + ClusterOp.ADD_MINOR_BLOCK_HEADER_REQUEST: ( + ClusterOp.ADD_MINOR_BLOCK_HEADER_RESPONSE, + _handle_add_minor_block_header, + ), + }, # op_rpc_map + name=name, + ) + self._loop_task = asyncio.create_task(self.active_and_loop_forever()) + + def get_connection_to_forward(self, metadata): + return None + + async def shutdown(self): + if self.state != ConnectionState.CLOSED: + self.close() + await self.wait_until_closed() + + +async def open_connection(host, port, name=None): + reader, writer = await asyncio.open_connection(host, port) + conn = MasterToSlaveConnection(reader, writer, name=name) + await conn.wait_until_active() + return conn + + +def parse_shards(s): + return [int(x.strip(), 0) for x in s.split(",") if x.strip()] + + +# --------------------------------------------------------------------------- +# Scenario sub-commands +# --------------------------------------------------------------------------- + + +async def do_rpc(args): + conn = await open_connection(args.host, args.port, name="py-master-rpc") + try: + req = Ping(b"", [], None) + await conn.write_rpc_request(ClusterOp.PING, req) + + # full_shard_key=0 addresses the root shard; the Go slave's + # communication-only backend returns success for any address. + address = Address(bytes.fromhex(args.address_hex), full_shard_key=0) + req = GetAccountDataRequest(address) + op, resp, _ = await conn.write_rpc_request(ClusterOp.GET_ACCOUNT_DATA_REQUEST, req) + print(f"RPC_OK error_code={resp.error_code}", flush=True) + return 0 if resp.error_code == 0 else 1 + finally: + await conn.shutdown() + + +async def do_peer(args): + conn = await open_connection(args.host, args.port, name="py-master-peer") + try: + # Handshake: master sends PING, slave replies PONG. + req = Ping(b"", [], None) + op, resp, _ = await conn.write_rpc_request(ClusterOp.PING, req) + + # Create two virtual cluster peer connections, then destroy only the + # first so the Go test can assert destroy(peer1) leaves peer2 intact. + for peer_id in args.cluster_peer_ids: + create_req = CreateClusterPeerConnectionRequest(peer_id) + op, create_resp, _ = await conn.write_rpc_request( + ClusterOp.CREATE_CLUSTER_PEER_CONNECTION_REQUEST, create_req + ) + if create_resp.error_code != 0: + print(f"CREATE_PEER error_code={create_resp.error_code}", file=sys.stderr, flush=True) + return 1 + print(f"PEER_CREATED {peer_id}", flush=True) + # Leave a window for the Go test to observe each peer before the next. + await asyncio.sleep(0.5) + + # Destroy only the first peer (fire-and-forget), then leave a window for + # the Go test to observe that peer2 survives peer1's destroy. + destroy = DestroyClusterPeerConnectionCommand(args.cluster_peer_ids[0]) + conn.write_command(ClusterOp.DESTROY_CLUSTER_PEER_CONNECTION_COMMAND, destroy, rpc_id=0) + await asyncio.sleep(args.hold) + print("PEER_DESTROYED", flush=True) + # Keep the connection alive after announcing the destroy: the Go test + # asserts peer2 still exists in the slave registry, but the master + # disconnect would tear down every peer, so the scenario must not + # shutdown until the assertion window has passed. + await asyncio.sleep(args.hold) + return 0 + finally: + await conn.shutdown() + + +async def do_peermsg(args): + conn = await open_connection(args.host, args.port, name="py-master-peermsg") + try: + # Handshake. + req = Ping(b"", [], None) + op, resp, _ = await conn.write_rpc_request(ClusterOp.PING, req) + + # Create the virtual cluster peer connection the message is routed to. + create_req = CreateClusterPeerConnectionRequest(args.cluster_peer_id) + op, create_resp, _ = await conn.write_rpc_request( + ClusterOp.CREATE_CLUSTER_PEER_CONNECTION_REQUEST, create_req + ) + if create_resp.error_code != 0: + print(f"CREATE_PEER error_code={create_resp.error_code}", file=sys.stderr, flush=True) + return 1 + print(f"PEER_CREATED {args.cluster_peer_id}", flush=True) + # Leave a window for the Go side to finish building the virtual PeerConn. + await asyncio.sleep(args.wait) + + # Send a P2P command addressed to (cluster_peer_id, branch). The empty + # transaction list serializes to a 4-byte zero length, byte-identical on + # both sides, so it is a minimal wire-valid payload. + cmd = NewTransactionListCommand() + metadata = ClusterMetadata(branch=Branch(args.branch), cluster_peer_id=args.cluster_peer_id) + conn.write_command(CommandOp.NEW_TRANSACTION_LIST, cmd, rpc_id=0, metadata=metadata) + await asyncio.sleep(args.wait) + print("PEER_MSG_SENT", flush=True) + return 0 + finally: + await conn.shutdown() + + +async def do_disconnect(args): + conn = await open_connection(args.host, args.port, name="py-master-disconnect") + try: + req = Ping(b"", [], None) + await conn.write_rpc_request(ClusterOp.PING, req) + await asyncio.sleep(0.2) + print("CONNECTED", flush=True) + # Closing the connection is the master disconnecting; the Go slave must + # observe it and shut down. + await asyncio.sleep(0.2) + await conn.shutdown() + print("DISCONNECT_SENT", flush=True) + return 0 + finally: + await conn.shutdown() + + +# Set lazily per addminorblock run; the handler signals it once the Go slave's +# request has been received and a response has been written. +_add_minor_block_handled = None + + +async def _handle_add_minor_block_header(self, request): + """Answer a Go slave's AddMinorBlockHeader request. + + Deliberately does NOT touch any MasterServer block-processing state. It + decodes via the real request serializer (already done by the connection), + echoes a real AddMinorBlockHeaderResponse, and lets the real RPC layer + encode and send it back. Verifies Go encode → Python decode → Python + encode → Go decode. + """ + resp = AddMinorBlockHeaderResponse( + error_code=0, + artificial_tx_config=ArtificialTxConfig( + target_root_block_time=10, target_minor_block_time=10 + ), + ) + ev = _add_minor_block_handled + if ev is not None: + ev.set() + return resp + + +async def do_add_minor_block(args): + global _add_minor_block_handled + conn = await open_connection(args.host, args.port, name="py-master-addminorblock") + try: + # Handshake: master sends PING and awaits the slave's PONG, confirming + # the Go slave's MasterConn is active before signalling the test. + req = Ping(b"", [], None) + await conn.write_rpc_request(ClusterOp.PING, req) + print("READY", flush=True) + + # Go slave now triggers SendMinorBlockHeaderToMaster; wait for the + # request to arrive and our response to be written. + _add_minor_block_handled = asyncio.Event() + await asyncio.wait_for(_add_minor_block_handled.wait(), timeout=args.wait) + # Give the response a moment to flush back to the Go slave. + await asyncio.sleep(0.1) + print("ADD_MINOR_BLOCK_OK error_code=0", flush=True) + return 0 + finally: + _add_minor_block_handled = None + await conn.shutdown() + + +# --------------------------------------------------------------------------- +# Full-master sub-command (heavy, loaded lazily) +# --------------------------------------------------------------------------- + + +def _patch_hostname(): + """Route hostname lookups to localhost (macOS may not resolve .local names).""" + import socket + + _original_gethostname = socket.gethostname + _original_gethostbyname = socket.gethostbyname + + def _patched_gethostname(): + return "localhost" + + def _patched_gethostbyname(name): + if name == "localhost": + return "127.0.0.1" + return _original_gethostbyname(name) + + socket.gethostname = _patched_gethostname + socket.gethostbyname = _patched_gethostbyname + + +def _stub_qkchash(): + """Stub the native qkchash library so master.main() runs without it. + + The native library is only needed for actual PoW mining, which is not part + of a bootstrap smoke test. + """ + + class _StubQkcHashNative: + def __init__(self, lib_path=None): + pass + + def hash(self, *args, **kwargs): + raise NotImplementedError("qkchash native library not available") + + def mine(self, *args, **kwargs): + raise NotImplementedError("qkchash native library not available") + + class _StubQkchash: + QkcHashNative = _StubQkcHashNative + + sys.modules["qkchash.qkchash"] = _StubQkchash() + + class _StubQkchashMiner: + def __init__(self, qkc_hash_native=None): + pass + + def mine(self, *args, **kwargs): + raise NotImplementedError("qkchash miner not available") + + def check_pow(self, *args, **kwargs): + return False + + def _stub_check_pow(header_hash, nonce, boundary, qkc_hash_native): + return False + + class _StubQkcpow: + QkchashMiner = _StubQkchashMiner + check_pow = _stub_check_pow + QKC_HASH_NATIVE = _StubQkcHashNative() + + sys.modules["qkchash.qkcpow"] = _StubQkcpow() + sys.modules["qkchash"] = type("_StubQkchashPackage", (), {})() + + +def run_bootstrap(): + """Patch environment, then run the real Python master. Uses the master's + own argument parser, so script argv is rewritten to drop the "bootstrap" + token and hand the remaining --cluster_config to master.main().""" + _patch_hostname() + _stub_qkchash() + + from quarkchain.cluster.cluster_config import ClusterConfig + + # Force SimpleNetwork so the master's P2P layer does not require RLPx. + ClusterConfig.use_p2p = lambda self: False + + from quarkchain.cluster.master import main + + sys.exit(main()) + + +def main(): + import argparse + + parser = argparse.ArgumentParser(description="Go Slave interop master harness") + subparsers = parser.add_subparsers(dest="command", required=True) + + subparsers.add_parser("bootstrap", help="run the full Python master (master.main())") + + for name in ("disconnect",): + p = subparsers.add_parser(name) + p.add_argument("host") + p.add_argument("port", type=int) + p.add_argument("master_shards") + + p = subparsers.add_parser("rpc") + p.add_argument("host") + p.add_argument("port", type=int) + p.add_argument("master_shards") + p.add_argument("address_hex") + + p = subparsers.add_parser("peer") + p.add_argument("host") + p.add_argument("port", type=int) + p.add_argument("master_shards") + p.add_argument("cluster_peer_ids", nargs="+", type=int) + p.add_argument("--hold", type=float, default=1.0) + + p = subparsers.add_parser("peermsg") + p.add_argument("host") + p.add_argument("port", type=int) + p.add_argument("master_shards") + p.add_argument("cluster_peer_id", type=int) + p.add_argument("branch", type=int) + p.add_argument("--wait", type=float, default=2.0) + + p = subparsers.add_parser("addminorblock") + p.add_argument("host") + p.add_argument("port", type=int) + p.add_argument("master_shards") + p.add_argument("--wait", type=float, default=15.0) + + args, extras = parser.parse_known_args() + + if args.command == "bootstrap": + # Ensure the master sees a clean argv: script name + its own args. + sys.argv = [sys.argv[0]] + extras + run_bootstrap() + return + + args.master_shards = parse_shards(args.master_shards) + if args.command == "rpc": + rc = asyncio.run(do_rpc(args)) + elif args.command == "peer": + rc = asyncio.run(do_peer(args)) + elif args.command == "peermsg": + rc = asyncio.run(do_peermsg(args)) + elif args.command == "disconnect": + rc = asyncio.run(do_disconnect(args)) + elif args.command == "addminorblock": + rc = asyncio.run(do_add_minor_block(args)) + else: + parser.print_help() + rc = 2 + + sys.exit(rc) + + +if __name__ == "__main__": + main() \ No newline at end of file