From 44b768b8c6197f1c4d6309b91901264048469558 Mon Sep 17 00:00:00 2001 From: jx12n Date: Sat, 13 Jun 2026 18:23:43 -0600 Subject: [PATCH] =?UTF-8?q?feat(m11):=20sharding=20=C3=97=20replication=20?= =?UTF-8?q?+=20rebalancing=20(m11p6=20L3-L5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- CHANGELOG.md | 33 ++ docs/planning/ROADMAP.md | 4 +- docs/planning/milestone-11/phase-6.md | 69 +++- docs/roadmap-to-cluster.md | 2 +- docs/runbooks/cluster.md | 84 +++++ docs/specs/14-scale-architecture.md | 2 +- k8s/cluster/topology-configmap.yaml | 13 + tidal-server/src/cluster/node.rs | 326 +++++++++++++++-- tidal-server/src/cluster/routes.rs | 2 + tidal-server/src/error.rs | 5 + tidal-server/src/openapi.rs | 3 + tidal-server/tests/cluster_sharding.rs | 456 ++++++++++++++++++++++++ tidal-server/tests/support/multiproc.rs | 242 +++++++++++++ tidal-stress/benches/hotpath.rs | 7 +- tidal-stress/src/main.rs | 47 +-- tidal-stress/src/workload.rs | 51 +-- 16 files changed, 1220 insertions(+), 126 deletions(-) create mode 100644 tidal-server/tests/cluster_sharding.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 94298bd..b6e7b6c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,39 @@ All notable changes to tidalDB will be documented in this file. ### Added +**Sharding × replication + rebalancing (m11p6) — the "replicated XOR sharded" split is over: S shard groups, each a replication group at RF with its own elected leader, leaders balanced across nodes; any gateway hash-routes** + +- **One write surface.** `/items`//`/embeddings`//`/signals` now hash-route the + entity to the owning shard group's leader (the engine's FNV-1a `ShardRouter`) + AND replicate at RF — sharding and replication at once. The old + `/sharded/*`-vs-leader split in `tidal-stress` (`WritePath::Leader|Sharded`) is + gone; it drives the one unified path, spreading writes round-robin across + gateways (or pinning one with `--leader-url`). `x-tidal-ack`/`x-tidal-seq`, + quorum await, and `NotLeader`/`QuorumTimeout` are per-group; `NotLeader` names + the group. +- **Rebalancing verbs (L3).** `POST /cluster/shards/{id}/transfer` moves one + group's leadership (the m11p4 fenced transfer scoped to the group); + `POST /cluster/shards/{id}/replicas` adds/removes a replica (the m11p5 join / + fenced-removal per group). A `?shard=` selector threads through every per-shard + admin verb (promote/heal/partition/catchup/reseed/members/join) and is + PROPAGATED on every intra-group forward/broadcast (`ShardReplica::admin_path`) + so the receiving sibling targets the same group. **S=1 is byte-for-byte** (no + selector emitted, no shard in the `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 (survivor groups keep theirs), reads never stop, and + every acked write is present on its shard's new leader (zero acked loss), across + random kill points. Plus a rebalance-verb test (transfer + `?shard=` promote + move exactly one group). Harness: `MultiProcCluster::start_sharded` + (per-(node,shard) ports, `shards:` emission, `agreed_shard_leaders`). +- **Throughput.** Local 3×3 release cluster sustains 3,000 quorum signal-writes/s + at 0% error with per-node CPU ≈30% and replication lag ~0 — generator-bound, + not engine-bound. The ≥5,000/s + ≥2.5×-single-shard scaling is Ref-A (k3s + Linux/`fdatasync`/multi-generator) — the standing access caveat since p1. +- Known follow-up (tracked): per-group-aware node readiness (today `is_ready` is + node-global across co-hosted groups) and cross-node read fan-out under PARTIAL + placement; the exit gate runs full placement, which these do not touch. + **Continuous correctness (m11p9) — new fault classes as REAL faults, first-class invariant checkers, soak regression gates, a Woodpecker nightly chaos+soak pipeline, and a guarantee→test matrix** - **Fault injection compiled out of production.** A non-default `fault-injection` diff --git a/docs/planning/ROADMAP.md b/docs/planning/ROADMAP.md index 80328bb..ddb0729 100644 --- a/docs/planning/ROADMAP.md +++ b/docs/planning/ROADMAP.md @@ -37,7 +37,7 @@ A single embeddable database can replace the 6-system content ranking stack by t | M8 | Distributed Fabric | Multi-region, multi-tenant replication keeps agent-memory semantics intact | Hosted tidalDB, cloud/edge deployments, shared agent substrate — **✅ COMPLETE**: in-process primitives + multi-node replication over real gRPC + true multi-process cluster mode (one process per region, real process isolation) with full tier-3 UAT (partition injection via TCP-proxy, clock-skew, rolling-upgrade, runbook verification); G1 + G2 resolved. Post-M8 follow-ups: quorum-ack writes, automatic failure detection / leader election | | M9 | Community Sync & Revocation | Local embeddable profiles can opt into community personalization and safely leave/purge contributions | Community personalization, federated taste graphs, shared feeds — ✅ COMPLETE (2026-06-06) | | M10 | Governance & Agent Rights | Community rules and agent-scoped permissions control what signals influence ranking | User-owned AI personalization at scale, policy-compliant agents — ✅ COMPLETE (2026-06-06) | -| M11 | Enterprise-Grade Cluster | The multi-process cluster becomes a system of record: fast unified log, quorum-durable, self-healing, horizontally scalable, secured, observable, chaos-tested | Paying-customer cluster deployments — **IN PROGRESS**: m11p1 ✅ + m11p2 ✅ + m11p3 ✅ + m11p4 ✅ (automatic failover: Raft-style election + term fencing + divergence quarantine — closes **G5**; v0.9 "Credible HA" wave complete 2026-06-12) + m11p5 ✅ (membership/discovery/elasticity: DNS peers, snapshot+stream reseed, kind-4 membership records, seed join, `k8s/cluster/` — 2026-06-12) + m11p7 ✅ (security hardening: gRPC mTLS default + zero-drop cert rotation, inter-node HTTP TLS + per-node signed tokens, admin audit log, per-principal rate limit — 2026-06-13) + m11p8 ✅ (observability + operations: completed `tidaldb_cluster_*` set incl. breaker/forwards/self-heal on the per-node `/metrics` listener + Grafana cluster row + alert group, request-id/tracing across hops, truthful status, self-driving heal, WAL PITR archival + `tidalctl` backup/restore, rolling-upgrade version handshake + Woodpecker release gate — closes **G-O** — 2026-06-13) + m11p9 ✅ (continuous correctness: new fault classes (disk-full / slow-fsync / asymmetric partition) as REAL faults behind a production-compiled-out `fault-injection` feature, first-class invariant checkers, `tidal-stress` soak with regression gates + JSON summary, a Woodpecker cron nightly chaos+soak pipeline, and a guarantee→test traceability matrix — closes the **G-C** apparatus; the 30-day-green calendar accrues nightly — 2026-06-13); m11p6 (sharding × replication) data plane in progress in [docs/roadmap-to-cluster.md](../roadmap-to-cluster.md) | +| M11 | Enterprise-Grade Cluster | The multi-process cluster becomes a system of record: fast unified log, quorum-durable, self-healing, horizontally scalable, secured, observable, chaos-tested | Paying-customer cluster deployments — **IN PROGRESS**: m11p1 ✅ + m11p2 ✅ + m11p3 ✅ + m11p4 ✅ (automatic failover: Raft-style election + term fencing + divergence quarantine — closes **G5**; v0.9 "Credible HA" wave complete 2026-06-12) + m11p5 ✅ (membership/discovery/elasticity: DNS peers, snapshot+stream reseed, kind-4 membership records, seed join, `k8s/cluster/` — 2026-06-12) + m11p7 ✅ (security hardening: gRPC mTLS default + zero-drop cert rotation, inter-node HTTP TLS + per-node signed tokens, admin audit log, per-principal rate limit — 2026-06-13) + m11p8 ✅ (observability + operations: completed `tidaldb_cluster_*` set incl. breaker/forwards/self-heal on the per-node `/metrics` listener + Grafana cluster row + alert group, request-id/tracing across hops, truthful status, self-driving heal, WAL PITR archival + `tidalctl` backup/restore, rolling-upgrade version handshake + Woodpecker release gate — closes **G-O** — 2026-06-13) + m11p9 ✅ (continuous correctness: new fault classes (disk-full / slow-fsync / asymmetric partition) as REAL faults behind a production-compiled-out `fault-injection` feature, first-class invariant checkers, `tidal-stress` soak with regression gates + JSON summary, a Woodpecker cron nightly chaos+soak pipeline, and a guarantee→test traceability matrix — closes the **G-C** apparatus; the 30-day-green calendar accrues nightly — 2026-06-13) + m11p6 ✅ (sharding × replication + rebalancing: ONE hash-routed + replicated write surface across S shard groups each at RF with its own elected leader; per-group rebalancing verbs (`/cluster/shards/{id}/transfer` + `/replicas`) and a `?shard=` admin selector; tier-3 3×3 kill-node exit gate — only the dead node's leaderships move, reads never stop, zero acked loss; ≥5,000/s + 2.5× scaling Ref-A-pending — 2026-06-13). **ALL NINE PHASES COMPLETE**; the v1.0 bar now waits only on the 30-day-green nightly calendar (m11p9) and the standing Ref-A/k3s throughput re-runs | ### Embeddable → Distributed Path @@ -3282,7 +3282,7 @@ Full gap analysis, measured baselines, phase specs, and exit gates live in | m11p3 | Quorum-acked writes: `ack=leader\|quorum` (topology default + `x-tidal-ack` override), durable-frontier reports, commit index, retryable-503 timeout semantics, ledger checker | ✅ **COMPLETE (2026-06-11)** — followers push durably-applied frontiers (`ReportApplied`, batch-level, decoupled from ship acks); commit index = k-th largest durable mark, leadership-scoped; `ack=quorum` writes await it asynchronously and 503 with laggard names on budget expiry; every cluster write returns `x-tidal-seq`. Exit gate: 100/100 leader-SIGKILL kill-points with ZERO acknowledged loss (frontier + content proofs on the promoted max-applied survivor); quorum throughput 3,600 signal-writes/s within SLO locally (79% of p1's 4,534/s leader-ack figure, gate ≥50%). Also fixed en route: follower blob applies group-committed (22× item-seed speedup), async quorum waits (thread-per-wait collapsed at 1k rps). See [milestone-11/phase-3.md](milestone-11/phase-3.md). | | m11p4 | Failure detection, election, term fencing — closes **G5** | ✅ **COMPLETE (2026-06-12)** — purpose-built election-only Raft (pre-vote + check-quorum + fenced transfer) over the existing transport, no raft crate: an elected leader's FIRST entry is a replicated kind-3 **term marker**, so `(lastLogTerm, lastLogIndex)` derives from one fsync stream (frontiers compared in the last joined term's STREAM numbering). Hard state `(term, voted_for)` in `data_dir/election_state` (corrupt → refuse boot; lost-with-WAL → forced follower), fsynced before any vote/claim. Term fencing on every RPC; the commit index folds only activation-term reports; a restarted ex-leader boots follower from durable state (the §1.4-1 split-brain incident closed by construction). Divergent suffixes QUARANTINE; promote is a fenced transfer that refuses lagging targets. Exit gates (local tier-3, 3 processes): elections 0.59–0.98s under ack=quorum kill loads with zero acked loss; a restarted partitioned ex-leader accepted 0 writes; 7 link flaps converged at term ≤15. See [milestone-11/phase-4.md](milestone-11/phase-4.md). | | m11p5 | Membership, discovery, elasticity (DNS, seed join, snapshot + stream catch-up) | ✅ **COMPLETE (2026-06-12)** — `grpc_addr` is now an **advertised** address (hostname or IP) split from an optional `grpc_bind`; `tidal-net`'s peer map retyped `SocketAddr → String` so hyper re-resolves DNS on every reconnect (the pod-rescheduled-onto-a-new-IP case). New server-streaming `FetchSnapshot` RPC ships `create_backup` as a manifest + BLAKE3-verified chunks (term-fenced, identify-or-refuse) into a boot-time staging-dir install (every crash window an idempotent redo); reseed is self-healing — a typed `x-tidal-catchup: snapshot-required` refusal (or the p4 quarantine) latches a durable `reseed_required` marker that runs on the next boot and clears the divergence gauge, no `wipe_data_dir`. Membership is data on the one log: a new **kind-4** `MembershipRecord` folds into a `ClusterMembership` cell; `POST /cluster/join` appends a quorum-committed Learner that auto-promotes to Voter; the even-n `majority()` fix (`(n+1).div_ceil(2)+1`) closes a latent dual-leader bug; the three-way term-join rule (`own < prev_log` → latch reseed) closes p4's pre-baseline-history hazard. Seed-join boot (`--seed`/`--advertise-*`/`--metrics`) + `k8s/cluster/` (one StatefulSet, headless Service, PDB) make `kubectl delete pod` the node-replace drill. Exit-gate mechanics proven by tier-3 `cluster_membership.rs` (`mp_seed_join_snapshot_catchup`, `mp_scale_3_5_3_under_load_zero_loss` — lost=0, p99 <2× across the joins, `mp_dns_hostname_topology_replicates`) and `cluster_reseed.rs`. 100k-item catch-up + k8s pod-reschedule drill remain Ref-A line items (k3s access pending). See [milestone-11/phase-5.md](milestone-11/phase-5.md). | -| m11p6 | Sharding × replication + rebalancing | Planned | +| m11p6 | Sharding × replication + rebalancing | ✅ 2026-06-13 — one hash-routed + replicated write surface across S groups (each RF, own leader); per-group `/cluster/shards/{id}/transfer`+`/replicas` rebalancing verbs + `?shard=` admin selector; tier-3 3×3 kill-node exit gate green (localized failover, reads never stop, zero acked loss). ≥5,000/s + 2.5× scaling Ref-A-pending (k3s). | | m11p7 | Security hardening (mTLS default, rotation, audit log) | ✅ **COMPLETE (2026-06-13)** — the cluster stops trusting the network. gRPC replication is served over a custom `tokio-rustls` acceptor fed a hot-swappable `DynamicCertResolver` (`ArcSwap`) — mTLS preserved exactly (`WebPkiClientVerifier`; a foreign/absent client cert fails the handshake before any RPC), plaintext now an explicit `insecure: true` + loud WARN. **Cert + bearer rotation without restart**: a content-hash poller (catches k8s `..data` symlink swaps inotify misses) atomically swaps the cert with in-flight sessions untouched — **zero dropped requests under load** (verified). Inter-node HTTP gains TLS (same resolver — one rotation covers both planes; `https` forwards + cluster-CA reqwest clients) + **per-node identity** via a signed `x-tidal-node-token` (keyed-BLAKE3 MAC under a shared cluster key — no new crypto dep); the `x-tidal-internal` marker is honored ONLY from a verified sibling (marker-without-token → 403), never an auth bypass. Admin verbs (promote/partition/heal/join/remove/reseed) emit a structured audit record (principal, term, target, outcome) to a `tidal_audit` target + optional `TIDAL_AUDIT_LOG` JSONL, operator-leg only. Per-principal HTTP rate limit (engine `RateLimiter`, nodes exempt, 429 + Retry-After). All TLS/identity/audit/limit opt-in (`grpc_tls` / cluster key / env) — absent ⇒ pre-m11p7 behavior byte-for-byte. `k8s/cluster/` gains cert-manager + `grpc_tls` topology + cluster-key Secret; `scripts/gen-cluster-certs.sh` for non-cert-manager. Exit gate verified: foreign pod rejected (gRPC `mtls.rs` + HTTP `cluster_security.rs`), zero-drop rotation under load, zero plaintext inter-node links. See [milestone-11/phase-7.md](milestone-11/phase-7.md). | | m11p8 | Observability + operations (complete metric set, self-driving heal, backup/PITR, rolling-upgrade gate) | ✅ **COMPLETE (2026-06-13)** — operable by someone who didn't build it (closes **G-O** + the operability half of **G-Op**, and incident §1.4-3). **Metrics**: completed the `tidaldb_cluster_*` set with breaker (`tidaldb_cluster_peer_breaker_state` + `_breaker_opens_total`, surfaced read-only from the `tidal-net` breaker) and forwards (`_forwards_total` / `_forward_failures_total`) + the self-heal series; multi-shard nodes serve ONE `/metrics` listener with the co-located siblings' series `shard="N"`-labeled (S=1 byte-identical). A 12-panel Grafana cluster row + 8-rule `tidaldb-cluster` alert group ship beside the standalone ones. **Tracing**: both cluster routers gained the request-id + `TraceLayer` stack; the id rides the forward hop. **Truthful status**: the leader's own `applied_events` (was 0 — now its flushed frontier) and the post-promote `ShardId(0)` lag-keying undercount fixed; `/cluster/status` gained `version`. **Self-driving heal**: a standing leader duty re-arms a stuck peer's backlog re-ship every ~3s so it converges through breaker resets with NO operator heal loop (`healing_peers` gauge + alert). **Backup/PITR**: `wal.archive_dir` archives sealed segments before compaction deletes them (gap-free); `tidalctl backup`/`restore` (BLAKE3-verified round-trip). **Rolling upgrade**: wire `build_version` handshake (N/N+1 by proto3 compat; `>= 2`-major WARNs, never rejects) + `mp_rolling_upgrade_no_loss_no_stall` promoted to the Woodpecker release gate. Exit gates green (dashboard answers the golden signals; rolling-upgrade gate passed; backup/restore + archival round-trips verified). See [milestone-11/phase-8.md](milestone-11/phase-8.md). | | m11p9 | Continuous correctness (nightly chaos CI, invariant checkers, soak) | ✅ **COMPLETE (2026-06-13)** — closes the **G-C** apparatus (the 30-consecutive-days-green half of the GA bar accrues nightly). New fault classes as REAL faults behind a production-compiled-out `fault-injection` feature: **slow-fsync** (`TIDAL_FAULT_FSYNC_DELAY_MS`) + **disk-full** (`TIDAL_FAULT_DISK_FULL_AFTER_BYTES` → real `ENOSPC`) WAL hooks (`tidal/src/fault.rs`); **asymmetric partition** via the harness's directed-edge proxies. First-class invariant checkers (`support/invariants.rs`): no-acked-loss `AckLedger` (the m11p3 ledger gate now consumes it), decay/feed parity, `assert_single_leader_per_term`, `MonotonicCounters`. New tier-3 `cluster_faults.rs` (disk-full / slow-fsync ×2 / asymmetric) **4/4 green** with zero acked loss. `tidal-stress` gained `--json-summary` + `--max-p99-ms`/`--max-error-pct`/`--fail-on-knee` regression gates (non-zero exit on breach — verified vs a real server). Nightly Woodpecker CRON pipeline (chaos suites with elevated kill-points + the gated soak) beside the push release gate. Every §2 guarantee → named test in [milestone-11/guarantee-traceability.md](milestone-11/guarantee-traceability.md). See [milestone-11/phase-9.md](milestone-11/phase-9.md). | diff --git a/docs/planning/milestone-11/phase-6.md b/docs/planning/milestone-11/phase-6.md index d6c1fff..07b53b3 100644 --- a/docs/planning/milestone-11/phase-6.md +++ b/docs/planning/milestone-11/phase-6.md @@ -1,4 +1,4 @@ -# m11p6 — Sharding × Replication + Rebalancing (IN PROGRESS) +# 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 @@ -236,12 +236,24 @@ rebalancing is explicitly "later"** — m11p6 ships the operator verbs. 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. -- [ ] L3 rebalancing verbs (operator shard move; reuse m11p5 per-group) -- [ ] L4 tidal-stress path collapse + nodes×shards tier-3 harness + - `cluster_sharding.rs` exit gate (3×3 kill-node + per-shard ledger; run for - real). 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). -- [ ] L5 docs (runbook/monitoring/roadmap/CHANGELOG/k8s/spec) + memory +- [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 @@ -252,8 +264,41 @@ 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`). -**Remaining for the headline exit gate:** the 3 shards × RF=3 tier-3 run over real -OS processes (kill any node → only its shard-leaderships move <10 s, reads never -stop; per-shard zero-acked-loss ledger) and the ≥5,000/s `tidal-stress` figure -(Ref-A-pending). These need the L4 harness extension (per-(node,shard) ports + -`shards:` emission) + the rebalance verbs (L3). +**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. diff --git a/docs/roadmap-to-cluster.md b/docs/roadmap-to-cluster.md index 87cd5d4..4c9c8e6 100644 --- a/docs/roadmap-to-cluster.md +++ b/docs/roadmap-to-cluster.md @@ -1,6 +1,6 @@ # Roadmap to an Enterprise-Grade Cluster -**Status:** ADOPTED as M11 — m11p1 ✅, m11p2 ✅, m11p3 ✅, m11p4 ✅, m11p5 ✅, m11p7 ✅, m11p8 ✅, m11p9 ✅ complete (m11p9 2026-06-13 — continuous correctness: new fault classes (disk-full / slow-fsync / asymmetric partition) as REAL faults behind a production-compiled-out `fault-injection` feature, first-class invariant checkers (no-acked-loss ledger, decay parity, single-leader-per-term, monotonic counters), a `tidal-stress` soak with PASS/FAIL regression gates + JSON summary, a Woodpecker cron nightly chaos+soak pipeline beside the push release gate, and a guarantee→test traceability matrix — closes **G-C** apparatus; the 30-day-green calendar accrues nightly) · m11p6 (sharding × replication) data plane in progress · **Date:** 2026-06-10 · **Baseline evidence:** +**Status:** ADOPTED as M11 — **ALL NINE PHASES COMPLETE.** m11p1–m11p5 ✅, m11p7 ✅, m11p8 ✅, m11p9 ✅ (2026-06-13 — continuous correctness: new fault classes (disk-full / slow-fsync / asymmetric partition) as REAL faults behind a production-compiled-out `fault-injection` feature, first-class invariant checkers (no-acked-loss ledger, decay parity, single-leader-per-term, monotonic counters), a `tidal-stress` soak with PASS/FAIL regression gates + JSON summary, a Woodpecker cron nightly chaos+soak pipeline beside the push release gate, and a guarantee→test traceability matrix — closes **G-C** apparatus; the 30-day-green calendar accrues nightly) · **m11p6 ✅ (2026-06-13 — sharding × replication + rebalancing: ONE hash-routed + replicated write surface across S shard groups, each a replication group at RF with its own elected leader; per-group `/cluster/shards/{id}/transfer`+`/replicas` rebalancing verbs and a `?shard=` admin selector propagated through intra-group forwards; tier-3 3-node × 3-shard × RF=3 kill-node exit gate green — only the dead node's leaderships move (<10s), reads never stop, zero acknowledged loss per group; ≥5,000/s + ≥2.5× single-shard scaling remain Ref-A/k3s-pending, local 3×3 sustains 3,000 quorum writes/s at 0% error with the engine at ~30% CPU)** · v1.0 now waits only on the 30-day-green nightly calendar + the standing Ref-A/k3s throughput re-runs · **Date:** 2026-06-10 · **Baseline evidence:** [stress-test-thepeach.md](ops/stress-test-thepeach.md), [cluster runbook](runbooks/cluster.md), [ROADMAP M8 Known Gaps](planning/ROADMAP.md) (G4/G5/G6), live k3s deployment (3 regions × 2-vCPU pods). diff --git a/docs/runbooks/cluster.md b/docs/runbooks/cluster.md index afb3f7e..dbaa6e9 100644 --- a/docs/runbooks/cluster.md +++ b/docs/runbooks/cluster.md @@ -331,6 +331,51 @@ every reconnect — a pod rescheduled onto a new IP is reached with no peer restarts. **TLS note:** SNI follows the URI host, so DNS peer names require DNS-SAN certs (see `grpc_tls` below); no code change. +### 3a. Sharding × replication (`shards:`, m11p6) + +Absent `shards:`, the cluster is **1 shard × RF = all regions** — one replicated +log, one elected leader, byte-for-byte everything above. Add an optional +`shards:` block to split the entity space into **S groups, each a replication +group at RF with its own elected leader**, leaders balanced across nodes. Writes +to `/items`//`/embeddings`//`/signals` hash-route (the engine's FNV-1a router) to +the owning group's leader and replicate at RF; any gateway accepts any write. + +```yaml +regions: # the NODE list (identity + addresses) + - { name: us-east, grpc_addr: "10.0.1.10:9601", http_addr: "10.0.1.10:9501" } + - { name: eu-west, grpc_addr: "10.0.2.10:9601", http_addr: "10.0.2.10:9501" } + - { name: ap-south, grpc_addr: "10.0.3.10:9601", http_addr: "10.0.3.10:9501" } +shards: # NEW — S groups (dense ids 0..S) + - id: 0 + leader: us-east # term-0 / preferred leader (balance: group i → node i) + replicas: # the RF nodes hosting this group + - { node: us-east } # grpc_addr optional → derived node base port + shard id + - { node: eu-west } + - { node: ap-south } + - { id: 1, leader: eu-west, replicas: [ {node: us-east}, {node: eu-west}, {node: ap-south} ] } + - { id: 2, leader: ap-south, replicas: [ {node: us-east}, {node: eu-west}, {node: ap-south} ] } +leader: us-east # legacy field, ignored once shards: is set +``` + +- **Ports & dirs.** A node hosting several groups binds one gRPC port PER group: + set `replicas[].grpc_addr` explicitly, or omit it to derive `node base port + + shard id`. Each group's data lives under `/shard-{id:05}/`; the + single-group legacy layout (`shards:` absent) keeps `` verbatim, so + existing clusters restart unchanged. +- **Placement.** `replicas` need not be every node (RF < N is valid). The exit + gate and the simplest production shape are **full placement** (every node + replicates every group); a strict-subset placement is supported for writes but + see the read caveat in [§7](#7-sharded-scatter-gather-api). +- **Validation.** Shard ids must be dense and unique; every `leader`/`replica.node` + names a declared region; a group needs ≥1 replica; co-hosted groups must not + collide on a derived gRPC port. + +**Per-shard admin (`?shard=`).** With `shards:` set, every cluster admin verb +takes an optional `?shard=` selecting which hosted group to act on; omitted = +the node's lowest-id hosted group (and the only group when S=1, so S=1 URLs are +unchanged). The selector is forwarded on intra-group hops, and `NotLeader` names +the group. See [§6a](#6a-rebalancing-m11p6) for the rebalancing verbs. + Fields: | Field | Single-process | Multi-process | Meaning | @@ -748,6 +793,45 @@ curl -X POST "$BASE/cluster/reseed" record to quorum-commit, then stop / delete the node — lowest ordinal last under a StatefulSet. +### 6a. Rebalancing (m11p6) + +With a `shards:` topology ([§3a](#3a-sharding--replication-shards-m11p6)) the +operator moves load between nodes per group. Automatic rate-limited rebalancing +is explicitly later; m11p6 ships the manual verbs (each reuses the proven +m11p4/m11p5 machinery, scoped to one group). + +**Move a group's leadership** (the rebalance-back-to-preferred path) — a fenced +transfer (catch-up drain → `TimeoutNow` → term+1 election), identical to +`/cluster/promote?shard=`: + +```bash +curl -X POST "$BASE/cluster/shards/0/transfer" \ + -H 'content-type: application/json' \ + -d '{"region":"eu-west"}' +# only group 0's leader moves; groups 1,2 are untouched +``` + +**Add / remove a replica of a group** — `add` runs the m11p5 join (the named node +joins group `{id}` as a Learner, catches up via snapshot+stream, auto-promotes to +Voter); `remove` runs the m11p5 fenced removal on that group's log: + +```bash +curl -X POST "$BASE/cluster/shards/2/replicas" \ + -H 'content-type: application/json' \ + -d '{"action":"add","name":"region-3","grpc_addr":"10.0.4.10:9603","http_addr":"10.0.4.10:9504"}' +``` + +> **Multi-group caveat (tracked follow-up).** A node's readiness is currently +> node-global across its co-hosted groups, so removing a node from ONE group of a +> node that hosts several would wrongly flip the whole node's readiness. Use the +> `replicas` verbs today for nodes that host a SINGLE group (or a brand-new node +> joining one group); per-group-aware readiness is the elasticity follow-up. + +**Inspect / target one group.** Every admin verb takes `?shard=`; +`/cluster/status/local` always lists a per-group `shards[]` array (leader, term, +role, applied, lag, commit index per group), and `/cluster/members?shard=` +returns that group's roster. + ## 7. Sharded scatter-gather API The `/sharded/*` routes hash-partition entities across regions diff --git a/docs/specs/14-scale-architecture.md b/docs/specs/14-scale-architecture.md index aaec8f3..17fc596 100644 --- a/docs/specs/14-scale-architecture.md +++ b/docs/specs/14-scale-architecture.md @@ -1,6 +1,6 @@ # Scale Architecture Specification -**Status:** Implemented (M0–M8); multi-node cluster mode is PARTIAL — see [CHANGELOG.md](../../CHANGELOG.md) known gaps (G1 in-process transport, G2 tier-3 tests, G3 hash inconsistency) +**Status:** Implemented (M0–M8); multi-node cluster mode shipped through M11 — quorum-durable writes (m11p3), automatic election/fencing (m11p4), dynamic membership (m11p5), and **sharding × replication (m11p6): S shard groups, each a replication group at RF with its own elected leader, any gateway hash-routing writes (the §4 Option-C shape), per-group rebalancing verbs.** The cluster `ShardReplica` (`tidal-server/src/cluster/node.rs`) is one group's machinery; a `ClusterNode` hosts a `BTreeMap>`. See [docs/roadmap-to-cluster.md](../roadmap-to-cluster.md) and [docs/planning/milestone-11/phase-6.md](../planning/milestone-11/phase-6.md). Original M8 known gaps G3 (shard-routing hash unification) and G6 (embedding 500→400) remain open. **Author:** tidalDB Engineering **Last Updated:** 2026-05-28 **Depends on:** Storage Engine (01), Entity Model (02), Signal System (03), Cohorts (05), Vector Retrieval (07) diff --git a/k8s/cluster/topology-configmap.yaml b/k8s/cluster/topology-configmap.yaml index a63c8c6..7c9dfd8 100644 --- a/k8s/cluster/topology-configmap.yaml +++ b/k8s/cluster/topology-configmap.yaml @@ -78,6 +78,19 @@ data: # Term-0 bootstrap leader only — post-election this field is dead config # (durable election_state governs; a restart always boots a follower). leader: tidaldb-0 + # ── Optional sharding × replication (m11p6) ────────────────────────────── + # Absent `shards:` ⇒ ONE group, RF = all pods (what this file ships). To + # scale WRITES horizontally, split the entity space into S groups, each a + # replication group at RF with its own elected leader, leaders balanced. A + # pod hosting several groups binds one gRPC port per group (omit + # replicas[].grpc_addr to derive `pod base port + shard id`); each group's + # data lives under /shard-{id:05}/. Full placement (every pod + # replicates every group) is the simplest shape. Operators rebalance with + # `POST /cluster/shards/{id}/transfer` and `/replicas` (see runbook §6a). + # shards: + # - { id: 0, leader: tidaldb-0, replicas: [ {node: tidaldb-0}, {node: tidaldb-1}, {node: tidaldb-2} ] } + # - { id: 1, leader: tidaldb-1, replicas: [ {node: tidaldb-0}, {node: tidaldb-1}, {node: tidaldb-2} ] } + # - { id: 2, leader: tidaldb-2, replicas: [ {node: tidaldb-0}, {node: tidaldb-1}, {node: tidaldb-2} ] } replication: # ack=quorum: a write succeeds once a MAJORITY of the replica set durably # holds it (m11p3). Callers can still override per-request with x-tidal-ack. diff --git a/tidal-server/src/cluster/node.rs b/tidal-server/src/cluster/node.rs index a10e984..c7c517f 100644 --- a/tidal-server/src/cluster/node.rs +++ b/tidal-server/src/cluster/node.rs @@ -55,7 +55,7 @@ use std::{ use axum::{ Json, Router, - extract::{Query, RawQuery, Request, State}, + extract::{Path, Query, RawQuery, Request, State}, http::{HeaderMap, StatusCode}, middleware::{self, Next}, response::{IntoResponse, Response}, @@ -200,6 +200,11 @@ struct ElectionBoot { /// /// A node holds one `ShardReplica` per group it replicates. It owns one region's /// in-group identity — the data-shard `group_shard` only namespaces its dir/port. +// The bool fields (`multi_shard`, `reseed_self_restart`, `seed_joiner`, +// `install_boot`) are INDEPENDENT boot/identity facts, not a state machine that +// would read better as an enum — an enum would force false either/or relations +// between orthogonal flags. Allow the bool count rather than contort the model. +#[allow(clippy::struct_excessive_bools)] pub struct ShardReplica { /// This process's region id (index into the topology declaration order). region: RegionId, @@ -210,6 +215,13 @@ pub struct ShardReplica { /// shard label. The in-group replication identity stays `region`-based /// (`shard_of_region`) — `group_shard` only distinguishes co-hosted groups. group_shard: ShardId, + /// Whether this node co-hosts more than one shard group (m11p6 S>1). When + /// true, this replica's intra-group admin forwards/broadcasts carry a + /// `?shard=` selector so the receiving sibling targets the SAME + /// group, and `NotLeader` names the group. `false` for the legacy single + /// group keeps the S=1 wire format byte-for-byte (no selector, no shard in + /// the error body) — set once at construction from the resolved group count. + multi_shard: bool, /// `Some` for the server's lifetime; taken on shutdown so the `TidalDb` is /// dropped (checkpoint + WAL fsync + thread join) deterministically. db: Option>, @@ -466,6 +478,7 @@ impl ShardReplica { hlc_offset_ms: i64, group: &ResolvedShardGroup, enable_metrics: bool, + multi_shard: bool, creds: Arc, ) -> Result { super::topology::validate_multiproc(topology, region_name)?; @@ -969,6 +982,7 @@ impl ShardReplica { region, region_name: region_name.to_string(), group_shard: group.shard, + multi_shard, db: Some(db), transport, ship_feed, @@ -1094,6 +1108,33 @@ impl ShardReplica { self.id_to_name.get(&id).map_or("unknown", String::as_str) } + /// This replica's data-shard for admin surfaces (m11p6): `Some(group_shard)` + /// when the node co-hosts several groups, `None` for the legacy single group. + /// The seam every per-shard admin surface (the `?shard=` selector, the + /// `NotLeader` body, the intra-group forward suffix) reads, so S=1 stays + /// byte-for-byte by construction (one place returns `None`). + const fn admin_shard(&self) -> Option { + if self.multi_shard { + Some(self.group_shard) + } else { + None + } + } + + /// An intra-group admin path with this replica's `?shard=` selector appended + /// when the node co-hosts several groups (m11p6). A forward/broadcast to a + /// sibling NODE must target the SAME group — the receiver's handler resolves + /// `replica_for(?shard)`. Returns `base` verbatim for the legacy single group + /// (the receiver's `replica_for(None)` resolves its sole group), keeping the + /// S=1 forward wire byte-for-byte. The query rides through [`peer_url`], which + /// appends the path verbatim. + fn admin_path(&self, base: &str) -> String { + self.admin_shard().map_or_else( + || base.to_string(), + |shard| format!("{base}?shard={}", shard.0), + ) + } + /// This node's current leadership view (`None` during an election). #[must_use] fn current_leader(&self) -> Option { @@ -1135,6 +1176,9 @@ impl ShardReplica { ), http_addr: self.leader_http(), term: self.election_term(), + // m11p6: name the group when this node co-hosts several (S>1). `None` + // for the legacy single group keeps the S=1 body byte-for-byte. + shard: self.admin_shard().map(|s| s.0), } } @@ -1284,7 +1328,9 @@ impl ShardReplica { let behind = reported_applied.is_none_or(|a| a < leader_flushed); let mut nudged = false; if behind && let Some(http_addr) = self.peer_http.get(&id) { - let url = super::forward::peer_url(http_addr, "/cluster/catchup"); + // m11p6: the nudge must hit the follower's replica of THIS group when + // it co-hosts several (`?shard=`); `admin_path` is a no-op for S=1. + let url = super::forward::peer_url(http_addr, &self.admin_path("/cluster/catchup")); let body = serde_json::json!({ "shard": shard_of_region(self.region).0, "from_seqno": reported_applied.unwrap_or(0) + 1, @@ -3607,6 +3653,7 @@ impl ClusterNode { hlc_offset_ms, group, enable_metrics, + !single, Arc::clone(&creds), )?; if enable_metrics { @@ -4036,6 +4083,9 @@ pub fn build_region_router( .route("/cluster/members", get(cluster_members)) .route("/cluster/members/remove", post(cluster_member_remove)) .route("/cluster/join", post(cluster_join)) + // m11p6 L3 rebalancing verbs (per-group, reusing the m11p5 machinery). + .route("/cluster/shards/{id}/replicas", post(shard_replicas)) + .route("/cluster/shards/{id}/transfer", post(shard_transfer)) .route( "/cluster/reconcile/snapshot", post(cluster_reconcile_snapshot), @@ -4306,12 +4356,14 @@ pub struct ShardStatusRow { #[allow(clippy::significant_drop_tightening)] pub async fn status_local( State(node): State>, + Query(sel): Query, ) -> std::result::Result, ClusterAppError> { - // m11p6: the flat fields mirror the first hosted group (S=1 byte-for-byte); - // the `shards` array carries every hosted group so an operator/the kill-node - // gate sees which shard-leaderships this node holds. + // m11p6: the flat fields mirror the selected group (the first hosted group by + // default — S=1 byte-for-byte); the `shards` array always carries every + // hosted group so an operator/the kill-node gate sees which shard-leaderships + // this node holds. let mut status = node - .replica_for(None)? + .replica_for(sel.shard_id())? .local_status() .map_err(ClusterAppError)?; status.shards = node.shard_status_rows(); @@ -4564,6 +4616,29 @@ pub struct RegionRequest { baseline: Option, } +/// The `?shard=` selector (m11p6) shared by every per-shard admin surface +/// (`/cluster/promote`, `/heal`, `/partition`, `/catchup`, `/reseed`, +/// `/members`, `/members/remove`, `/join`, `/status/local`). +/// +/// Absent ⇒ the FIRST hosted group — exact for the legacy single group (S=1, +/// byte-for-byte) and the gateway default. `?shard=N` targets that group via +/// [`ClusterNode::replica_for`] (400 if this node hosts no replica of it). An +/// intra-group forward carries the selector forward ([`ShardReplica::admin_path`]) +/// so the receiving sibling resolves the SAME group. +#[derive(Debug, Default, Deserialize)] +pub struct ShardSelector { + /// The data-shard group id to target. Absent for the legacy single group. + #[serde(default)] + shard: Option, +} + +impl ShardSelector { + /// The selected shard as a [`ShardId`], or `None` for the default group. + fn shard_id(&self) -> Option { + self.shard.map(ShardId) + } +} + /// Promote a region to leader across the cluster (m11p2). /// /// * **internal marker present**: apply the leadership change locally only @@ -4597,9 +4672,10 @@ pub struct RegionRequest { pub async fn cluster_promote( State(node): State>, headers: HeaderMap, + Query(sel): Query, Json(req): Json, ) -> std::result::Result, ClusterAppError> { - let state = node.replica_for(None)?; + let state = node.replica_for(sel.shard_id())?; if is_internal(&headers) { // Marked fan-out leg (the LEGACY term-0 protocol): apply locally and // terminate. promote_local fences this once the cluster is @@ -4711,7 +4787,7 @@ pub async fn cluster_promote( // live leader is known; otherwise campaign directly (the dead- // leader failover drill). let sanctioned = if let Some(addr) = state.leader_http_addr() { - let url = peer_url(&addr, "/cluster/promote"); + let url = peer_url(&addr, &state.admin_path("/cluster/promote")); let auth = forwarded_auth(&headers); let body = serde_json::json!({ "region": req.region }); // A relayed operator hop: the leader runs the full fenced @@ -4742,7 +4818,7 @@ pub async fn cluster_promote( req.region )))); }; - let url = peer_url(&addr, "/cluster/promote"); + let url = peer_url(&addr, &state.admin_path("/cluster/promote")); let auth = forwarded_auth(&headers); let body = serde_json::json!({ "region": req.region }); // A relayed operator hop: the target runs the full protocol @@ -4828,7 +4904,7 @@ pub async fn cluster_promote( req.region )))); }; - let url = peer_url(&http_addr, "/cluster/promote"); + let url = peer_url(&http_addr, &state.admin_path("/cluster/promote")); let auth = forwarded_auth(&headers); let body = serde_json::json!({ "region": req.region }); let baseline = match forward_json_with_headers( @@ -4872,7 +4948,13 @@ pub async fn cluster_promote( // below being all-peers — the target simply re-activates from the same // baseline, which is a no-op for its peers' frontiers. let fan_body = serde_json::json!({ "region": req.region, "baseline": baseline }); - let outcome = broadcast_to_peers(&state, "/cluster/promote", &fan_body, &headers).await; + let outcome = broadcast_to_peers( + &state, + &state.admin_path("/cluster/promote"), + &fan_body, + &headers, + ) + .await; Ok(Json(serde_json::json!({ "ok": true, "leader": req.region, @@ -4929,9 +5011,10 @@ pub struct CatchupRequest { pub async fn cluster_catchup( State(node): State>, headers: HeaderMap, + Query(sel): Query, Json(req): Json, ) -> std::result::Result { - let state = node.replica_for(None)?; + let state = node.replica_for(sel.shard_id())?; if !is_internal(&headers) { return Err(ClusterAppError(ServerError::BadRequest( "/cluster/catchup is internal; the x-tidal-internal marker is required \ @@ -4970,8 +5053,9 @@ pub async fn cluster_catchup( pub async fn cluster_reseed( State(node): State>, headers: HeaderMap, + Query(sel): Query, ) -> std::result::Result { - let state = node.replica_for(None)?; + let state = node.replica_for(sel.shard_id())?; let term = state.election_term(); // The resume seqno: this node's applied frontier (against the current // leader's shard) + 1 — the first seqno past what it has durably applied. @@ -5062,8 +5146,9 @@ pub struct MembersResponse { #[allow(clippy::significant_drop_tightening)] pub async fn cluster_members( State(node): State>, + Query(sel): Query, ) -> std::result::Result { - let state = node.replica_for(None)?; + let state = node.replica_for(sel.shard_id())?; let roster = state.membership().roster(); let members = roster .members @@ -5116,39 +5201,58 @@ pub async fn cluster_members( pub async fn cluster_member_remove( State(node): State>, headers: HeaderMap, + Query(sel): Query, Json(req): Json, ) -> std::result::Result { - let state = node.replica_for(None)?; + let state = node.replica_for(sel.shard_id())?; let term = state.election_term(); let target = req.region.clone(); - // Non-leader: forward to the leader (the conf-change must run there). + // Non-leader: forward to the leader (the conf-change must run there). The + // forward carries the `?shard=` selector so the leader resolves the SAME + // group (m11p6; `admin_path` is a no-op for S=1). if !is_internal(&headers) && !state.is_leader() { - let result = forward_write(&state, "/cluster/members/remove", &req, &headers).await; + let result = forward_write( + &state, + &state.admin_path("/cluster/members/remove"), + &req, + &headers, + ) + .await; node.audit_admin(&headers, "member_remove", &target, term, &result); return result; } - let name = req.region.clone(); - let state_for_job = Arc::clone(&state); + let result = do_remove(&state, &req.region).await; + node.audit_admin(&headers, "member_remove", &target, term, &result); + result +} + +/// Leader-side member removal on a resolved group (m11p6): the write-pool +/// submit + response shape shared by `/cluster/members/remove` and the +/// `/cluster/shards/{id}/replicas` (remove) rebalance verb. The caller has +/// already resolved leadership/forwarding and audits the outcome. +async fn do_remove( + state: &Arc, + name: &str, +) -> std::result::Result { + let name_owned = name.to_string(); + let state_for_job = Arc::clone(state); // Runs on the write pool: the append + bounded same-term commit wait blocks. let outcome = state .write_pool - .submit(move || Ok::<_, ServerError>(state_for_job.handle_remove(&name))) + .submit(move || Ok::<_, ServerError>(state_for_job.handle_remove(&name_owned))) .await .map_err(ClusterAppError)?; - let result = match outcome { + match outcome { RemoveOutcome::Removed { version } => Ok(( StatusCode::OK, - Json(serde_json::json!({ "removed": req.region, "membership_version": version })), + Json(serde_json::json!({ "removed": name, "membership_version": version })), ) .into_response()), RemoveOutcome::NotPresent => Err(ClusterAppError(ServerError::Cluster(format!( - "region '{}' is not a current member (unknown or already removed)", - req.region + "region '{name}' is not a current member (unknown or already removed)" )))), RemoveOutcome::Refused(reason) => Err(ClusterAppError(ServerError::Cluster(reason))), - }; - node.audit_admin(&headers, "member_remove", &target, term, &result); - result + } } /// Join the cluster over HTTP (m11p5 §3.3): the operator/manual wrapper over the @@ -5171,13 +5275,15 @@ pub async fn cluster_member_remove( pub async fn cluster_join( State(node): State>, headers: HeaderMap, + Query(sel): Query, Json(req): Json, ) -> std::result::Result { - let state = node.replica_for(None)?; + let state = node.replica_for(sel.shard_id())?; let term = state.election_term(); let target = req.name.clone(); if !is_internal(&headers) && !state.is_leader() { - let result = forward_write(&state, "/cluster/join", &req, &headers).await; + let result = + forward_write(&state, &state.admin_path("/cluster/join"), &req, &headers).await; node.audit_admin(&headers, "join", &target, term, &result); return result; } @@ -5187,13 +5293,26 @@ pub async fn cluster_join( http_addr: req.http_addr.clone(), capabilities: tidal_net::CAP_KIND4_MEMBERSHIP, }; - let state_for_job = Arc::clone(&state); + let result = do_join(&state, ask).await; + node.audit_admin(&headers, "join", &target, term, &result); + result +} + +/// Leader-side join on a resolved group (m11p6): the write-pool submit + +/// response shape shared by `/cluster/join` and the +/// `/cluster/shards/{id}/replicas` (add) rebalance verb. The caller has already +/// resolved leadership/forwarding and audits the outcome. +async fn do_join( + state: &Arc, + ask: tidal_net::JoinAsk, +) -> std::result::Result { + let state_for_job = Arc::clone(state); let outcome = state .write_pool .submit(move || Ok::<_, ServerError>(state_for_job.handle_join(&ask))) .await .map_err(ClusterAppError)?; - let result = if outcome.accepted { + if outcome.accepted { Ok(( StatusCode::OK, Json(serde_json::json!({ @@ -5209,9 +5328,7 @@ pub async fn cluster_join( "join refused: {} (leader: {})", outcome.refusal_reason, outcome.leader_region )))) - }; - node.audit_admin(&headers, "join", &target, term, &result); - result + } } /// `POST /cluster/join` request body (the HTTP wrapper). @@ -5253,13 +5370,20 @@ pub struct JoinHttpRequest { pub async fn cluster_partition( State(node): State>, headers: HeaderMap, + Query(sel): Query, Json(req): Json, ) -> std::result::Result { - let state = node.replica_for(None)?; + let state = node.replica_for(sel.shard_id())?; let term = state.election_term(); let target = req.region.clone(); if !is_internal(&headers) && !state.is_leader() { - let result = forward_write(&state, "/cluster/partition", &req, &headers).await; + let result = forward_write( + &state, + &state.admin_path("/cluster/partition"), + &req, + &headers, + ) + .await; node.audit_admin(&headers, "partition", &target, term, &result); return result; } @@ -5301,13 +5425,15 @@ pub async fn cluster_partition( pub async fn cluster_heal( State(node): State>, headers: HeaderMap, + Query(sel): Query, Json(req): Json, ) -> std::result::Result { - let state = node.replica_for(None)?; + let state = node.replica_for(sel.shard_id())?; let term = state.election_term(); let target = req.region.clone(); if !is_internal(&headers) && !state.is_leader() { - let result = forward_write(&state, "/cluster/heal", &req, &headers).await; + let result = + forward_write(&state, &state.admin_path("/cluster/heal"), &req, &headers).await; node.audit_admin(&headers, "heal", &target, term, &result); return result; } @@ -5331,6 +5457,130 @@ pub async fn cluster_heal( result } +// ── Rebalancing verbs (m11p6 L3) ─────────────────────────────────────────────── + +/// `POST /cluster/shards/{id}/replicas` request body (m11p6 rebalancing). +#[derive(Serialize, Deserialize, ToSchema)] +pub struct ShardReplicaChange { + /// `"add"` — seed a node into this group's roster as a Learner (the m11p5 + /// join flow per-group; the joiner catches up via snapshot+stream and the + /// leader auto-promotes it to Voter). `"remove"` — the m11p5 fenced removal + /// of a replica from this group. + #[schema(example = "add")] + action: String, + /// The region/node name to add to or remove from this shard group. + #[schema(example = "ap-south")] + name: String, + /// The joining node's advertised gRPC address (required for `add`). + #[serde(default)] + grpc_addr: Option, + /// The joining node's advertised HTTP address (required for `add`). + #[serde(default)] + http_addr: Option, +} + +/// Add or remove a replica of one shard group (m11p6 rebalancing). +/// +/// The roadmap's `POST /cluster/shards/{id}/replicas` operator verb. It is the +/// per-group projection of the m11p5 membership flow: `add` runs the leader-side +/// join (Learner → snapshot+stream catch-up → auto-promotion to Voter) against +/// group `{id}`; `remove` runs the fenced removal on group `{id}`'s own log. +/// Both reuse the existing `/cluster/join` / `/cluster/members/remove` handlers +/// with the `?shard={id}` selector, so the non-leader forward, the conf-change +/// gate, the audit record, and the snapshot+stream cutover are the proven m11p5 +/// machinery — unchanged, just scoped to one group. +#[utoipa::path( + post, + path = "/cluster/shards/{id}/replicas", + tag = "cluster", + params(("id" = u16, Path, description = "Target shard group id")), + request_body = ShardReplicaChange, + responses( + (status = 200, description = "Replica added (Learner) or removed (Removed record committed)"), + (status = 400, description = "Unknown action/region, or this node hosts no replica of the group"), + (status = 401, description = "Missing or invalid API key"), + (status = 503, description = "Not the group leader / conf-change held; retry"), + ), + security(("bearerAuth" = [])), +)] +pub async fn shard_replicas( + State(node): State>, + Path(shard): Path, + headers: HeaderMap, + Json(req): Json, +) -> std::result::Result { + let sel = ShardSelector { shard: Some(shard) }; + match req.action.as_str() { + "add" => { + let (Some(grpc_addr), Some(http_addr)) = (req.grpc_addr, req.http_addr) else { + return Err(ClusterAppError(ServerError::BadRequest( + "shard replica 'add' requires grpc_addr and http_addr".into(), + ))); + }; + cluster_join( + State(node), + headers, + Query(sel), + Json(JoinHttpRequest { + name: req.name, + grpc_addr, + http_addr, + }), + ) + .await + } + "remove" => { + cluster_member_remove( + State(node), + headers, + Query(sel), + Json(RegionRequest { + region: req.name, + baseline: None, + }), + ) + .await + } + other => Err(ClusterAppError(ServerError::BadRequest(format!( + "unknown shard replica action '{other}' (expected 'add' or 'remove')" + )))), + } +} + +/// Transfer one shard group's leadership to a named replica (m11p6 rebalancing). +/// +/// The roadmap's `POST /cluster/shards/{id}/transfer` operator verb — the +/// rebalance-back-to-preferred-leader path (design §5/§6). It is the `REST` +/// projection of `/cluster/promote` scoped to group `{id}`: the same fenced +/// transfer (catch-up drain → `TimeoutNow` → term+1 election), forwarding, and +/// audit, reached by delegating to [`cluster_promote`] with the `?shard={id}` +/// selector so none of that delicate election logic is duplicated. +#[utoipa::path( + post, + path = "/cluster/shards/{id}/transfer", + tag = "cluster", + params(("id" = u16, Path, description = "Target shard group id")), + request_body = RegionRequest, + responses( + (status = 200, description = "Leadership transferred to the named replica"), + (status = 400, description = "Unknown region, or this node hosts no replica of the group"), + (status = 401, description = "Missing or invalid API key"), + (status = 503, description = "Transfer did not complete; the group keeps its current leader"), + ), + security(("bearerAuth" = [])), +)] +pub async fn shard_transfer( + State(node): State>, + Path(shard): Path, + headers: HeaderMap, + Json(req): Json, +) -> std::result::Result { + let sel = ShardSelector { shard: Some(shard) }; + cluster_promote(State(node), headers, Query(sel), Json(req)) + .await + .map(IntoResponse::into_response) +} + // ── Data routes ────────────────────────────────────────────────────────────── /// Create an item on the cluster (m11p2: items ride the one replicated log). diff --git a/tidal-server/src/cluster/routes.rs b/tidal-server/src/cluster/routes.rs index b00c1ee..9742437 100644 --- a/tidal-server/src/cluster/routes.rs +++ b/tidal-server/src/cluster/routes.rs @@ -903,11 +903,13 @@ impl IntoResponse for ClusterAppError { leader, http_addr, term, + shard, } => serde_json::json!({ "error": self.0.to_string(), "leader": leader, "leader_http_addr": http_addr, "term": term, + "shard": shard, }), ServerError::LeaderUnreachable { leader, diff --git a/tidal-server/src/error.rs b/tidal-server/src/error.rs index d692cd8..7eba4ab 100644 --- a/tidal-server/src/error.rs +++ b/tidal-server/src/error.rs @@ -49,6 +49,11 @@ pub enum ServerError { /// The responder's election term (m11p4): lets a forwarder tell a /// stale answer from a fresh one during election churn. term: u64, + /// The shard group this not-leader answer is for (m11p6). `None` for the + /// legacy single group (S=1, byte-for-byte). Names the group in the body + /// so a client retrying a per-shard write learns which group rejected it + /// and which `?shard=` to re-target. + shard: Option, }, /// A region-pinned read named a region this process does not own (multi- /// process cluster mode). Maps to 400 naming the region; task 03 upgrades diff --git a/tidal-server/src/openapi.rs b/tidal-server/src/openapi.rs index d6bd174..07ab6ae 100644 --- a/tidal-server/src/openapi.rs +++ b/tidal-server/src/openapi.rs @@ -199,6 +199,8 @@ pub struct ClusterApiDoc; crate::cluster::node::cluster_partition, crate::cluster::node::cluster_heal, crate::cluster::node::cluster_reseed, + crate::cluster::node::shard_replicas, + crate::cluster::node::shard_transfer, crate::cluster::node::cluster_reconcile, crate::cluster::node::cluster_reconcile_snapshot, crate::cluster::node::create_item, @@ -226,6 +228,7 @@ pub struct ClusterApiDoc; crate::cluster::node::AggregatedStatusResponse, crate::cluster::node::AggregatedRegionStatus, crate::cluster::node::RegionRequest, + crate::cluster::node::ShardReplicaChange, crate::cluster::node::HardNegRequest, crate::cluster::node::ReconcileResponse, crate::cluster::node::ReconcileSnapshotResponse, diff --git a/tidal-server/tests/cluster_sharding.rs b/tidal-server/tests/cluster_sharding.rs new file mode 100644 index 0000000..c365222 --- /dev/null +++ b/tidal-server/tests/cluster_sharding.rs @@ -0,0 +1,456 @@ +//! Tier-3 sharding × replication suite (m11p6 L4, REAL multi-process cluster). +//! +//! The m11p6 exit gate, run over real OS processes — 3 shard groups × RF=3, every +//! node a replica of every group, each group its own elected leader (balanced: +//! group `s` led by node `s`). Two pillars: +//! +//! 1. **Kill-node failover localizes to the dead node's groups** — SIGKILL a node +//! under concurrent `ack=quorum` load and prove: (a) ONLY the groups that node +//! led re-elect a new leader (the groups led by survivors keep theirs); (b) the +//! re-election completes inside the failover budget (<10s); (c) READS NEVER STOP +//! (a concurrent `/feed` poller on a survivor sees zero failures across the +//! window — reads are served from local replicas, no leader needed); and +//! (d) ZERO acknowledged-write loss per group — every item 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). +//! +//! 2. **The L3 rebalance verbs move exactly one group** — `POST +//! /cluster/shards/{id}/transfer` and `/cluster/promote?shard=` move one +//! group's leadership and leave the others untouched, and `POST +//! /cluster/shards/{id}/replicas` (remove) runs a per-group fenced conf-change. +//! +//! Kill-point count: `TIDAL_SHARDING_KILLPOINTS` (default 3 for CI; the recorded +//! exit-gate run sweeps more — see docs/planning/milestone-11/phase-6.md). +//! +//! Run: `cargo test -p tidal-server --features cluster-e2e --test cluster_sharding -- --nocapture` + +#![cfg(feature = "cluster-e2e")] +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::cast_possible_truncation, + clippy::cast_precision_loss, + clippy::too_many_lines +)] + +mod support; + +use std::collections::BTreeMap; +use std::sync::{ + Arc, + atomic::{AtomicBool, AtomicU64, Ordering}, +}; +use std::time::{Duration, Instant}; + +use support::{ + invariants::{AckLedger, item_token, post_acked}, + multiproc::{MultiProcCluster, convergence_budget}, +}; +use tidaldb::{replication::shard::ShardRouter, schema::EntityId}; + +/// Fast election (mirrors `cluster_election.rs`): heartbeat 100ms, timeout +/// 500–1000ms, lease 350ms — so a SIGKILL failover completes well inside the +/// budget. Applied to every group (each `ShardReplica` reads this block). +const FAST_ELECTION_YAML: &str = "election:\n heartbeat_interval_ms: 100\n election_timeout_min_ms: 500\n election_timeout_max_ms: 1000\n leader_lease_ms: 350"; + +/// "Failover < 10s p99" — the roadmap's m11p6 bound (also p4's). +const FAILOVER_BUDGET: Duration = Duration::from_secs(10); + +const NODES: usize = 3; +const SHARDS: usize = 3; + +/// CI-default kill points; the exit-gate run sets `TIDAL_SHARDING_KILLPOINTS` higher. +fn killpoints() -> usize { + std::env::var("TIDAL_SHARDING_KILLPOINTS") + .ok() + .and_then(|v| v.parse().ok()) + .filter(|&n| n > 0) + .unwrap_or(3) +} + +/// Find the node index whose region name matches `name`. +fn idx_of(cluster: &MultiProcCluster, name: &str) -> usize { + (0..cluster.len()) + .find(|&i| cluster.region_name(i) == name) + .unwrap_or_else(|| panic!("no node named {name}")) +} + +/// The m11p6 headline exit gate over real OS processes: kill a node under +/// `ack=quorum` load and prove only ITS shard-leaderships move (<10s), reads +/// never stop, and zero acknowledged writes are lost — across several kill points. +#[test] +fn mp_sharded_kill_node_moves_only_its_leaderships_zero_loss() { + let mut cluster = MultiProcCluster::start_sharded(NODES, SHARDS, Some(FAST_ELECTION_YAML)); + // The gateway's entity→shard hash (the same FNV-1a router every node routes by). + let router = ShardRouter::hash(SHARDS as u16).expect("build shard router"); + + // Balanced placement: each group `s` is led (term 0) by node `s`. + let initial = cluster.wait_shard_leaders_agreed(convergence_budget()); + for s in 0..SHARDS as u16 { + assert_eq!( + initial[&s], + cluster.region_name(usize::from(s)), + "group {s} should boot led by node {s} (balanced placement)" + ); + } + + let mut entity: u64 = 1; + for round in 0..killpoints() { + // Wait for a fully-converged steady state (all groups agree their leader). + let before = cluster.wait_shard_leaders_agreed(convergence_budget()); + // Target the leader of group `round % SHARDS` — "kill any node". + let target_shard = (round % SHARDS) as u16; + let target_name = before[&target_shard].clone(); + let target_idx = idx_of(&cluster, &target_name); + // The groups this node currently leads (its leaderships must move) and the + // rest (must stay put — the localization property). + let led: Vec = (0..SHARDS as u16) + .filter(|s| before[s] == target_name) + .collect(); + let others: Vec = (0..SHARDS as u16) + .filter(|s| before[s] != target_name) + .collect(); + // Write + read through a SURVIVOR gateway so the client's entry node never + // dies mid-request (it forwards group-leader writes; reads serve locally). + let survivor = (0..NODES) + .find(|&i| i != target_idx) + .expect("a live survivor gateway distinct from the kill target"); + let survivor_base = cluster.node(survivor); + + // ── Concurrent reads that must never stop ───────────────────────────── + let stop = Arc::new(AtomicBool::new(false)); + let read_failures = Arc::new(AtomicU64::new(0)); + let read_ok = Arc::new(AtomicU64::new(0)); + let reader = { + let stop = Arc::clone(&stop); + let fails = Arc::clone(&read_failures); + let oks = Arc::clone(&read_ok); + let base = survivor_base.clone(); + std::thread::spawn(move || { + let client = reqwest::blocking::Client::builder() + .timeout(Duration::from_secs(3)) + .build() + .unwrap(); + while !stop.load(Ordering::Acquire) { + match client + .get(format!("{base}/feed?profile=for_you&limit=24")) + .send() + { + Ok(r) if r.status().is_success() => { + oks.fetch_add(1, Ordering::Relaxed); + } + _ => { + fails.fetch_add(1, Ordering::Relaxed); + } + } + std::thread::sleep(Duration::from_millis(50)); + } + }) + }; + + // ── Two writer threads: ack=quorum items+views, entities spread across + // all groups by the gateway hash. Record only what the client saw acked. + let mut writers = Vec::new(); + for w in 0..2u64 { + let stop = Arc::clone(&stop); + let base = survivor_base.clone(); + let first = entity + w * 100_000; + writers.push(std::thread::spawn(move || { + let client = reqwest::blocking::Client::builder() + .timeout(Duration::from_secs(3)) + .build() + .unwrap(); + let mut acked: Vec<(u64, u64, Option)> = Vec::new(); + let mut e = first; + while !stop.load(Ordering::Acquire) { + let item_seq = post_acked( + &client, + &base, + "/items", + "quorum", + &serde_json::json!({ + "entity_id": e, "metadata": { "title": item_token(e) } + }), + ); + let view_seq = post_acked( + &client, + &base, + "/signals", + "quorum", + &serde_json::json!({ "entity_id": e, "signal": "view", "weight": 1.0 }), + ); + if let Some(s) = item_seq { + acked.push((e, s, view_seq)); + } + e += 1; + } + acked + })); + } + + // Pseudo-random kill point per round (reproducible — no Math.random here). + std::thread::sleep(Duration::from_millis(200 + (round as u64 * 131) % 500)); + cluster.kill_hard(target_idx); + stop.store(true, Ordering::Release); + + // ── Collect the acked ledger and the per-shard coverage. ────────────── + let mut ledger = AckLedger::new(); + let mut per_shard: BTreeMap = BTreeMap::new(); + for wj in writers { + for (e, item_seq, view_seq) in wj.join().expect("writer thread") { + let s = router.route(EntityId::new(e)); + *per_shard.entry(s.0).or_default() += 1; + ledger.record(e, item_seq, view_seq); + } + } + reader.join().expect("reader thread"); + + // ── (a)+(b) Failover localizes to the killed node's groups, 0, + "round {round}: the read poller never got a single 2xx — it was not exercising reads" + ); + + // A round with ZERO acked writes proves nothing — the failover would have + // had no acknowledged state to lose. The writers run hundreds of quorum + // writes in the pre-kill window, so an empty ledger means a setup fault + // (writers never got a 2xx), not a passing round. The ledger spans every + // group the gateway hash routed to (logged per-shard); the loss proof + // below probes each acked write on ITS shard's new leader, so coverage of + // a killed node's groups is whatever genuinely routed there this round. + assert!( + !ledger.is_empty(), + "round {round}: no acked writes recorded — the kill tested nothing \ + (writers saw 0 quorum acks before the kill at {target_name})" + ); + println!( + "round {round}: killed {target_name} (led {led:?}); failover -> {after:?}; \ + {} acked writes, per-shard {per_shard:?}; reads {} ok / 0 failed", + ledger.len(), + read_ok.load(Ordering::Relaxed) + ); + + // ── (d) ZERO acknowledged-write loss: every acked item is present on ITS + // shard's NEW leader. A quorum ack means the write committed (a majority + // held it durably), and the m11p4 vote restriction guarantees the + // elected leader holds every committed write — so the post-failover + // shard leader is the authoritative place to prove presence (a still + // catching-up follower is not). Poll past the text index's 2s + // auto-commit (the leader scatter-reads all its hosted groups). + for w in ledger.writes() { + let shard = router.route(EntityId::new(w.entity_id)).0; + let leader_name = after.get(&shard).expect("post-failover leader for shard"); + let leader_idx = idx_of(&cluster, leader_name); + assert!( + item_present(&cluster, leader_idx, w.entity_id, Duration::from_secs(15)), + "round {round} (killed {target_name}): ACKNOWLEDGED LOSS — item {} \ + (seq {}, shard {shard}, acked) is missing on shard {shard}'s new leader \ + {leader_name} (node {leader_idx})", + w.entity_id, + w.item_seq + ); + } + + // Restart the killed node so the cluster is whole for the next kill point + // (one node down at a time keeps every group's quorum intact). The doubled + // budget covers BOTH the rejoiner's boot+catch-up AND any group whose + // leadership is still settling — stacked, not just convergence. + entity += 10_000; + cluster.restart(target_idx, &[]); + let _ = cluster.wait_shard_leaders_agreed(convergence_budget() + convergence_budget()); + } +} + +/// The L3 rebalance verbs over real processes: `POST /cluster/shards/{id}/transfer` +/// and `/cluster/promote?shard=` move EXACTLY one group's leadership, and the +/// per-group remove verb runs a fenced conf-change — proving `?shard=` selection +/// and the per-group reuse of the m11p4/m11p5 machinery end to end. +#[test] +fn mp_sharded_rebalance_verbs_move_one_group() { + let cluster = MultiProcCluster::start_sharded(NODES, SHARDS, Some(FAST_ELECTION_YAML)); + let before = cluster.wait_shard_leaders_agreed(convergence_budget()); + for s in 0..SHARDS as u16 { + assert_eq!(before[&s], cluster.region_name(usize::from(s))); + } + let node0 = cluster.region_name(0).to_string(); + let node1 = cluster.region_name(1).to_string(); + + // ── Transfer group 0's leadership node0 -> node1 (RESTful rebalance verb). ── + let resp = cluster.post( + 0, + "/cluster/shards/0/transfer", + &serde_json::json!({ "region": node1 }), + ); + assert!( + resp.status().is_success(), + "shards/0/transfer must succeed, got {}", + resp.status() + ); + wait_until(FAILOVER_BUDGET, || { + cluster + .agreed_shard_leaders() + .is_some_and(|m| m[&0] == node1 && m[&1] == before[&1] && m[&2] == before[&2]) + }); + let mid = cluster + .agreed_shard_leaders() + .expect("agreed after transfer"); + assert_eq!( + mid[&0], node1, + "group 0 leadership must have moved to node1" + ); + assert_eq!(mid[&1], before[&1], "group 1 leadership must be untouched"); + assert_eq!(mid[&2], before[&2], "group 2 leadership must be untouched"); + + // ── Move it back with the shard-scoped promote (`?shard=` selector). ────── + let resp = cluster.post( + 0, + "/cluster/promote?shard=0", + &serde_json::json!({ "region": node0 }), + ); + assert!( + resp.status().is_success(), + "promote?shard=0 must succeed, got {}", + resp.status() + ); + wait_until(FAILOVER_BUDGET, || { + cluster + .agreed_shard_leaders() + .is_some_and(|m| m[&0] == node0) + }); + + // ── The per-group replica verb is wired + `?shard=`-scoped: assert its input + // validation and that `/cluster/members?shard=` selects the right group's + // roster. (The live add/remove conf-change reuses the m11p5 machinery + // per-group — proven by the membership suite — but a node's readiness is + // still node-global across its co-hosted groups, so removing a node from + // ONE group of a multi-group node is a tracked S>1 follow-up, NOT asserted + // here. See docs/planning/milestone-11/phase-6.md.) + let bad_action = cluster.post( + 0, + "/cluster/shards/0/replicas", + &serde_json::json!({ "action": "frobnicate", "name": node1 }), + ); + assert_eq!( + bad_action.status().as_u16(), + 400, + "an unknown shard-replica action must be a 400" + ); + let add_missing_addrs = cluster.post( + 0, + "/cluster/shards/0/replicas", + &serde_json::json!({ "action": "add", "name": "region-9" }), + ); + assert_eq!( + add_missing_addrs.status().as_u16(), + 400, + "add without grpc_addr/http_addr must be a 400" + ); + // The `?shard=` selector resolves a hosted group's roster (full placement ⇒ + // every group lists all three nodes); an unhosted shard id is a 400. + let roster = cluster.get_json(0, "/cluster/members?shard=2"); + assert_eq!( + roster["members"].as_array().map(Vec::len), + Some(NODES), + "group 2's roster must list every node under full placement" + ); + let unhosted = cluster.post( + 0, + "/cluster/shards/9/transfer", + &serde_json::json!({ "region": node0 }), + ); + assert_eq!( + unhosted.status().as_u16(), + 400, + "targeting a shard this node does not host must be a 400" + ); +} + +/// Whether item `entity` is searchable on node `idx` within `budget` — the +/// content presence probe (`/search?query=`), polling past the text +/// index's ~2s auto-commit. A clean 2xx-with-no-hit through the whole budget is +/// genuine absence (returns false); transport/non-2xx is retried until the +/// budget, so a just-promoted leader still warming up is not charged as absence +/// prematurely. +fn item_present(cluster: &MultiProcCluster, idx: usize, entity: u64, budget: Duration) -> bool { + let token = item_token(entity); + let base = cluster.node(idx); + let client = reqwest::blocking::Client::builder() + .timeout(Duration::from_secs(4)) + .build() + .unwrap(); + let deadline = Instant::now() + budget; + loop { + if let Ok(resp) = client + .get(format!("{base}/search?query={token}&limit=5")) + .send() + && resp.status().is_success() + { + let body: serde_json::Value = resp.json().unwrap_or(serde_json::Value::Null); + let hit = body["items"].as_array().is_some_and(|items| { + items + .iter() + .any(|it| it["entity_id"].as_u64() == Some(entity)) + }); + if hit { + return true; + } + } + if Instant::now() > deadline { + return false; + } + std::thread::sleep(Duration::from_millis(200)); + } +} + +/// Poll `cond` every 100ms until it returns true or `budget` elapses; returns the +/// final value of `cond` (so a caller can assert it true with context). +fn wait_until(budget: Duration, mut cond: impl FnMut() -> bool) -> bool { + let deadline = Instant::now() + budget; + loop { + if cond() { + return true; + } + if Instant::now() > deadline { + return false; + } + std::thread::sleep(Duration::from_millis(100)); + } +} diff --git a/tidal-server/tests/support/multiproc.rs b/tidal-server/tests/support/multiproc.rs index 7017354..277f11e 100644 --- a/tidal-server/tests/support/multiproc.rs +++ b/tidal-server/tests/support/multiproc.rs @@ -206,6 +206,11 @@ struct RegionPlan { grpc: SocketAddr, http: SocketAddr, data_dir: PathBuf, + /// Per-shard gRPC bind ports for the m11p6 sharded harness: index `s` is the + /// real loopback port this node's replica of shard group `s` binds. Empty for + /// the single-group (`shards:` absent) harness — that path uses [`grpc`] + /// verbatim and stays byte-for-byte the pre-m11p6 topology. + shard_grpc: Vec, } /// A running single-region process. @@ -283,6 +288,10 @@ pub struct MultiProcCluster { log: String, extra_env: HashMap>, client: reqwest::blocking::Client, + /// Shard-group count (m11p6). `1` for the single-group harness (`shards:` + /// absent); `S` for a [`start_sharded`](MultiProcCluster::start_sharded) + /// cluster. Used by the per-shard status helpers. + shards: usize, /// Held so the tempdir (configs + per-node data dirs) outlives every process. _tmp: tempfile::TempDir, } @@ -332,6 +341,7 @@ impl MultiProcCluster { grpc: free_addr(), http: free_addr(), data_dir, + shard_grpc: Vec::new(), } }) .collect(); @@ -360,6 +370,95 @@ impl MultiProcCluster { client: reqwest::blocking::Client::builder() .build() .expect("build blocking client"), + shards: 1, + _tmp: tmp, + }; + + for i in 0..harness.plans.len() { + let child = harness.spawn_process(&bin, i, &[]); + let plan = &harness.plans[i]; + harness.nodes.push(NodeHandle { + name: plan.name.clone(), + http: plan.http, + process: Some(child), + }); + } + + std::thread::sleep(SETTLE); + let deadline = Instant::now() + boot_budget(); + for i in 0..harness.nodes.len() { + harness.wait_health_inner(i, deadline); + } + drop(spawn_guard); + harness + } + + /// Spawn an `nodes`-process cluster of `shards` shard groups at FULL placement + /// (m11p6): every node replicates every group (RF = `nodes`), and group `s`'s + /// term-0 leader is node `s` (balanced placement). `topology_extra` is appended + /// to every per-process topology file (e.g. a fast-election `election:` block). + /// + /// Each `(node, shard)` binds its own real loopback gRPC port, so a node hosts + /// `shards` `ShardReplica`s, each peering with the same group's replicas on the + /// other nodes. This is the harness the m11p6 exit gate runs on: kill any node + /// → only its shard-leaderships re-elect, the rest keep serving. + /// + /// # Panics + /// + /// Panics if `shards < 1`, `nodes < 2`, `shards > nodes` (balanced placement + /// needs a distinct preferred-leader node per group), files cannot be written, + /// or any process does not become healthy within [`boot_budget`]. + #[must_use] + pub fn start_sharded(nodes: usize, shards: usize, topology_extra: Option<&str>) -> Self { + assert!(nodes >= 2, "need at least 2 nodes for a cluster"); + assert!(shards >= 1, "need at least 1 shard group"); + assert!( + shards <= nodes, + "balanced placement needs a distinct preferred-leader node per group \ + (shards {shards} > nodes {nodes})" + ); + let spawn_guard = spawn_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + + let tmp = tempfile::tempdir().expect("create temp dir"); + let bin = tidal_server_bin(); + + // Per node: one HTTP gateway port + one real gRPC port per shard group it + // hosts (full placement ⇒ `shards` of them). `grpc` (the node base) is + // shard 0's port, matching the topology `regions[].grpc_addr` convention. + let plans: Vec = (0..nodes) + .map(|i| { + let data_dir = tmp.path().join(format!("region-{i}")); + std::fs::create_dir_all(&data_dir).expect("create per-node data dir"); + let shard_grpc: Vec = (0..shards).map(|_| free_addr()).collect(); + RegionPlan { + name: region_name(i), + grpc: shard_grpc[0], + http: free_addr(), + data_dir, + shard_grpc, + } + }) + .collect(); + + let schema_path = write_schema(tmp.path()); + let topology_paths: Vec = (0..nodes) + .map(|i| write_sharded_topology_for(tmp.path(), &plans, i, shards, topology_extra)) + .collect(); + + let mut harness = Self { + plans, + nodes: Vec::new(), + schema_path, + topology_paths, + rewrite: identity_rewrite(), + log: "warn".into(), + extra_env: HashMap::new(), + client: reqwest::blocking::Client::builder() + .build() + .expect("build blocking client"), + shards, _tmp: tmp, }; @@ -739,6 +838,7 @@ impl MultiProcCluster { grpc, http, data_dir, + shard_grpc: Vec::new(), }); self.topology_paths.push(knob_path); self.nodes.push(NodeHandle { @@ -961,6 +1061,80 @@ impl MultiProcCluster { None } + // ── Per-shard leadership (m11p6 sharded harness) ─────────────────────────── + + /// The shard-group count this harness booted (1 for the single-group harness). + #[must_use] + pub const fn shard_count(&self) -> usize { + self.shards + } + + /// The leader EVERY live node agrees on for each shard group, or `None` if any + /// live node disagrees, reports a null leader, or is unreachable. The + /// agreed-leader read used to detect the converged steady state and to detect a + /// failover (the map changes for exactly the killed node's groups). + /// + /// Reads the per-shard `shards[]` rows in `/cluster/status/local`: for each + /// group it collects every live node's reported leader and accepts the group + /// only when all of them are present and identical. + #[must_use] + pub fn agreed_shard_leaders(&self) -> Option> { + let live: Vec = self.live_indices(); + if live.is_empty() { + return None; + } + let mut per_shard: HashMap = HashMap::new(); + for &idx in &live { + let st = self.local_status(idx)?; + let rows = st["shards"].as_array()?; + // `shards[]` is one row per group this node HOSTS — a constant under + // full placement (every node hosts every group), so a short array + // means the node is still booting/unreachable (its status raced the + // listener), not a leadership-convergence lag. Treating it as + // not-yet-agreed (wait) is correct: a just-restarted rejoiner reports + // a short/empty array until its groups open. (Partial placement would + // make this per-node; the exit gate is full placement.) + if rows.len() != self.shards { + return None; + } + for row in rows { + let shard = u16::try_from(row["shard"].as_u64()?).ok()?; + let leader = row["leader"].as_str()?.to_string(); + match per_shard.get(&shard) { + Some(seen) if seen != &leader => return None, // disagreement + Some(_) => {} + None => { + per_shard.insert(shard, leader); + } + } + } + } + (per_shard.len() == self.shards).then_some(per_shard) + } + + /// Block until every live node agrees on a (non-null) leader for every shard + /// group, returning the agreed `shard -> leader` map. The sharded analogue of + /// [`wait_leader_agreed`](Self::wait_leader_agreed). + /// + /// # Panics + /// + /// Panics on timeout, dumping the last partial view. + #[must_use] + pub fn wait_shard_leaders_agreed(&self, timeout: Duration) -> HashMap { + let deadline = Instant::now() + timeout; + loop { + if let Some(map) = self.agreed_shard_leaders() { + return map; + } + assert!( + Instant::now() <= deadline, + "shard leaders did not converge within {timeout:?}; last partial view: {:?}", + self.agreed_shard_leaders() + ); + std::thread::sleep(POLL_INTERVAL); + } + } + /// Best-effort hint for a failing node, to make boot failures diagnosable in /// the panic message. Node logs go to /dev/null by default (re-run with /// `TIDAL_TEST_NODE_LOGS=inherit` to see them), so we surface the bind @@ -1068,6 +1242,74 @@ fn write_topology_for( path } +/// Write the sharded topology file process `idx` loads (m11p6 full placement): +/// a `regions:` list (every node) plus a `shards:` block declaring `shards` +/// groups, each replicated across every node and led (term-0) by node `s`. +/// +/// Each `(node, shard)` replica entry carries its own real loopback `grpc_addr` +/// (`plans[node].shard_grpc[shard]`), so the engine binds a distinct socket per +/// hosted group — no derived-port collisions. With identity addressing every +/// process's file is byte-identical (the kill-node exit gate uses SIGKILL, not +/// TCP-proxy partitions, so no per-observer rewrite is needed). The legacy +/// top-level `leader:` is set but unused once `shards:` is present. +fn write_sharded_topology_for( + dir: &Path, + plans: &[RegionPlan], + idx: usize, + shards: usize, + extra: Option<&str>, +) -> PathBuf { + // Full placement: balanced leaders need a distinct node per group, and every + // node must have one bound port per group. Assert both up front so a future + // refactor that desyncs `shards` from the allocation fails loudly here, not as + // an out-of-bounds index mid-write or an opaque engine bind error. + assert!( + shards <= plans.len(), + "balanced placement needs nodes ({}) >= shards ({shards})", + plans.len() + ); + for plan in plans { + assert!( + plan.shard_grpc.len() >= shards, + "node {} has {} shard ports, need {shards}", + plan.name, + plan.shard_grpc.len() + ); + } + let path = dir.join(format!("topology-{idx}.yaml")); + let mut body = String::from("regions:\n"); + for plan in plans { + // The node's base grpc_addr is shard 0's port (the topology convention); + // each group sets its own explicit grpc_addr in the `shards:` block below. + let _ = writeln!(body, " - name: {}", plan.name); + let _ = writeln!(body, " grpc_addr: \"{}\"", plan.shard_grpc[0]); + let _ = writeln!(body, " http_addr: \"{}\"", plan.http); + } + let _ = writeln!(body, "shards:"); + for s in 0..shards { + let _ = writeln!(body, " - id: {s}"); + // Balanced placement: group `s`'s term-0 leader is node `s`. + let _ = writeln!(body, " leader: {}", plans[s].name); + let _ = writeln!(body, " replicas:"); + for plan in plans { + let _ = writeln!( + body, + " - {{ node: {}, grpc_addr: \"{}\" }}", + plan.name, plan.shard_grpc[s] + ); + } + } + // Legacy field — required by the loader, ignored once `shards:` is present. + let _ = writeln!(body, "leader: {}", plans[0].name); + if let Some(extra) = extra { + let _ = writeln!(body, "{extra}"); + } + let mut f = std::fs::File::create(&path).expect("create sharded topology file"); + f.write_all(body.as_bytes()) + .expect("write sharded topology file"); + path +} + /// Write the shared schema file. Matches the signal set the in-process route /// tests use (`view` decayed, `like` decayed, `hide` permanent for `/hardnegs`) /// so the harness asserts against identical engine behavior, plus a `title` text diff --git a/tidal-stress/benches/hotpath.rs b/tidal-stress/benches/hotpath.rs index bd22410..21d6dc7 100644 --- a/tidal-stress/benches/hotpath.rs +++ b/tidal-stress/benches/hotpath.rs @@ -14,19 +14,18 @@ use std::time::Duration; use criterion::{Criterion, criterion_group, criterion_main}; use tidal_stress::metrics::LatencyHistogram; -use tidal_stress::workload::{Workload, WritePath, parse_mix}; +use tidal_stress::workload::{Workload, parse_mix}; fn make_workload(mix: &str) -> Workload { Workload::new( - // Three read gateways (round-robin), leader-pinned writes => write_bases - // len == 1, the headline `--leader-url` path (exercises the F4 short-circuit). + // Three read gateways (round-robin), writes pinned to one gateway => + // write_bases len == 1, the headline `--leader-url` path. vec![ "http://10.0.0.1:9500".into(), "http://10.0.0.2:9500".into(), "http://10.0.0.3:9500".into(), ], vec!["http://10.0.0.1:9500".into()], - WritePath::Leader, parse_mix(mix).expect("mix preset parses"), 10_000, // corpus 50_000, // users diff --git a/tidal-stress/src/main.rs b/tidal-stress/src/main.rs index 29aa3df..7c09333 100644 --- a/tidal-stress/src/main.rs +++ b/tidal-stress/src/main.rs @@ -21,7 +21,7 @@ use tidal_stress::error::{Result, StressError}; use tidal_stress::metrics::{self, StageStats}; use tidal_stress::scheduler::{Stage, parse_ramp, run_stage}; use tidal_stress::summary::{GateConfig, RunSummary, StageSummary}; -use tidal_stress::workload::{OpKind, Workload, WritePath, parse_mix}; +use tidal_stress::workload::{OpKind, Workload, parse_mix}; #[derive(Parser)] #[command( @@ -38,8 +38,10 @@ struct Cli { #[arg(long = "target", required = true)] targets: Vec, - /// Pin leader-path writes (/signals,/items,/embeddings) to this URL to avoid - /// the follower→leader forward hop. Ignored for --write-path sharded. + /// Pin all writes (`/signals`,`/items`,`/embeddings`) to this one gateway URL + /// instead of spreading them round-robin across `--target`s. The write still + /// hash-routes to the owning shard group's leader from there (m11p6), so this + /// only fixes the entry gateway — useful to isolate one node's forward cost. #[arg(long)] leader_url: Option, @@ -65,11 +67,6 @@ struct Cli { #[arg(long, default_value = "peach")] mix: String, - /// Write surface: `leader` (replicated, single-leader funnel) or `sharded` - /// (hash-partitioned across regions, not replicated). - #[arg(long, default_value = "leader")] - write_path: String, - /// Items (id 1..=N) + embeddings to seed before the ramp. #[arg(long, default_value_t = 10_000)] corpus: u64, @@ -175,15 +172,6 @@ async fn run() -> Result<()> { "at least one --target is required".into(), )); } - let write_path = match cli.write_path.as_str() { - "leader" => WritePath::Leader, - "sharded" => WritePath::Sharded, - other => { - return Err(StressError::Target(format!( - "unknown --write-path '{other}'" - ))); - } - }; let mix = parse_mix(&cli.mix)?; let stages = parse_ramp(&cli.ramp, cli.stage_secs)?; if let Some(ack) = cli.ack.as_deref() @@ -215,26 +203,27 @@ async fn run() -> Result<()> { cli.ack.clone(), )?); - // The leader URL anchors seeding (items must broadcast from the leader so - // every region's /feed can rank them) and, for the leader write path, the - // pinned write target when given. + // The seed URL anchors corpus seeding (any gateway accepts items and + // hash-routes them to their shard leader, m11p6) — default to the first + // target, or the pinned gateway when `--leader-url` is given. let leader_url = cli .leader_url .clone() .unwrap_or_else(|| cli.targets[0].clone()); let read_bases = cli.targets.clone(); - let write_bases = match (write_path, &cli.leader_url) { - (WritePath::Leader, Some(url)) => vec![url.clone()], - _ => cli.targets.clone(), - }; + // m11p6: ONE write path. Every write hash-routes from any gateway, so spread + // them round-robin across the targets unless `--leader-url` pins one gateway. + let write_bases = cli + .leader_url + .clone() + .map_or_else(|| cli.targets.clone(), |url| vec![url]); println!("tidal-stress — thepeach feed workload"); println!(" targets : {}", cli.targets.join(", ")); println!( - " write path : {} ({})", - cli.write_path, + " write path : /signals,/items,/embeddings — hash-routed + replicated ({})", if cli.leader_url.is_some() { - "leader pinned" + "one gateway pinned" } else { "round-robin gateways" } @@ -293,7 +282,6 @@ async fn run() -> Result<()> { let workload = Arc::new(Workload::new( read_bases, write_bases, - write_path, mix, cli.corpus, cli.users, @@ -472,8 +460,7 @@ fn print_verdict( ); } println!( - " scaling levers : write path = {} (try --write-path sharded to remove the single-leader funnel);\n leader pod cpu \"2\" caps the write pool at ~2 workers — more CPU on the leader raises it (clamp max 8).", - cli.write_path, + " scaling levers : writes hash-route across shard groups (m11p6 — add shards to scale write throughput horizontally);\n each group's leader pod cpu \"2\" caps its write pool at ~2 workers — more CPU per node raises it (clamp max 8)." ); if cli.embedding_dim == 128 { println!( diff --git a/tidal-stress/src/workload.rs b/tidal-stress/src/workload.rs index f61294e..d14ab05 100644 --- a/tidal-stress/src/workload.rs +++ b/tidal-stress/src/workload.rs @@ -79,37 +79,16 @@ impl OpKind { } } -/// Which write surface the signal/item/embedding writes target. -#[derive(Clone, Copy, PartialEq, Eq)] -pub enum WritePath { - /// `/signals` etc. — every write funnels to the single leader (us-east); a - /// write to a follower adds a forward hop. Replicated (the real feedback loop). - Leader, - /// `/sharded/*` — hash-partitioned across all 3 regions, no leader funnel. - /// NOT replicated (single owner per entity) — a horizontal-scale comparison. - Sharded, -} - -impl WritePath { - fn signal_path(self) -> &'static str { - match self { - Self::Leader => "/signals", - Self::Sharded => "/sharded/signals", - } - } - fn item_path(self) -> &'static str { - match self { - Self::Leader => "/items", - Self::Sharded => "/sharded/items", - } - } - fn embedding_path(self) -> &'static str { - match self { - Self::Leader => "/embeddings", - Self::Sharded => "/sharded/embeddings", - } - } -} +/// The replicated write surface (m11p6: ONE path). Before m11p6 the tool chose +/// between `/signals` (replicated, single-leader funnel) and `/sharded/*` +/// (hash-partitioned, NOT replicated) to compare scaling shapes. m11p6 ended +/// that split: `/signals`//`/items`//`/embeddings` now hash-route to the owning +/// shard group's leader AND replicate at RF — sharding × replication, one +/// surface. Any gateway accepts the write and routes it, so the tool spreads +/// writes across every `--target` (or pins them with `--leader-url`). +const SIGNAL_PATH: &str = "/signals"; +const ITEM_PATH: &str = "/items"; +const EMBEDDING_PATH: &str = "/embeddings"; #[derive(Clone, Copy)] pub enum HttpMethod { @@ -220,7 +199,6 @@ impl Mix { pub struct Workload { reads: RoundRobin, writes: RoundRobin, - write_path: WritePath, mix: Mix, feed_profiles: Vec<(f64, &'static str)>, // cumulative profile_total: f64, @@ -255,7 +233,6 @@ impl Workload { pub fn new( read_bases: Vec, write_bases: Vec, - write_path: WritePath, mix_weights: Vec<(OpKind, f64)>, corpus: u64, users: u64, @@ -281,7 +258,6 @@ impl Workload { Self { reads: RoundRobin::new(read_bases), writes: RoundRobin::new(write_bases), - write_path, mix: Mix::from_weights(&mix_weights), feed_profiles: cum, profile_total: acc, @@ -369,7 +345,7 @@ impl Workload { OpKind::SignalSkip => ("skip", 1.0), _ => ("view", 1.0), }; - let url = format!("{}{}", self.writes.pick(), self.write_path.signal_path()); + let url = format!("{}{SIGNAL_PATH}", self.writes.pick()); let body = Body::Signal { entity_id: self.pick_item(rng), signal: name, @@ -384,7 +360,7 @@ impl Workload { } OpKind::RegisterItem => { let id = self.pick_item(rng); - let url = format!("{}{}", self.writes.pick(), self.write_path.item_path()); + let url = format!("{}{ITEM_PATH}", self.writes.pick()); let body = Body::Item { entity_id: id, metadata: ItemMetadata { @@ -400,7 +376,7 @@ impl Workload { } } OpKind::RegisterEmbedding => { - let url = format!("{}{}", self.writes.pick(), self.write_path.embedding_path()); + let url = format!("{}{EMBEDDING_PATH}", self.writes.pick()); let body = Body::Embedding { entity_id: self.pick_item(rng), values: self.random_embedding(rng), @@ -504,7 +480,6 @@ mod tests { let wl = Workload::new( vec!["http://x".into()], vec!["http://x".into()], - WritePath::Leader, parse_mix("peach").expect("mix"), 1000, 100,