Files
zeekay ff1ee17c4f feat(network): Quasar-replicated WAL frames for Base (Phase 1)
Core runtime for network/ per docs/NETWORK.md. One env var
(BASE_NETWORK=quasar) opts Base into PQ-native HA replication via
luxfi/[email protected] Quasar DAG; one DAG per shard. Standalone
mode (default) returns a no-op Network with zero overhead.

Package surface:

  network.FromEnv()                      — parse BASE_* env, build Network
  network.Network.InstallWALHook(c, id)  — install modernc commit hook
  network.Network.WriterFor(shardID)     — router owner lookup
  network.Network.MembersFor(shardID)    — router replica set
  network.Network.Metrics()              — Prom collectors

Frame = {ShardID, Seq, PrevSeq, Ts, Salt, Payload, Cksm, Bytes}.
Content-addressed by (Salt, Cksm) so apply is idempotent across
pod restarts. Frame.Bytes is the exact serialised wire form —
archive replay never re-signs.

Quasar adapter (quasar.go): the upstream public API is Engine-based
(Submit(Block)/Finalized() <-chan Block), not the shard-scoped
Submit(shardID, frame) the design doc assumed. Adapter wraps one
Engine per shard; Block.ChainID = SHA256(shardID), Block.Data =
Frame.encode(). threshold < 2 routes to NewTestEngine for
single-node paths; production multi-node uses NewEngine with
k = floor(N/2)+1.

modernc.org/sqlite exposes RegisterCommitHook, not sqlite3_wal_hook
— same semantics for our use. core/base_network.go bridges
sqlite.HookRegisterer to network.HookRegisterer via a one-line
adapter so the network package stays driver-agnostic.

Hook into core/base.go: one field (network baseNetwork), one call
at end of Bootstrap (app.attachNetwork(ctx)), one call in
ResetBootstrapState (app.network.Stop). Standalone path unchanged.

Metrics: every collector in the doc (base_shards, base_active_shards,
base_wal_lag_bytes, base_hot_shards, base_lease_contentions,
base_archive_lag_bytes) plus frames/errors counters for the
produce/finalize loop. Callers bind to their own registry;
prometheus.DefaultRegisterer untouched.

Tests — 20 pass, 0 fail:
  wal_test.go           frame encode/decode roundtrip, tamper, idempotency key
  router_test.go        HRW determinism, balance, quorum math
  network_test.go       FromEnv standalone/quasar, fail-fast validation
  integration_test.go   3 in-process nodes share memoryHub transport; A and B
                        write different shards; all 3 finalise both frames
                        and increment FramesIngested on the non-writer.
  core/ regression      384 PASS / 2 FAIL — identical to pre-change baseline.

go.mod: +github.com/luxfi/consensus v1.22.85.
2026-04-22 13:39:01 -07:00

90 lines
2.1 KiB
Go

package network
import (
"bytes"
"testing"
)
func TestFrameEncodeDecodeRoundTrip(t *testing.T) {
f := newFrame("shard-42", 7, 6, []byte("hello"))
b := f.encode()
got, err := decodeFrame(b)
if err != nil {
t.Fatalf("decodeFrame: %v", err)
}
if got.ShardID != f.ShardID {
t.Errorf("ShardID: got %q want %q", got.ShardID, f.ShardID)
}
if got.Seq != f.Seq || got.PrevSeq != f.PrevSeq {
t.Errorf("seq: got (%d,%d) want (%d,%d)", got.Seq, got.PrevSeq, f.Seq, f.PrevSeq)
}
if got.Salt != f.Salt {
t.Errorf("salt mismatch")
}
if got.Cksm != f.Cksm {
t.Errorf("cksm mismatch")
}
if !bytes.Equal(got.Payload, f.Payload) {
t.Errorf("payload mismatch")
}
if !bytes.Equal(got.Bytes, b) {
t.Errorf("Bytes field should carry raw input back")
}
if err := got.Valid(); err != nil {
t.Errorf("Valid() after round-trip: %v", err)
}
}
func TestFrameChecksumDetectsTamper(t *testing.T) {
f := newFrame("s", 1, 0, []byte("payload"))
b := f.encode()
// Corrupt one payload byte.
b[len(b)-33] ^= 0x01
got, err := decodeFrame(b)
if err != nil {
t.Fatalf("decodeFrame: %v", err)
}
if err := got.Valid(); err == nil {
t.Error("expected Valid() to fail on tampered payload")
}
}
func TestFrameDecodeRejectsTruncated(t *testing.T) {
f := newFrame("s", 1, 0, []byte("abc"))
b := f.encode()
_, err := decodeFrame(b[:len(b)-1])
if err == nil {
t.Error("expected error on truncated input")
}
}
func TestFrameApplyKeyStable(t *testing.T) {
f := newFrame("s", 1, 0, []byte("x"))
k1 := f.ApplyKey()
// Re-decode and check key stability.
g, _ := decodeFrame(f.encode())
k2 := g.ApplyKey()
if k1 != k2 {
t.Error("ApplyKey must be stable across encode/decode")
}
}
func TestFrameApplyKeyUnique(t *testing.T) {
a := newFrame("s", 1, 0, []byte("x"))
b := newFrame("s", 1, 0, []byte("x"))
if a.ApplyKey() == b.ApplyKey() {
t.Error("two frames with independent salts must not share ApplyKey")
}
}
func TestFrameDecodeRejectsBadVersion(t *testing.T) {
f := newFrame("s", 1, 0, nil)
b := f.encode()
b[0] = 0xFE
if _, err := decodeFrame(b); err == nil {
t.Error("expected version-mismatch error")
}
}