mirror of
https://github.com/hanzoai/base.git
synced 2026-10-21 03:41:50 +00:00
zeekay
658d7d181d
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.
90 lines
2.9 KiB
Go
90 lines
2.9 KiB
Go
package core
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
|
|
"github.com/hanzoai/base/network"
|
|
"github.com/hanzoai/dbx"
|
|
sqlite "modernc.org/sqlite"
|
|
)
|
|
|
|
// baseNetwork is a local alias for network.Network so the BaseApp field can
|
|
// use a short name without shadowing the imported package.
|
|
type baseNetwork = network.Network
|
|
|
|
// attachNetwork is the one hook point into BaseApp that network integration
|
|
// requires. It is called at the end of initDataDB — standalone mode (the
|
|
// default) is a fast-path return with zero cost.
|
|
//
|
|
// When BASE_NETWORK=quasar, it builds a Network from the environment,
|
|
// starts it, and installs a commit hook on one pooled concurrentDB
|
|
// connection per shard key. Today we register a single hook against the
|
|
// default shard ("_") so baseline replication works; multi-shard
|
|
// registration is driven by per-request shard resolution upstream.
|
|
func (app *BaseApp) attachNetwork(ctx context.Context) error {
|
|
net, err := network.FromEnv()
|
|
if err != nil {
|
|
return fmt.Errorf("network from env: %w", err)
|
|
}
|
|
if !net.Enabled() {
|
|
return nil
|
|
}
|
|
if err := net.Start(ctx); err != nil {
|
|
return fmt.Errorf("network start: %w", err)
|
|
}
|
|
app.network = net
|
|
|
|
db := app.concurrentDB
|
|
if db == nil {
|
|
return nil
|
|
}
|
|
return installCommitHook(ctx, db, net, "_")
|
|
}
|
|
|
|
// installCommitHook reserves one *sql.Conn from the pool and wires the
|
|
// network commit hook onto its driver conn. The conn is retained for the
|
|
// lifetime of the process so the hook outlives pool churn.
|
|
//
|
|
// Failure here is soft: we log but do not block Bootstrap — a broken
|
|
// transport is less harmful than a refusal to start.
|
|
func installCommitHook(ctx context.Context, db dbx.Builder, net network.Network, shardID string) error {
|
|
sqlDB, ok := unwrapSQL(db)
|
|
if !ok {
|
|
return fmt.Errorf("network: db is not *sql.DB (got %T)", db)
|
|
}
|
|
raw, err := sqlDB.Conn(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("network: reserve conn: %w", err)
|
|
}
|
|
return raw.Raw(func(driverConn any) error {
|
|
reg, ok := driverConn.(sqlite.HookRegisterer)
|
|
if !ok {
|
|
return fmt.Errorf("network: driver conn %T has no commit hook", driverConn)
|
|
}
|
|
return net.InstallWALHook(commitHookAdapter{reg}, shardID)
|
|
})
|
|
}
|
|
|
|
// commitHookAdapter narrows sqlite.HookRegisterer to the network package's
|
|
// own HookRegisterer shape, which takes func() int32 directly rather than
|
|
// the modernc-typed CommitHookFn. This adapter is the only place in Base
|
|
// that the sqlite driver type leaks into the integration surface.
|
|
type commitHookAdapter struct{ inner sqlite.HookRegisterer }
|
|
|
|
func (c commitHookAdapter) RegisterCommitHook(cb func() int32) {
|
|
c.inner.RegisterCommitHook(sqlite.CommitHookFn(cb))
|
|
}
|
|
|
|
// unwrapSQL pulls the underlying *sql.DB out of a dbx.Builder. dbx.DB
|
|
// exposes DB() explicitly; other builders (transactions, wrappers) return
|
|
// false so the caller can degrade gracefully.
|
|
func unwrapSQL(b dbx.Builder) (*sql.DB, bool) {
|
|
type hasDB interface{ DB() *sql.DB }
|
|
if v, ok := b.(hasDB); ok {
|
|
return v.DB(), true
|
|
}
|
|
return nil, false
|
|
}
|