diff --git a/qkc/cluster/slave/compat_test.go b/qkc/cluster/slave/compat_test.go index 6907d4e094..79e97c21e2 100644 --- a/qkc/cluster/slave/compat_test.go +++ b/qkc/cluster/slave/compat_test.go @@ -4,7 +4,9 @@ package slave import ( "bufio" + "bytes" "context" + "encoding/binary" "fmt" "net" "os" @@ -12,12 +14,122 @@ import ( "path/filepath" "runtime" "strings" + "sync" "testing" "time" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/qkc/cluster/wire" ) +// startPythonMaster starts the Python master.py subprocess and returns the +// TCP port it listens on, a function to retrieve captured stdout lines, and a +// cleanup function. The peer listens on a random port (port=0) and prints +// "PORT:" to stdout when ready. +func startPythonMaster(t *testing.T, extraArgs ...string) (int, func() []string, func()) { + t.Helper() + + _, filename, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("cannot get caller path") + } + pyScript := filepath.Join(filepath.Dir(filename), "testdata", "pyproto", "master.py") + + if _, err := exec.LookPath("python3"); err != nil { + t.Skip("python3 not found in PATH") + } + if _, err := os.Stat(pyScript); err != nil { + t.Skipf("master.py not found at %s", pyScript) + } + + args := []string{pyScript, "--port", "0", "--id", "py-master", "--shards", "1,2"} + args = append(args, extraArgs...) + + cmd := exec.Command("python3", args...) + stdout, err := cmd.StdoutPipe() + if err != nil { + t.Fatalf("stdout pipe: %v", err) + } + cmd.Stderr = os.Stderr + + if err := cmd.Start(); err != nil { + t.Fatalf("start python master: %v", err) + } + + portCh := make(chan int, 1) + var outputLines []string + var outputMu sync.Mutex + go func() { + scanner := bufio.NewScanner(stdout) + for scanner.Scan() { + line := scanner.Text() + outputMu.Lock() + outputLines = append(outputLines, line) + outputMu.Unlock() + if strings.HasPrefix(line, "PORT:") { + var port int + if _, err := fmt.Sscanf(line, "PORT:%d", &port); err == nil { + portCh <- port + } + } + } + }() + + var port int + select { + case port = <-portCh: + case <-time.After(5 * time.Second): + cmd.Process.Kill() + cmd.Wait() + t.Fatal("timeout waiting for python master port") + } + + getOutput := func() []string { + outputMu.Lock() + defer outputMu.Unlock() + out := make([]string, len(outputLines)) + copy(out, outputLines) + return out + } + + cleanup := func() { + cmd.Process.Kill() + cmd.Wait() + } + + return port, getOutput, cleanup +} + +// dialPythonMaster starts python master.py and dials the port it listens on, +// wrapping the connection in a MasterConn. Returns the MasterConn, a function +// to retrieve captured stdout lines, and a cleanup function. +func dialPythonMaster(t *testing.T) (*MasterConn, func() []string, func()) { + t.Helper() + + port, getOutput, cleanupPy := startPythonMaster(t) + + addr := fmt.Sprintf("127.0.0.1:%d", port) + mc, err := NewMasterConn( + addr, + 0, + []byte("go-slave"), + []uint32{0x00010001, 0x00020001}, + log.New(), + ) + if err != nil { + cleanupPy() + t.Fatalf("create MasterConn: %v", err) + } + mc.Start() + + cleanup := func() { + mc.Close() + cleanupPy() + } + + return mc, getOutput, cleanup +} + // startPythonPeer starts a Python protocol peer subprocess and returns the // TCP port and a cleanup function. The peer listens on a random port (port=0) // and prints "PORT:" to stdout when ready. @@ -317,3 +429,84 @@ func TestPythonCompat_PoolReconnect(t *testing.T) { t.Fatalf("pool size after reconnect: got %d, want 1", pool.OutboundSize()) } } + +// --------------------------------------------------------------------------- +// Test: Python Master -> Go Slave full handshake + RPC flow +// +// Validates: Python MasterConnection behavior against Go MasterConn. +// Python sends PING, GetEcoInfoListRequest, AddRootBlockRequest, and +// DestroyClusterPeerConnectionCommand. Go must decode the 12-byte +// ClusterMetadata frames, dispatch to the correct handlers, and return +// protocol-compatible responses. +// --------------------------------------------------------------------------- +func TestPythonCompat_MasterFullFlow(t *testing.T) { + mc, getOutput, cleanup := dialPythonMaster(t) + defer cleanup() + + // Wait for the Python master to finish its scripted exchange. + select { + case <-mc.WaitUntilClosed(): + case <-time.After(15 * time.Second): + output := getOutput() + t.Fatalf("MasterConn did not close after Python master finished; output=%v", output) + } + + // Allow a moment for the scanner goroutine to drain the Python stdout pipe. + time.Sleep(100 * time.Millisecond) + + output := getOutput() + expected := []string{ + "PONG_OK id=676f2d736c617665", // hex of "go-slave" + "ECO_OK error_code=0", + "ROOT_OK error_code=0", + "DESTROY_OK", + "PONG_OK id=676f2d736c617665", + "DISCONNECTED", + } + for _, exp := range expected { + found := false + for _, line := range output { + if line == exp { + found = true + break + } + } + if !found { + t.Fatalf("expected output line %q not found in %v", exp, output) + } + } +} + +// --------------------------------------------------------------------------- +// Test: Python-generated ClusterMetadata frame layout +// +// Validates: the 12-byte ClusterMetadata encoding (4-byte branch + 8-byte +// cluster_peer_id) is the same on both sides of the wire. +// --------------------------------------------------------------------------- +func TestPythonCompat_MasterFrameLayout(t *testing.T) { + // This is a static golden-vector test: we compare Go's wire format against + // the documented Python frame layout without requiring a Python subprocess. + meta := wire.ClusterMetadata{Branch: 0x01020304, ClusterPeerID: 0x1122334455667788} + frame := &wire.Frame{ + Meta: meta, + Opcode: byte(wire.ClusterOpPing), + RPCID: 1, + Payload: []byte{0xAA, 0xBB}, + } + + var buf bytes.Buffer + if err := wire.WriteFrame(&buf, frame); err != nil { + t.Fatalf("WriteFrame: %v", err) + } + + wireBytes := buf.Bytes() + if len(wireBytes) != 4+12+1+8+2 { + t.Fatalf("frame length: got %d, want %d", len(wireBytes), 4+12+1+8+2) + } + if binary.BigEndian.Uint32(wireBytes[4:8]) != meta.Branch { + t.Fatalf("branch mismatch") + } + if binary.BigEndian.Uint64(wireBytes[8:16]) != meta.ClusterPeerID { + t.Fatalf("cluster_peer_id mismatch") + } +} diff --git a/qkc/cluster/slave/master_conn.go b/qkc/cluster/slave/master_conn.go new file mode 100644 index 0000000000..fbf34f59f9 --- /dev/null +++ b/qkc/cluster/slave/master_conn.go @@ -0,0 +1,550 @@ +// Copyright 2026-2027, QuarkChain. + +package slave + +import ( + "context" + "fmt" + "io" + "net" + + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/qkc/cluster/wire" + "github.com/ethereum/go-ethereum/qkc/serialize" +) + +// MasterConn represents the slave-side TCP connection to the cluster master. +// It corresponds to Python's quarkchain.cluster.slave.MasterConnection and uses +// 12-byte ClusterMetadata framing. +// +// Architecture: +// +// MasterConn embeds *rpcConn +// +// All master→slave ClusterOp handlers are registered during construction. +// Business handlers that depend on unported components (Shard, StateDB, etc.) +// are implemented as protocol-compatible stubs that return valid responses. +type MasterConn struct { + *rpcConn + + localID []byte + localFullShardIDList []uint32 +} + +// NewMasterConn dials the master at addr and returns a MasterConn. +// maxPayloadSize controls frame payload size limit; 0 disables the limit. +// localID and localFullShardIDList identify this slave and are used in PONG. +func NewMasterConn(addr string, maxPayloadSize uint32, localID []byte, localFullShardIDList []uint32, logger log.Logger) (*MasterConn, error) { + conn, err := net.DialTimeout("tcp", addr, defaultDialTimeout) + if err != nil { + return nil, fmt.Errorf("dial master %s: %w", addr, err) + } + return newMasterConn(conn, maxPayloadSize, localID, localFullShardIDList, logger), nil +} + +// NewMasterConnFromConn wraps an accepted net.Conn as a MasterConn. +// maxPayloadSize controls frame payload size limit; 0 disables the limit. +func NewMasterConnFromConn(conn net.Conn, maxPayloadSize uint32, localID []byte, localFullShardIDList []uint32, logger log.Logger) *MasterConn { + return newMasterConn(conn, maxPayloadSize, localID, localFullShardIDList, logger) +} + +func newMasterConn(conn net.Conn, maxPayloadSize uint32, localID []byte, localFullShardIDList []uint32, logger log.Logger) *MasterConn { + readFrame := func(r io.Reader) (*wire.Frame, error) { + return wire.ReadFrame(r, maxPayloadSize) + } + mc := &MasterConn{ + rpcConn: newRPCConn(conn, readFrame, wire.WriteFrame, logger), + localID: append([]byte(nil), localID...), + localFullShardIDList: append([]uint32(nil), localFullShardIDList...), + } + + mc.registerOpSerializers() + mc.registerHandlers() + + return mc +} + +// registerOpSerializers registers serializers for every opcode in Python's +// CLUSTER_OP_SERIALIZER_MAP. This covers master→slave, slave→master and +// slave→slave opcodes so outbound RPC responses can be deserialized if needed. +func (mc *MasterConn) registerOpSerializers() { + mc.rpcConn.RegisterOpSerializers(map[byte]*OpSerializer{ + // §1 Cluster initialisation + byte(wire.ClusterOpPing): OpSerializerFor[wire.PingRequest, wire.PongResponse](), + byte(wire.ClusterOpPong): OpSerializerFor[wire.PongResponse, wire.PingRequest](), + byte(wire.ClusterOpConnectToSlavesRequest): OpSerializerFor[wire.ConnectToSlavesRequest, wire.ConnectToSlavesResponse](), + byte(wire.ClusterOpConnectToSlavesResponse): OpSerializerFor[wire.ConnectToSlavesResponse, wire.ConnectToSlavesRequest](), + byte(wire.ClusterOpAddRootBlockRequest): OpSerializerFor[wire.AddRootBlockRequest, wire.AddRootBlockResponse](), + byte(wire.ClusterOpAddRootBlockResponse): OpSerializerFor[wire.AddRootBlockResponse, wire.AddRootBlockRequest](), + byte(wire.ClusterOpGetEcoInfoListRequest): OpSerializerFor[wire.GetEcoInfoListRequest, wire.GetEcoInfoListResponse](), + byte(wire.ClusterOpGetEcoInfoListResponse): OpSerializerFor[wire.GetEcoInfoListResponse, wire.GetEcoInfoListRequest](), + byte(wire.ClusterOpGetNextBlockToMineRequest): OpSerializerFor[wire.GetNextBlockToMineRequest, wire.GetNextBlockToMineResponse](), + byte(wire.ClusterOpGetNextBlockToMineResponse): OpSerializerFor[wire.GetNextBlockToMineResponse, wire.GetNextBlockToMineRequest](), + byte(wire.ClusterOpGetUnconfirmedHeadersRequest): OpSerializerFor[wire.GetUnconfirmedHeadersRequest, wire.GetUnconfirmedHeadersResponse](), + byte(wire.ClusterOpGetUnconfirmedHeadersResponse): OpSerializerFor[wire.GetUnconfirmedHeadersResponse, wire.GetUnconfirmedHeadersRequest](), + byte(wire.ClusterOpGetAccountDataRequest): OpSerializerFor[wire.GetAccountDataRequest, wire.GetAccountDataResponse](), + byte(wire.ClusterOpGetAccountDataResponse): OpSerializerFor[wire.GetAccountDataResponse, wire.GetAccountDataRequest](), + byte(wire.ClusterOpAddTransactionRequest): OpSerializerFor[wire.AddTransactionRequest, wire.AddTransactionResponse](), + byte(wire.ClusterOpAddTransactionResponse): OpSerializerFor[wire.AddTransactionResponse, wire.AddTransactionRequest](), + + // §2 Slave → Master (mining) + byte(wire.ClusterOpAddMinorBlockHeaderRequest): OpSerializerFor[wire.AddMinorBlockHeaderRequest, wire.AddMinorBlockHeaderResponse](), + byte(wire.ClusterOpAddMinorBlockHeaderResponse): OpSerializerFor[wire.AddMinorBlockHeaderResponse, wire.AddMinorBlockHeaderRequest](), + + // §3 Slave ↔ Slave (xshard direct) + byte(wire.ClusterOpAddXshardTxListRequest): OpSerializerFor[wire.AddXshardTxListRequest, wire.AddXshardTxListResponse](), + byte(wire.ClusterOpAddXshardTxListResponse): OpSerializerFor[wire.AddXshardTxListResponse, wire.AddXshardTxListRequest](), + + // §4 Master → Slave (sync / virtual conns) + byte(wire.ClusterOpSyncMinorBlockListRequest): OpSerializerFor[wire.SyncMinorBlockListRequest, wire.SyncMinorBlockListResponse](), + byte(wire.ClusterOpSyncMinorBlockListResponse): OpSerializerFor[wire.SyncMinorBlockListResponse, wire.SyncMinorBlockListRequest](), + byte(wire.ClusterOpAddMinorBlockRequest): OpSerializerFor[wire.AddMinorBlockRequest, wire.AddMinorBlockResponse](), + byte(wire.ClusterOpAddMinorBlockResponse): OpSerializerFor[wire.AddMinorBlockResponse, wire.AddMinorBlockRequest](), + byte(wire.ClusterOpCreateClusterPeerConnectionRequest): OpSerializerFor[wire.CreateClusterPeerConnectionRequest, wire.CreateClusterPeerConnectionResponse](), + byte(wire.ClusterOpCreateClusterPeerConnectionResponse): OpSerializerFor[wire.CreateClusterPeerConnectionResponse, wire.CreateClusterPeerConnectionRequest](), + byte(wire.ClusterOpDestroyClusterPeerConnectionCommand): OpSerializerFor[wire.DestroyClusterPeerConnectionCommand, wire.DestroyClusterPeerConnectionCommand](), + byte(wire.ClusterOpGetMinorBlockRequest): OpSerializerFor[wire.GetMinorBlockRequest, wire.GetMinorBlockResponse](), + byte(wire.ClusterOpGetMinorBlockResponse): OpSerializerFor[wire.GetMinorBlockResponse, wire.GetMinorBlockRequest](), + byte(wire.ClusterOpGetTransactionRequest): OpSerializerFor[wire.GetTransactionRequest, wire.GetTransactionResponse](), + byte(wire.ClusterOpGetTransactionResponse): OpSerializerFor[wire.GetTransactionResponse, wire.GetTransactionRequest](), + + // §5 Slave ↔ Slave (xshard batch) + byte(wire.ClusterOpBatchAddXshardTxListRequest): OpSerializerFor[wire.BatchAddXshardTxListRequest, wire.BatchAddXshardTxListResponse](), + byte(wire.ClusterOpBatchAddXshardTxListResponse): OpSerializerFor[wire.BatchAddXshardTxListResponse, wire.BatchAddXshardTxListRequest](), + + // §6 Master → Slave (JSON-RPC-like) + byte(wire.ClusterOpExecuteTransactionRequest): OpSerializerFor[wire.ExecuteTransactionRequest, wire.ExecuteTransactionResponse](), + byte(wire.ClusterOpExecuteTransactionResponse): OpSerializerFor[wire.ExecuteTransactionResponse, wire.ExecuteTransactionRequest](), + byte(wire.ClusterOpGetTransactionReceiptRequest): OpSerializerFor[wire.GetTransactionReceiptRequest, wire.GetTransactionReceiptResponse](), + byte(wire.ClusterOpGetTransactionReceiptResponse): OpSerializerFor[wire.GetTransactionReceiptResponse, wire.GetTransactionReceiptRequest](), + byte(wire.ClusterOpMineRequest): OpSerializerFor[wire.MineRequest, wire.MineResponse](), + byte(wire.ClusterOpMineResponse): OpSerializerFor[wire.MineResponse, wire.MineRequest](), + byte(wire.ClusterOpGenTxRequest): OpSerializerFor[wire.GenTxRequest, wire.GenTxResponse](), + byte(wire.ClusterOpGenTxResponse): OpSerializerFor[wire.GenTxResponse, wire.GenTxRequest](), + byte(wire.ClusterOpGetTransactionListByAddressRequest): OpSerializerFor[wire.GetTransactionListByAddressRequest, wire.GetTransactionListByAddressResponse](), + byte(wire.ClusterOpGetTransactionListByAddressResponse): OpSerializerFor[wire.GetTransactionListByAddressResponse, wire.GetTransactionListByAddressRequest](), + byte(wire.ClusterOpGetLogRequest): OpSerializerFor[wire.GetLogRequest, wire.GetLogResponse](), + byte(wire.ClusterOpGetLogResponse): OpSerializerFor[wire.GetLogResponse, wire.GetLogRequest](), + byte(wire.ClusterOpEstimateGasRequest): OpSerializerFor[wire.EstimateGasRequest, wire.EstimateGasResponse](), + byte(wire.ClusterOpEstimateGasResponse): OpSerializerFor[wire.EstimateGasResponse, wire.EstimateGasRequest](), + byte(wire.ClusterOpGetStorageRequest): OpSerializerFor[wire.GetStorageRequest, wire.GetStorageResponse](), + byte(wire.ClusterOpGetStorageResponse): OpSerializerFor[wire.GetStorageResponse, wire.GetStorageRequest](), + byte(wire.ClusterOpGetCodeRequest): OpSerializerFor[wire.GetCodeRequest, wire.GetCodeResponse](), + byte(wire.ClusterOpGetCodeResponse): OpSerializerFor[wire.GetCodeResponse, wire.GetCodeRequest](), + byte(wire.ClusterOpGasPriceRequest): OpSerializerFor[wire.GasPriceRequest, wire.GasPriceResponse](), + byte(wire.ClusterOpGasPriceResponse): OpSerializerFor[wire.GasPriceResponse, wire.GasPriceRequest](), + byte(wire.ClusterOpGetWorkRequest): OpSerializerFor[wire.GetWorkRequest, wire.GetWorkResponse](), + byte(wire.ClusterOpGetWorkResponse): OpSerializerFor[wire.GetWorkResponse, wire.GetWorkRequest](), + byte(wire.ClusterOpSubmitWorkRequest): OpSerializerFor[wire.SubmitWorkRequest, wire.SubmitWorkResponse](), + byte(wire.ClusterOpSubmitWorkResponse): OpSerializerFor[wire.SubmitWorkResponse, wire.SubmitWorkRequest](), + + // §7 Slave → Master (block list) + byte(wire.ClusterOpAddMinorBlockHeaderListRequest): OpSerializerFor[wire.AddMinorBlockHeaderListRequest, wire.AddMinorBlockHeaderListResponse](), + byte(wire.ClusterOpAddMinorBlockHeaderListResponse): OpSerializerFor[wire.AddMinorBlockHeaderListResponse, wire.AddMinorBlockHeaderListRequest](), + + // §8 Master → Slave (JRPC & staking) + byte(wire.ClusterOpCheckMinorBlockRequest): OpSerializerFor[wire.CheckMinorBlockRequest, wire.CheckMinorBlockResponse](), + byte(wire.ClusterOpCheckMinorBlockResponse): OpSerializerFor[wire.CheckMinorBlockResponse, wire.CheckMinorBlockRequest](), + byte(wire.ClusterOpGetAllTransactionsRequest): OpSerializerFor[wire.GetAllTransactionsRequest, wire.GetAllTransactionsResponse](), + byte(wire.ClusterOpGetAllTransactionsResponse): OpSerializerFor[wire.GetAllTransactionsResponse, wire.GetAllTransactionsRequest](), + byte(wire.ClusterOpGetRootChainStakesRequest): OpSerializerFor[wire.GetRootChainStakesRequest, wire.GetRootChainStakesResponse](), + byte(wire.ClusterOpGetRootChainStakesResponse): OpSerializerFor[wire.GetRootChainStakesResponse, wire.GetRootChainStakesRequest](), + byte(wire.ClusterOpGetTotalBalanceRequest): OpSerializerFor[wire.GetTotalBalanceRequest, wire.GetTotalBalanceResponse](), + byte(wire.ClusterOpGetTotalBalanceResponse): OpSerializerFor[wire.GetTotalBalanceResponse, wire.GetTotalBalanceRequest](), + }) +} + +// registerHandlers registers all master→slave RPC handlers and marks the +// fire-and-forget opcodes as non-RPC. +func (mc *MasterConn) registerHandlers() { + mc.rpcConn.RegisterTypedHandlers(map[byte]TypedHandler{ + byte(wire.ClusterOpPing): mc.handlePing, + byte(wire.ClusterOpConnectToSlavesRequest): mc.handleConnectToSlaves, + byte(wire.ClusterOpMineRequest): mc.handleMine, + byte(wire.ClusterOpGenTxRequest): mc.handleGenTx, + byte(wire.ClusterOpAddRootBlockRequest): mc.handleAddRootBlock, + byte(wire.ClusterOpGetEcoInfoListRequest): mc.handleGetEcoInfoList, + byte(wire.ClusterOpGetNextBlockToMineRequest): mc.handleGetNextBlockToMine, + byte(wire.ClusterOpAddMinorBlockRequest): mc.handleAddMinorBlock, + byte(wire.ClusterOpGetUnconfirmedHeadersRequest): mc.handleGetUnconfirmedHeaders, + byte(wire.ClusterOpGetAccountDataRequest): mc.handleGetAccountData, + byte(wire.ClusterOpAddTransactionRequest): mc.handleAddTransaction, + byte(wire.ClusterOpCreateClusterPeerConnectionRequest): mc.handleCreateClusterPeerConnection, + byte(wire.ClusterOpDestroyClusterPeerConnectionCommand): mc.handleDestroyClusterPeerConnection, + byte(wire.ClusterOpGetMinorBlockRequest): mc.handleGetMinorBlock, + byte(wire.ClusterOpGetTransactionRequest): mc.handleGetTransaction, + byte(wire.ClusterOpSyncMinorBlockListRequest): mc.handleSyncMinorBlockList, + byte(wire.ClusterOpExecuteTransactionRequest): mc.handleExecuteTransaction, + byte(wire.ClusterOpGetTransactionReceiptRequest): mc.handleGetTransactionReceipt, + byte(wire.ClusterOpGetTransactionListByAddressRequest): mc.handleGetTransactionListByAddress, + byte(wire.ClusterOpGetLogRequest): mc.handleGetLogs, + byte(wire.ClusterOpEstimateGasRequest): mc.handleEstimateGas, + byte(wire.ClusterOpGetStorageRequest): mc.handleGetStorageAt, + byte(wire.ClusterOpGetCodeRequest): mc.handleGetCode, + byte(wire.ClusterOpGasPriceRequest): mc.handleGasPrice, + byte(wire.ClusterOpGetWorkRequest): mc.handleGetWork, + byte(wire.ClusterOpSubmitWorkRequest): mc.handleSubmitWork, + byte(wire.ClusterOpCheckMinorBlockRequest): mc.handleCheckMinorBlock, + byte(wire.ClusterOpGetAllTransactionsRequest): mc.handleGetAllTransactions, + byte(wire.ClusterOpGetRootChainStakesRequest): mc.handleGetRootChainStakes, + byte(wire.ClusterOpGetTotalBalanceRequest): mc.handleGetTotalBalance, + }) + + mc.rpcConn.RegisterNonRPCOps([]byte{ + byte(wire.ClusterOpDestroyClusterPeerConnectionCommand), + }) +} + +// rawBytes is a helper that returns a non-nil *wire.RawBytes pointer. +func rawBytes(b []byte) *wire.RawBytes { + rb := wire.RawBytes(b) + return &rb +} + +// emptyRawBytes returns a non-nil *wire.RawBytes pointing to an empty slice. +func emptyRawBytes() *wire.RawBytes { + return rawBytes([]byte{}) +} + +// LocalID returns this slave's ID used in PONG responses. +func (mc *MasterConn) LocalID() []byte { + return append([]byte(nil), mc.localID...) +} + +// LocalFullShardIDList returns this slave's full shard ID list used in PONG responses. +func (mc *MasterConn) LocalFullShardIDList() []uint32 { + return append([]uint32(nil), mc.localFullShardIDList...) +} + +// handlePing responds to the master's PING with this slave's identity. +// Python: MasterConnection.handle_ping -> Pong(self.slave_server.id, ...). +func (mc *MasterConn) handlePing(req any) (any, error) { + // TODO: when core.RootBlock is ported, use ping.root_tip to drive shard creation. + _ = req.(*wire.PingRequest) + + return &wire.PongResponse{ + ID: append([]byte(nil), mc.localID...), + FullShardIDList: append([]uint32(nil), mc.localFullShardIDList...), + }, nil +} + +// handleConnectToSlaves accepts a list of slaves to connect to. +// Python: returns ConnectToSlavesResponse with one empty bytes result per slave. +func (mc *MasterConn) handleConnectToSlaves(req any) (any, error) { + r := req.(*wire.ConnectToSlavesRequest) + + // TODO: delegate to SlaveServer.slave_connection_manager.connect_to_slave. + resultList := make([]wire.PrependedSizeBytes4, len(r.SlaveInfoList)) + for i := range resultList { + resultList[i] = wire.PrependedSizeBytes4{} + } + return &wire.ConnectToSlavesResponse{ResultList: resultList}, nil +} + +// handleMine starts or stops mining. +// Python: MineResponse(error_code=0). +func (mc *MasterConn) handleMine(req any) (any, error) { + _ = req.(*wire.MineRequest) + // TODO: delegate to SlaveServer.start_mining / stop_mining. + return &wire.MineResponse{ErrorCode: 0}, nil +} + +// handleGenTx generates transactions. +// Python: GenTxResponse(error_code=0). +func (mc *MasterConn) handleGenTx(req any) (any, error) { + _ = req.(*wire.GenTxRequest) + // TODO: delegate to SlaveServer.create_transactions. + return &wire.GenTxResponse{ErrorCode: 0}, nil +} + +// handleAddRootBlock processes a root block from the master. +// Python: returns AddRootBlockResponse(error_code=0, switched=False) on success. +func (mc *MasterConn) handleAddRootBlock(req any) (any, error) { + _ = req.(*wire.AddRootBlockRequest) + // TODO: delegate to shard.add_root_block and SlaveServer.create_shards. + return &wire.AddRootBlockResponse{ErrorCode: 0, Switched: false}, nil +} + +// handleGetEcoInfoList returns economic info for all initialized shards. +// Python: returns empty list when no shards are initialized. +func (mc *MasterConn) handleGetEcoInfoList(req any) (any, error) { + _ = req.(*wire.GetEcoInfoListRequest) + // TODO: collect real EcoInfo from shard states. + return &wire.GetEcoInfoListResponse{ErrorCode: 0, EcoInfoList: []wire.EcoInfo{}}, nil +} + +// handleGetNextBlockToMine returns a block template for the requested branch. +// Python requires the shard to exist; without shard runtime we return not-found. +func (mc *MasterConn) handleGetNextBlockToMine(req any) (any, error) { + _ = req.(*wire.GetNextBlockToMineRequest) + // TODO: delegate to shard.state.create_block_to_mine. + return &wire.GetNextBlockToMineResponse{ErrorCode: 1, Block: emptyRawBytes()}, nil +} + +// handleAddMinorBlock adds a JRPC-mined minor block. +// Python: returns AddMinorBlockResponse(error_code=0) on success. +func (mc *MasterConn) handleAddMinorBlock(req any) (any, error) { + _ = req.(*wire.AddMinorBlockRequest) + // TODO: deserialize MinorBlock and delegate to shard.add_block. + return &wire.AddMinorBlockResponse{ErrorCode: 0}, nil +} + +// handleGetUnconfirmedHeaders returns unconfirmed headers per shard. +// Python: returns empty list when no shards are initialized. +func (mc *MasterConn) handleGetUnconfirmedHeaders(req any) (any, error) { + _ = req.(*wire.GetUnconfirmedHeadersRequest) + // TODO: collect real HeadersInfo from shard states. + return &wire.GetUnconfirmedHeadersResponse{ErrorCode: 0, HeadersInfoList: []wire.HeadersInfo{}}, nil +} + +// handleGetAccountData returns account data across shards. +// Python: returns empty list when there are no shards for the address. +func (mc *MasterConn) handleGetAccountData(req any) (any, error) { + _ = req.(*wire.GetAccountDataRequest) + // TODO: delegate to SlaveServer.get_account_data. + return &wire.GetAccountDataResponse{ErrorCode: 0, AccountBranchDataList: []wire.AccountBranchData{}}, nil +} + +// handleAddTransaction adds a transaction to the tx pool. +// Python: returns AddTransactionResponse(error_code=0) on success. +func (mc *MasterConn) handleAddTransaction(req any) (any, error) { + _ = req.(*wire.AddTransactionRequest) + // TODO: delegate to SlaveServer.add_tx. + return &wire.AddTransactionResponse{ErrorCode: 0}, nil +} + +// handleCreateClusterPeerConnection creates virtual peer connections for all shards. +// Python: returns CreateClusterPeerConnectionResponse(error_code=0) on success. +func (mc *MasterConn) handleCreateClusterPeerConnection(req any) (any, error) { + _ = req.(*wire.CreateClusterPeerConnectionRequest) + // TODO: create PeerShardConnection instances and wire with the dispatcher (PR6). + return &wire.CreateClusterPeerConnectionResponse{ErrorCode: 0}, nil +} + +// handleDestroyClusterPeerConnection is a fire-and-forget command to tear down +// a virtual peer connection. No response is sent. +func (mc *MasterConn) handleDestroyClusterPeerConnection(req any) (any, error) { + _ = req.(*wire.DestroyClusterPeerConnectionCommand) + // TODO: notify dispatcher / close peer shard connections (PR6). + return nil, nil +} + +// handleGetMinorBlock fetches a minor block by hash or height. +// Python returns error_code=1 with an empty block when not found. +func (mc *MasterConn) handleGetMinorBlock(req any) (any, error) { + _ = req.(*wire.GetMinorBlockRequest) + // TODO: delegate to SlaveServer.get_minor_block_by_hash / by_height. + return &wire.GetMinorBlockResponse{ + ErrorCode: 1, + MinorBlock: emptyRawBytes(), + ExtraInfo: nil, + }, nil +} + +// handleGetTransaction fetches a transaction by hash. +// Python returns error_code=1 with an empty block when not found. +func (mc *MasterConn) handleGetTransaction(req any) (any, error) { + _ = req.(*wire.GetTransactionRequest) + // TODO: delegate to SlaveServer.get_transaction_by_hash. + return &wire.GetTransactionResponse{ + ErrorCode: 1, + MinorBlock: emptyRawBytes(), + Index: 0, + }, nil +} + +// handleSyncMinorBlockList downloads and applies a list of minor blocks. +// Python returns error_code=0 with empty data when the input list is empty. +func (mc *MasterConn) handleSyncMinorBlockList(req any) (any, error) { + r := req.(*wire.SyncMinorBlockListRequest) + _ = r + // TODO: delegate to SlaveServer.add_block_list_for_sync. + return &wire.SyncMinorBlockListResponse{ + ErrorCode: 0, + BlockCoinbaseMap: emptyRawBytes(), + ShardStats: nil, + }, nil +} + +// handleExecuteTransaction executes a transaction and returns the result. +// Python returns error_code=1 when execution fails (e.g. shard missing). +func (mc *MasterConn) handleExecuteTransaction(req any) (any, error) { + _ = req.(*wire.ExecuteTransactionRequest) + // TODO: delegate to SlaveServer.execute_tx. + return &wire.ExecuteTransactionResponse{ErrorCode: 1, Result: []byte{}}, nil +} + +// handleGetTransactionReceipt fetches a transaction receipt. +// Python returns error_code=1 with empty block/receipt when not found. +func (mc *MasterConn) handleGetTransactionReceipt(req any) (any, error) { + _ = req.(*wire.GetTransactionReceiptRequest) + // TODO: delegate to SlaveServer.get_transaction_receipt. + return &wire.GetTransactionReceiptResponse{ + ErrorCode: 1, + MinorBlock: emptyRawBytes(), + Index: 0, + Receipt: emptyRawBytes(), + }, nil +} + +// handleGetTransactionListByAddress returns transactions for an address. +// Python returns error_code=1 with empty lists when the shard is missing. +func (mc *MasterConn) handleGetTransactionListByAddress(req any) (any, error) { + _ = req.(*wire.GetTransactionListByAddressRequest) + // TODO: delegate to SlaveServer.get_transaction_list_by_address. + return &wire.GetTransactionListByAddressResponse{ + ErrorCode: 1, + TxList: []wire.TransactionDetail{}, + Next: []byte{}, + }, nil +} + +// handleGetLogs returns logs matching the filter. +// Python returns error_code=1 with empty logs when the shard is missing. +func (mc *MasterConn) handleGetLogs(req any) (any, error) { + _ = req.(*wire.GetLogRequest) + // TODO: delegate to SlaveServer.get_logs. + return &wire.GetLogResponse{ErrorCode: 1, Logs: []*wire.RawBytes{}}, nil +} + +// handleEstimateGas estimates gas for a transaction. +// Python returns error_code=1 when estimation fails (e.g. shard missing). +func (mc *MasterConn) handleEstimateGas(req any) (any, error) { + _ = req.(*wire.EstimateGasRequest) + // TODO: delegate to SlaveServer.estimate_gas. + return &wire.EstimateGasResponse{ErrorCode: 1, Result: 0}, nil +} + +// handleGetStorageAt reads storage at the given address/key. +// Python returns error_code=1 with a zero result when the shard is missing. +func (mc *MasterConn) handleGetStorageAt(req any) (any, error) { + _ = req.(*wire.GetStorageRequest) + // TODO: delegate to SlaveServer.get_storage_at. + return &wire.GetStorageResponse{ErrorCode: 1, Result: [wire.HashLength]byte{}}, nil +} + +// handleGetCode reads code at the given address. +// Python returns error_code=1 with empty bytes when the shard is missing. +func (mc *MasterConn) handleGetCode(req any) (any, error) { + _ = req.(*wire.GetCodeRequest) + // TODO: delegate to SlaveServer.get_code. + return &wire.GetCodeResponse{ErrorCode: 1, Result: []byte{}}, nil +} + +// handleGasPrice returns the gas price for a token on a branch. +// Python returns error_code=1 with result 0 when the shard is missing. +func (mc *MasterConn) handleGasPrice(req any) (any, error) { + _ = req.(*wire.GasPriceRequest) + // TODO: delegate to SlaveServer.gas_price. + return &wire.GasPriceResponse{ErrorCode: 1, Result: 0}, nil +} + +// handleGetWork returns mining work. +// Python returns error_code=1 when work cannot be produced. +func (mc *MasterConn) handleGetWork(req any) (any, error) { + _ = req.(*wire.GetWorkRequest) + // TODO: delegate to SlaveServer.get_work. + return &wire.GetWorkResponse{ErrorCode: 1}, nil +} + +// handleSubmitWork submits mining work. +// Python returns error_code=1, success=False when submission fails. +func (mc *MasterConn) handleSubmitWork(req any) (any, error) { + _ = req.(*wire.SubmitWorkRequest) + // TODO: delegate to SlaveServer.submit_work. + return &wire.SubmitWorkResponse{ErrorCode: 1, Success: false}, nil +} + +// handleCheckMinorBlock validates a minor block header. +// Python returns CheckMinorBlockResponse(error_code=0) when the block is valid, +// and error_code=errno.EBADMSG when the shard is missing or validation fails. +// This stub returns ErrorCode=1 to signal "not implemented / cannot validate". +func (mc *MasterConn) handleCheckMinorBlock(req any) (any, error) { + _ = req.(*wire.CheckMinorBlockRequest) + // TODO: delegate to shard.check_minor_block_by_header. + return &wire.CheckMinorBlockResponse{ErrorCode: 1}, nil +} + +// handleGetAllTransactions returns all transactions in the mempool. +// Python returns error_code=1 with empty lists when the shard is missing. +func (mc *MasterConn) handleGetAllTransactions(req any) (any, error) { + _ = req.(*wire.GetAllTransactionsRequest) + // TODO: delegate to SlaveServer.get_all_transactions. + return &wire.GetAllTransactionsResponse{ + ErrorCode: 1, + TxList: []wire.TransactionDetail{}, + Next: []byte{}, + }, nil +} + +// handleGetRootChainStakes reads root-chain stake info. +// Python returns GetRootChainStakesResponse(0, stakes, signer). +func (mc *MasterConn) handleGetRootChainStakes(req any) (any, error) { + _ = req.(*wire.GetRootChainStakesRequest) + // TODO: delegate to SlaveServer.get_root_chain_stakes. + return &wire.GetRootChainStakesResponse{ + ErrorCode: 0, + Stakes: serialize.BigUint{}, + Signer: [20]byte{}, + }, nil +} + +// handleGetTotalBalance returns the total token balance across accounts. +// Python catches exceptions and returns GetTotalBalanceResponse(1, 0, b""). +func (mc *MasterConn) handleGetTotalBalance(req any) (any, error) { + _ = req.(*wire.GetTotalBalanceRequest) + // TODO: delegate to SlaveServer.get_total_balance. + return &wire.GetTotalBalanceResponse{ + ErrorCode: 1, + TotalBalance: serialize.BigUint{}, + Next: []byte{}, + }, nil +} + +// SetForwarder installs a raw-frame forwarder hook for peer traffic +// (cluster_peer_id != 0). This is used by the dispatcher in PR6. +func (mc *MasterConn) SetForwarder(f func(*wire.Frame) bool) { + mc.rpcConn.SetForwarder(f) +} + +// SendRPCMeta sends a request with ClusterMetadata and waits for the response. +// It is the primitive used by all typed outbound methods. +func (mc *MasterConn) SendRPCMeta(ctx context.Context, opcode byte, payload []byte, meta wire.ClusterMetadata) (*wire.Frame, error) { + return mc.rpcConn.SendRPCMeta(ctx, opcode, payload, meta) +} + +// SendAddMinorBlockHeader sends AddMinorBlockHeaderRequest to the master and +// returns the parsed response. +func (mc *MasterConn) SendAddMinorBlockHeader(ctx context.Context, req *wire.AddMinorBlockHeaderRequest) (*wire.AddMinorBlockHeaderResponse, error) { + payload, err := serializeBytes(req) + if err != nil { + return nil, fmt.Errorf("serialize AddMinorBlockHeaderRequest: %w", err) + } + frame, err := mc.SendRPCMeta(ctx, byte(wire.ClusterOpAddMinorBlockHeaderRequest), payload, wire.ClusterMetadata{}) + if err != nil { + return nil, err + } + var resp wire.AddMinorBlockHeaderResponse + if err := deserializeBytes(frame.Payload, &resp); err != nil { + return nil, fmt.Errorf("deserialize AddMinorBlockHeaderResponse: %w", err) + } + return &resp, nil +} + +// SendAddMinorBlockHeaderList sends AddMinorBlockHeaderListRequest to the master +// and returns the parsed response. +func (mc *MasterConn) SendAddMinorBlockHeaderList(ctx context.Context, req *wire.AddMinorBlockHeaderListRequest) (*wire.AddMinorBlockHeaderListResponse, error) { + payload, err := serializeBytes(req) + if err != nil { + return nil, fmt.Errorf("serialize AddMinorBlockHeaderListRequest: %w", err) + } + frame, err := mc.SendRPCMeta(ctx, byte(wire.ClusterOpAddMinorBlockHeaderListRequest), payload, wire.ClusterMetadata{}) + if err != nil { + return nil, err + } + var resp wire.AddMinorBlockHeaderListResponse + if err := deserializeBytes(frame.Payload, &resp); err != nil { + return nil, fmt.Errorf("deserialize AddMinorBlockHeaderListResponse: %w", err) + } + return &resp, nil +} diff --git a/qkc/cluster/slave/master_conn_test.go b/qkc/cluster/slave/master_conn_test.go new file mode 100644 index 0000000000..4af852017a --- /dev/null +++ b/qkc/cluster/slave/master_conn_test.go @@ -0,0 +1,854 @@ +// Copyright 2026-2027, QuarkChain. + +package slave + +import ( + "bytes" + "context" + "encoding/binary" + "net" + "reflect" + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/qkc/account" + "github.com/ethereum/go-ethereum/qkc/cluster/wire" + "github.com/ethereum/go-ethereum/qkc/serialize" +) + +// newMasterTestConnPair creates a pair of MasterConns connected over a local TCP +// socket. The caller is responsible for calling cleanup. +func newMasterTestConnPair(t *testing.T) (client, server *MasterConn, cleanup func()) { + t.Helper() + return newMasterTestConnPairWithIdentity( + t, + []byte("go-slave-client"), []uint32{0x00010001}, + []byte("go-slave-server"), []uint32{0x00010001, 0x00020001}, + ) +} + +func newMasterTestConnPairWithIdentity( + t *testing.T, + clientID []byte, clientShards []uint32, + serverID []byte, serverShards []uint32, +) (client, server *MasterConn, cleanup func()) { + t.Helper() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + + var serverConn net.Conn + var acceptErr error + accepted := make(chan struct{}) + go func() { + defer close(accepted) + serverConn, acceptErr = ln.Accept() + ln.Close() + }() + + clientConn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + <-accepted + if acceptErr != nil { + t.Fatalf("accept: %v", acceptErr) + } + + logger := log.New() + client = NewMasterConnFromConn(clientConn, 0, clientID, clientShards, logger) + server = NewMasterConnFromConn(serverConn, 0, serverID, serverShards, logger) + cleanup = func() { + client.Close() + server.Close() + } + return +} + +// writeRawMasterFrame writes a raw ClusterMetadata frame directly to the +// underlying TCP connection, bypassing the connection's frame writer. +func writeRawMasterFrame(t *testing.T, conn net.Conn, frame *wire.Frame) { + t.Helper() + if err := wire.WriteFrame(conn, frame); err != nil { + t.Fatalf("write raw frame: %v", err) + } +} + +// hasHandler reports whether the connection has a typed handler for opcode. +func hasHandler(c *MasterConn, opcode byte) bool { + rv := reflect.ValueOf(c.rpcConn).Elem() + handlers := rv.FieldByName("typedHandlers").MapKeys() + for _, k := range handlers { + if k.Uint() == uint64(opcode) { + return true + } + } + return false +} + +// hasSerializer reports whether the connection has an OpSerializer for opcode. +func hasSerializer(c *MasterConn, opcode byte) bool { + rv := reflect.ValueOf(c.rpcConn).Elem() + serializers := rv.FieldByName("serializers").MapKeys() + for _, k := range serializers { + if k.Uint() == uint64(opcode) { + return true + } + } + return false +} + +// TestMasterConn_AllMasterHandlersRegistered verifies that every master→slave +// request opcode has a handler registered and that the fire-and-forget opcode +// is marked as non-RPC. +func TestMasterConn_AllMasterHandlersRegistered(t *testing.T) { + _, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + masterRPCOps := []wire.ClusterOp{ + wire.ClusterOpPing, + wire.ClusterOpConnectToSlavesRequest, + wire.ClusterOpMineRequest, + wire.ClusterOpGenTxRequest, + wire.ClusterOpAddRootBlockRequest, + wire.ClusterOpGetEcoInfoListRequest, + wire.ClusterOpGetNextBlockToMineRequest, + wire.ClusterOpAddMinorBlockRequest, + wire.ClusterOpGetUnconfirmedHeadersRequest, + wire.ClusterOpGetAccountDataRequest, + wire.ClusterOpAddTransactionRequest, + wire.ClusterOpCreateClusterPeerConnectionRequest, + wire.ClusterOpGetMinorBlockRequest, + wire.ClusterOpGetTransactionRequest, + wire.ClusterOpSyncMinorBlockListRequest, + wire.ClusterOpExecuteTransactionRequest, + wire.ClusterOpGetTransactionReceiptRequest, + wire.ClusterOpGetTransactionListByAddressRequest, + wire.ClusterOpGetLogRequest, + wire.ClusterOpEstimateGasRequest, + wire.ClusterOpGetStorageRequest, + wire.ClusterOpGetCodeRequest, + wire.ClusterOpGasPriceRequest, + wire.ClusterOpGetWorkRequest, + wire.ClusterOpSubmitWorkRequest, + wire.ClusterOpCheckMinorBlockRequest, + wire.ClusterOpGetAllTransactionsRequest, + wire.ClusterOpGetRootChainStakesRequest, + wire.ClusterOpGetTotalBalanceRequest, + } + + for _, op := range masterRPCOps { + if !hasHandler(server, byte(op)) { + t.Fatalf("missing handler for opcode 0x%02x (%v)", op, op) + } + } + + if !isNonRPC(server, byte(wire.ClusterOpDestroyClusterPeerConnectionCommand)) { + t.Fatalf("DESTROY_CLUSTER_PEER_CONNECTION_COMMAND is not marked as non-RPC") + } +} + +// isNonRPC reports whether opcode is registered as fire-and-forget. +func isNonRPC(c *MasterConn, opcode byte) bool { + rv := reflect.ValueOf(c.rpcConn).Elem() + nonRPCOps := rv.FieldByName("nonRPCOps").MapKeys() + for _, k := range nonRPCOps { + if k.Uint() == uint64(opcode) { + return true + } + } + return false +} + +// TestMasterConn_AllSerializersRegistered verifies that every ClusterOp defined +// in wire/opcode.go has a registered OpSerializer. +func TestMasterConn_AllSerializersRegistered(t *testing.T) { + _, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + for op := wire.ClusterOpPing; op <= wire.ClusterOpGetTotalBalanceResponse; op++ { + if op == 0x9C { // 28 is intentionally skipped in Python + continue + } + if !hasSerializer(server, byte(op)) { + t.Fatalf("missing serializer for opcode 0x%02x (%v)", op, op) + } + } +} + +// TestMasterConn_Ping verifies the master→slave PING handshake. +func TestMasterConn_Ping(t *testing.T) { + clientID := []byte("go-slave-client") + clientShards := []uint32{0x00010001} + serverID := []byte("go-slave-server") + serverShards := []uint32{0x00010001, 0x00020001} + + client, server, cleanup := newMasterTestConnPairWithIdentity(t, clientID, clientShards, serverID, serverShards) + defer cleanup() + + server.Start() + client.Start() + + pingPayload, err := serialize.SerializeToBytes(&wire.PingRequest{ + ID: []byte("master"), + FullShardIDList: []uint32{0x00010001}, + RootTip: nil, + }) + if err != nil { + t.Fatalf("serialize ping: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + resp, err := client.SendRPCMeta(ctx, byte(wire.ClusterOpPing), pingPayload, wire.ClusterMetadata{}) + if err != nil { + t.Fatalf("send ping: %v", err) + } + if resp.Opcode != byte(wire.ClusterOpPong) { + t.Fatalf("expected opcode 0x%x, got 0x%x", wire.ClusterOpPong, resp.Opcode) + } + + var pong wire.PongResponse + if err := serialize.Deserialize(serialize.NewByteBuffer(resp.Payload), &pong); err != nil { + t.Fatalf("deserialize pong: %v", err) + } + if string(pong.ID) != string(serverID) { + t.Fatalf("pong id mismatch: got %s, expected %s", pong.ID, serverID) + } + if len(pong.FullShardIDList) != len(serverShards) { + t.Fatalf("pong shard list mismatch: got %v", pong.FullShardIDList) + } +} + +// TestMasterConn_RPCRoundTrip verifies request/response dispatch for a +// representative set of master→slave RPCs. +func TestMasterConn_RPCRoundTrip(t *testing.T) { + client, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + server.Start() + client.Start() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + cases := []struct { + name string + opcode wire.ClusterOp + req any + resp any + respOpcode wire.ClusterOp + }{ + { + name: "add_root_block", + opcode: wire.ClusterOpAddRootBlockRequest, + req: &wire.AddRootBlockRequest{RootBlock: emptyRawBytes(), ExpectSwitch: false}, + resp: &wire.AddRootBlockResponse{}, + respOpcode: wire.ClusterOpAddRootBlockResponse, + }, + { + name: "get_eco_info_list", + opcode: wire.ClusterOpGetEcoInfoListRequest, + req: &wire.GetEcoInfoListRequest{}, + resp: &wire.GetEcoInfoListResponse{}, + respOpcode: wire.ClusterOpGetEcoInfoListResponse, + }, + { + name: "add_transaction", + opcode: wire.ClusterOpAddTransactionRequest, + req: &wire.AddTransactionRequest{Tx: emptyRawBytes()}, + resp: &wire.AddTransactionResponse{}, + respOpcode: wire.ClusterOpAddTransactionResponse, + }, + { + name: "get_minor_block", + opcode: wire.ClusterOpGetMinorBlockRequest, + req: &wire.GetMinorBlockRequest{Branch: 0x00010001, Height: 1, NeedExtraInfo: false}, + resp: &wire.GetMinorBlockResponse{}, + respOpcode: wire.ClusterOpGetMinorBlockResponse, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + payload, err := serialize.SerializeToBytes(tc.req) + if err != nil { + t.Fatalf("serialize request: %v", err) + } + + frame, err := client.SendRPCMeta(ctx, byte(tc.opcode), payload, wire.ClusterMetadata{Branch: 0x00010001}) + if err != nil { + t.Fatalf("send rpc: %v", err) + } + if frame.Opcode != byte(tc.respOpcode) { + t.Fatalf("expected response opcode 0x%x, got 0x%x", tc.respOpcode, frame.Opcode) + } + if frame.Meta.Branch != 0x00010001 { + t.Fatalf("metadata branch not preserved: got %d", frame.Meta.Branch) + } + + if err := serialize.Deserialize(serialize.NewByteBuffer(frame.Payload), tc.resp); err != nil { + t.Fatalf("deserialize response: %v", err) + } + }) + } +} + +// TestMasterConn_NonRPCDispatch verifies that the fire-and-forget +// DESTROY_CLUSTER_PEER_CONNECTION_COMMAND is accepted with rpc_id == 0 and does +// not produce a response or close the connection. +func TestMasterConn_NonRPCDispatch(t *testing.T) { + client, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + server.Start() + client.Start() + + payload, err := serialize.SerializeToBytes(&wire.DestroyClusterPeerConnectionCommand{ClusterPeerID: 42}) + if err != nil { + t.Fatalf("serialize command: %v", err) + } + + // Write a non-RPC frame directly; no response should come back, but the + // connection must remain usable for a subsequent RPC. + writeRawMasterFrame(t, client.conn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001}, + Opcode: byte(wire.ClusterOpDestroyClusterPeerConnectionCommand), + RPCID: 0, + Payload: payload, + }) + + // Give the server a moment to process the command. + time.Sleep(50 * time.Millisecond) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{ + ID: []byte("master"), + FullShardIDList: []uint32{0x00010001}, + }) + resp, err := client.SendRPCMeta(ctx, byte(wire.ClusterOpPing), pingPayload, wire.ClusterMetadata{}) + if err != nil { + t.Fatalf("ping after non-rpc command failed: %v", err) + } + if resp.Opcode != byte(wire.ClusterOpPong) { + t.Fatalf("expected pong, got opcode 0x%x", resp.Opcode) + } +} + +// TestMasterConn_NonRPCWithNonZeroRPCID verifies that a non-RPC command with a +// non-zero rpc_id causes the server to close the connection. +func TestMasterConn_NonRPCWithNonZeroRPCID(t *testing.T) { + client, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + server.Start() + client.Start() + + payload, _ := serialize.SerializeToBytes(&wire.DestroyClusterPeerConnectionCommand{ClusterPeerID: 42}) + + writeRawMasterFrame(t, client.conn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001}, + Opcode: byte(wire.ClusterOpDestroyClusterPeerConnectionCommand), + RPCID: 1, // non-RPC must have rpc_id == 0 + Payload: payload, + }) + + select { + case <-server.WaitUntilClosed(): + case <-time.After(2 * time.Second): + t.Fatal("server did not close after non-rpc with non-zero rpc_id") + } +} + +// TestMasterConn_Forwarder verifies that frames with cluster_peer_id != 0 are +// routed through the forwarder hook and are not dispatched locally. +func TestMasterConn_Forwarder(t *testing.T) { + client, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + var forwardedMu sync.Mutex + var forwarded []*wire.Frame + server.SetForwarder(func(frame *wire.Frame) bool { + if frame.Meta.ClusterPeerID == 0 { + return false + } + forwardedMu.Lock() + forwarded = append(forwarded, frame) + forwardedMu.Unlock() + return true + }) + + server.Start() + client.Start() + + payload, _ := serialize.SerializeToBytes(&wire.GetMinorBlockRequest{Branch: 0x00010001, Height: 1}) + + // Peer-originated frame: cluster_peer_id != 0. + writeRawMasterFrame(t, client.conn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001, ClusterPeerID: 123}, + Opcode: byte(wire.ClusterOpGetMinorBlockRequest), + RPCID: 7, + Payload: payload, + }) + + time.Sleep(100 * time.Millisecond) + + forwardedMu.Lock() + count := len(forwarded) + forwardedMu.Unlock() + if count != 1 { + t.Fatalf("expected 1 forwarded frame, got %d", count) + } + + // Connection should still be open; a subsequent master RPC works. + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{ID: []byte("m"), FullShardIDList: []uint32{1}}) + resp, err := client.SendRPCMeta(ctx, byte(wire.ClusterOpPing), pingPayload, wire.ClusterMetadata{}) + if err != nil { + t.Fatalf("ping after forwarded frame failed: %v", err) + } + if resp.Opcode != byte(wire.ClusterOpPong) { + t.Fatalf("expected pong, got 0x%x", resp.Opcode) + } +} + +// TestMasterConn_UnsupportedOpcodeClosesConnection verifies that an opcode +// without a registered handler and rpc_id == 0 causes the server to close the +// connection. +func TestMasterConn_UnsupportedOpcodeClosesConnection(t *testing.T) { + client, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + server.Start() + client.Start() + + // 0x01 is a CommandOp with no handler registered on the master side. + // rpc_id must be 0 for the server to treat it as a non-RPC unsupported + // command and close the connection. + writeRawMasterFrame(t, client.conn, &wire.Frame{ + Meta: wire.ClusterMetadata{}, + Opcode: 0x01, + RPCID: 0, + Payload: []byte("payload"), + }) + + select { + case <-server.WaitUntilClosed(): + case <-time.After(2 * time.Second): + t.Fatal("server did not close after unsupported opcode") + } +} + +// TestMasterConn_RPCIDMonotonic verifies that duplicate RPC IDs cause the +// server to close the connection. +func TestMasterConn_RPCIDMonotonic(t *testing.T) { + client, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + server.Start() + client.Start() + + pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{ + ID: []byte("master"), + FullShardIDList: []uint32{0x00010001}, + }) + + // Manually send two PING frames with the same RPC ID. + writeRawMasterFrame(t, client.conn, &wire.Frame{ + Meta: wire.ClusterMetadata{}, + Opcode: byte(wire.ClusterOpPing), + RPCID: 1, + Payload: pingPayload, + }) + writeRawMasterFrame(t, client.conn, &wire.Frame{ + Meta: wire.ClusterMetadata{}, + Opcode: byte(wire.ClusterOpPing), + RPCID: 1, // duplicate + Payload: pingPayload, + }) + + select { + case <-server.WaitUntilClosed(): + case <-time.After(2 * time.Second): + t.Fatal("server did not close after duplicate rpc_id") + } +} + +// TestMasterConn_RPCIDDecreasing verifies that a decreasing RPC ID causes the +// server to close the connection. +func TestMasterConn_RPCIDDecreasing(t *testing.T) { + client, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + server.Start() + client.Start() + + pingPayload, _ := serialize.SerializeToBytes(&wire.PingRequest{ + ID: []byte("master"), + FullShardIDList: []uint32{0x00010001}, + }) + + writeRawMasterFrame(t, client.conn, &wire.Frame{ + Meta: wire.ClusterMetadata{}, + Opcode: byte(wire.ClusterOpPing), + RPCID: 2, + Payload: pingPayload, + }) + writeRawMasterFrame(t, client.conn, &wire.Frame{ + Meta: wire.ClusterMetadata{}, + Opcode: byte(wire.ClusterOpPing), + RPCID: 1, // decreasing + Payload: pingPayload, + }) + + select { + case <-server.WaitUntilClosed(): + case <-time.After(2 * time.Second): + t.Fatal("server did not close after decreasing rpc_id") + } +} + +// TestMasterConn_CloseWakesPendingRPC verifies that Close wakes all pending +// outbound RPCs with ErrConnectionClosed. +func TestMasterConn_CloseWakesPendingRPC(t *testing.T) { + client, _, cleanup := newMasterTestConnPair(t) + defer cleanup() + + // Server intentionally left unstarted so it never replies. + client.Start() + + var wg sync.WaitGroup + wg.Add(1) + errChan := make(chan error, 1) + go func() { + wg.Done() + _, err := client.SendRPCMeta(context.Background(), byte(wire.ClusterOpPing), []byte("ping"), wire.ClusterMetadata{}) + errChan <- err + }() + + wg.Wait() + client.Close() + + select { + case err := <-errChan: + if err != ErrConnectionClosed { + t.Fatalf("expected ErrConnectionClosed, got %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("pending RPC was not woken by Close") + } +} + +// TestMasterConn_OutboundRPCMeta verifies that outbound RPCs from the slave +// encode ClusterMetadata correctly on the wire. +func TestMasterConn_OutboundRPCMeta(t *testing.T) { + client, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + // Server echoes the request opcode + 1 and preserves metadata. + server.RegisterTypedHandlers(map[byte]TypedHandler{ + byte(wire.ClusterOpGetEcoInfoListRequest): func(req any) (any, error) { + _ = req.(*wire.GetEcoInfoListRequest) + return &wire.GetEcoInfoListResponse{ErrorCode: 0}, nil + }, + }) + + server.Start() + client.Start() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + meta := wire.ClusterMetadata{Branch: 0x00010001, ClusterPeerID: 99} + payload, _ := serialize.SerializeToBytes(&wire.GetEcoInfoListRequest{}) + resp, err := client.SendRPCMeta(ctx, byte(wire.ClusterOpGetEcoInfoListRequest), payload, meta) + if err != nil { + t.Fatalf("send rpc: %v", err) + } + if resp.Opcode != byte(wire.ClusterOpGetEcoInfoListResponse) { + t.Fatalf("unexpected response opcode 0x%x", resp.Opcode) + } + if resp.Meta.Branch != meta.Branch || resp.Meta.ClusterPeerID != meta.ClusterPeerID { + t.Fatalf("response metadata mismatch: got %+v, want %+v", resp.Meta, meta) + } +} + +// TestMasterConn_SendAddMinorBlockHeader verifies the typed outbound helper. +func TestMasterConn_SendAddMinorBlockHeader(t *testing.T) { + client, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + server.RegisterTypedHandlers(map[byte]TypedHandler{ + byte(wire.ClusterOpAddMinorBlockHeaderRequest): func(req any) (any, error) { + r := req.(*wire.AddMinorBlockHeaderRequest) + if r.TxCount != 5 { + t.Fatalf("unexpected tx_count: %d", r.TxCount) + } + return &wire.AddMinorBlockHeaderResponse{ErrorCode: 0}, nil + }, + }) + + server.Start() + client.Start() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + req := &wire.AddMinorBlockHeaderRequest{ + MinorBlockHeader: emptyRawBytes(), + TxCount: 5, + XShardTxCount: 0, + CoinbaseAmountMap: emptyRawBytes(), + ShardStats: wire.ShardStats{Branch: 0x00010001}, + } + resp, err := client.SendAddMinorBlockHeader(ctx, req) + if err != nil { + t.Fatalf("SendAddMinorBlockHeader: %v", err) + } + if resp.ErrorCode != 0 { + t.Fatalf("unexpected error_code: %d", resp.ErrorCode) + } +} + +// TestMasterConn_12ByteMetadata verifies that ClusterMetadata is encoded as +// 4-byte branch followed by 8-byte cluster_peer_id. +func TestMasterConn_12ByteMetadata(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + + var serverConn net.Conn + var acceptErr error + accepted := make(chan struct{}) + go func() { + defer close(accepted) + serverConn, acceptErr = ln.Accept() + ln.Close() + }() + + clientConn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatalf("dial: %v", err) + } + <-accepted + if acceptErr != nil { + t.Fatalf("accept: %v", acceptErr) + } + defer clientConn.Close() + defer serverConn.Close() + + mc := NewMasterConnFromConn(clientConn, 0, []byte("s"), []uint32{1}, log.New()) + defer mc.Close() + + // Read the raw first frame written by the client to inspect metadata layout. + go func() { + // Accept but do not respond; we only need the wire bytes. + buf := make([]byte, 1024) + _, _ = serverConn.Read(buf) + }() + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + + payload, _ := serialize.SerializeToBytes(&wire.GetEcoInfoListRequest{}) + // This RPC will time out because the fake server does not reply, but the + // request bytes are still written to the wire before the timeout. + _, _ = mc.SendRPCMeta(ctx, byte(wire.ClusterOpGetEcoInfoListRequest), payload, wire.ClusterMetadata{Branch: 0x01020304, ClusterPeerID: 0x1122334455667788}) + + // Read back what the client wrote from serverConn using a fresh connection + // is not straightforward; instead verify the metadata marshal helper. + meta := wire.ClusterMetadata{Branch: 0x01020304, ClusterPeerID: 0x1122334455667788} + b := wire.MarshalClusterMetadata(meta) + if len(b) != 12 { + t.Fatalf("metadata length: got %d, want 12", len(b)) + } + if binary.BigEndian.Uint32(b[0:4]) != meta.Branch { + t.Fatalf("branch mismatch") + } + if binary.BigEndian.Uint64(b[4:12]) != meta.ClusterPeerID { + t.Fatalf("cluster_peer_id mismatch") + } +} + +// TestMasterConn_StubResponsesAreValidBytes verifies that every master handler +// stub returns a response that can be serialized. +func TestMasterConn_StubResponsesAreValidBytes(t *testing.T) { + _, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + server.Start() + + cases := []struct { + opcode wire.ClusterOp + req any + resp any + }{ + {wire.ClusterOpPing, &wire.PingRequest{ID: []byte("m"), FullShardIDList: []uint32{1}}, &wire.PongResponse{}}, + {wire.ClusterOpConnectToSlavesRequest, &wire.ConnectToSlavesRequest{SlaveInfoList: []wire.SlaveInfo{}}, &wire.ConnectToSlavesResponse{}}, + {wire.ClusterOpMineRequest, &wire.MineRequest{}, &wire.MineResponse{}}, + {wire.ClusterOpGenTxRequest, &wire.GenTxRequest{Tx: emptyRawBytes()}, &wire.GenTxResponse{}}, + {wire.ClusterOpAddRootBlockRequest, &wire.AddRootBlockRequest{RootBlock: emptyRawBytes()}, &wire.AddRootBlockResponse{}}, + {wire.ClusterOpGetEcoInfoListRequest, &wire.GetEcoInfoListRequest{}, &wire.GetEcoInfoListResponse{}}, + {wire.ClusterOpGetNextBlockToMineRequest, &wire.GetNextBlockToMineRequest{Address: account.Address{}}, &wire.GetNextBlockToMineResponse{}}, + {wire.ClusterOpAddMinorBlockRequest, &wire.AddMinorBlockRequest{MinorBlockData: []byte{}}, &wire.AddMinorBlockResponse{}}, + {wire.ClusterOpGetUnconfirmedHeadersRequest, &wire.GetUnconfirmedHeadersRequest{}, &wire.GetUnconfirmedHeadersResponse{}}, + {wire.ClusterOpGetAccountDataRequest, &wire.GetAccountDataRequest{}, &wire.GetAccountDataResponse{}}, + {wire.ClusterOpAddTransactionRequest, &wire.AddTransactionRequest{Tx: emptyRawBytes()}, &wire.AddTransactionResponse{}}, + {wire.ClusterOpCreateClusterPeerConnectionRequest, &wire.CreateClusterPeerConnectionRequest{ClusterPeerID: 1}, &wire.CreateClusterPeerConnectionResponse{}}, + {wire.ClusterOpGetMinorBlockRequest, &wire.GetMinorBlockRequest{}, &wire.GetMinorBlockResponse{}}, + {wire.ClusterOpGetTransactionRequest, &wire.GetTransactionRequest{}, &wire.GetTransactionResponse{}}, + {wire.ClusterOpSyncMinorBlockListRequest, &wire.SyncMinorBlockListRequest{MinorBlockHashList: [][wire.HashLength]byte{}}, &wire.SyncMinorBlockListResponse{}}, + {wire.ClusterOpExecuteTransactionRequest, &wire.ExecuteTransactionRequest{Tx: emptyRawBytes()}, &wire.ExecuteTransactionResponse{}}, + {wire.ClusterOpGetTransactionReceiptRequest, &wire.GetTransactionReceiptRequest{}, &wire.GetTransactionReceiptResponse{}}, + {wire.ClusterOpGetTransactionListByAddressRequest, &wire.GetTransactionListByAddressRequest{}, &wire.GetTransactionListByAddressResponse{}}, + {wire.ClusterOpGetLogRequest, &wire.GetLogRequest{}, &wire.GetLogResponse{}}, + {wire.ClusterOpEstimateGasRequest, &wire.EstimateGasRequest{Tx: emptyRawBytes()}, &wire.EstimateGasResponse{}}, + {wire.ClusterOpGetStorageRequest, &wire.GetStorageRequest{}, &wire.GetStorageResponse{}}, + {wire.ClusterOpGetCodeRequest, &wire.GetCodeRequest{}, &wire.GetCodeResponse{}}, + {wire.ClusterOpGasPriceRequest, &wire.GasPriceRequest{}, &wire.GasPriceResponse{}}, + {wire.ClusterOpGetWorkRequest, &wire.GetWorkRequest{}, &wire.GetWorkResponse{}}, + {wire.ClusterOpSubmitWorkRequest, &wire.SubmitWorkRequest{}, &wire.SubmitWorkResponse{}}, + {wire.ClusterOpCheckMinorBlockRequest, &wire.CheckMinorBlockRequest{MinorBlockHeader: emptyRawBytes()}, &wire.CheckMinorBlockResponse{}}, + {wire.ClusterOpGetAllTransactionsRequest, &wire.GetAllTransactionsRequest{}, &wire.GetAllTransactionsResponse{}}, + {wire.ClusterOpGetRootChainStakesRequest, &wire.GetRootChainStakesRequest{}, &wire.GetRootChainStakesResponse{}}, + {wire.ClusterOpGetTotalBalanceRequest, &wire.GetTotalBalanceRequest{}, &wire.GetTotalBalanceResponse{}}, + } + + for _, tc := range cases { + // Serialize the request bytes. + reqBytes, err := serialize.SerializeToBytes(tc.req) + if err != nil { + t.Fatalf("serialize request for opcode 0x%x: %v", tc.opcode, err) + } + + // Ask the server to process the request by writing a raw frame. + // We use a fresh connection per case to avoid ordering issues. + client, srv, cleanupPair := newMasterTestConnPair(t) + srv.Start() + + writeRawMasterFrame(t, client.conn, &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x00010001}, + Opcode: byte(tc.opcode), + RPCID: 1, + Payload: reqBytes, + }) + + // Read the raw response from the connection. + clientConn := client.conn + clientConn.SetReadDeadline(time.Now().Add(2 * time.Second)) + frame, err := wire.ReadFrame(clientConn, 0) + if err != nil { + t.Fatalf("read response for opcode 0x%x: %v", tc.opcode, err) + } + if frame.Opcode != byte(tc.opcode)+1 { + t.Fatalf("opcode 0x%x: expected response opcode 0x%x, got 0x%x", tc.opcode, byte(tc.opcode)+1, frame.Opcode) + } + if err := serialize.Deserialize(serialize.NewByteBuffer(frame.Payload), tc.resp); err != nil { + t.Fatalf("deserialize response for opcode 0x%x: %v", tc.opcode, err) + } + + cleanupPair() + } +} + +// TestMasterConn_EmptyPayloadDeserialization verifies that request types with +// empty bodies deserialize correctly. +func TestMasterConn_EmptyPayloadDeserialization(t *testing.T) { + client, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + // Only start the server; the client connection is used as a bare socket so + // we can observe the raw response frame without the client's readLoop + // competing for bytes. + server.Start() + + // Empty payload should deserialize to an empty GetEcoInfoListRequest. + writeRawMasterFrame(t, client.conn, &wire.Frame{ + Meta: wire.ClusterMetadata{}, + Opcode: byte(wire.ClusterOpGetEcoInfoListRequest), + RPCID: 1, + Payload: []byte{}, + }) + + // Read the response from the same connection the client wrote on. + client.conn.SetReadDeadline(time.Now().Add(2 * time.Second)) + frame, err := wire.ReadFrame(client.conn, 0) + if err != nil { + t.Fatalf("read response: %v", err) + } + if frame.Opcode != byte(wire.ClusterOpGetEcoInfoListResponse) { + t.Fatalf("expected GetEcoInfoListResponse, got 0x%x", frame.Opcode) + } +} + +// TestMasterConn_MetadataPreserved verifies that request metadata is echoed +// back in the response. +func TestMasterConn_MetadataPreserved(t *testing.T) { + client, server, cleanup := newMasterTestConnPair(t) + defer cleanup() + + server.Start() + client.Start() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + meta := wire.ClusterMetadata{Branch: 0xDEADBEEF, ClusterPeerID: 0xCAFEBABECAFEBABE} + payload, _ := serialize.SerializeToBytes(&wire.GetEcoInfoListRequest{}) + resp, err := client.SendRPCMeta(ctx, byte(wire.ClusterOpGetEcoInfoListRequest), payload, meta) + if err != nil { + t.Fatalf("send rpc: %v", err) + } + if resp.Meta != meta { + t.Fatalf("metadata not preserved: got %+v, want %+v", resp.Meta, meta) + } +} + +// TestMasterConn_FrameWireLayout verifies the full ClusterMetadata frame layout +// written by MasterConn matches the Python protocol. +func TestMasterConn_FrameWireLayout(t *testing.T) { + var buf bytes.Buffer + frame := &wire.Frame{ + Meta: wire.ClusterMetadata{Branch: 0x01020304, ClusterPeerID: 0x1122334455667788}, + Opcode: byte(wire.ClusterOpPing), + RPCID: 0xAABBCCDDEEFF0011, + Payload: []byte{0xAA, 0xBB}, + } + if err := wire.WriteFrame(&buf, frame); err != nil { + t.Fatalf("WriteFrame: %v", err) + } + + wireBytes := buf.Bytes() + if len(wireBytes) != 4+12+1+8+2 { + t.Fatalf("frame length: got %d, want %d", len(wireBytes), 4+12+1+8+2) + } + + if got := binary.BigEndian.Uint32(wireBytes[0:4]); got != 2 { + t.Fatalf("payload_len: got %d, want 2", got) + } + if got := binary.BigEndian.Uint32(wireBytes[4:8]); got != frame.Meta.Branch { + t.Fatalf("branch mismatch: got 0x%x", got) + } + if got := binary.BigEndian.Uint64(wireBytes[8:16]); got != frame.Meta.ClusterPeerID { + t.Fatalf("cluster_peer_id mismatch: got 0x%x", got) + } + if wireBytes[16] != frame.Opcode { + t.Fatalf("opcode mismatch: got 0x%x", wireBytes[16]) + } + if got := binary.BigEndian.Uint64(wireBytes[17:25]); got != frame.RPCID { + t.Fatalf("rpc_id mismatch: got 0x%x", got) + } + if !bytes.Equal(wireBytes[25:], frame.Payload) { + t.Fatalf("payload mismatch: got %x", wireBytes[25:]) + } +} diff --git a/qkc/cluster/slave/testdata/pyproto/master.py b/qkc/cluster/slave/testdata/pyproto/master.py new file mode 100644 index 0000000000..5cc2b728d5 --- /dev/null +++ b/qkc/cluster/slave/testdata/pyproto/master.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +"""Minimal MasterConnection protocol peer for Go compatibility tests. + +This peer implements the master-to-slave protocol that MasterConn needs: + - Frame read/write (12-byte ClusterMetadata, matching ReadFrame/WriteFrame) + - PING/PONG identity exchange (ClusterOp 0x81/0x82) + - RPC request/response for a representative set of master->slave opcodes + - Fire-and-forget command dispatch + +Usage: + python3 master.py --port 0 --id "master" --shards "1,2" + +Output: + PORT: Printed when listening + PONG_OK id= Printed when PONG is received + ECO_OK error_code= Printed when GetEcoInfoListResponse is received + ROOT_OK error_code= Printed when AddRootBlockResponse is received + DESTROY_OK Printed after DESTROY command (no response expected) + DISCONNECTED Printed when connection closes + +Behavior: + - Listens on TCP, accepts one connection + - Sends PING (rpc_id=1), waits for PONG + - Sends GetEcoInfoListRequest (rpc_id=2), waits for GetEcoInfoListResponse + - Sends AddRootBlockRequest (rpc_id=3), waits for AddRootBlockResponse + - Sends DestroyClusterPeerConnectionCommand (rpc_id=0) + - Sends a second PING (rpc_id=4) to verify the connection is still alive + - Closes connection and exits +""" +import argparse +import socket +import struct +import sys + +from master_frame import read_master_frame, write_master_frame +from messages import serialize_ping_request, parse_pong_response + +CLUSTER_OP_BASE = 0x80 + +CLUSTER_OP_PING = 1 + CLUSTER_OP_BASE +CLUSTER_OP_PONG = 2 + CLUSTER_OP_BASE + +# Master -> Slave opcodes +CLUSTER_OP_GET_ECO_INFO_LIST_REQUEST = 7 + CLUSTER_OP_BASE +CLUSTER_OP_GET_ECO_INFO_LIST_RESPONSE = 8 + CLUSTER_OP_BASE +CLUSTER_OP_ADD_ROOT_BLOCK_REQUEST = 5 + CLUSTER_OP_BASE +CLUSTER_OP_ADD_ROOT_BLOCK_RESPONSE = 6 + CLUSTER_OP_BASE +CLUSTER_OP_DESTROY_CLUSTER_PEER_CONNECTION_COMMAND = 27 + CLUSTER_OP_BASE + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--port', type=int, required=True) + parser.add_argument('--id', type=str, required=True) + parser.add_argument('--shards', type=str, required=True) + args = parser.parse_args() + + master_id = args.id.encode('utf-8') + shard_list = [int(s) for s in args.shards.split(',')] + + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server.bind(('127.0.0.1', args.port)) + server.listen(1) + + actual_port = server.getsockname()[1] + print(f"PORT:{actual_port}", flush=True) + + conn, _ = server.accept() + + try: + # 1. PING -> PONG + _send_ping(conn, master_id, shard_list) + + # 2. GetEcoInfoListRequest -> GetEcoInfoListResponse + _send_rpc( + conn, + CLUSTER_OP_GET_ECO_INFO_LIST_REQUEST, + CLUSTER_OP_GET_ECO_INFO_LIST_RESPONSE, + 2, + b'', + 'ECO_OK', + ) + + # 3. AddRootBlockRequest -> AddRootBlockResponse + add_root_block_payload = b'\x00\x00\x00\x00' + b'\x00' # empty root block + expect_switch=False + _send_rpc( + conn, + CLUSTER_OP_ADD_ROOT_BLOCK_REQUEST, + CLUSTER_OP_ADD_ROOT_BLOCK_RESPONSE, + 3, + add_root_block_payload, + 'ROOT_OK', + ) + + # 4. Fire-and-forget DestroyClusterPeerConnectionCommand + write_master_frame( + conn, + CLUSTER_OP_DESTROY_CLUSTER_PEER_CONNECTION_COMMAND, + 0, + struct.pack('>Q', 42), + branch=0x00010001, + ) + print("DESTROY_OK", flush=True) + + # 5. Second PING to confirm the connection is still alive after the + # fire-and-forget command. + _send_ping(conn, master_id, shard_list) + + except (ConnectionError, BrokenPipeError, OSError) as e: + print(f"ERROR: {e}", flush=True) + sys.exit(1) + finally: + conn.close() + server.close() + print("DISCONNECTED", flush=True) + + +def _send_ping(conn, master_id, shard_list): + ping_payload = serialize_ping_request(master_id, shard_list) + write_master_frame(conn, CLUSTER_OP_PING, 1, ping_payload) + + frame = read_master_frame(conn) + if frame is None: + print("ERROR: no pong received", flush=True) + sys.exit(1) + + if frame['opcode'] != CLUSTER_OP_PONG: + print(f"ERROR: expected PONG(0x{CLUSTER_OP_PONG:02x}), got 0x{frame['opcode']:02x}", flush=True) + sys.exit(1) + if frame['rpc_id'] != 1: + print(f"ERROR: expected rpc_id 1, got {frame['rpc_id']}", flush=True) + sys.exit(1) + + peer_id_recv, _ = parse_pong_response(frame['payload']) + print(f"PONG_OK id={peer_id_recv.hex()}", flush=True) + + +def _send_rpc(conn, req_opcode, resp_opcode, rpc_id, payload, ok_label): + write_master_frame(conn, req_opcode, rpc_id, payload, branch=0x00010001) + + frame = read_master_frame(conn) + if frame is None: + print(f"ERROR: no response for opcode 0x{req_opcode:02x}", flush=True) + sys.exit(1) + + if frame['opcode'] != resp_opcode: + print(f"ERROR: expected 0x{resp_opcode:02x}, got 0x{frame['opcode']:02x}", flush=True) + sys.exit(1) + if frame['rpc_id'] != rpc_id: + print(f"ERROR: expected rpc_id {rpc_id}, got {frame['rpc_id']}", flush=True) + sys.exit(1) + + # First field of these responses is a uint32 error_code. + if len(frame['payload']) < 4: + print(f"ERROR: response payload too short for {ok_label}", flush=True) + sys.exit(1) + error_code = struct.unpack('>I', frame['payload'][0:4])[0] + print(f"{ok_label} error_code={error_code}", flush=True) + + +if __name__ == '__main__': + main() diff --git a/qkc/cluster/slave/testdata/pyproto/master_frame.py b/qkc/cluster/slave/testdata/pyproto/master_frame.py new file mode 100644 index 0000000000..4ea9b3c945 --- /dev/null +++ b/qkc/cluster/slave/testdata/pyproto/master_frame.py @@ -0,0 +1,53 @@ +"""Frame read/write for master-slave protocol (12-byte ClusterMetadata). + +Wire format: [4B payload_len][4B branch][8B cluster_peer_id][1B opcode][8B rpc_id][payload] + +This matches Go's qkc/cluster/wire ReadFrame/WriteFrame with ClusterMetadata. +""" +import struct + + +def read_master_frame(conn): + """Read one master frame from conn. + + Returns a dict with keys: branch, cluster_peer_id, opcode, rpc_id, payload. + Returns None on EOF. + """ + header = conn.recv(25) # 4 + 12 + 1 + 8 + if not header: + return None + if len(header) < 25: + raise ConnectionError("truncated master frame header") + + payload_len = struct.unpack('>I', header[0:4])[0] + branch = struct.unpack('>I', header[4:8])[0] + cluster_peer_id = struct.unpack('>Q', header[8:16])[0] + opcode = header[16] + rpc_id = struct.unpack('>Q', header[17:25])[0] + + payload = b'' + while len(payload) < payload_len: + chunk = conn.recv(payload_len - len(payload)) + if not chunk: + raise ConnectionError("truncated master frame payload") + payload += chunk + + return { + 'branch': branch, + 'cluster_peer_id': cluster_peer_id, + 'opcode': opcode, + 'rpc_id': rpc_id, + 'payload': payload, + } + + +def write_master_frame(conn, opcode, rpc_id, payload, branch=0, cluster_peer_id=0): + """Write one master frame to conn.""" + header = ( + struct.pack('>I', len(payload)) + + struct.pack('>I', branch) + + struct.pack('>Q', cluster_peer_id) + + bytes([opcode]) + + struct.pack('>Q', rpc_id) + ) + conn.sendall(header + payload)