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.
21 lines
403 B
Go
21 lines
403 B
Go
package replicate
|
|
|
|
import "testing"
|
|
|
|
func TestSyncPolicy_String(t *testing.T) {
|
|
cases := []struct {
|
|
p SyncPolicy
|
|
want string
|
|
}{
|
|
{SyncAll, "all"},
|
|
{SyncQuorumW2, "quorum_w2"},
|
|
{SyncLocalOnly, "local_only"},
|
|
{SyncPolicy(99), "unknown"},
|
|
}
|
|
for _, c := range cases {
|
|
if got := c.p.String(); got != c.want {
|
|
t.Errorf("SyncPolicy(%d).String() = %q want %q", c.p, got, c.want)
|
|
}
|
|
}
|
|
}
|