mirror of
https://github.com/hanzoai/replicate.git
synced 2026-10-30 18:23:20 +00:00
zeekayandGitHub
9e9292df0d
* 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.
55 lines
1.6 KiB
Go
55 lines
1.6 KiB
Go
package replicate
|
||
|
||
import (
|
||
"encoding/hex"
|
||
"strings"
|
||
|
||
"github.com/zeebo/blake3"
|
||
)
|
||
|
||
// ShardedPrefix returns a 2-hex-char shard segment derived from a
|
||
// keyed blake3 of the input. Use it to bucket per-symbol or per-
|
||
// tenant S3 prefixes across the 256-shard space so a single S3
|
||
// partition doesn't bottleneck on AWS's ~3500 PUT/s/prefix limit.
|
||
//
|
||
// Example, sharding 12K markets at 100 PUT/s each:
|
||
//
|
||
// prefix := replicate.ShardedPrefix("AAPL")
|
||
// // prefix => "a3"
|
||
// key := fmt.Sprintf("markets/%s/AAPL/%016x.ltx", prefix, txid)
|
||
//
|
||
// 256 shards × 3500 PUT/s/shard ≈ 900K PUT/s aggregate — well above
|
||
// the cold-tier write rate even at 12K markets full-throttle.
|
||
//
|
||
// The blake3 key is a constant string so the same input always lands
|
||
// on the same shard across processes + restarts. Changing the key
|
||
// resharded the entire keyspace; treat it as a one-way migration.
|
||
func ShardedPrefix(symbol string) string {
|
||
h := blake3.NewDeriveKey("github.com/hanzoai/replicate.ShardedPrefix v1")
|
||
_, _ = h.WriteString(symbol)
|
||
var sum [1]byte
|
||
d := h.Digest()
|
||
_, _ = d.Read(sum[:])
|
||
return hex.EncodeToString(sum[:])
|
||
}
|
||
|
||
// ShardedKey is a convenience builder: returns
|
||
// "<rootSegment>/<shard>/<symbol>/<suffix>". Use it to keep the call-
|
||
// site layout consistent across producers.
|
||
func ShardedKey(rootSegment, symbol, suffix string) string {
|
||
var b strings.Builder
|
||
b.Grow(len(rootSegment) + 1 + 2 + 1 + len(symbol) + 1 + len(suffix))
|
||
if rootSegment != "" {
|
||
b.WriteString(rootSegment)
|
||
b.WriteByte('/')
|
||
}
|
||
b.WriteString(ShardedPrefix(symbol))
|
||
b.WriteByte('/')
|
||
b.WriteString(symbol)
|
||
if suffix != "" {
|
||
b.WriteByte('/')
|
||
b.WriteString(suffix)
|
||
}
|
||
return b.String()
|
||
}
|