change cluster lib

This commit is contained in:
iteye 2026-07-02 15:15:56 +08:00
parent 43d0e0e580
commit edf7f7b473
3 changed files with 84 additions and 44 deletions

View file

@ -1,6 +1,6 @@
// Copyright 2026-2027, QuarkChain.
//
// Package slave implements a binary frame codec compatible with pyquarkchain.
// Package wire implements a binary frame codec compatible with pyquarkchain.
//
// Wire format (per frame):
//
@ -10,7 +10,7 @@
// metadata size depends on metaSize parameter:
// - 12 bytes: ClusterMetadata (branch uint32 + cluster_peer_id uint64) for master↔slave
// - 0 bytes: no metadata for slave↔slave traffic
package slave
package wire
import (
"encoding/binary"
@ -19,17 +19,10 @@ import (
"io"
)
// Metadata is the routing header for ClusterMetadata traffic.
// Wire representation: 12 bytes (4B branch + 8B cluster_peer_id).
type Metadata struct {
Branch uint32
ClusterPeerID uint64
}
// Frame is a complete protocol frame.
// Wire layout after metadata: [1B opcode][8B rpc_id][N bytes payload]
type Frame struct {
Meta Metadata
Meta ClusterMetadata
Opcode byte
RPCID uint64
Payload []byte
@ -42,12 +35,12 @@ const (
frameHeader = 4 // payload_len prefix
)
// ReadFrame reads a frame with 12-byte metadata and no payload-length cap.
// ReadFrame reads a frame with 12-byte ClusterMetadata and no payload-length cap.
func ReadFrame(r io.Reader) (*Frame, error) {
return readFrame(r, metaSize, 0)
}
// ReadFrameWithMaxPayload reads a frame with 12-byte metadata and rejects frames
// ReadFrameWithMaxPayload reads a frame with 12-byte ClusterMetadata and rejects frames
// whose payload_len exceeds maxPayloadLen before allocation.
func ReadFrameWithMaxPayload(r io.Reader, maxPayloadLen uint32) (*Frame, error) {
if maxPayloadLen == 0 {
@ -83,16 +76,16 @@ func readFrame(r io.Reader, metaSize int, maxPayloadLen uint32) (*Frame, error)
}
// 2. Read metadata (size depends on metadata_class)
var meta Metadata
var meta ClusterMetadata
if metaSize > 0 {
if metaSize != 12 {
return nil, fmt.Errorf("unsupported metaSize %d (only 0 or 12 supported)", metaSize)
return nil, fmt.Errorf("unsupported metaSize %d (only 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{
meta = ClusterMetadata{
Branch: binary.BigEndian.Uint32(metaBuf[0:4]),
ClusterPeerID: binary.BigEndian.Uint64(metaBuf[4:12]),
}
@ -113,7 +106,7 @@ func readFrame(r io.Reader, metaSize int, maxPayloadLen uint32) (*Frame, error)
}, nil
}
// WriteFrame serializes f with 12-byte metadata and writes it to w.
// WriteFrame serializes f with 12-byte ClusterMetadata and writes it to w.
func WriteFrame(w io.Writer, f *Frame) error {
return writeFrameWithMetaSize(w, f, metaSize)
}
@ -152,22 +145,3 @@ func writeFrameWithMetaSize(w io.Writer, f *Frame, metaSize int) error {
_, 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
}

View file

@ -1,6 +1,6 @@
// Copyright 2026-2027, QuarkChain.
package slave
package wire
import (
"bytes"
@ -11,6 +11,7 @@ import (
var (
// Test vectors serialized by pyquarkchain serializers using synthetic test values.
// NOT real production network data.
//
// Payload content (Ping/Pong commands):
// - id = "id" (ASCII, 2 bytes)
@ -116,9 +117,9 @@ func TestRoundTrip_Meta(t *testing.T) {
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")}},
{"with_meta", &Frame{Meta: ClusterMetadata{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{}}},
{"zero_meta", &Frame{Meta: ClusterMetadata{}, Opcode: 0x81, RPCID: 1, Payload: []byte{}}},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
@ -154,7 +155,7 @@ func TestRoundTrip_NoMeta(t *testing.T) {
func TestWireFormatLayout(t *testing.T) {
f := &Frame{
Meta: Metadata{Branch: 1, ClusterPeerID: 0x1122334455667788},
Meta: ClusterMetadata{Branch: 1, ClusterPeerID: 0x1122334455667788},
Opcode: 0x42,
RPCID: 0xDEADBEEFCAFEBABE,
Payload: []byte{0xAA, 0xBB, 0xCC},
@ -183,9 +184,9 @@ func TestWireFormatLayout(t *testing.T) {
func TestMultiFrameStream(t *testing.T) {
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},
{Meta: ClusterMetadata{Branch: 0, ClusterPeerID: 0}, Opcode: 0x81, RPCID: 0, Payload: []byte("ping")},
{Meta: ClusterMetadata{Branch: 2, ClusterPeerID: 999}, Opcode: 0x05, RPCID: 100, Payload: []byte("block_data")},
{Meta: ClusterMetadata{Branch: 1, ClusterPeerID: 0}, Opcode: 0x03, RPCID: 200, Payload: nil},
}
var stream bytes.Buffer
for _, f := range frames {
@ -251,7 +252,34 @@ func TestReadFrameNoMeta_PayloadLimit(t *testing.T) {
}
}
func writeFrameForTest(meta Metadata, opcode byte, rpcID uint64, payload []byte) []byte {
func TestClusterMetadata_RoundTrip(t *testing.T) {
cases := []ClusterMetadata{
{Branch: 0, ClusterPeerID: 0},
{Branch: 1, ClusterPeerID: 12345},
{Branch: 0xFFFFFFFF, ClusterPeerID: 0xFFFFFFFFFFFFFFFF},
}
for _, m := range cases {
wire := MarshalClusterMetadata(m)
got, err := UnmarshalClusterMetadata(wire)
if err != nil {
t.Fatalf("UnmarshalClusterMetadata(%+v): %v", m, err)
}
if got != m {
t.Errorf("round-trip mismatch: got %+v, want %+v", got, m)
}
}
}
func TestUnmarshalClusterMetadata_InvalidLength(t *testing.T) {
for _, n := range []int{0, 4, 8, 11, 13, 16} {
b := make([]byte, n)
if _, err := UnmarshalClusterMetadata(b); err == nil {
t.Errorf("expected error for %d-byte input", n)
}
}
}
func writeFrameForTest(meta ClusterMetadata, 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()

View file

@ -0,0 +1,38 @@
// Copyright 2026-2027, QuarkChain.
package wire
import (
"encoding/binary"
"fmt"
)
// ClusterMetadata is the routing header for intra-cluster (master↔slave) traffic.
// Wire representation: 12 bytes (4B branch + 8B cluster_peer_id).
//
// Matches pyquarkchain's ClusterMetadata class (protocol.py).
// The base Metadata in pyquarkchain is the 0-byte variant used by direct
// slave-to-slave connections, handled by ReadFrameNoMeta / WriteFrameNoMeta.
type ClusterMetadata struct {
Branch uint32
ClusterPeerID uint64
}
// MarshalClusterMetadata serializes ClusterMetadata into its 12-byte wire representation.
func MarshalClusterMetadata(m ClusterMetadata) []byte {
buf := make([]byte, metaSize)
binary.BigEndian.PutUint32(buf[0:4], m.Branch)
binary.BigEndian.PutUint64(buf[4:12], m.ClusterPeerID)
return buf
}
// UnmarshalClusterMetadata deserializes a 12-byte wire representation into ClusterMetadata.
func UnmarshalClusterMetadata(b []byte) (ClusterMetadata, error) {
if len(b) != metaSize {
return ClusterMetadata{}, fmt.Errorf("metadata must be %d bytes, got %d", metaSize, len(b))
}
return ClusterMetadata{
Branch: binary.BigEndian.Uint32(b[0:4]),
ClusterPeerID: binary.BigEndian.Uint64(b[4:12]),
}, nil
}