44b768b8c6
End the "replicated XOR sharded" split: S shard groups, each a
replication group at RF with its own elected leader, leaders balanced
across nodes; any gateway hash-routes.
- One unified write surface: /items,/embeddings,/signals hash-route to
the owning shard group's leader (ShardRouter FNV-1a) AND replicate at
RF. x-tidal-ack/x-tidal-seq, quorum await, NotLeader/QuorumTimeout are
per-group; NotLeader names the group.
- Rebalance verbs (L3): POST /cluster/shards/{id}/transfer (fenced
leadership move) + /cluster/shards/{id}/replicas (add/remove replica).
A ?shard= selector threads through every per-shard admin verb and is
propagated on intra-group forwards (ShardReplica::admin_path). S=1 is
byte-for-byte (no selector, no shard in NotLeader body).
- Tier-3 exit gate (cluster_sharding.rs): 3 nodes × 3 shards × RF=3 over
real OS processes — SIGKILL a node under ack=quorum load → only its
shard-leaderships re-elect, reads never stop, zero acked loss across
random kill points; plus a rebalance-verb test. Harness:
MultiProcCluster::start_sharded.
- tidal-stress drives the single path (WritePath::Leader|Sharded gone),
spreading writes round-robin across gateways or pinning --leader-url.
- Throughput: local 3×3 sustains 3,000 quorum signal-writes/s @ 0% err,
~30% CPU, lag ~0 (generator-bound). ≥5,000/s + ≥2.5× scaling is Ref-A.
Known follow-up (tracked): per-group-aware node readiness and cross-node
read fan-out under PARTIAL placement.
305 lines
19 KiB
Markdown
305 lines
19 KiB
Markdown
# m11p6 — Sharding × Replication + Rebalancing (COMPLETE — 2026-06-13)
|
||
|
||
Phase spec and exit gate: [docs/roadmap-to-cluster.md §4/m11p6](../../roadmap-to-cluster.md).
|
||
Closes the ROADMAP **write-scaling** gap ("EITHER replicated (1 leader for
|
||
everything) OR sharded (no replication)").
|
||
Predecessors: p1 (ship queue/perf), p2 (one log), p3 (quorum/commit-index),
|
||
p4 (election/fencing), p5 (membership/discovery/elasticity).
|
||
|
||
**Goal:** writes scale horizontally **without** giving up replication. The
|
||
"replicated XOR sharded" split ends: `S` shard groups, each a replication group
|
||
at RF (default 3) with its **own WAL/relay/commit-index and its own elected
|
||
leader**, leaders balanced across nodes; regions demote to placement labels;
|
||
any gateway hash-routes to the shard leader (write) or nearest replica (read);
|
||
rebalancing = snapshot + stream + fenced cutover.
|
||
|
||
## The load-bearing realization (why this is tractable, not a rewrite)
|
||
|
||
A single `TidalDb` is **already a complete, self-contained shard** (scale-arch
|
||
spec §1; single-process mode already runs N real `TidalDb` in one process). And
|
||
**today's cluster is exactly "1 shard × RF=N"** — every region replicates one
|
||
log under one leader. So m11p6 = **run today's whole single-shard cluster
|
||
machinery `N` times, once per data-shard group**, where each group spans only
|
||
its replica nodes, in its own data subdir + gRPC port. The data-shard number
|
||
`S` is purely a *gateway-routing + directory + port* concern; **inside a group,
|
||
"shard id == region id" is preserved unchanged**.
|
||
|
||
Consequences:
|
||
- **Engine replication (relay/commit/election/ship/receiver/membership): NO
|
||
semantic change** — each runs per-group exactly as today's whole cluster.
|
||
- **`tidal-net`: NO change** — each group owns its own `GrpcTransport` bound to
|
||
a distinct port, peering only with that group's replica nodes.
|
||
- **`S=1` is byte-for-byte today's behavior** — every existing tier-3 suite
|
||
(quorum, membership, election, region, runbook, chaos) runs in `S=1` mode and
|
||
stays green by construction.
|
||
- The work concentrates in **`tidal-server`**: decompose the per-shard
|
||
machinery out of `RegionClusterState` into `ShardReplica`; the node hosts
|
||
`BTreeMap<ShardId, Arc<ShardReplica>>`; add hash-routing to the right group's
|
||
leader. Plus topology schema, route unification, rebalancing, stress/harness,
|
||
the exit gate, docs.
|
||
|
||
## Design (as adopted)
|
||
|
||
### 1. Two types: process node + shard replica
|
||
|
||
> **As-built note (decided during L1):** the §1-era naming below was superseded.
|
||
> The process node is a NEW type **`ClusterNode`** (not "keep the name
|
||
> `RegionClusterState`"); the former `RegionClusterState` was **renamed to
|
||
> `ShardReplica`** (it already WAS one group's machinery). The responsibilities
|
||
> are exactly as described — only the node's type name changed from the planned
|
||
> `RegionClusterState` to `ClusterNode`.
|
||
|
||
- **`ClusterNode` (NEW process node)** = the **process node** and the axum
|
||
`State`. Owns the node identity (`region`/`region_name`), the gateway
|
||
`ShardRouter` + `node_http` map, the shared forward client, and
|
||
`groups: BTreeMap<ShardId, Arc<ShardReplica>>` (the groups this node hosts)
|
||
plus a `placement` map of every group (to forward writes for groups it does
|
||
NOT host). The HTTP handlers stay free functions over
|
||
`State<Arc<ClusterNode>>`; each hash-routes the target group and either calls
|
||
a local `ShardReplica` method or forwards to that group's leader.
|
||
- **`ShardReplica` (the renamed `RegionClusterState`)** = one shard group's full
|
||
replication machinery, essentially the per-shard fields + methods of the old
|
||
`RegionClusterState`:
|
||
`db: Arc<TidalDb>`, `ship_feed`, `ship_queue`, `commit` + `commit_watch`
|
||
bridge, `election_runtime`/`election_store`/`election_boot`,
|
||
`membership: MembershipView` + `membership_store`, `stream_baseline`,
|
||
snapshot source + `reseed_marker_store`, `leader: RwLock<Option<RegionId>>`,
|
||
`activation_prev`, `partitioned`, `deferred_retires`, `ack_default`,
|
||
`quorum_timeout`, and its **own `GrpcTransport`** bound to this group's port.
|
||
Construction = today's `RegionClusterState::new` body, parameterized by
|
||
`(shard, replica nodes, bootstrap leader, group data subdir, group grpc port)`.
|
||
|
||
### 2. Shard identity, ports, and data dirs
|
||
|
||
- **In-group identity is region-id based** (preserved from today). `ShardReplica`
|
||
for data-shard `S` on node `N` opens `TidalDb` with
|
||
`NodeConfig{ shard_id: ShardId(N.region_id), peer_shards: <other group-S
|
||
nodes' region ids>, .. }`. Two groups on one node use the same `shard_id`
|
||
value but live in separate data dirs + separate transports, so nothing
|
||
collides. The data-shard `S` never enters the replication wire.
|
||
- **gRPC ports:** each replica entry may set an explicit `grpc_addr`/`grpc_bind`
|
||
per `(node, shard)`; otherwise derive `node.base_port + S`. Legacy `S=0` →
|
||
offset 0 → the region's declared address verbatim.
|
||
- **Data dirs:** explicit `shards:` → `<data_dir>/shard-{S:05}/` per hosted
|
||
group. Legacy (synthesized 1-shard) → `<data_dir>` **verbatim** (existing
|
||
on-disk clusters restart unchanged).
|
||
- **`/metrics`:** the node owns ONE listener on `metrics_addr`; the engine's
|
||
per-`TidalDb` metrics HTTP server is suppressed in cluster mode. The node
|
||
renders each group's `cluster_metrics` with a `shard="S"` label and the
|
||
per-shard engine metrics likewise.
|
||
|
||
### 3. Topology schema (backward compatible)
|
||
|
||
`TopologySpec` gains optional `shards: Option<Vec<ShardSpec>>` and `RegionSpec`
|
||
gains optional `zone: Option<String>` (placement/read-affinity label), both
|
||
`#[serde(default)]`.
|
||
|
||
```yaml
|
||
nodes: # `regions:` still accepted as the node list
|
||
- { name: us-east, grpc_addr: ..., http_addr: ..., zone: az-a }
|
||
shards: # NEW — optional
|
||
- id: 0
|
||
leader: us-east # term-0 / preferred leader (balances placement)
|
||
replicas: # nodes hosting this group's RF replicas
|
||
- { node: us-east } # grpc_addr/grpc_bind optional → derive base+S
|
||
- { node: eu-west }
|
||
- { node: ap-south }
|
||
- { id: 1, leader: eu-west, replicas: [...] }
|
||
- { id: 2, leader: ap-south, replicas: [...] }
|
||
```
|
||
|
||
**Absent `shards:` ⇒ legacy synthesis:** one `ShardSpec{ id: 0,
|
||
leader: topology.leader, replicas: <every region, grpc verbatim> }`. This is
|
||
literally today's "1 shard × RF=all-regions," so the model unifies cleanly.
|
||
|
||
Validation: shard ids dense & unique; each `leader`/`replica.node` names a
|
||
declared region; RF ≥ 1; a node's shards do not collide on derived ports.
|
||
|
||
### 4. Routing (the new gateway layer)
|
||
|
||
Any node, any write to entity `E`:
|
||
1. `S = ShardRouter.route(E)` (the engine's FNV-1a hash; the same router the
|
||
`/sharded/*` path already uses).
|
||
2. If this node hosts group `S` **and leads it** → apply locally on
|
||
`ShardReplica[S]` (the existing stage→complete→ship→await-quorum path).
|
||
3. If this node hosts group `S` but follows → forward to group `S`'s leader
|
||
(resolved from `ShardReplica[S].leader`), to the leader node's `http_addr`.
|
||
4. If this node does not host group `S` → forward to any replica node of `S`
|
||
(from topology); that node routes to the leader. Term fencing + the
|
||
`NotLeader` retry bound the forward chain.
|
||
|
||
Reads route to a local replica of `S` if present, else nearest (zone) / any
|
||
replica. **Scatter-gather merges over shard GROUPS:** one replica per group,
|
||
totals **summed** (entity-sharded groups are disjoint — no dedup-by-max), the
|
||
existing degraded/`unavailable_shards`/deadline semantics retained.
|
||
|
||
**Route unification:** `/items`, `/embeddings`, `/signals` become the shard-
|
||
routed surface (they hash-route instead of forwarding to a single global
|
||
leader); `/sharded/*` is retained as an alias of the same path. `x-tidal-ack`,
|
||
`x-tidal-seq`, quorum await, and `NotLeader`/`QuorumTimeout` all become
|
||
per-shard; `NotLeader` names the shard and its leader.
|
||
|
||
### 5. Leadership balance
|
||
|
||
`ShardSpec.leader` is each group's term-0 / preferred leader; the operator (and
|
||
the test harness) set shard `i`'s leader to node `i` for balanced placement.
|
||
Each group boot-classifies its own leader exactly like today's `topology.leader`,
|
||
but per group. On failover, the group's election picks the surviving max-applied
|
||
voter (m11p4 protocol, unchanged); only the dead node's groups elect. An
|
||
optional leadership-transfer-back-to-preferred is the rebalance path, not the
|
||
availability mechanism.
|
||
|
||
### 6. Rebalancing (operator-triggered)
|
||
|
||
Shard move / replica change reuses m11p5 verbatim, **per group**: a conf-change
|
||
on the group's own log adds a Learner replica (a new node, or an existing node
|
||
gaining a replica of group `S`), which catches up via `FetchSnapshot` +
|
||
`StreamSegments` (per-group, already per-instance once §1 lands), auto-promotes
|
||
Learner→Voter, then the operator may transfer leadership; removing a replica is
|
||
the m11p5 fenced removal on that group's log. Endpoints:
|
||
`POST /cluster/shards/{id}/replicas` (add/remove) and
|
||
`POST /cluster/shards/{id}/transfer`. Per the roadmap, **automatic rate-limited
|
||
rebalancing is explicitly "later"** — m11p6 ships the operator verbs.
|
||
|
||
## Exit gate (from the roadmap)
|
||
|
||
- 3 shards × RF=3: **≥5,000 quorum signals/s** (≥2.5× single-shard p3).
|
||
- Kill any node → **only its shard-leaderships move (<10 s), reads never stop**.
|
||
- `tidal-stress` sharded-vs-replicated comparison **collapses into one path**.
|
||
|
||
## Layer plan (each keeps the workspace green)
|
||
|
||
- **L0** — this doc + topology schema (`shards:`/`zone:` + validation + legacy
|
||
synthesis). `S=1` unchanged.
|
||
- **L1** — extract `ShardReplica`; node hosts a one-entry map; suppress engine
|
||
metrics server in cluster mode + node aggregates. `S=1` byte-for-byte.
|
||
- **L2** — multi-shard construction + gateway hash-routing + per-shard forward +
|
||
route unification + scatter-gather over groups.
|
||
- **L3** — operator rebalancing verbs.
|
||
- **L4** — tidal-stress path collapse + nodes×shards tier-3 harness +
|
||
`cluster_sharding.rs` exit gate (run for real).
|
||
- **L5** — docs (runbook/monitoring/roadmap/CHANGELOG/k8s/spec) + memory.
|
||
|
||
## Status
|
||
|
||
- [x] L0 topology schema (`shards:`/`zone:` + resolver + validation; legacy
|
||
synthesis) + this doc — 20 topology tests green incl. 5 new shard tests.
|
||
- [x] L1 `RegionClusterState` → `ShardReplica` rename + group-parameterized
|
||
`ShardReplica::new(.., group, enable_metrics)` (leader/peers/voters/data-
|
||
dir/metrics from the resolved group; in-group identity stays region-based).
|
||
**S=1 byte-for-byte green** (116 lib + cluster_region 11 + cluster_routes 6
|
||
+ cluster_grpc 2). NB: the `ClusterNode` process wrapper + multi-shard
|
||
hosting moved to L2 (inseparable from the routing/handler rewiring).
|
||
- [x] L2 **DONE + green** (the sharding × replication data plane). `ClusterNode`
|
||
(the axum State + process handle) hosts `BTreeMap<ShardId, Arc<ShardReplica>>`,
|
||
built by `ClusterNode::new(topology, region, schema, profiles, data_dir,
|
||
hlc)` (resolves groups; opens one `ShardReplica` per hosted group — own data
|
||
subdir `shard-<id>/` + derived/explicit gRPC port for `S>1`; node data dir
|
||
verbatim + metrics owner for `S=1`). `ServeState for ClusterNode` starts
|
||
every group's election driver and shuts each down via `Arc::try_unwrap`.
|
||
Entity writes (`/items`,`/embeddings`,`/signals`,`/hardnegs`) hash-route via
|
||
`route_entity` → local replica (existing leader/forward path) or
|
||
`forward_to_group_node` (remote group). Reads (`/feed`,`/search`) scatter
|
||
in-process over hosted groups and merge (sum totals, score-sort — S=1
|
||
unchanged). `/cluster/status/local` gains a per-shard `shards[]` array;
|
||
`region_health` aggregates all groups.
|
||
**Verified:** in-process **2 nodes × 2 shards × RF=2** test
|
||
(`region_sharded_writes_route_per_shard_and_reads_scatter`) — writes route
|
||
per-shard (A forwards shard-1 writes to B), each shard replicates to its
|
||
follower, reads scatter over both groups and return all items from BOTH
|
||
shards, over real gRPC. **S=1 byte-for-byte:** 116 lib + cluster_region 12 +
|
||
cluster_routes 6 + cluster_grpc 2 + tier-3 `cluster_multiproc` 5 (real OS
|
||
processes: replication <2s, leader-crash failover <10s). clippy `-D` + fmt
|
||
clean. **Deferred to L2-followups (tracked):** per-shard admin verbs
|
||
(`?shard=` on promote/heal/partition/etc.) and `/sharded/*`→unified-path
|
||
aliasing — currently the admin verbs + `/sharded/*` target the first hosted
|
||
group (correct for `S=1`; multi-shard admin needs the selector).
|
||
|
||
**Known L2 limitations — `S>1` only, FULL-placement assumed (tracked for L4,
|
||
NOT silently dropped):** these are correct for `S=1` and for the headline
|
||
exit-gate shape (3×3 RF=3, where every node hosts every group), and the writes
|
||
proven by the 2×2 test are unaffected — but a genuinely PARTIAL placement (a
|
||
node hosting a strict subset of groups) hits them:
|
||
1. **Unified reads are local-only.** `/feed`//`/search` scatter over the
|
||
groups THIS node hosts (`hosted_dbs`); a node that does not host every group
|
||
returns a strict-subset corpus with a 200 and no `degraded` flag. Complete
|
||
only under full placement; cross-node read fan-out (reuse the `/sharded/*`
|
||
`HttpShardContext`) is the L4 item. Use `/sharded/*` for partial placements.
|
||
2. **No cross-shard diversity re-rank.** The scatter merge keeps each group's
|
||
own diversity pass then score-merges; it does not re-run diversity ACROSS
|
||
groups, so an `S>1` `/feed` can be less diverse than `S=1`. Summing totals
|
||
over disjoint groups is exact for cardinality; a cross-shard MMR re-rank
|
||
(engine entry point) is the L4 follow-up.
|
||
3. **`S>1` observability is one-group-deep.** Only the metrics-owner group
|
||
binds the engine `/metrics` listener; the other hosted groups' per-shard
|
||
`cluster_metrics` (ship/relay/commit/quorum) are not rendered. The
|
||
kill-node gate uses `/cluster/status/local` (and now `/cluster/status`)
|
||
`shards[]`, which IS per-group; node-level per-shard `/metrics` rendering
|
||
(design §2) is the L4/p8 item.
|
||
- [x] L3 rebalancing verbs (operator shard move; reuse m11p5 per-group):
|
||
`POST /cluster/shards/{id}/transfer` (the fenced-transfer machinery scoped
|
||
to one group via `?shard=`) + `POST /cluster/shards/{id}/replicas`
|
||
(add/remove = the m11p5 join / fenced-removal per group), AND the `?shard=`
|
||
selector wired through every per-shard admin verb
|
||
(promote/heal/partition/catchup/reseed/members/join) with the selector
|
||
PROPAGATED on every intra-group forward/broadcast (`ShardReplica::admin_path`)
|
||
so the receiving sibling targets the same group; `NotLeader` now names the
|
||
group. S=1 stays byte-for-byte (no selector emitted).
|
||
- [x] L4 tidal-stress path collapse (the `WritePath::Leader|Sharded` split is
|
||
gone — one hash-routed + replicated `/signals`//`/items`//`/embeddings`
|
||
surface) + nodes×shards tier-3 harness (`MultiProcCluster::start_sharded`
|
||
+ per-(node,shard) ports + `shards:` emission + `agreed_shard_leaders`)
|
||
+ `cluster_sharding.rs` exit gate (3 nodes × 3 shards × RF=3, real OS
|
||
processes — recorded below). NB: the ≥5,000/s throughput sub-gate is a
|
||
Ref-A line item — k3s access has blocked the Ref-A runs since p1 (same
|
||
standing caveat); the local figure is recorded below.
|
||
- [x] L5 docs (runbook/monitoring/roadmap/CHANGELOG/spec) + memory
|
||
|
||
## Exit-gate evidence
|
||
|
||
**Data-plane correctness (local, in-process, real gRPC) — DONE.** The 2×2 test
|
||
proves the replicated-XOR-sharded split is gone: a single gateway routes writes
|
||
to the owning shard's leader (forwarding cross-node), each shard group replicates
|
||
independently, and corpus reads merge across groups. The per-shard election +
|
||
quorum + commit machinery is the unchanged m11p4/m11p3 code instantiated per
|
||
group, and S=1 failover is re-proven over real OS processes (`cluster_multiproc`).
|
||
|
||
**Headline exit gate — DONE (local, real OS processes).** `cluster_sharding.rs`
|
||
(tier-3, `cluster-e2e`) boots 3 nodes × 3 shards × RF=3 (every node a replica of
|
||
every group; group `s` led by node `s`) and proves, across
|
||
`TIDAL_SHARDING_KILLPOINTS` random kill points under concurrent `ack=quorum`
|
||
load:
|
||
|
||
| Property | Result |
|
||
|---|---|
|
||
| **Failover localizes** | SIGKILL a node → ONLY the groups it led re-elect; groups led by survivors keep their leader. Verified including the worst case where a node had accumulated ALL THREE leaderships (round 2: killing it redistributed shard 0→eu-west, 1→us-east, 2→eu-west). |
|
||
| **<10s failover** | every re-election completed inside the 10s budget (fast-election block, like p4). |
|
||
| **Reads never stop** | a concurrent `/feed` poller on a survivor saw **0 failures** across every failover window (reads serve from local replicas — no leader needed). |
|
||
| **Per-shard zero acked loss** | every write the client saw a 2xx + `x-tidal-seq` for is present afterwards on **its shard's NEW leader** (the m11p4 vote restriction guarantees the elected leader holds every committed write); proven across all groups, per kill point. |
|
||
| **Rebalance verbs** | `mp_sharded_rebalance_verbs_move_one_group`: `POST /cluster/shards/0/transfer` and `/cluster/promote?shard=0` move EXACTLY group 0's leadership (groups 1/2 untouched); the `?shard=` selector resolves per-group rosters; bad action / missing-addrs / unhosted-shard are 400s. |
|
||
|
||
**Throughput sub-gate — Ref-A-pending (k3s), local figure recorded.** On a local
|
||
release-build 3×3 cluster (`/tmp/m11p6-bench`, `wal.batch_timeout_ms: 2`), the
|
||
unified `ack=quorum` write path sustained **3,000 signal-writes/s within SLO**
|
||
(writes mix, 0% error, p99 ≈95ms, replication lag ≤3) with **per-node CPU ≈30%
|
||
and replication lag ~0** — the cluster has clear headroom. The knee at 4,000 rps
|
||
was the SINGLE open-loop `tidal-stress` generator hitting its in-flight cap
|
||
(~25k shed, "never sent"), the same client-side / connection-establishment wall
|
||
p1 measured (~5k rps), NOT the engine. The **≥5,000/s absolute and the ≥2.5×
|
||
single-shard scaling are genuinely Ref-A** (Linux `fdatasync`, multi-node,
|
||
multiple load sources): they cannot be shown on one macOS laptop where the
|
||
generator + `F_FULLFSYNC` floor is the limiter — the standing k3s-access caveat
|
||
since p1. The horizontal-scaling mechanism is in place (writes hash-route to S
|
||
independent group leaders); demonstrating the 2.5× requires the Ref-A harness.
|
||
|
||
### Known follow-up (S>1, tracked — not silently dropped)
|
||
|
||
The per-group replica **remove** verb is wired and reuses the m11p5 fenced
|
||
conf-change, but a node's READINESS is still node-global across its co-hosted
|
||
groups (`is_ready` ANDs every hosted group), so removing a node from ONE group of
|
||
a multi-group node would wrongly flip the whole node's readiness. Per-group-aware
|
||
readiness (and runtime instantiation of a brand-new group on a node) is the
|
||
elasticity follow-up; the exit-gate (full placement) and the transfer/`?shard=`
|
||
rebalance paths do not touch it. `cluster_sharding.rs` therefore asserts the
|
||
remove verb's wiring + input validation, not a live multi-group removal.
|