mirror of
https://github.com/hanzoai/base.git
synced 2026-10-21 03:41:50 +00:00
zeekay
2f97904e71
Phase 1 of the Base Network archive tier: witness-only validator
appends quasar-finalised WAL frames to object storage as per-shard
.lbn segments. PITR replay recovers SQLite state at any txseq.
Scope:
- network/archive.go Archive interface + URL dispatch
(s3://…|gs://…|off)
- network/segment.go .lbn encoder/decoder. Magic "LBN1",
CRC32-IEEE footer. Forwards-compat via
magic bump on any format change.
- network/archive_writer.go Goroutine-per-shard buffer with
size+time flush, exponential-backoff
retry against RetryDeadline, and
never-drop re-buffering on timeout.
- network/archive_s3.go minio-go/v7 backend. Works against AWS
S3 and any MinIO-protocol endpoint —
hanzoai/s3 self-hosted is one such.
Static creds from AWS_ACCESS_KEY_ID /
SECRET, else IAM/IRSA via
credentials.NewIAM().
- network/archive_gcs.go cloud.google.com/go/storage backend.
ADC by default; Workload Identity in
k8s, GOOGLE_APPLICATION_CREDENTIALS
for local dev.
- hack/pitr-restore.go CLI replay tool. Usage:
pitr-restore --archive gs://…/svc \
--shard <id> --to-seq <n> \
--out file.db
Writes a SQLite file with one row per
frame; Base's real apply path replays
into the schema.
Segment layout: "LBN1" | shardIDLen u16 | shardID | startSeq u64 |
frameCount u32 | (frameLen u32 | frameBytes){N} | crc32 footer.
PQ signature on each frame is preserved byte-for-byte — the
archive never re-signs.
Storage key layout per docs/NETWORK.md:
<svc>/<shard>/<16-digit-seq-prefix>/<20-digit-start-seq>.lbn
Config knobs (defaults): SegmentTargetBytes=8MiB, FlushInterval=10s,
RetryDeadline=5m. lagBytes gauge reflects backlog awaiting flush.
Metrics binding: BindArchiveMetrics(*Metrics) wires archive callbacks
onto the core Metrics.ArchiveLagBytes gauge + ApplyErrors counter.
Archive stays decoupled from prometheus/client_golang.
Tests: 12 archive-specific tests pass alongside the 20 network tests
from Phase 1 (32/32 total). Integration tests (`//go:build integration`)
cover 1000-frame S3 round-trip against local MinIO and PITR CLI
restore of 100 frames at cut-off seq 50 producing a SQLite file
with exactly 50 rows.
70 lines
1.6 KiB
Go
70 lines
1.6 KiB
Go
//go:build integration && gcs
|
|
|
|
package network
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestGCSIntegrationRoundTrip(t *testing.T) {
|
|
if os.Getenv("STORAGE_EMULATOR_HOST") == "" {
|
|
t.Skip("STORAGE_EMULATOR_HOST unset; start fake-gcs-server per the header of archive_integration_test.go")
|
|
}
|
|
bucket := os.Getenv("BASE_ARCHIVE_TEST_BUCKET")
|
|
if bucket == "" {
|
|
bucket = "base-archive-gcs-test"
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
|
defer cancel()
|
|
|
|
a, err := NewArchive(ctx, ArchiveConfig{
|
|
URL: fmt.Sprintf("gs://%s", bucket),
|
|
SegmentTargetBytes: 4096,
|
|
FlushInterval: 100 * time.Millisecond,
|
|
RetryDeadline: 30 * time.Second,
|
|
}, "svc-gitest", nil)
|
|
if err != nil {
|
|
t.Fatalf("NewArchive gs: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = a.Close() })
|
|
|
|
const n = 200
|
|
for i := uint64(1); i <= n; i++ {
|
|
f := newFrame("shard-g", i, i-1, []byte(fmt.Sprintf("p%d", i)))
|
|
if err := a.Append(ctx, "shard-g", i, f.encode()); err != nil {
|
|
t.Fatalf("append: %v", err)
|
|
}
|
|
}
|
|
if err := a.Close(); err != nil {
|
|
t.Fatalf("close: %v", err)
|
|
}
|
|
|
|
a2, err := NewArchive(ctx, ArchiveConfig{URL: fmt.Sprintf("gs://%s", bucket)}, "svc-gitest", nil)
|
|
if err != nil {
|
|
t.Fatalf("reopen: %v", err)
|
|
}
|
|
t.Cleanup(func() { _ = a2.Close() })
|
|
|
|
it, err := a2.Range(ctx, "shard-g", 1, n)
|
|
if err != nil {
|
|
t.Fatalf("range: %v", err)
|
|
}
|
|
var got uint64
|
|
for f, err := range it {
|
|
if err != nil {
|
|
t.Fatalf("iter: %v", err)
|
|
}
|
|
got++
|
|
if f.Seq != got {
|
|
t.Fatalf("seq: want %d got %d", got, f.Seq)
|
|
}
|
|
}
|
|
if got != n {
|
|
t.Fatalf("want %d frames, got %d", n, got)
|
|
}
|
|
}
|