mirror of
https://github.com/hanzoai/base.git
synced 2026-10-21 03:41:50 +00:00
zeekay
ff1ee17c4f
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.
89 lines
2.1 KiB
Go
89 lines
2.1 KiB
Go
package network
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"encoding/binary"
|
|
"sort"
|
|
"sync"
|
|
)
|
|
|
|
// router is the consistent-hash ring over live members. For a shardID the
|
|
// ring returns a deterministic ordered list of members — the first is the
|
|
// writer, the first `replication` are the shard's subset.
|
|
//
|
|
// Member churn: rebuild is cheap (sort a small slice). In k8s the operator
|
|
// keeps the BASE_PEERS list current and we rebuild on a pod-id change.
|
|
type router struct {
|
|
mu sync.RWMutex
|
|
members []NodeID
|
|
replication int
|
|
}
|
|
|
|
func newRouter(members []NodeID, replication int) *router {
|
|
r := &router{replication: replication}
|
|
r.setMembers(members)
|
|
return r
|
|
}
|
|
|
|
func (r *router) setMembers(members []NodeID) {
|
|
out := append([]NodeID(nil), members...)
|
|
sort.Slice(out, func(i, j int) bool { return out[i] < out[j] })
|
|
r.mu.Lock()
|
|
r.members = out
|
|
r.mu.Unlock()
|
|
}
|
|
|
|
// ownerOf returns the writer member for shardID.
|
|
func (r *router) ownerOf(shardID string) NodeID {
|
|
ms := r.membersFor(shardID)
|
|
if len(ms) == 0 {
|
|
return ""
|
|
}
|
|
return ms[0]
|
|
}
|
|
|
|
// membersFor returns up to `replication` members for shardID, ordered by
|
|
// rendezvous hash. Deterministic: every node computes the same list from
|
|
// the same member set.
|
|
func (r *router) membersFor(shardID string) []NodeID {
|
|
r.mu.RLock()
|
|
members := r.members
|
|
n := r.replication
|
|
r.mu.RUnlock()
|
|
|
|
if len(members) == 0 {
|
|
return nil
|
|
}
|
|
if n > len(members) {
|
|
n = len(members)
|
|
}
|
|
|
|
scored := make([]scored, 0, len(members))
|
|
for _, m := range members {
|
|
scored = append(scored, scored0(shardID, m))
|
|
}
|
|
sort.Slice(scored, func(i, j int) bool { return scored[i].score < scored[j].score })
|
|
|
|
out := make([]NodeID, 0, n)
|
|
for i := 0; i < n; i++ {
|
|
out = append(out, scored[i].id)
|
|
}
|
|
return out
|
|
}
|
|
|
|
type scored struct {
|
|
id NodeID
|
|
score uint64
|
|
}
|
|
|
|
// scored0 computes the rendezvous score (lower = preferred) as the first
|
|
// 8 bytes of SHA-256(shardID || member). Classic HRW hashing.
|
|
func scored0(shardID string, m NodeID) scored {
|
|
h := sha256.New()
|
|
h.Write([]byte(shardID))
|
|
h.Write([]byte{0})
|
|
h.Write([]byte(m))
|
|
sum := h.Sum(nil)
|
|
return scored{id: m, score: binary.BigEndian.Uint64(sum[:8])}
|
|
}
|