mirror of
https://github.com/hanzoai/replicate.git
synced 2026-10-30 18:23:20 +00:00
zeekayandGitHub
f3e1dd3d6f
* peer: add ReplicaClient + Server for ZAP-based LTX streaming
ZAP-native peer-to-peer LTX frame streaming for W=2 hot path. Owner
pushes one frame, standby fsyncs + ACKs same-AZ in ~200us. Replaces
S3 PUT in PlaceOrder ACK (50ms ceiling) with peer ACK (~1ms p99).
Wire protocol (4 opcodes in caller-supplied namespace):
0xA020 OpWALFrame [8B mkt][4B txid][4B len][N body]
0xA021 OpWALFrameAck [1B code][8B mkt][4B txid]
0xA022 OpSnapshotReq [8B mkt][8B from_txid]
0xA023 OpSnapshotResp [1B code][8B mkt][8B txid][N body]
ReplicaClient satisfies replicate.ReplicaClient so the existing
DB/Replica plumbing routes a peer replica identically to S3. LTXFiles
+ OpenLTXFile + DeleteLTXFiles are no-ops by design: peer is the live
channel, not the backfill source. New-pod cold-start uses Snapshot
+ then tails Sevenframes; that path stays on S3.
Server registers OpWALFrame + OpSnapshotReq handlers via an injected
Dispatcher interface so the peer package is namespace-agnostic. ATS's
atsrpc.Dispatcher satisfies the contract.
6 unit tests (race-clean):
- TestReplicaClient_WriteAck — happy path, body integrity
- TestReplicaClient_ApplyFailureSurfacesAsError — fsync error -> AckError
- TestReplicaClient_PeerCloseRejects — closed server returns ErrCodeNotStandby
- TestReplicaClient_SnapshotResp — OpSnapshotReq round trip
- TestRegister_RejectsDuplicate — last-write-wins is rejected
- TestWireRoundTrip — encode/decode symmetry
* sync_policy: SyncPolicy enum for W=2 quorum semantics
Type-level expression of the W=2 quorum contract. SyncQuorumW2 means:
return after local fsync + one peer ACK; remaining replicas (S3) lag
async. SyncAll is the historical single-or-multi replica behavior;
SyncLocalOnly is dev-mode only.
Caller (ATS Journal) consumes the enum to choose its push strategy.
The replicate.DB doesn't need to interpret it directly — the journal
sequences peer push before triggering S3 sync — so we keep DB as the
single-replica abstraction and let the journal own quorum logic.
1 unit test covering String() across all values + the unknown branch.
* sharded_prefix: 256-shard helper for S3 cold-tier writes
AWS S3 throttles at ~3500 PUT/s per prefix. 12K markets at 100 PUT/s
each (cold tier) exceeds a single-prefix budget. ShardedPrefix(symbol)
returns one of 256 stable 2-hex shards (blake3-keyed) so the cluster
spreads cold-tier writes across the partition space:
256 shards x 3500 PUT/s = ~900K PUT/s aggregate cluster-wide.
ShardedKey is the layout-canonical builder:
ShardedKey('markets', 'AAPL', '0123.ltx')
=> 'markets/a3/AAPL/0123.ltx'
5 unit tests:
- Deterministic (same input -> same shard across runs)
- Distribution (>200 of 256 buckets covered over 4000 symbols)
- Layout (4-segment shape)
- NoRoot / NoSuffix edge cases
* deps: bump Go to 1.26.4 across go.mod, Dockerfiles, GH Actions
Workspace-wide sync. luxfi/node already shipped on 1.26.4 in v1.30.6
(commit 121aca1fa9); this is the cross-repo catch-up.
* replicate: sync.Pool for LTX frame buffers
WriteLTXFile previously did io.ReadAll(r) per push — at NYSE-class
throughput (~80M pushes/sec aggregate across a 16-pod cluster), that
allocates O(frame_size_bytes) on every Push and churns the allocator.
readPooled reuses up to 64KB byte slices via sync.Pool. L0 frames
(typical 4-32KB) fit in one pooled buffer; oversize compaction
outputs (L1+) bypass the pool with a single io.ReadAll fallback so
the pool's working set stays bounded.
Steady-state allocations per Push drop from O(frame_size) to O(1).
GC pressure drops linearly with batch size (one frame per batch in
the new ATS batched-accept path, not per order).
64KB ceiling matches the L0 frame cap baked into the SQLite WAL → LTX
pipeline (4KB page × ~16 pages per checkpointed WAL frame).
* peer: panic recovery + pool oversize drop (F-22 F-05)
F-22 CRITICAL — wrap onWALFrame and onSnapshotReq in defer/recover.
A panic inside FrameSink.Apply (e.g. nil-deref while decoding a
malformed LTX body) previously propagated up through the ZAP
dispatch goroutine and DoS'd the standby's peer receive surface.
Now: panic is logged with full stack and the handler returns
ErrCodeServerInternal — owner retries, standby keeps serving.
F-05 HIGH — readPooled grew the cached buffer past ltxPoolCap on
the first outlier frame (>64KB compaction output). Subsequent
Get/Put cycles preserved the inflated cap, putting every steady-
state push back through the outlier branch. Now: putPooled drops
any buffer with cap > ltxPoolCap; the pool's working-set stays
bounded.
Tests: TestPeerServer_PanicIn{WALFrame,Snapshot}HandlerDoesNotKillServer,
TestPeerFramePool_DropsOversizeBuffers, TestPutPooled_DropsOversize.
All pass under -race.
88 lines
2.2 KiB
Go
88 lines
2.2 KiB
Go
package replicate
|
|
|
|
import (
|
|
"regexp"
|
|
"strings"
|
|
"testing"
|
|
)
|
|
|
|
var hexShard = regexp.MustCompile(`^[0-9a-f]{2}$`)
|
|
|
|
func TestShardedPrefix_Deterministic(t *testing.T) {
|
|
for _, sym := range []string{"AAPL", "MSFT", "NVDA", "BTC", "ETH"} {
|
|
a := ShardedPrefix(sym)
|
|
b := ShardedPrefix(sym)
|
|
if a != b {
|
|
t.Errorf("%s: prefix not stable %q vs %q", sym, a, b)
|
|
}
|
|
if !hexShard.MatchString(a) {
|
|
t.Errorf("%s: prefix %q not 2-hex", sym, a)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestShardedPrefix_Distribution(t *testing.T) {
|
|
// Synthetic load: hit 4000 unique symbols, count shards. Each of
|
|
// the 256 buckets should appear at least once on a uniformly-keyed
|
|
// hash; we don't enforce uniform spread (the test would be flaky)
|
|
// but we do enforce coverage of >200 buckets so we know the hash
|
|
// is actually mixing.
|
|
buckets := map[string]int{}
|
|
for i := 0; i < 4000; i++ {
|
|
buckets[ShardedPrefix(symbolize(i))]++
|
|
}
|
|
if len(buckets) < 200 {
|
|
t.Errorf("expected >200 unique shard buckets, got %d", len(buckets))
|
|
}
|
|
}
|
|
|
|
func TestShardedKey_Layout(t *testing.T) {
|
|
key := ShardedKey("markets", "AAPL", "000000000000007b.ltx")
|
|
parts := strings.Split(key, "/")
|
|
if len(parts) != 4 {
|
|
t.Fatalf("expected 4 segments, got %d (%v)", len(parts), parts)
|
|
}
|
|
if parts[0] != "markets" {
|
|
t.Errorf("segment[0]: %q", parts[0])
|
|
}
|
|
if !hexShard.MatchString(parts[1]) {
|
|
t.Errorf("segment[1] not 2-hex: %q", parts[1])
|
|
}
|
|
if parts[2] != "AAPL" {
|
|
t.Errorf("segment[2]: %q", parts[2])
|
|
}
|
|
if parts[3] != "000000000000007b.ltx" {
|
|
t.Errorf("segment[3]: %q", parts[3])
|
|
}
|
|
}
|
|
|
|
func TestShardedKey_NoRoot(t *testing.T) {
|
|
key := ShardedKey("", "AAPL", "file.ltx")
|
|
if strings.HasPrefix(key, "/") {
|
|
t.Errorf("unexpected leading slash: %q", key)
|
|
}
|
|
if !strings.HasSuffix(key, "/AAPL/file.ltx") {
|
|
t.Errorf("layout: %q", key)
|
|
}
|
|
}
|
|
|
|
func TestShardedKey_NoSuffix(t *testing.T) {
|
|
key := ShardedKey("markets", "AAPL", "")
|
|
if strings.HasSuffix(key, "/") {
|
|
t.Errorf("trailing slash: %q", key)
|
|
}
|
|
if !strings.HasSuffix(key, "/AAPL") {
|
|
t.Errorf("layout: %q", key)
|
|
}
|
|
}
|
|
|
|
func symbolize(i int) string {
|
|
// Generate distinct 5-char codes: covers ~11.8M for the 4000 used.
|
|
var b [5]byte
|
|
for k := 0; k < 5; k++ {
|
|
b[k] = byte('A' + (i % 26))
|
|
i /= 26
|
|
}
|
|
return string(b[:])
|
|
}
|