mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-21 20:26:41 +00:00
qkc/slave: add binary frame codec
This commit is contained in:
iteye
parent
4160ab81ee
commit
58b66d610e
2 changed files with 609 additions and 0 deletions
249
qkc/slave/frame.go
Normal file
249
qkc/slave/frame.go
Normal file
|
|
@ -0,0 +1,249 @@
|
|||
// Package slave: binary frame codec compatible with pyquarkchain protocol.py.
|
||||
//
|
||||
// Wire format (per-frame):
|
||||
//
|
||||
// [4B payload_len] [metaSize B metadata] [1B opcode] [8B rpc_id] [payload_len bytes]
|
||||
//
|
||||
// This matches Python's Connection.read_metadata_and_raw_data /
|
||||
// Connection.write_raw_data exactly:
|
||||
//
|
||||
// Python: protocol.py lines 285-308
|
||||
//
|
||||
// Metadata sizes (matching Python Metadata subclasses):
|
||||
//
|
||||
// ClusterMetadata.get_byte_size() = 12 (branch 4B + cluster_peer_id 8B)
|
||||
// Used for master ↔ slave traffic.
|
||||
//
|
||||
// Metadata.get_byte_size() = 0 (base class)
|
||||
// Used for slave ↔ slave traffic (SlaveConnection).
|
||||
//
|
||||
// P2PMetadata.get_byte_size() = 4 (branch 4B)
|
||||
// Used for inter-cluster P2P. Handled by Python master only;
|
||||
// Go slave never sends or receives P2PMetadata frames.
|
||||
//
|
||||
// payload_len definition (matching Python line 305):
|
||||
//
|
||||
// cmd_length_bytes = (len(raw_data) - 8 - 1).to_bytes(4, "big")
|
||||
//
|
||||
// That is, payload_len = len(raw_data) - 9, where raw_data is
|
||||
// [1B opcode][8B rpc_id][N bytes payload].
|
||||
package slave
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
)
|
||||
|
||||
// Metadata is the 12-byte frame header carrying routing information.
|
||||
// Matches Python's ClusterMetadata:
|
||||
//
|
||||
// class ClusterMetadata(Metadata):
|
||||
// FIELDS = [("branch", Branch), ("cluster_peer_id", uint64)]
|
||||
// @staticmethod
|
||||
// def get_byte_size(): return 12
|
||||
type Metadata struct {
|
||||
Branch uint32 // shard identifier (Python: Branch = uint32)
|
||||
ClusterPeerID uint64 // 0 = cluster RPC (master commands), ≠0 = specific external peer
|
||||
}
|
||||
|
||||
// Frame is a complete protocol frame.
|
||||
// raw_data layout on wire: [1B opcode][8B rpc_id][N bytes payload]
|
||||
type Frame struct {
|
||||
Meta Metadata
|
||||
Opcode byte
|
||||
RPCID uint64
|
||||
Payload []byte
|
||||
}
|
||||
|
||||
const (
|
||||
metaSize = 12 // ClusterMetadata.get_byte_size() = 12 (branch 4B + cluster_peer_id 8B)
|
||||
opcodeSize = 1
|
||||
rpcIDSize = 8
|
||||
frameHeader = 4 // payload_len prefix
|
||||
totalOverhead = frameHeader + metaSize + opcodeSize + rpcIDSize // 4+12+1+8 = 25
|
||||
)
|
||||
|
||||
// ReadFrame reads a frame with 12-byte ClusterMetadata.
|
||||
//
|
||||
// Matches Python's Connection.read_metadata_and_raw_data (protocol.py lines 285-300)
|
||||
// with metadata_class = ClusterMetadata (get_byte_size() == 12).
|
||||
//
|
||||
// Used for master ↔ slave traffic.
|
||||
// For slave ↔ slave traffic, use ReadFrameNoMeta (0-byte metadata).
|
||||
func ReadFrame(r io.Reader) (*Frame, error) {
|
||||
return readFrameWithMetaSize(r, metaSize)
|
||||
}
|
||||
|
||||
// ReadFrameNoMeta reads a frame with 0-byte Metadata.
|
||||
//
|
||||
// Matches Python's SlaveConnection which uses metadata_class = Metadata
|
||||
// (get_byte_size() == 0).
|
||||
//
|
||||
// Used for slave ↔ slave direct TCP traffic.
|
||||
func ReadFrameNoMeta(r io.Reader) (*Frame, error) {
|
||||
return readFrameWithMetaSize(r, 0)
|
||||
}
|
||||
|
||||
// readFrameWithMetaSize is the underlying frame reader.
|
||||
//
|
||||
// Wire layout (matching Python read_metadata_and_raw_data):
|
||||
//
|
||||
// size_bytes = await read_fully(4) → 4B payload_len (big-endian)
|
||||
// metadata_bytes = await read_fully(metaSize) → metaSize B metadata
|
||||
// raw_data_without_size = await read_fully(1+8+size) → opcode + rpc_id + payload
|
||||
func readFrameWithMetaSize(r io.Reader, metaSize int) (*Frame, error) {
|
||||
// 1. Read 4-byte big-endian payload length
|
||||
// Python: size_bytes = await self.__read_fully(4, allow_eof=True)
|
||||
// size = int.from_bytes(size_bytes, byteorder="big")
|
||||
var payloadLen uint32
|
||||
if err := binary.Read(r, binary.BigEndian, &payloadLen); err != nil {
|
||||
return nil, fmt.Errorf("reading frame length: %w", err)
|
||||
}
|
||||
|
||||
// 2. Read metadata (size depends on metadata_class)
|
||||
// Python: metadata_bytes = await self.__read_fully(self.metadata_class.get_byte_size())
|
||||
// metadata = self.metadata_class.deserialize(metadata_bytes)
|
||||
var meta Metadata
|
||||
if metaSize > 0 {
|
||||
if metaSize != 12 {
|
||||
return nil, fmt.Errorf("unsupported metaSize %d (only 0 or 12 supported)", metaSize)
|
||||
}
|
||||
metaBuf := make([]byte, metaSize)
|
||||
if _, err := io.ReadFull(r, metaBuf); err != nil {
|
||||
return nil, fmt.Errorf("reading metadata: %w", err)
|
||||
}
|
||||
meta = Metadata{
|
||||
Branch: binary.BigEndian.Uint32(metaBuf[0:4]),
|
||||
ClusterPeerID: binary.BigEndian.Uint64(metaBuf[4:12]),
|
||||
}
|
||||
}
|
||||
|
||||
// 3. Read raw_data_without_size: [1B opcode][8B rpc_id][N bytes payload]
|
||||
// Python: raw_data_without_size = await self.__read_fully(1 + 8 + size)
|
||||
bodySize := opcodeSize + rpcIDSize + int(payloadLen)
|
||||
body := make([]byte, bodySize)
|
||||
if _, err := io.ReadFull(r, body); err != nil {
|
||||
return nil, fmt.Errorf("reading frame body (payload_len=%d): %w", payloadLen, err)
|
||||
}
|
||||
|
||||
return &Frame{
|
||||
Meta: meta,
|
||||
Opcode: body[0],
|
||||
RPCID: binary.BigEndian.Uint64(body[1:9]),
|
||||
Payload: body[9:],
|
||||
}, nil
|
||||
}
|
||||
|
||||
// WriteFrame serializes f with 12-byte ClusterMetadata and writes it to w.
|
||||
//
|
||||
// Matches Python's Connection.write_raw_data (protocol.py lines 302-308)
|
||||
// with metadata_class = ClusterMetadata.
|
||||
//
|
||||
// Used for master ↔ slave traffic.
|
||||
func WriteFrame(w io.Writer, f *Frame) error {
|
||||
return writeFrameWithMetaSize(w, f, metaSize)
|
||||
}
|
||||
|
||||
// WriteFrameNoMeta writes a frame with 0-byte Metadata.
|
||||
//
|
||||
// Matches Python's SlaveConnection which uses metadata_class = Metadata
|
||||
// (get_byte_size() == 0).
|
||||
//
|
||||
// Used for slave ↔ slave traffic.
|
||||
func WriteFrameNoMeta(w io.Writer, f *Frame) error {
|
||||
return writeFrameWithMetaSize(w, f, 0)
|
||||
}
|
||||
|
||||
// writeFrameWithMetaSize serializes f with the given metadata size and writes
|
||||
// it to w.
|
||||
//
|
||||
// Wire layout (matching Python write_raw_data, protocol.py lines 302-308):
|
||||
//
|
||||
// cmd_length_bytes = (len(raw_data) - 8 - 1).to_bytes(4, "big")
|
||||
// self.writer.write(cmd_length_bytes) → 4B payload_len
|
||||
// self.writer.write(metadata.serialize()) → metaSize B
|
||||
// self.writer.write(raw_data) → [1B opcode][8B rpc_id][payload]
|
||||
func writeFrameWithMetaSize(w io.Writer, f *Frame, metaSize int) error {
|
||||
payloadLen := uint32(len(f.Payload))
|
||||
if int(payloadLen) != len(f.Payload) {
|
||||
return errors.New("payload too large")
|
||||
}
|
||||
|
||||
// Build the buffer in one write (Python writes in 3 chunks, but the wire
|
||||
// bytes are identical).
|
||||
total := frameHeader + metaSize + opcodeSize + rpcIDSize + int(payloadLen)
|
||||
buf := make([]byte, total)
|
||||
|
||||
// Frame length (payload only): matches Python's cmd_length_bytes
|
||||
binary.BigEndian.PutUint32(buf[0:frameHeader], payloadLen)
|
||||
|
||||
// Metadata
|
||||
if metaSize > 0 {
|
||||
binary.BigEndian.PutUint32(buf[frameHeader:frameHeader+4], f.Meta.Branch)
|
||||
binary.BigEndian.PutUint64(buf[frameHeader+4:frameHeader+metaSize], f.Meta.ClusterPeerID)
|
||||
}
|
||||
|
||||
// Opcode
|
||||
buf[frameHeader+metaSize] = f.Opcode
|
||||
|
||||
// RPC ID (big-endian, matches Python: rpc_id.to_bytes(8, "big"))
|
||||
binary.BigEndian.PutUint64(buf[frameHeader+metaSize+opcodeSize:frameHeader+metaSize+opcodeSize+rpcIDSize], f.RPCID)
|
||||
|
||||
// Payload
|
||||
copy(buf[frameHeader+metaSize+opcodeSize+rpcIDSize:], f.Payload)
|
||||
|
||||
_, err := w.Write(buf)
|
||||
return err
|
||||
}
|
||||
|
||||
// MarshalMetadata serializes Metadata into its 12-byte wire representation.
|
||||
func MarshalMetadata(m Metadata) []byte {
|
||||
buf := make([]byte, metaSize)
|
||||
binary.BigEndian.PutUint32(buf[0:4], m.Branch)
|
||||
binary.BigEndian.PutUint64(buf[4:12], m.ClusterPeerID)
|
||||
return buf
|
||||
}
|
||||
|
||||
// UnmarshalMetadata deserializes a 12-byte wire representation into Metadata.
|
||||
func UnmarshalMetadata(b []byte) (Metadata, error) {
|
||||
if len(b) != metaSize {
|
||||
return Metadata{}, fmt.Errorf("metadata must be %d bytes, got %d", metaSize, len(b))
|
||||
}
|
||||
return Metadata{
|
||||
Branch: binary.BigEndian.Uint32(b[0:4]),
|
||||
ClusterPeerID: binary.BigEndian.Uint64(b[4:12]),
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ── Convenience wrappers (used in tests) ─────────────────────────────────────
|
||||
|
||||
// ReadFrameFromReader wraps r in a bufio.Reader.
|
||||
func ReadFrameFromReader(r io.Reader) (*Frame, error) {
|
||||
return ReadFrame(bufio.NewReader(r))
|
||||
}
|
||||
|
||||
// WriteFrameToWriter wraps w with a bufio.Writer and flushes.
|
||||
func WriteFrameToWriter(w io.Writer, frame *Frame) error {
|
||||
bw := bufio.NewWriter(w)
|
||||
if err := WriteFrame(bw, frame); err != nil {
|
||||
return err
|
||||
}
|
||||
return bw.Flush()
|
||||
}
|
||||
|
||||
// ReadFrameNoMetaFromReader wraps r for ReadFrameNoMeta.
|
||||
func ReadFrameNoMetaFromReader(r io.Reader) (*Frame, error) {
|
||||
return ReadFrameNoMeta(bufio.NewReader(r))
|
||||
}
|
||||
|
||||
// WriteFrameNoMetaToWriter wraps w for WriteFrameNoMeta.
|
||||
func WriteFrameNoMetaToWriter(w io.Writer, frame *Frame) error {
|
||||
bw := bufio.NewWriter(w)
|
||||
if err := WriteFrameNoMeta(bw, frame); err != nil {
|
||||
return err
|
||||
}
|
||||
return bw.Flush()
|
||||
}
|
||||
360
qkc/slave/frame_test.go
Normal file
360
qkc/slave/frame_test.go
Normal file
|
|
@ -0,0 +1,360 @@
|
|||
package slave
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"encoding/hex"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// =============================================================================
|
||||
// Python compatibility reference vectors
|
||||
//
|
||||
// Each vector below is the exact wire bytes a Python peer would send/receive
|
||||
// for a given frame. Source: qkc/quarkchain/protocol.py
|
||||
// - Connection.write_raw_data (lines 302-308)
|
||||
// - Connection.read_metadata_and_raw_data (lines 285-300)
|
||||
//
|
||||
// Wire layout: [4B payload_len] [metaSize B metadata] [1B opcode] [8B rpc_id] [payload]
|
||||
//
|
||||
// metadata_class:
|
||||
// ClusterMetadata (12B) for master↔slave traffic
|
||||
// Metadata (0B) for slave↔slave traffic
|
||||
// =============================================================================
|
||||
|
||||
// pingMasterWire: meta=(branch=0, peer=0), opcode=0x81 (PING), rpc_id=1, payload=empty
|
||||
//
|
||||
// Equivalent Python: write_raw_command(op=ClusterOp.PING, cmd_data=b"", rpc_id=1, metadata=ClusterMetadata(0, 0))
|
||||
// payload_len = 0
|
||||
// 00000000 | 00000000 0000000000000000 | 81 | 0000000000000001
|
||||
var pingMasterWire = "00000000" + "00000000" + "0000000000000000" + "81" + "0000000000000001"
|
||||
|
||||
// pongMasterWire: meta=(branch=0, peer=0), opcode=0x82 (PONG), rpc_id=1, payload=empty
|
||||
//
|
||||
// Equivalent Python: write_raw_command(op=ClusterOp.PONG, cmd_data=b"", rpc_id=1, metadata=ClusterMetadata(0, 0))
|
||||
// 00000000 | 00000000 0000000000000000 | 82 | 0000000000000001
|
||||
var pongMasterWire = "00000000" + "00000000" + "0000000000000000" + "82" + "0000000000000001"
|
||||
|
||||
// peerNewBlockWire: meta=(branch=1, peer=12345), opcode=0x01 (NEW_MINOR_BLOCK_HEADER_LIST), rpc_id=0, payload=12B
|
||||
//
|
||||
// cluster_peer_id=12345=0x3039, rpc_id=0 (NON-RPC fire-and-forget)
|
||||
// payload = 02 00 00 00 (list len=2) + a1b2c3d4 + e5f60718
|
||||
// 0000000c | 00000001 0000000000003039 | 01 | 0000000000000000 | 02000000a1b2c3d4e5f60718
|
||||
var peerNewBlockWire = "0000000c" +
|
||||
"00000001" + "0000000000003039" +
|
||||
"01" + "0000000000000000" +
|
||||
"02000000a1b2c3d4e5f60718"
|
||||
|
||||
// xshardWire: 0-byte metadata, opcode=0x93 (ADD_XSHARD_TX_LIST_REQUEST), rpc_id=42, payload=56B
|
||||
//
|
||||
// Used for slave↔slave direct TCP (Python SlaveConnection, metadata_class=Metadata)
|
||||
// payload_len = 56 = 0x38
|
||||
// payload = 01 00 00 00 (list len=1) + ff*32 + 00*20
|
||||
// 00000038 | (no meta) | 93 | 000000000000002a | 01000000 + ff×32 + 00×20
|
||||
var xshardWire = "00000038" +
|
||||
"93" + "000000000000002a" +
|
||||
"01000000" +
|
||||
strings.Repeat("ff", 32) +
|
||||
strings.Repeat("00", 20)
|
||||
|
||||
// largePayloadWire: meta=(branch=3, peer=0), opcode=0x10, rpc_id=7, payload=10000×0xAB
|
||||
//
|
||||
// payload_len = 10000 = 0x2710
|
||||
// 00002710 | 00000003 0000000000000000 | 10 | 0000000000000007 | ab×10000
|
||||
var largePayloadWire = "00002710" +
|
||||
"00000003" + "0000000000000000" +
|
||||
"10" + "0000000000000007" +
|
||||
strings.Repeat("ab", 10000)
|
||||
|
||||
// =============================================================================
|
||||
// ReadFrame tests — parse Python-generated wire bytes
|
||||
// =============================================================================
|
||||
|
||||
func TestPythonRead_Ping(t *testing.T) {
|
||||
wire, _ := hex.DecodeString(pingMasterWire)
|
||||
f, err := ReadFrame(bytes.NewReader(wire))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFrame: %v", err)
|
||||
}
|
||||
if f.Opcode != 0x81 {
|
||||
t.Errorf("Opcode: got 0x%02x, want 0x81", f.Opcode)
|
||||
}
|
||||
if f.RPCID != 1 {
|
||||
t.Errorf("RPCID: got %d, want 1", f.RPCID)
|
||||
}
|
||||
if f.Meta != (Metadata{Branch: 0, ClusterPeerID: 0}) {
|
||||
t.Errorf("Meta: got %+v", f.Meta)
|
||||
}
|
||||
if len(f.Payload) != 0 {
|
||||
t.Errorf("Payload len: got %d, want 0", len(f.Payload))
|
||||
}
|
||||
}
|
||||
|
||||
func TestPythonRead_Pong(t *testing.T) {
|
||||
wire, _ := hex.DecodeString(pongMasterWire)
|
||||
f, err := ReadFrame(bytes.NewReader(wire))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFrame: %v", err)
|
||||
}
|
||||
if f.Opcode != 0x82 {
|
||||
t.Errorf("Opcode: got 0x%02x, want 0x82", f.Opcode)
|
||||
}
|
||||
if f.RPCID != 1 {
|
||||
t.Errorf("RPCID: got %d, want 1", f.RPCID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPythonRead_PeerNewBlock(t *testing.T) {
|
||||
wire, _ := hex.DecodeString(peerNewBlockWire)
|
||||
f, err := ReadFrame(bytes.NewReader(wire))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFrame: %v", err)
|
||||
}
|
||||
if f.Opcode != 0x01 {
|
||||
t.Errorf("Opcode: got 0x%02x, want 0x01", f.Opcode)
|
||||
}
|
||||
if f.RPCID != 0 {
|
||||
t.Errorf("RPCID: got %d, want 0 (non-RPC)", f.RPCID)
|
||||
}
|
||||
if f.Meta.Branch != 1 {
|
||||
t.Errorf("Branch: got %d, want 1", f.Meta.Branch)
|
||||
}
|
||||
if f.Meta.ClusterPeerID != 12345 {
|
||||
t.Errorf("ClusterPeerID: got %d, want 12345", f.Meta.ClusterPeerID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPythonRead_Xshard(t *testing.T) {
|
||||
wire, _ := hex.DecodeString(xshardWire)
|
||||
f, err := ReadFrameNoMeta(bytes.NewReader(wire))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFrameNoMeta: %v", err)
|
||||
}
|
||||
if f.Opcode != 0x93 {
|
||||
t.Errorf("Opcode: got 0x%02x, want 0x93", f.Opcode)
|
||||
}
|
||||
if f.RPCID != 42 {
|
||||
t.Errorf("RPCID: got %d, want 42", f.RPCID)
|
||||
}
|
||||
if len(f.Payload) != 56 {
|
||||
t.Errorf("Payload len: got %d, want 56", len(f.Payload))
|
||||
}
|
||||
}
|
||||
|
||||
func TestPythonRead_LargePayload(t *testing.T) {
|
||||
wire, _ := hex.DecodeString(largePayloadWire)
|
||||
f, err := ReadFrame(bytes.NewReader(wire))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFrame: %v", err)
|
||||
}
|
||||
if f.Opcode != 0x10 {
|
||||
t.Errorf("Opcode: got 0x%02x, want 0x10", f.Opcode)
|
||||
}
|
||||
if f.RPCID != 7 {
|
||||
t.Errorf("RPCID: got %d, want 7", f.RPCID)
|
||||
}
|
||||
if f.Meta.Branch != 3 {
|
||||
t.Errorf("Branch: got %d, want 3", f.Meta.Branch)
|
||||
}
|
||||
if len(f.Payload) != 10000 {
|
||||
t.Errorf("Payload len: got %d, want 10000", len(f.Payload))
|
||||
}
|
||||
for i, b := range f.Payload {
|
||||
if b != 0xAB {
|
||||
t.Errorf("Payload[%d]: got 0x%02x, want 0xAB", i, b)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// WriteFrame tests — Go's wire output must match Python's byte-for-byte.
|
||||
// This is the strongest compatibility test: any wire-format drift is caught.
|
||||
// =============================================================================
|
||||
|
||||
func TestPythonWrite_Ping(t *testing.T) {
|
||||
want, _ := hex.DecodeString(pingMasterWire)
|
||||
got := writeFrameForTest(Metadata{Branch: 0, ClusterPeerID: 0}, 0x81, 1, nil)
|
||||
if !bytes.Equal(got, want) {
|
||||
t.Errorf("WriteFrame mismatch:\n got %x\n want %x", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPythonWrite_PeerNewBlock(t *testing.T) {
|
||||
want, _ := hex.DecodeString(peerNewBlockWire)
|
||||
payload := []byte{0x02, 0x00, 0x00, 0x00, 0xa1, 0xb2, 0xc3, 0xd4, 0xe5, 0xf6, 0x07, 0x18}
|
||||
got := writeFrameForTest(Metadata{Branch: 1, ClusterPeerID: 12345}, 0x01, 0, payload)
|
||||
if !bytes.Equal(got, want) {
|
||||
t.Errorf("WriteFrame mismatch:\n got %x\n want %x", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPythonWrite_Xshard(t *testing.T) {
|
||||
want, _ := hex.DecodeString(xshardWire)
|
||||
payload := append([]byte{0x01, 0x00, 0x00, 0x00}, bytes.Repeat([]byte{0xff}, 32)...)
|
||||
payload = append(payload, bytes.Repeat([]byte{0x00}, 20)...)
|
||||
got := writeFrameNoMetaForTest(0x93, 42, payload)
|
||||
if !bytes.Equal(got, want) {
|
||||
t.Errorf("WriteFrameNoMeta mismatch:\n got %x\n want %x", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPythonWrite_LargePayload(t *testing.T) {
|
||||
want, _ := hex.DecodeString(largePayloadWire)
|
||||
payload := bytes.Repeat([]byte{0xAB}, 10000)
|
||||
got := writeFrameForTest(Metadata{Branch: 3, ClusterPeerID: 0}, 0x10, 7, payload)
|
||||
if !bytes.Equal(got, want) {
|
||||
t.Errorf("WriteFrame mismatch (large):\n got %d bytes\n want %d bytes", len(got), len(want))
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// Write+Read round-trip — defensive tests independent of the Python reference
|
||||
// =============================================================================
|
||||
|
||||
func TestRoundTrip_Meta(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
f *Frame
|
||||
}{
|
||||
{"empty", &Frame{Opcode: 1, RPCID: 0, Payload: nil}},
|
||||
{"with_meta", &Frame{Meta: Metadata{Branch: 5, ClusterPeerID: 999}, Opcode: 0x10, RPCID: 7, Payload: []byte("hello")}},
|
||||
{"large_rpc_id", &Frame{Opcode: 0xC4, RPCID: 0xFFFFFFFFFFFFFFFF, Payload: []byte("x")}},
|
||||
{"zero_meta", &Frame{Meta: Metadata{}, Opcode: 0x81, RPCID: 1, Payload: []byte{}}},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
wire := writeFrameForTest(tc.f.Meta, tc.f.Opcode, tc.f.RPCID, tc.f.Payload)
|
||||
got, err := ReadFrame(bytes.NewReader(wire))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFrame: %v", err)
|
||||
}
|
||||
if got.Opcode != tc.f.Opcode || got.RPCID != tc.f.RPCID || got.Meta != tc.f.Meta {
|
||||
t.Errorf("mismatch: got %+v, want %+v", got, tc.f)
|
||||
}
|
||||
if !bytes.Equal(got.Payload, tc.f.Payload) {
|
||||
t.Errorf("payload mismatch")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRoundTrip_NoMeta(t *testing.T) {
|
||||
original := &Frame{Opcode: 0x93, RPCID: 99, Payload: []byte("xshard-data")}
|
||||
wire := writeFrameNoMetaForTest(original.Opcode, original.RPCID, original.Payload)
|
||||
got, err := ReadFrameNoMeta(bytes.NewReader(wire))
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFrameNoMeta: %v", err)
|
||||
}
|
||||
if got.Opcode != original.Opcode || got.RPCID != original.RPCID {
|
||||
t.Errorf("mismatch: got %+v", got)
|
||||
}
|
||||
if !bytes.Equal(got.Payload, original.Payload) {
|
||||
t.Errorf("payload mismatch")
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// Wire-format layout — verify byte-level structure with hand-computed expected
|
||||
// values, independent of any Python reference.
|
||||
// =============================================================================
|
||||
|
||||
func TestWireFormatLayout(t *testing.T) {
|
||||
f := &Frame{
|
||||
Meta: Metadata{Branch: 1, ClusterPeerID: 0x1122334455667788},
|
||||
Opcode: 0x42,
|
||||
RPCID: 0xDEADBEEFCAFEBABE,
|
||||
Payload: []byte{0xAA, 0xBB, 0xCC},
|
||||
}
|
||||
wire := writeFrameForTest(f.Meta, f.Opcode, f.RPCID, f.Payload)
|
||||
|
||||
// payload_len = 3
|
||||
if got := binary.BigEndian.Uint32(wire[0:4]); got != 3 {
|
||||
t.Errorf("payload_len: got %d, want 3", got)
|
||||
}
|
||||
// branch = 1
|
||||
if got := binary.BigEndian.Uint32(wire[4:8]); got != 1 {
|
||||
t.Errorf("branch: got %d, want 1", got)
|
||||
}
|
||||
// cluster_peer_id = 0x1122334455667788
|
||||
if got := binary.BigEndian.Uint64(wire[8:16]); got != 0x1122334455667788 {
|
||||
t.Errorf("cluster_peer_id: got 0x%x", got)
|
||||
}
|
||||
// opcode = 0x42
|
||||
if wire[16] != 0x42 {
|
||||
t.Errorf("opcode: got 0x%02x, want 0x42", wire[16])
|
||||
}
|
||||
// rpc_id = 0xDEADBEEFCAFEBABE
|
||||
if got := binary.BigEndian.Uint64(wire[17:25]); got != 0xDEADBEEFCAFEBABE {
|
||||
t.Errorf("rpc_id: got 0x%x", got)
|
||||
}
|
||||
// payload
|
||||
if !bytes.Equal(wire[25:28], []byte{0xAA, 0xBB, 0xCC}) {
|
||||
t.Errorf("payload: got %x", wire[25:28])
|
||||
}
|
||||
}
|
||||
|
||||
func TestMultiFrameStream(t *testing.T) {
|
||||
// 3 consecutive frames on a single stream, matching Python's back-to-back
|
||||
// write_raw_data() calls on the same TCP connection.
|
||||
frames := []*Frame{
|
||||
{Meta: Metadata{Branch: 0, ClusterPeerID: 0}, Opcode: 0x81, RPCID: 0, Payload: []byte("ping")},
|
||||
{Meta: Metadata{Branch: 2, ClusterPeerID: 999}, Opcode: 0x05, RPCID: 100, Payload: []byte("block_data")},
|
||||
{Meta: Metadata{Branch: 1, ClusterPeerID: 0}, Opcode: 0x03, RPCID: 200, Payload: nil},
|
||||
}
|
||||
var stream bytes.Buffer
|
||||
for _, f := range frames {
|
||||
stream.Write(writeFrameForTest(f.Meta, f.Opcode, f.RPCID, f.Payload))
|
||||
}
|
||||
|
||||
reader := bytes.NewReader(stream.Bytes())
|
||||
for i, want := range frames {
|
||||
got, err := ReadFrame(reader)
|
||||
if err != nil {
|
||||
t.Fatalf("frame %d: %v", i, err)
|
||||
}
|
||||
if got.Opcode != want.Opcode || got.RPCID != want.RPCID || got.Meta != want.Meta {
|
||||
t.Errorf("frame %d mismatch: got %+v, want %+v", i, got, want)
|
||||
}
|
||||
if !bytes.Equal(got.Payload, want.Payload) {
|
||||
t.Errorf("frame %d payload mismatch", i)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// Error handling
|
||||
// =============================================================================
|
||||
|
||||
func TestReadFrame_EOF(t *testing.T) {
|
||||
if _, err := ReadFrame(bytes.NewReader(nil)); err == nil {
|
||||
t.Error("expected error on empty stream")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadFrame_Truncated(t *testing.T) {
|
||||
// payload_len says 100 bytes, but we only give 4 bytes
|
||||
hdr := make([]byte, 4)
|
||||
binary.BigEndian.PutUint32(hdr, 100)
|
||||
if _, err := ReadFrame(bytes.NewReader(hdr)); err == nil {
|
||||
t.Error("expected error on truncated frame")
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// Helpers
|
||||
// =============================================================================
|
||||
|
||||
// writeFrameForTest is a thin wrapper that returns the wire bytes directly.
|
||||
func writeFrameForTest(meta Metadata, opcode byte, rpcID uint64, payload []byte) []byte {
|
||||
var buf bytes.Buffer
|
||||
_ = WriteFrame(&buf, &Frame{Meta: meta, Opcode: opcode, RPCID: rpcID, Payload: payload})
|
||||
return buf.Bytes()
|
||||
}
|
||||
|
||||
func writeFrameNoMetaForTest(opcode byte, rpcID uint64, payload []byte) []byte {
|
||||
var buf bytes.Buffer
|
||||
_ = WriteFrameNoMeta(&buf, &Frame{Opcode: opcode, RPCID: rpcID, Payload: payload})
|
||||
return buf.Bytes()
|
||||
}
|
||||
Loading…
Reference in a new issue