From 3bfde53b90efbafd4ae047a33aab6bbe8231a528 Mon Sep 17 00:00:00 2001 From: jx12n Date: Fri, 12 Jun 2026 23:06:41 -0600 Subject: [PATCH] =?UTF-8?q?feat(m11):=20data-plane=20sharding=20=C3=97=20r?= =?UTF-8?q?eplication=20(m11p6=20L0-L2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ClusterNode hosts a BTreeMap>: writes hash-route to the owning shard leader, reads scatter over shard groups. In-group shard==region preserved so the engine and tidal-net are untouched; S=1 stays byte-for-byte (today's cluster is a 1-shard × RF=N group). Topology grows shard-group awareness; membership, election, forward, reseed, and join_boot thread ShardId through. Proven by an in-process 2×2 RF=2 gRPC test plus S=1 parity, incl. tier-3 real-OS-process failover. clippy/fmt clean. --- docs/planning/milestone-11/phase-6.md | 259 ++++++ tidal-server/src/cluster/election_driver.rs | 8 +- tidal-server/src/cluster/forward.rs | 47 +- tidal-server/src/cluster/join_boot.rs | 21 +- tidal-server/src/cluster/membership.rs | 123 ++- tidal-server/src/cluster/mod.rs | 28 +- tidal-server/src/cluster/node.rs | 950 ++++++++++++++++---- tidal-server/src/cluster/reseed.rs | 18 +- tidal-server/src/cluster/topology.rs | 431 ++++++++- tidal-server/src/main.rs | 30 +- tidal-server/tests/cluster_grpc.rs | 2 + tidal-server/tests/cluster_region.rs | 350 +++++++- tidal-server/tests/cluster_routes.rs | 16 +- tidal-server/tests/support/multiproc.rs | 2 +- 14 files changed, 2030 insertions(+), 255 deletions(-) create mode 100644 docs/planning/milestone-11/phase-6.md diff --git a/docs/planning/milestone-11/phase-6.md b/docs/planning/milestone-11/phase-6.md new file mode 100644 index 0000000..d6c1fff --- /dev/null +++ b/docs/planning/milestone-11/phase-6.md @@ -0,0 +1,259 @@ +# m11p6 — Sharding × Replication + Rebalancing (IN PROGRESS) + +Phase spec and exit gate: [docs/roadmap-to-cluster.md §4/m11p6](../../roadmap-to-cluster.md). +Closes the ROADMAP **write-scaling** gap ("EITHER replicated (1 leader for +everything) OR sharded (no replication)"). +Predecessors: p1 (ship queue/perf), p2 (one log), p3 (quorum/commit-index), +p4 (election/fencing), p5 (membership/discovery/elasticity). + +**Goal:** writes scale horizontally **without** giving up replication. The +"replicated XOR sharded" split ends: `S` shard groups, each a replication group +at RF (default 3) with its **own WAL/relay/commit-index and its own elected +leader**, leaders balanced across nodes; regions demote to placement labels; +any gateway hash-routes to the shard leader (write) or nearest replica (read); +rebalancing = snapshot + stream + fenced cutover. + +## The load-bearing realization (why this is tractable, not a rewrite) + +A single `TidalDb` is **already a complete, self-contained shard** (scale-arch +spec §1; single-process mode already runs N real `TidalDb` in one process). And +**today's cluster is exactly "1 shard × RF=N"** — every region replicates one +log under one leader. So m11p6 = **run today's whole single-shard cluster +machinery `N` times, once per data-shard group**, where each group spans only +its replica nodes, in its own data subdir + gRPC port. The data-shard number +`S` is purely a *gateway-routing + directory + port* concern; **inside a group, +"shard id == region id" is preserved unchanged**. + +Consequences: +- **Engine replication (relay/commit/election/ship/receiver/membership): NO + semantic change** — each runs per-group exactly as today's whole cluster. +- **`tidal-net`: NO change** — each group owns its own `GrpcTransport` bound to + a distinct port, peering only with that group's replica nodes. +- **`S=1` is byte-for-byte today's behavior** — every existing tier-3 suite + (quorum, membership, election, region, runbook, chaos) runs in `S=1` mode and + stays green by construction. +- The work concentrates in **`tidal-server`**: decompose the per-shard + machinery out of `RegionClusterState` into `ShardReplica`; the node hosts + `BTreeMap>`; add hash-routing to the right group's + leader. Plus topology schema, route unification, rebalancing, stress/harness, + the exit gate, docs. + +## Design (as adopted) + +### 1. Two types: process node + shard replica + +> **As-built note (decided during L1):** the §1-era naming below was superseded. +> The process node is a NEW type **`ClusterNode`** (not "keep the name +> `RegionClusterState`"); the former `RegionClusterState` was **renamed to +> `ShardReplica`** (it already WAS one group's machinery). The responsibilities +> are exactly as described — only the node's type name changed from the planned +> `RegionClusterState` to `ClusterNode`. + +- **`ClusterNode` (NEW process node)** = the **process node** and the axum + `State`. Owns the node identity (`region`/`region_name`), the gateway + `ShardRouter` + `node_http` map, the shared forward client, and + `groups: BTreeMap>` (the groups this node hosts) + plus a `placement` map of every group (to forward writes for groups it does + NOT host). The HTTP handlers stay free functions over + `State>`; each hash-routes the target group and either calls + a local `ShardReplica` method or forwards to that group's leader. +- **`ShardReplica` (the renamed `RegionClusterState`)** = one shard group's full + replication machinery, essentially the per-shard fields + methods of the old + `RegionClusterState`: + `db: Arc`, `ship_feed`, `ship_queue`, `commit` + `commit_watch` + bridge, `election_runtime`/`election_store`/`election_boot`, + `membership: MembershipView` + `membership_store`, `stream_baseline`, + snapshot source + `reseed_marker_store`, `leader: RwLock>`, + `activation_prev`, `partitioned`, `deferred_retires`, `ack_default`, + `quorum_timeout`, and its **own `GrpcTransport`** bound to this group's port. + Construction = today's `RegionClusterState::new` body, parameterized by + `(shard, replica nodes, bootstrap leader, group data subdir, group grpc port)`. + +### 2. Shard identity, ports, and data dirs + +- **In-group identity is region-id based** (preserved from today). `ShardReplica` + for data-shard `S` on node `N` opens `TidalDb` with + `NodeConfig{ shard_id: ShardId(N.region_id), peer_shards: , .. }`. Two groups on one node use the same `shard_id` + value but live in separate data dirs + separate transports, so nothing + collides. The data-shard `S` never enters the replication wire. +- **gRPC ports:** each replica entry may set an explicit `grpc_addr`/`grpc_bind` + per `(node, shard)`; otherwise derive `node.base_port + S`. Legacy `S=0` → + offset 0 → the region's declared address verbatim. +- **Data dirs:** explicit `shards:` → `/shard-{S:05}/` per hosted + group. Legacy (synthesized 1-shard) → `` **verbatim** (existing + on-disk clusters restart unchanged). +- **`/metrics`:** the node owns ONE listener on `metrics_addr`; the engine's + per-`TidalDb` metrics HTTP server is suppressed in cluster mode. The node + renders each group's `cluster_metrics` with a `shard="S"` label and the + per-shard engine metrics likewise. + +### 3. Topology schema (backward compatible) + +`TopologySpec` gains optional `shards: Option>` and `RegionSpec` +gains optional `zone: Option` (placement/read-affinity label), both +`#[serde(default)]`. + +```yaml +nodes: # `regions:` still accepted as the node list + - { name: us-east, grpc_addr: ..., http_addr: ..., zone: az-a } +shards: # NEW — optional + - id: 0 + leader: us-east # term-0 / preferred leader (balances placement) + replicas: # nodes hosting this group's RF replicas + - { node: us-east } # grpc_addr/grpc_bind optional → derive base+S + - { node: eu-west } + - { node: ap-south } + - { id: 1, leader: eu-west, replicas: [...] } + - { id: 2, leader: ap-south, replicas: [...] } +``` + +**Absent `shards:` ⇒ legacy synthesis:** one `ShardSpec{ id: 0, +leader: topology.leader, replicas: }`. This is +literally today's "1 shard × RF=all-regions," so the model unifies cleanly. + +Validation: shard ids dense & unique; each `leader`/`replica.node` names a +declared region; RF ≥ 1; a node's shards do not collide on derived ports. + +### 4. Routing (the new gateway layer) + +Any node, any write to entity `E`: +1. `S = ShardRouter.route(E)` (the engine's FNV-1a hash; the same router the + `/sharded/*` path already uses). +2. If this node hosts group `S` **and leads it** → apply locally on + `ShardReplica[S]` (the existing stage→complete→ship→await-quorum path). +3. If this node hosts group `S` but follows → forward to group `S`'s leader + (resolved from `ShardReplica[S].leader`), to the leader node's `http_addr`. +4. If this node does not host group `S` → forward to any replica node of `S` + (from topology); that node routes to the leader. Term fencing + the + `NotLeader` retry bound the forward chain. + +Reads route to a local replica of `S` if present, else nearest (zone) / any +replica. **Scatter-gather merges over shard GROUPS:** one replica per group, +totals **summed** (entity-sharded groups are disjoint — no dedup-by-max), the +existing degraded/`unavailable_shards`/deadline semantics retained. + +**Route unification:** `/items`, `/embeddings`, `/signals` become the shard- +routed surface (they hash-route instead of forwarding to a single global +leader); `/sharded/*` is retained as an alias of the same path. `x-tidal-ack`, +`x-tidal-seq`, quorum await, and `NotLeader`/`QuorumTimeout` all become +per-shard; `NotLeader` names the shard and its leader. + +### 5. Leadership balance + +`ShardSpec.leader` is each group's term-0 / preferred leader; the operator (and +the test harness) set shard `i`'s leader to node `i` for balanced placement. +Each group boot-classifies its own leader exactly like today's `topology.leader`, +but per group. On failover, the group's election picks the surviving max-applied +voter (m11p4 protocol, unchanged); only the dead node's groups elect. An +optional leadership-transfer-back-to-preferred is the rebalance path, not the +availability mechanism. + +### 6. Rebalancing (operator-triggered) + +Shard move / replica change reuses m11p5 verbatim, **per group**: a conf-change +on the group's own log adds a Learner replica (a new node, or an existing node +gaining a replica of group `S`), which catches up via `FetchSnapshot` + +`StreamSegments` (per-group, already per-instance once §1 lands), auto-promotes +Learner→Voter, then the operator may transfer leadership; removing a replica is +the m11p5 fenced removal on that group's log. Endpoints: +`POST /cluster/shards/{id}/replicas` (add/remove) and +`POST /cluster/shards/{id}/transfer`. Per the roadmap, **automatic rate-limited +rebalancing is explicitly "later"** — m11p6 ships the operator verbs. + +## Exit gate (from the roadmap) + +- 3 shards × RF=3: **≥5,000 quorum signals/s** (≥2.5× single-shard p3). +- Kill any node → **only its shard-leaderships move (<10 s), reads never stop**. +- `tidal-stress` sharded-vs-replicated comparison **collapses into one path**. + +## Layer plan (each keeps the workspace green) + +- **L0** — this doc + topology schema (`shards:`/`zone:` + validation + legacy + synthesis). `S=1` unchanged. +- **L1** — extract `ShardReplica`; node hosts a one-entry map; suppress engine + metrics server in cluster mode + node aggregates. `S=1` byte-for-byte. +- **L2** — multi-shard construction + gateway hash-routing + per-shard forward + + route unification + scatter-gather over groups. +- **L3** — operator rebalancing verbs. +- **L4** — tidal-stress path collapse + nodes×shards tier-3 harness + + `cluster_sharding.rs` exit gate (run for real). +- **L5** — docs (runbook/monitoring/roadmap/CHANGELOG/k8s/spec) + memory. + +## Status + +- [x] L0 topology schema (`shards:`/`zone:` + resolver + validation; legacy + synthesis) + this doc — 20 topology tests green incl. 5 new shard tests. +- [x] L1 `RegionClusterState` → `ShardReplica` rename + group-parameterized + `ShardReplica::new(.., group, enable_metrics)` (leader/peers/voters/data- + dir/metrics from the resolved group; in-group identity stays region-based). + **S=1 byte-for-byte green** (116 lib + cluster_region 11 + cluster_routes 6 + + cluster_grpc 2). NB: the `ClusterNode` process wrapper + multi-shard + hosting moved to L2 (inseparable from the routing/handler rewiring). +- [x] L2 **DONE + green** (the sharding × replication data plane). `ClusterNode` + (the axum State + process handle) hosts `BTreeMap>`, + built by `ClusterNode::new(topology, region, schema, profiles, data_dir, + hlc)` (resolves groups; opens one `ShardReplica` per hosted group — own data + subdir `shard-/` + derived/explicit gRPC port for `S>1`; node data dir + verbatim + metrics owner for `S=1`). `ServeState for ClusterNode` starts + every group's election driver and shuts each down via `Arc::try_unwrap`. + Entity writes (`/items`,`/embeddings`,`/signals`,`/hardnegs`) hash-route via + `route_entity` → local replica (existing leader/forward path) or + `forward_to_group_node` (remote group). Reads (`/feed`,`/search`) scatter + in-process over hosted groups and merge (sum totals, score-sort — S=1 + unchanged). `/cluster/status/local` gains a per-shard `shards[]` array; + `region_health` aggregates all groups. + **Verified:** in-process **2 nodes × 2 shards × RF=2** test + (`region_sharded_writes_route_per_shard_and_reads_scatter`) — writes route + per-shard (A forwards shard-1 writes to B), each shard replicates to its + follower, reads scatter over both groups and return all items from BOTH + shards, over real gRPC. **S=1 byte-for-byte:** 116 lib + cluster_region 12 + + cluster_routes 6 + cluster_grpc 2 + tier-3 `cluster_multiproc` 5 (real OS + processes: replication <2s, leader-crash failover <10s). clippy `-D` + fmt + clean. **Deferred to L2-followups (tracked):** per-shard admin verbs + (`?shard=` on promote/heal/partition/etc.) and `/sharded/*`→unified-path + aliasing — currently the admin verbs + `/sharded/*` target the first hosted + group (correct for `S=1`; multi-shard admin needs the selector). + + **Known L2 limitations — `S>1` only, FULL-placement assumed (tracked for L4, + NOT silently dropped):** these are correct for `S=1` and for the headline + exit-gate shape (3×3 RF=3, where every node hosts every group), and the writes + proven by the 2×2 test are unaffected — but a genuinely PARTIAL placement (a + node hosting a strict subset of groups) hits them: + 1. **Unified reads are local-only.** `/feed`//`/search` scatter over the + groups THIS node hosts (`hosted_dbs`); a node that does not host every group + returns a strict-subset corpus with a 200 and no `degraded` flag. Complete + only under full placement; cross-node read fan-out (reuse the `/sharded/*` + `HttpShardContext`) is the L4 item. Use `/sharded/*` for partial placements. + 2. **No cross-shard diversity re-rank.** The scatter merge keeps each group's + own diversity pass then score-merges; it does not re-run diversity ACROSS + groups, so an `S>1` `/feed` can be less diverse than `S=1`. Summing totals + over disjoint groups is exact for cardinality; a cross-shard MMR re-rank + (engine entry point) is the L4 follow-up. + 3. **`S>1` observability is one-group-deep.** Only the metrics-owner group + binds the engine `/metrics` listener; the other hosted groups' per-shard + `cluster_metrics` (ship/relay/commit/quorum) are not rendered. The + kill-node gate uses `/cluster/status/local` (and now `/cluster/status`) + `shards[]`, which IS per-group; node-level per-shard `/metrics` rendering + (design §2) is the L4/p8 item. +- [ ] 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 + +## Exit-gate evidence + +**Data-plane correctness (local, in-process, real gRPC) — DONE.** The 2×2 test +proves the replicated-XOR-sharded split is gone: a single gateway routes writes +to the owning shard's leader (forwarding cross-node), each shard group replicates +independently, and corpus reads merge across groups. The per-shard election + +quorum + commit machinery is the unchanged m11p4/m11p3 code instantiated per +group, and S=1 failover is re-proven over real OS processes (`cluster_multiproc`). + +**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). diff --git a/tidal-server/src/cluster/election_driver.rs b/tidal-server/src/cluster/election_driver.rs index 3538c5f..b0c3696 100644 --- a/tidal-server/src/cluster/election_driver.rs +++ b/tidal-server/src/cluster/election_driver.rs @@ -3,7 +3,7 @@ //! The pure state machine lives in the engine //! ([`tidaldb::replication::ElectionState`]); this module owns everything the //! machine deliberately does not — timers, RPC fan-outs, durable persistence, -//! and the leadership transitions on [`RegionClusterState`]: +//! and the leadership transitions on [`ShardReplica`]: //! //! * **One driver thread per node** ticks the machine every //! [`TICK_INTERVAL`] and drains the [`ElectionNetEvent`] inbox (vote @@ -39,7 +39,7 @@ use tidaldb::replication::{ shard::{RegionId, ShardId}, }; -use super::{node::RegionClusterState, topology::shard_of_region}; +use super::{node::ShardReplica, topology::shard_of_region}; /// Driver tick cadence: the machine's own deadlines (election timeout, /// heartbeat due, lease) are all ≥ 100ms-scale, so 50ms keeps every deadline @@ -62,7 +62,7 @@ pub struct ElectionRuntime { quarantined: AtomicBool, /// The node, for transitions and log-position reads. Weak: the runtime /// is owned BY the node; a strong ref would leak both. - node: Weak, + node: Weak, net: ElectionNet, /// The driver's inbox for async RPC outcomes. inbox_tx: mpsc::Sender, @@ -697,7 +697,7 @@ fn decide_join(term: u64, own: LogPosition, prev_log: LogPosition) -> JoinDecisi /// spawn the driver thread. Returns the runtime handle the node stores (for /// status reads and shutdown). pub fn start( - node: &Arc, + node: &Arc, machine: ElectionState, store: tidaldb::replication::ElectionStore, hooks_cell: &Arc>>, diff --git a/tidal-server/src/cluster/forward.rs b/tidal-server/src/cluster/forward.rs index 26f7adb..d6a2809 100644 --- a/tidal-server/src/cluster/forward.rs +++ b/tidal-server/src/cluster/forward.rs @@ -21,7 +21,7 @@ //! concurrently with the marker set, collecting per-peer success/failure so the //! handler can report `{"replicated_to": n, "failed": [..]}` honestly. //! -//! The shared [`reqwest::Client`] lives on [`RegionClusterState`] (connection +//! The shared [`reqwest::Client`] lives on [`ShardReplica`] (connection //! pooling across requests); its timeouts are tight (connect 1s, request 5s) so //! a dead peer degrades a forward into a clean 503/partial result instead of //! hanging an axum worker. Operation-specific budgets override the client @@ -32,7 +32,11 @@ use std::time::Duration; -use axum::http::{HeaderMap, StatusCode, header::AUTHORIZATION}; +use axum::{ + Json, + http::{HeaderMap, HeaderName, HeaderValue, StatusCode, header::AUTHORIZATION}, + response::{IntoResponse, Response}, +}; use serde::Serialize; /// Header name for the internal-propagation marker. @@ -97,7 +101,7 @@ pub const STATUS_PEER_TIMEOUT: Duration = Duration::from_millis(500); pub const BROADCAST_PEER_TIMEOUT: Duration = Duration::from_secs(2); /// Build the shared forwarding client with the standard connect/request -/// timeouts. One per node (held on [`RegionClusterState`]) so connections pool +/// timeouts. One per node (held on [`ShardReplica`]) so connections pool /// across requests. /// /// # Errors @@ -151,6 +155,43 @@ pub struct ForwardedResponse { pub body: serde_json::Value, } +/// Rebuild an axum [`Response`] from a [`ForwardedResponse`], relaying the +/// replicated-log verdict headers (`x-tidal-seq` / `x-tidal-deduplicated`) the +/// peer set. This is the ONE place a forwarded write's quorum/dedup verdict is +/// re-attached to the client-facing response, so the follower→leader hop +/// (`forward_write`) and the cross-shard gateway hop (`forward_to_group_node`) +/// cannot drift on the relay contract. +#[must_use] +pub fn relay_forwarded(resp: ForwardedResponse) -> Response { + let mut response = (resp.status, Json(resp.body)).into_response(); + if let Some(seq) = resp.seq + && let Ok(value) = HeaderValue::from_str(&seq) + { + response + .headers_mut() + .insert(HeaderName::from_static(SEQ_HEADER), value); + } + if resp.deduplicated { + response.headers_mut().insert( + HeaderName::from_static(DEDUP_HEADER), + HeaderValue::from_static(DEDUP_HEADER_VALUE), + ); + } + response +} + +/// The `x-tidal-ack` override (m11p3) as a forward passthrough entry, if the +/// caller set one — so a forwarded write reaches the leader with the CALLER's +/// ack mode, not the gateway's default. Empty when absent. +#[must_use] +pub fn ack_passthrough(headers: &HeaderMap) -> Vec<(&'static str, String)> { + headers + .get(ACK_HEADER) + .and_then(|v| v.to_str().ok()) + .map(|v| vec![(ACK_HEADER, v.to_owned())]) + .unwrap_or_default() +} + /// Forward a JSON request to one peer and relay its status + body back. /// /// `auth` is the caller's verbatim `Authorization` header (passed through so the diff --git a/tidal-server/src/cluster/join_boot.rs b/tidal-server/src/cluster/join_boot.rs index dcd3d66..30deba9 100644 --- a/tidal-server/src/cluster/join_boot.rs +++ b/tidal-server/src/cluster/join_boot.rs @@ -20,21 +20,21 @@ //! serves from 1). //! 5. **Synthesize a topology** from the JOIN-RESPONSE roster (this node's //! `--region` = its assigned id), carrying the local file's behavioral knob -//! blocks (§3.5). `RegionClusterState::new` builds the transport/ship/election +//! blocks (§3.5). `ShardReplica::new` builds the transport/ship/election //! peer tables from THIS, not the local topology's `regions:` list. //! //! A **restart of a joiner** (the data dir already has a WAL + a membership //! cache) skips the seed loop: it boots from the cache roster (no leader //! discovery, no fetch) and the runtime catch-up self-heal converges it. The //! authoritative roster once the WAL opens is the recovered `ClusterMembership` -//! cell (§3.6 precedence, resolved inside `RegionClusterState::new`); the cache +//! cell (§3.6 precedence, resolved inside `ShardReplica::new`); the cache //! and the synthesized topology only feed the pre-open peer-table construction, //! and the cache is rewritten from the cell after every applied record. //! //! # A joiner is a LEARNER //! //! The join response role is `Learner`. The synthesized topology marks the -//! joiner's role so `RegionClusterState`'s era-0 view (when no cell exists yet) +//! joiner's role so `ShardReplica`'s era-0 view (when no cell exists yet) //! does not mistake it for a voter — but in practice a snapshot install always //! carries the leader's WAL with the kind-4 Learner record, so the view boots //! `from_record` with the correct role. `ElectionState` boots with `self` NOT in @@ -69,14 +69,14 @@ const BACKOFF_MAX: Duration = Duration::from_secs(5); const DEFAULT_JOIN_WINDOW_MS: u64 = 120_000; /// The synthesized topology a seed-join boot constructs for -/// `RegionClusterState::new` (§3.4). +/// `ShardReplica::new` (§3.4). /// /// The join-response roster becomes the `regions:` list, the local file's knob /// blocks are carried verbatim, and this node is included at its assigned id. pub struct SeedJoinBoot { /// The synthesized topology (roster from the join, knobs from the local file). pub topology: TopologySpec, - /// This node's region name (echoed for `RegionClusterState::new`). + /// This node's region name (echoed for `ShardReplica::new`). pub region: String, } @@ -117,7 +117,7 @@ pub fn seed_join_boot(input: &SeedJoinInput<'_>) -> Result { // §3.6 precedence for the PRE-OPEN loop: a restart of a joiner whose data dir // already carries a WAL boots from the cache WITHOUT the seed. The // authoritative roster once the WAL opens is the recovered cell (resolved - // inside `RegionClusterState::new`); this cache only seeds the peer tables. + // inside `ShardReplica::new`); this cache only seeds the peer tables. let wal_exists = input .data_dir .join("wal") @@ -589,7 +589,7 @@ fn adopt_term(data_dir: &Path, term: u64) -> Result<()> { Ok(()) } -/// Synthesize the topology `RegionClusterState::new` consumes (§3.4): the +/// Synthesize the topology `ShardReplica::new` consumes (§3.4): the /// join-response roster as `regions:` (in id order so the positional `RegionId` /// MATCHES the assigned member id — the seam every durable id depends on), this /// node's `--region` carried, the local file's knob blocks copied verbatim @@ -644,6 +644,7 @@ fn synthesize_topology( } else { None }, + zone: None, }); } else { // A burned id with no member entry at all (a gap that should not @@ -656,6 +657,7 @@ fn synthesize_topology( http_addr: Some(format!("http://0.0.0.0:{}", 1u16.wrapping_add(id))), grpc_tls: None, metrics_addr: None, + zone: None, }); } } @@ -694,6 +696,9 @@ fn synthesize_topology( replication: clone_replication(&input.knobs.replication), wal: clone_wal(&input.knobs.wal), election: clone_election(&input.knobs.election), + // Seed-join synthesizes the legacy single group (the joiner learns its + // shard assignment from the membership log, not the synthesized file). + shards: None, }) } @@ -770,6 +775,7 @@ mod tests { http_addr: Some("127.0.0.1:9001".into()), grpc_tls: None, metrics_addr: None, + zone: None, }], leader: "us-east".into(), write_workers: None, @@ -781,6 +787,7 @@ mod tests { }, wal: WalSpec::default(), election: ElectionSpec::default(), + shards: None, } } diff --git a/tidal-server/src/cluster/membership.rs b/tidal-server/src/cluster/membership.rs index b07974a..fdae7c8 100644 --- a/tidal-server/src/cluster/membership.rs +++ b/tidal-server/src/cluster/membership.rs @@ -42,7 +42,7 @@ use std::sync::RwLock; use tidaldb::replication::shard::{RegionId, ShardId}; use tidaldb::wal::format::{MemberEntry, MemberRole, MembershipRecord}; -use super::topology::TopologySpec; +use super::topology::{ResolvedShardGroup, TopologySpec}; /// A coherent snapshot of the roster the view derived (era 0 or a kind-4 /// record). All the lookup maps are built from this so they can never drift. @@ -158,35 +158,44 @@ pub struct ApplyPlan { } impl MembershipView { - /// Build the era-0 view from the topology positional tables — byte-for-byte - /// today's behavior. `name_to_id` is the topology declaration-order map (the - /// node already computed it); `peer_http`/`peer_grpc` are the - /// `build_peer_tables` outputs. The roster is reconstructed so the view's - /// `voter_ids`/`learner_ids`/status surfaces match the implicit topology - /// roster (every declared region is a Voter, ids = positional). + /// Build the era-0 view for one shard `group` from its resolved replicas. + /// `name_to_id`/`id_to_name` stay CLUSTER-WIDE (the node's declaration-order + /// maps) so a stale cross-shard forward still resolves a non-replica region's + /// NAME; `peer_http`/`peer_grpc` are the group-scoped `build_group_peer_tables` + /// outputs. The roster lists exactly the GROUP's replicas (every one a Voter, + /// id = positional `RegionId`, gRPC from the resolved replica address, HTTP + /// from the hosting node's spec) so `voter_ids`/`voter_count`/`live_ids` and + /// the status surfaces agree with the (group-scoped) peer/quorum/election sets + /// that the rest of `ShardReplica::new` built. For the legacy single group + /// (`resolve_shard_groups` synthesizes every region as a replica, addresses + /// verbatim) this is byte-for-byte the pre-m11p6 all-regions-Voter roster. #[must_use] pub fn era0( self_region: RegionId, + group: &ResolvedShardGroup, topology: &TopologySpec, name_to_id: &HashMap, id_to_name: &HashMap, peer_http: &HashMap, peer_grpc: &HashMap, ) -> Self { - // Reconstruct the implicit era-0 roster from the topology: every region - // is a Voter, id = its positional RegionId, addresses from the spec. - let mut members: Vec = topology - .regions + // Reconstruct the era-0 roster from THIS GROUP's replicas (NOT every + // topology region): each replica is a Voter, id = positional RegionId, + // gRPC from the resolved replica address, HTTP from the hosting node's + // RegionSpec (ResolvedReplica carries no HTTP — one HTTP server per node). + let mut members: Vec = group + .replicas .iter() - .map(|spec| { - let id = name_to_id[&spec.name].0; - MemberEntry { - id, - name: spec.name.clone(), - grpc_addr: spec.grpc_addr.clone().unwrap_or_default(), - http_addr: spec.http_addr.clone().unwrap_or_default(), - role: MemberRole::Voter, - } + .map(|r| MemberEntry { + id: r.region.0, + name: r.name.clone(), + grpc_addr: r.grpc_addr.clone(), + http_addr: topology + .regions + .get(usize::from(r.region.0)) + .and_then(|n| n.http_addr.clone()) + .unwrap_or_default(), + role: MemberRole::Voter, }) .collect(); members.sort_unstable_by_key(|m| m.id); @@ -702,7 +711,7 @@ pub fn plan_activation_reappend(roster: &Roster, term: u64) -> MembershipRecord #[cfg(test)] #[allow(clippy::unwrap_used)] mod tests { - use super::super::topology::{RegionSpec, TopologySpec}; + use super::super::topology::{RegionSpec, ShardReplicaSpec, ShardSpec, TopologySpec}; use super::*; fn topo(names: &[&str]) -> TopologySpec { @@ -717,6 +726,7 @@ mod tests { http_addr: Some(format!("http://{n}.svc:9501")), grpc_tls: None, metrics_addr: None, + zone: None, }) .collect(), write_workers: None, @@ -724,11 +734,15 @@ mod tests { replication: super::super::topology::ReplicationSpec::default(), wal: super::super::topology::WalSpec::default(), election: super::super::topology::ElectionSpec::default(), + shards: None, } } fn era0_view(self_idx: u16, names: &[&str]) -> MembershipView { let t = topo(names); + // Legacy single group: every region is a replica (the synthesized + // group is byte-for-byte the pre-m11p6 all-regions roster). + let groups = t.resolve_shard_groups().expect("legacy synthesis"); let mut name_to_id = HashMap::new(); let mut id_to_name = HashMap::new(); let mut peer_http = HashMap::new(); @@ -744,6 +758,7 @@ mod tests { } MembershipView::era0( RegionId(self_idx), + &groups[0], &t, &name_to_id, &id_to_name, @@ -752,6 +767,72 @@ mod tests { ) } + /// An RF < N group's era-0 roster lists EXACTLY the group's replicas, not + /// every region — so `voter_ids`/`voter_count` and `all_regions_for_status` + /// agree with the group-scoped peer/quorum/election sets (the FIX-A guard + /// against an N-voter majority on an RF-replica group). + #[test] + fn era0_roster_scoped_to_group_replicas_for_rf_lt_n() { + // Three regions, one shard of RF=2 (us-east + ap-south; eu-west is NOT a + // replica of this group). name_to_id stays cluster-wide (all three). + let mut t = topo(&["us", "eu", "ap"]); + t.shards = Some(vec![ShardSpec { + id: 0, + leader: Some("us".into()), + replicas: vec![ + ShardReplicaSpec { + node: "us".into(), + grpc_addr: None, + grpc_bind: None, + }, + ShardReplicaSpec { + node: "ap".into(), + grpc_addr: None, + grpc_bind: None, + }, + ], + }]); + let groups = t.resolve_shard_groups().expect("resolve RF=2 group"); + let mut name_to_id = HashMap::new(); + let mut id_to_name = HashMap::new(); + for (i, n) in ["us", "eu", "ap"].iter().enumerate() { + let rid = RegionId(u16::try_from(i).unwrap()); + name_to_id.insert((*n).to_string(), rid); + id_to_name.insert(rid, (*n).to_string()); + } + // self = us (region 0); the only peer in THIS group is ap (region 2). + let mut peer_http = HashMap::new(); + let mut peer_grpc = HashMap::new(); + peer_http.insert(RegionId(2), "http://ap.svc:9501".to_string()); + peer_grpc.insert(ShardId(2), "ap.svc:9500".to_string()); + let v = MembershipView::era0( + RegionId(0), + &groups[0], + &t, + &name_to_id, + &id_to_name, + &peer_http, + &peer_grpc, + ); + // The roster is the 2 replicas, NOT all 3 regions — eu-west is absent. + assert_eq!( + v.roster().voter_ids(), + vec![RegionId(0), RegionId(2)], + "era-0 roster must be the group's RF replicas, not every region" + ); + assert_eq!(v.roster().members.len(), 2); + assert!( + v.roster().members.iter().all(|m| m.name != "eu"), + "the non-replica region must not appear in the group roster" + ); + // all_regions_for_status reflects the group, not the cluster. + let all = v.all_regions_for_status(); + assert_eq!(all.len(), 2); + assert!(all.iter().all(|(_, name, _)| name != "eu")); + // name_to_id stays cluster-wide: a stale forward to eu still resolves. + assert_eq!(v.name_to_id("eu"), Some(RegionId(1))); + } + fn member(id: u16, name: &str, role: MemberRole) -> MemberEntry { MemberEntry { id, diff --git a/tidal-server/src/cluster/mod.rs b/tidal-server/src/cluster/mod.rs index 58c6bfe..7b3dae2 100644 --- a/tidal-server/src/cluster/mod.rs +++ b/tidal-server/src/cluster/mod.rs @@ -7,12 +7,15 @@ //! name ↔ [`RegionId`] mapping. Replication between regions still traverses a //! real [`GrpcTransport`] on loopback (not in-process channels), but a crash //! takes the whole "cluster" down — there is no process isolation. -//! * **Multi-process** ([`RegionClusterState`], the m8p10 mode, selected by -//! `--region`): this process owns exactly ONE region — one [`TidalDb`], one -//! [`GrpcTransport`] whose server binds this region's `grpc_addr` and whose -//! peers are every sibling region's real `grpc_addr` — and peers with sibling -//! processes over real gRPC. Process isolation is real; quorum-ack writes and -//! automatic failure detection are still future work. +//! * **Multi-process** ([`ClusterNode`], the m8p10→m11p6 mode, selected by +//! `--region`): this process is a [`ClusterNode`] hosting a +//! `BTreeMap>` — one [`ShardReplica`] (one +//! [`TidalDb`] + [`GrpcTransport`] + election/commit/membership) per shard +//! group it replicates. Entity writes hash-route to the owning group's leader +//! (apply locally or forward); corpus reads scatter over hosted groups. With +//! `shards:` absent it holds ONE group spanning every region — byte-for-byte +//! the original one-region-per-process node. Process isolation is real; +//! quorum-ack writes (m11p3) and elected failover (m11p4) are wired per group. //! //! Both modes are gated behind the same explicit operator opt-in //! ([`ensure_experimental_enabled`]). @@ -26,7 +29,8 @@ //! * [`transport`] — single-process self-loop gRPC transport wiring //! * [`state`] — [`ClusterState`] (single-process) + the experimental gate //! * [`routes`] — the single-process router + handlers -//! * [`node`] — [`RegionClusterState`] (multi-process) + its router/handlers +//! * [`node`] — [`ClusterNode`] (the process) hosting [`ShardReplica`] groups +//! (multi-process) + its gateway router/handlers //! //! [`GrpcTransport`]: tidal_net::GrpcTransport //! [`RegionId`]: tidaldb::replication::shard::RegionId @@ -39,7 +43,7 @@ pub(crate) mod forward; /// joins an existing cluster by contacting a `--seed`. /// /// Public so the binary's `run_seed_join_cluster` can invoke `seed_join_boot` -/// before `RegionClusterState::new`. +/// before `ShardReplica::new`. pub mod join_boot; /// The membership runtime (m11p5 §3.1–§3.3): effective roster, the one fenced /// apply path, conf-change gates, and leader-side join/remove planning. @@ -48,7 +52,7 @@ pub(crate) mod node; /// Boot-time snapshot install + swap-recovery (m11p5 §2.2–§2.7). /// /// Public so the binary's `run_region_cluster` can invoke `run_boot_install` -/// before `RegionClusterState::new`. +/// before `ShardReplica::new`. pub mod reseed; pub(crate) mod routes; pub(crate) mod snapshot; @@ -58,12 +62,12 @@ mod transport; // ── Public API (preserved across the split) ───────────────────────────────── -pub use node::{RegionClusterState, build_region_router}; +pub use node::{ClusterNode, ShardReplica, build_region_router}; pub use routes::build_cluster_router; pub use state::{ClusterMode, ClusterState, EXPERIMENTAL_CLUSTER_ENV, ensure_experimental_enabled}; pub use topology::{ - ElectionSpec, GrpcTlsSpec, RegionSpec, ReplicationSpec, TimeoutsSpec, TopologySpec, WalSpec, - load_topology, validate_multiproc, + ElectionSpec, GrpcTlsSpec, RegionSpec, ReplicationSpec, ShardReplicaSpec, ShardSpec, + TimeoutsSpec, TopologySpec, WalSpec, load_topology, validate_multiproc, }; // The OpenAPI documents (`crate::openapi`) reach the handlers/DTOs through diff --git a/tidal-server/src/cluster/node.rs b/tidal-server/src/cluster/node.rs index a8ea992..ce2056b 100644 --- a/tidal-server/src/cluster/node.rs +++ b/tidal-server/src/cluster/node.rs @@ -1,15 +1,27 @@ -//! `RegionClusterState`: a TRUE one-region-per-process cluster node (m8p10). +//! `ClusterNode` (the process) hosting `ShardReplica`s (the shard groups), m11p6. //! -//! Unlike [`ClusterState`](super::ClusterState) (every region in one process), -//! a `RegionClusterState` owns exactly ONE region: +//! Two types, one file: //! -//! * one [`TidalDb`] (`NodeRole::Single`, `shard = ShardId(region)`, -//! `peer_shards = siblings`) — direct-writable and promotable; -//! * one [`GrpcTransport`] whose gRPC server binds THIS region's `grpc_addr` -//! and whose peer table points at every SIBLING region's real `grpc_addr`; -//! * an **always-on** segment receiver (`db.start_replication`) — leadership can -//! move, so an inbound segment is legal on any node after a promote elsewhere; -//! * a [`SignalRelay`]-backed leader write path with a leader check. +//! * [`ClusterNode`] is the OS process and the axum `State`. It owns node +//! identity, the shared forward client, the gateway [`ShardRouter`], and a +//! `BTreeMap>` of the groups this node replicates. +//! Every entity write hash-routes through it to the owning group's leader +//! (applying locally or forwarding); corpus reads scatter over its hosted +//! groups. For the legacy single group (`shards:` absent) it holds exactly +//! ONE `ShardReplica` spanning every region — byte-for-byte the pre-m11p6 +//! one-region-per-process node. +//! * [`ShardReplica`] is ONE shard group's full replication machinery (formerly +//! `RegionClusterState`, when a process WAS exactly one group). A node hosts N +//! of them, not one. Each owns: +//! * one [`TidalDb`] (`NodeRole::Single`, `shard = ShardId(region)`, +//! `peer_shards = the group's other replica nodes`) — direct-writable and +//! promotable. The in-group identity stays region-id based; the data-shard +//! only namespaces its data subdir + gRPC port; +//! * one [`GrpcTransport`] whose server binds THIS group's resolved gRPC +//! address and whose peers are the group's OTHER replica nodes; +//! * an **always-on** segment receiver (`db.start_replication`) — leadership +//! can move, so an inbound segment is legal on any node after a promote; +//! * a [`SignalRelay`]-backed leader write path with a leader check. //! //! Without `--region` the existing single-process [`ClusterState`] runs //! byte-for-byte unchanged. @@ -33,7 +45,7 @@ //! nothing to ship or serve catch-up from. use std::{ - collections::{HashMap, HashSet}, + collections::{BTreeMap, HashMap, HashSet}, sync::{ Arc, RwLock, Weak, atomic::{AtomicBool, AtomicU64, Ordering}, @@ -65,7 +77,7 @@ use tidaldb::{ replication::{ CommitIndex, ReseedMarker, ReseedMarkerStore, ReseedReason, ShipQueue, Transport, WalFeedSource, - shard::{RegionId, ShardId}, + shard::{RegionId, ShardId, ShardRouter}, }, schema::{EntityId, Schema, Timestamp}, wal::feed::WalShipFeed, @@ -81,7 +93,7 @@ use super::{ }, reseed, routes::{ClusterAppError, ScatterGatherInfo, ShardedFeedResponse, ShardedSearchResponse}, - topology::{TopologySpec, shard_of_region}, + topology::{ResolvedShardGroup, TopologySpec, shard_of_region}, transport::{GRPC_READY_TIMEOUT, grpc_server_ready, resolve_grpc_bind_addr}, }; use crate::{ @@ -167,9 +179,9 @@ const COMMIT_BRIDGE_WAKE_INTERVAL: Duration = Duration::from_secs(1); /// for tests via `TIDAL_REMOVE_DELIVERY_GRACE_MS`. const REMOVE_DELIVERY_GRACE_DEFAULT_MS: u64 = 30_000; -/// The election driver's boot bundle: prepared in [`RegionClusterState::new`] +/// The election driver's boot bundle: prepared in [`ShardReplica::new`] /// (where the durable classification runs), consumed by -/// [`RegionClusterState::start_election_driver`] once the node is in its +/// [`ShardReplica::start_election_driver`] once the node is in its /// final `Arc`. struct ElectionBoot { config: tidaldb::replication::ElectionConfig, @@ -179,12 +191,20 @@ struct ElectionBoot { topology_leader: RegionId, } -/// A multi-process cluster node owning exactly one region. -pub struct RegionClusterState { +/// One shard group's full replication machinery, hosted inside a [`ClusterNode`]. +/// +/// 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. +pub struct ShardReplica { /// This process's region id (index into the topology declaration order). region: RegionId, /// This region's human-readable name. region_name: String, + /// The data-shard group this replica serves (m11p6): the gateway hash + /// output, the `ClusterNode` `shards` map key, and the metrics/status + /// shard label. The in-group replication identity stays `region`-based + /// (`shard_of_region`) — `group_shard` only distinguishes co-hosted groups. + group_shard: ShardId, /// `Some` for the server's lifetime; taken on shutdown so the `TidalDb` is /// dropped (checkpoint + WAL fsync + thread join) deterministically. db: Option>, @@ -332,7 +352,7 @@ pub struct RegionClusterState { membership: Arc, /// The transport's `JoinCluster` hooks cell (m11p5 §3.3), late-bound by /// [`Self::start_election_driver`] with a leader-side join adapter holding a - /// `Weak`. + /// `Weak`. join_hooks_cell: Arc>>, /// Guards the auto-promotion duty (m11p5 §3.3) so at most ONE promotion /// evaluation runs at a time: the leader tick fires every 50ms, but a single @@ -386,7 +406,7 @@ struct DeferredRetire { deadline: Instant, } -impl RegionClusterState { +impl ShardReplica { /// Build the single region named `region_name` from `topology`. /// /// `RegionId`s are assigned by topology declaration order — the IDENTICAL @@ -418,7 +438,7 @@ impl RegionClusterState { /// runs first and proves both names are declared. // Linear construction sequence (validate → db → sources → transport → // receiver → queue); splitting it would scatter the ordering invariants. - #[allow(clippy::too_many_lines)] + #[allow(clippy::too_many_lines, clippy::too_many_arguments)] pub fn new( topology: &TopologySpec, region_name: &str, @@ -426,6 +446,8 @@ impl RegionClusterState { profiles: Vec, data_dir: Option, hlc_offset_ms: i64, + group: &ResolvedShardGroup, + enable_metrics: bool, ) -> Result { super::topology::validate_multiproc(topology, region_name)?; @@ -455,9 +477,11 @@ impl RegionClusterState { let region = *name_to_id .get(region_name) .expect("validate_multiproc proved region_name is declared"); - let leader = *name_to_id - .get(&topology.leader) - .expect("validate_multiproc proved the leader is declared"); + // m11p6: leadership is per shard GROUP — the term-0 / preferred leader of + // THIS group, not a single cluster-wide leader. For the legacy single + // group this is `topology.leader` (the resolver synthesizes it), so S=1 + // is byte-for-byte. In-group peer identity stays region-id based. + let leader = group.leader; let my_shard = shard_of_region(region); // m11p4 boot classification (phase-4.md §1): durable election state @@ -520,7 +544,7 @@ impl RegionClusterState { None }; - // Sibling region shards + their gRPC/HTTP addresses. + // This group's OTHER replica nodes + their gRPC/HTTP addresses (m11p6). let PeerTables { peer_shards, peer_grpc, @@ -528,7 +552,7 @@ impl RegionClusterState { my_grpc_spec, my_grpc_bind_spec, my_tls, - } = build_peer_tables(topology, &name_to_id, region); + } = build_group_peer_tables(group, topology, region); // One TidalDb owning this region (NodeRole::Single, shard = this region, // peer_shards = siblings). Persistent (validated above); the HLC @@ -542,6 +566,7 @@ impl RegionClusterState { hlc_offset_ms, my_shard, &peer_shards, + enable_metrics, )?; let db = Arc::new(db); @@ -589,7 +614,7 @@ impl RegionClusterState { let applied_sink_cell = Arc::clone(&sources.applied_sink); let election_hooks_cell = Arc::clone(&sources.election); // m11p5 §2.4: the snapshot-required refusal sink cell, late-bound by - // `start_election_driver` (the sink holds a `Weak`, + // `start_election_driver` (the sink holds a `Weak`, // so it can only be built once the node is in its final Arc). Until set, // a snapshot-required catch-up trailer falls back to log + retry. let snapshot_required_cell = Arc::clone(&sources.snapshot_required); @@ -598,7 +623,7 @@ impl RegionClusterState { // set, `FetchSnapshot` answers Unimplemented. let snapshot_source_cell = Arc::clone(&sources.snapshots); // m11p5 §3.3: the `JoinCluster` hooks cell, late-bound by - // `start_election_driver` (the adapter holds a `Weak` + // `start_election_driver` (the adapter holds a `Weak` // for the same reason as the snapshot-required sink). Until set, // `JoinCluster` answers Unimplemented. let join_hooks_cell = Arc::clone(&sources.join); @@ -824,9 +849,13 @@ impl RegionClusterState { config: super::election_driver::election_config( &topology.election, region, - name_to_id - .values() - .copied() + // m11p6: the election's voter set is THIS group's replica nodes + // (excluding self), not every region. For the legacy single + // group this is every other region — byte-for-byte. + group + .replicas + .iter() + .map(|r| r.region) .filter(|&r| r != region) .collect(), true, @@ -858,6 +887,7 @@ impl RegionClusterState { } None => Arc::new(super::membership::MembershipView::era0( region, + group, topology, &name_to_id, &id_to_name, @@ -915,6 +945,7 @@ impl RegionClusterState { Ok(Self { region, region_name: region_name.to_string(), + group_shard: group.shard, db: Some(db), transport, ship_feed, @@ -2815,11 +2846,14 @@ impl RegionClusterState { membership_version: self.membership_version(), membership_term: self.membership.term(), membership_role: self.role_in_roster().to_string(), + // Populated by `ClusterNode::status_local` (it owns the group set); + // a bare per-replica status carries only its own row implicitly. + shards: Vec::new(), }) } } -impl Drop for RegionClusterState { +impl Drop for ShardReplica { fn drop(&mut self) { self.shutdown(); } @@ -2885,19 +2919,22 @@ struct PeerTables { my_tls: Option, } -/// Resolve the sibling-region address tables for `region` from `topology`. +/// Resolve the address tables for THIS region's replica of one shard `group` +/// (m11p6). Peers are the group's OTHER replica nodes (not all topology +/// regions): their gRPC comes from the group's resolved replica addresses +/// (explicit or port-derived), their HTTP from each node's single +/// `RegionSpec.http_addr` (one HTTP server per node serves all its shards). +/// THIS replica's own gRPC advertise/bind comes from the group's resolved +/// self-replica; TLS from this node's `RegionSpec`. /// -/// `validate_multiproc` (run first) proves every region declares both a -/// `grpc_addr` and an `http_addr` (the gRPC one hostname-tolerant, m11p5), so -/// the table construction below never elides a peer silently. Peer gRPC -/// addresses are NOT parsed here — they pass through as advertised strings so -/// the transport's lazy channels re-resolve DNS on every reconnect. This is -/// why the function is now infallible: the `SocketAddr` parse (the only thing -/// that could fail) moved to the bind resolver and the config-level peer-string -/// check, leaving nothing here to reject. -fn build_peer_tables( +/// In-group peer identity stays region-id based (`shard_of_region(rid)` = +/// `ShardId(rid.0)`), exactly as today's single-group cluster — the data-shard +/// `group.shard` only namespaces the directory/port, never the wire identity. +/// For the legacy single group (RF = all regions, port offset 0) this yields +/// byte-for-byte the pre-m11p6 tables. +fn build_group_peer_tables( + group: &ResolvedShardGroup, topology: &TopologySpec, - name_to_id: &HashMap, region: RegionId, ) -> PeerTables { let mut peer_shards = Vec::new(); @@ -2906,11 +2943,13 @@ fn build_peer_tables( let mut my_grpc_spec: Option = None; let mut my_grpc_bind_spec: Option = None; let mut my_tls: Option = None; - for region_spec in &topology.regions { - let rid = name_to_id[®ion_spec.name]; + for replica in &group.replicas { + let rid = replica.region; + // This node's own TLS comes from its RegionSpec (one cert per node). + let region_spec = &topology.regions[usize::from(rid.0)]; if rid == region { - my_grpc_spec.clone_from(®ion_spec.grpc_addr); - my_grpc_bind_spec.clone_from(®ion_spec.grpc_bind); + my_grpc_spec = Some(replica.grpc_addr.clone()); + my_grpc_bind_spec.clone_from(&replica.grpc_bind); my_tls = region_spec .grpc_tls .as_ref() @@ -2918,14 +2957,7 @@ fn build_peer_tables( continue; } peer_shards.push(shard_of_region(rid)); - // The advertised address passes through verbatim — no SocketAddr parse. - // `validate_multiproc` already proved it is a syntactic host:port. - // `unwrap_or_default` cannot lose a real address here: a missing - // grpc_addr would have failed validation, so this only guards the - // never-taken None branch (an empty string would be caught by the - // transport config's own peer-address validation downstream). - let grpc = region_spec.grpc_addr.clone().unwrap_or_default(); - peer_grpc.insert(shard_of_region(rid), grpc); + peer_grpc.insert(shard_of_region(rid), replica.grpc_addr.clone()); if let Some(http) = ®ion_spec.http_addr { peer_http.insert(rid, http.clone()); } @@ -2954,6 +2986,7 @@ fn open_region_db( hlc_offset_ms: i64, my_shard: ShardId, peer_shards: &[ShardId], + enable_metrics: bool, ) -> Result { let mut builder = TidalDb::builder() .with_schema(schema) @@ -2974,7 +3007,10 @@ fn open_region_db( if let Some(timeout_ms) = topology.wal.batch_timeout_ms { builder = builder.wal_batch_timeout(Duration::from_millis(timeout_ms)); } - if let Some(metrics_addr) = topology.metrics_addr_of(region_name) { + // m11p6: at most ONE hosted shard per node binds the engine's `/metrics` + // server (N TidalDb instances would otherwise fight for one `metrics_addr`). + // ClusterNode designates the metrics-owning shard; the others open without. + if enable_metrics && let Some(metrics_addr) = topology.metrics_addr_of(region_name) { builder = builder.enable_metrics(metrics_addr); } let db = builder.open().map_err(ServerError::Tidal)?; @@ -3005,11 +3041,11 @@ fn open_region_db( /// off the async runtime so `Drop` never blocks a reactor. struct StagedWriteTicket { staged: Option, - node: Arc, + node: Arc, } impl StagedWriteTicket { - const fn new(staged: StagedSignal, node: Arc) -> Self { + const fn new(staged: StagedSignal, node: Arc) -> Self { Self { staged: Some(staged), node, @@ -3208,7 +3244,7 @@ impl SegmentSource for NodeSegmentSource { /// Build the node's two shared forwarding clients off the reactor. /// /// Both build their own runtime internally and assert they are NOT inside one; -/// `RegionClusterState::new` runs on a dedicated `std::thread`, so building them +/// `ShardReplica::new` runs on a dedicated `std::thread`, so building them /// here is sound. The async client carries the standard connect/request timeouts; /// the blocking client (for the `/sharded/*` detached-thread fetch) carries NO /// global timeout — the per-shard deadline is applied per request so a slow shard @@ -3223,6 +3259,413 @@ fn build_forwarding_clients() -> Result<(reqwest::Client, reqwest::blocking::Cli Ok((client, blocking_client)) } +// ── ClusterNode: the process handle hosting N shard-group replicas (m11p6) ─── + +/// The on-disk data subdir for shard group `shard` (zero-padded to 5 digits). +/// The ONE authoritative formatter for the per-group directory layout — a +/// durability contract any ops tooling (backup, `tidalctl`, the L3 rebalance +/// stager) must format identically to find an existing group's data. +fn shard_subdir(shard: ShardId) -> String { + format!("shard-{:05}", shard.0) +} + +/// Where an entity-routed write must go: a locally-hosted replica of the +/// entity's shard group, or a remote node hosting it (this node hosts no +/// replica of that group). +enum EntityRoute { + /// This node hosts the entity's shard group — apply/forward via this + /// replica's own leadership view (the existing per-group write path). + Local(Arc), + /// This node hosts no replica of the entity's group — forward to a replica + /// node, which routes to the group's leader. + Remote { + /// The target shard group (for the log/error). + shard: ShardId, + /// Ordered HTTP bases to try: the group's believed leader first, then + /// its other replicas (deduped). The gateway walks them on a CONNECT + /// failure so a single dead replica does not 503 a write the rest of the + /// group's quorum can still serve. Empty ⇒ no replica is addressable. + candidates: Vec, + }, +} + +/// The cluster process node (m11p6). +/// +/// One OS process hosting a replica of each shard group it is assigned to. It +/// owns the shared write/forward clients and node identity, holds one +/// [`ShardReplica`] per hosted group, and routes each request by entity hash to +/// the right group's leader. For the legacy single group (`shards:` absent) it +/// holds exactly one replica spanning every region — byte-for-byte the pre-m11p6 +/// process. +pub struct ClusterNode { + /// Every shard group in the cluster, resolved (for routing a write to a + /// group this node does NOT host). + placement: BTreeMap, + /// The shard groups THIS node hosts a replica of, keyed by data-shard id. + groups: BTreeMap>, + /// Entity → data-shard router (`Single` for one group, `Hash(S)` otherwise). + router: ShardRouter, + /// Node region id → public HTTP base (cross-node forward to a shard leader). + node_http: HashMap, + /// Shared async forwarding client for the cross-shard gateway hop. + client: reqwest::Client, + /// Flipped on shutdown so `/health` reports not-ready while draining. + shutting_down: AtomicBool, +} + +impl ClusterNode { + /// Build the cluster node: resolve the shard-group assignment, open one + /// [`ShardReplica`] per group this node hosts (each in its own data subdir + /// and gRPC port for `S > 1`; the node data dir verbatim for `S == 1`), and + /// wire the gateway router. Same argument shape as the pre-m11p6 + /// `ShardReplica::new` so the boot paths and tests need no extra plumbing. + /// + /// # Errors + /// + /// Propagates [`ShardReplica::new`] failures, or [`ServerError`] when the + /// topology is invalid, this region hosts no group, or a per-group data + /// subdir cannot be created. + /// + /// # Panics + /// + /// Does not panic on caller input: the one internal `expect` on the region + /// name is guarded by [`validate_multiproc`], which runs first and proves + /// the name is declared. + #[allow(clippy::needless_pass_by_value)] // schema/profiles are cloned per group + pub fn new( + topology: &TopologySpec, + region_name: &str, + schema: Schema, + profiles: Vec, + data_dir: Option, + hlc_offset_ms: i64, + ) -> Result { + super::topology::validate_multiproc(topology, region_name)?; + + let mut name_to_id = HashMap::new(); + for (i, r) in topology.regions.iter().enumerate() { + let id = RegionId(u16::try_from(i).map_err(|_| { + ServerError::SchemaConfig("topology declares more than 65535 regions".into()) + })?); + name_to_id.insert(r.name.clone(), id); + } + let region = *name_to_id + .get(region_name) + .expect("validate_multiproc proved region_name is declared"); + + let resolved = topology.resolve_shard_groups()?; + let single = resolved.len() == 1; + + let mut groups: BTreeMap> = BTreeMap::new(); + let mut metrics_owner_assigned = false; + for group in &resolved { + if !group.replicas.iter().any(|r| r.region == region) { + continue; // this node does not host this group + } + // S=1 uses the node data dir verbatim (existing clusters restart + // unchanged); S>1 isolates each group under `shard-/`. + let group_dir = match &data_dir { + Some(d) if single => Some(d.clone()), + Some(d) => { + let sub = d.join(shard_subdir(group.shard)); + std::fs::create_dir_all(&sub).map_err(|e| ServerError::io(&sub, e))?; + Some(sub) + } + None => None, + }; + // At most ONE hosted group binds the engine's `/metrics` server + // (N TidalDb instances would otherwise fight for one `metrics_addr`). + let enable_metrics = !metrics_owner_assigned; + let replica = ShardReplica::new( + topology, + region_name, + schema.clone(), + profiles.clone(), + group_dir, + hlc_offset_ms, + group, + enable_metrics, + )?; + if enable_metrics { + metrics_owner_assigned = true; + } + groups.insert(group.shard, Arc::new(replica)); + } + if groups.is_empty() { + return Err(ServerError::Cluster(format!( + "region '{region_name}' is not a replica of any shard group" + ))); + } + + let router = if single { + ShardRouter::single() + } else { + let n = u16::try_from(resolved.len()) + .map_err(|_| ServerError::SchemaConfig("more than 65535 shard groups".into()))?; + ShardRouter::hash(n).map_err(|e| { + ServerError::Cluster(format!("build shard router for {n} groups: {e}")) + })? + }; + + // Node id → public HTTP base for the cross-node forward. Reuses the + // already-validated `name_to_id` (the one overflow-checked region→id + // map) rather than re-deriving the positional id a second time. + let mut node_http = HashMap::new(); + for r in &topology.regions { + if let Some(h) = &r.http_addr { + node_http.insert(name_to_id[&r.name], h.clone()); + } + } + let placement: BTreeMap = + resolved.into_iter().map(|g| (g.shard, g)).collect(); + // The router's shard space and the placement map are two derivations of + // the same resolved set; `resolve_shard_groups` guarantees dense ids in + // `[0, len)`, so every `route()` output is a placement key. Assert it so + // an L3 change that mutates the group set (split/merge) cannot let + // `route()` return a ShardId absent from `placement`. + debug_assert!( + router + .all_shards() + .iter() + .all(|s| placement.contains_key(s)), + "router shard space must be covered by placement" + ); + let (client, _blocking) = build_forwarding_clients()?; + + tracing::info!( + region = region_name, + hosted_groups = groups.len(), + total_groups = placement.len(), + "cluster node started (m11p6: one replica per hosted shard group)" + ); + + Ok(Self { + placement, + groups, + router, + node_http, + client, + shutting_down: AtomicBool::new(false), + }) + } + + /// Start the per-group election driver on every hosted replica (called once + /// the node is in its final `Arc`, like the single-node `started` hook). + pub fn start_all_elections(&self) { + for replica in self.groups.values() { + replica.start_election_driver(); + } + } + + /// Route an entity-scoped write to the replica that owns its shard group. + fn route_entity(&self, entity_id: u64) -> EntityRoute { + let shard = self.router.route(EntityId::new(entity_id)); + if let Some(replica) = self.groups.get(&shard) { + return EntityRoute::Local(Arc::clone(replica)); + } + EntityRoute::Remote { + shard, + candidates: self.forward_candidates(shard), + } + } + + /// Ordered HTTP bases to forward a write for a group this node does NOT host: + /// the group's believed leader first (placement's term-0/preferred leader), + /// then its other replicas, deduped and skipping any without a known HTTP + /// base. The receiver re-routes to the CURRENT leader, so any live replica + /// suffices — leader-first just minimizes the extra hop, and the ordered list + /// lets the gateway fail over on a connect error instead of pinning a write + /// to one dead replica. + fn forward_candidates(&self, shard: ShardId) -> Vec { + let Some(group) = self.placement.get(&shard) else { + return Vec::new(); + }; + let mut seen = HashSet::new(); + std::iter::once(group.leader) + .chain(group.replicas.iter().map(|r| r.region)) + .filter(|r| seen.insert(*r)) + .filter_map(|r| self.node_http.get(&r).cloned()) + .collect() + } + + /// Resolve the [`ShardReplica`] for an admin/read/status surface. `None` + /// returns the FIRST hosted group — exact for `S=1` (the sole group), and the + /// gateway's default for surfaces that do not yet take a `?shard=` selector + /// (the L3 per-shard admin work passes `Some(shard)`). `Some(s)` returns the + /// keyed group or a 400 if this node hosts no replica of it. This is the one + /// seam the deferred `?shard=` selector threads through — handlers never reach + /// into `self.groups` directly. + fn replica_for( + &self, + shard: Option, + ) -> std::result::Result, ClusterAppError> { + if let Some(s) = shard { + return self.groups.get(&s).cloned().ok_or_else(|| { + ClusterAppError(ServerError::BadRequest(format!( + "shard {} is not hosted by this node", + s.0 + ))) + }); + } + self.groups.values().next().cloned().ok_or_else(|| { + ClusterAppError(ServerError::Cluster( + "node hosts no shard group (new() rejects empty)".into(), + )) + }) + } + + /// Borrow every hosted replica (status aggregation, health, shutdown). + fn hosted(&self) -> impl Iterator> { + self.groups.values() + } + + /// The `TidalDb` of every hosted group, for an in-process read scatter + /// (m11p6: each group holds a disjoint entity subset, so a corpus-wide + /// `/feed`//`/search` queries every local group and merges). A group that is + /// shutting down is skipped (its `db_arc` errors). For `S=1` this is the one + /// db — byte-for-byte today's single read. + fn hosted_dbs(&self) -> Vec> { + self.groups + .values() + .filter_map(|r| r.db_arc().ok()) + .collect() + } + + /// A per-shard status row for every hosted group (m11p6 status surface). + fn shard_status_rows(&self) -> Vec { + self.groups + .values() + .filter_map(|r| { + let s = r.local_status().ok()?; + Some(ShardStatusRow { + shard: r.group_shard.0, + is_leader: s.is_leader, + leader: s.leader, + term: s.term, + role: s.role, + applied_events: s.applied_events, + lag_events: s.lag_events, + commit_index: s.commit_index, + }) + }) + .collect() + } + + /// Resolve an entity write to its local replica, or forward it to a node + /// hosting the group. The single gateway-routing seam for every entity-write + /// handler: `Continue(replica)` ⇒ apply on the local per-group path, + /// `Break(resp)` ⇒ the relayed cross-node response to return verbatim. + async fn route_or_forward( + &self, + key: u64, + path: &str, + body: &B, + headers: &HeaderMap, + ) -> std::ops::ControlFlow, Arc> + { + match self.route_entity(key) { + EntityRoute::Local(replica) => std::ops::ControlFlow::Continue(replica), + EntityRoute::Remote { shard, candidates } => std::ops::ControlFlow::Break( + self.forward_to_group_node(&candidates, shard, path, body, headers) + .await, + ), + } + } + + /// Forward an entity write to a remote node that hosts its group (this node + /// hosts no replica). `internal=false` so the receiver re-routes to its + /// group's CURRENT leader; the caller's `x-tidal-ack` and the verdict headers + /// relay. Walks `candidates` (leader-first) on a CONNECT failure so one dead + /// replica does not 503 a write the group's surviving quorum can serve; a + /// real verdict (a 2xx/NotLeader/QuorumTimeout STATUS) comes back as `Ok` and + /// is relayed immediately, never retried. + async fn forward_to_group_node( + &self, + candidates: &[String], + shard: ShardId, + path: &str, + body: &B, + headers: &HeaderMap, + ) -> std::result::Result { + if candidates.is_empty() { + return Err(ClusterAppError(ServerError::Unavailable(format!( + "shard {} has no reachable replica to route the write to", + shard.0 + )))); + } + let auth = forwarded_auth(headers); + let passthrough = forward::ack_passthrough(headers); + let mut last_err = String::new(); + for http in candidates { + let url = peer_url(http, path); + match forward_json_with_headers( + &self.client, + &url, + body, + auth.as_deref(), + false, + &passthrough, + ) + .await + { + // A real verdict (status received): relay it, do not try another + // replica. The receiver already re-routed to its current leader. + Ok(resp) => return Ok(forward::relay_forwarded(resp)), + // A connect/transport failure: this replica is unreachable — + // try the next candidate (the group's quorum may still be live). + Err(e) => last_err = format!("{url}: {e}"), + } + } + Err(ClusterAppError(ServerError::Unavailable(format!( + "shard {} unreachable: all {} replica candidates failed (last {last_err})", + shard.0, + candidates.len() + )))) + } + + /// Ready iff not draining AND every hosted group is ready (m11p6: a node + /// serves traffic only when all its shard replicas can). + fn is_ready(&self) -> bool { + !self.shutting_down.load(Ordering::Acquire) && self.hosted().all(|r| r.is_ready()) + } + + /// Flip `/health` to not-ready and propagate to every hosted replica. + pub fn set_shutting_down(&self) { + self.shutting_down.store(true, Ordering::Release); + for replica in self.hosted() { + replica.set_shutting_down(); + } + } + + /// True once shutdown began. + #[must_use] + pub fn is_shutting_down(&self) -> bool { + self.shutting_down.load(Ordering::Acquire) + } + + /// Deterministically shut down every hosted replica (checkpoint, WAL fsync, + /// and thread join per group). Reclaims sole ownership of each via + /// `try_unwrap` — the election drivers and transport sources hold only + /// `Weak`, which `serve_state` already relies on for the single-group path. + pub fn shutdown(&mut self) { + self.shutting_down.store(true, Ordering::Release); + for (shard, replica) in std::mem::take(&mut self.groups) { + match Arc::try_unwrap(replica) { + Ok(mut r) => r.shutdown(), + Err(arc) => { + arc.set_shutting_down(); + tracing::warn!( + shard = shard.0, + strong = Arc::strong_count(&arc), + "shard replica still referenced at shutdown; signalled drain but \ + could not run its deterministic close (a request Arc outlived drain)" + ); + } + } + } + } +} + // ── Router ────────────────────────────────────────────────────────────────── /// Build the multi-process region router. @@ -3233,7 +3676,7 @@ fn build_forwarding_clients() -> Result<(reqwest::Client, reqwest::blocking::Cli /// broadcast leader→peers, `/cluster/status` aggregates every region, the /// `/cluster/reconcile*` pair exchanges CRDT snapshots, and `/sharded/*` fans /// out across processes. -pub fn build_region_router(state: Arc, api_key: Option>) -> Router { +pub fn build_region_router(node: Arc, api_key: Option>) -> Router { let public = Router::new() .route("/health", get(region_health)) .route("/health/startup", get(crate::health::health_startup)) @@ -3241,7 +3684,7 @@ pub fn build_region_router(state: Arc, api_key: Option, api_key: Option protected.layer(middleware::from_fn(move |req: Request, next: Next| { @@ -3295,22 +3738,33 @@ pub fn build_region_router(state: Arc, api_key: Option>, + State(node): State>, ) -> std::result::Result<(StatusCode, Json), ClusterAppError> { + let state = node.replica_for(None)?; // m11p5 §4 readiness: 503 while shutting down, quarantined, or an install - // boot has not yet first-converged (sticky-ready after). A restarted - // PVC-retained voter keeps today's behavior. - if !state.is_ready() { - let cause = if state.is_shutting_down() { + // boot has not yet first-converged (sticky-ready after). m11p6: the node is + // ready only when EVERY hosted shard group is ready (aggregate). + if !node.is_ready() { + // Inspect the group ACTUALLY keeping the node unready (S>1: not + // necessarily the first hosted group) so the 503 cause is accurate; + // fall back to the default group if only the node-level drain flag is set. + let unready = node + .hosted() + .find(|r| !r.is_ready()) + .cloned() + .unwrap_or_else(|| Arc::clone(&state)); + let cause = if unready.is_shutting_down() { "shutting down" - } else if state + } else if unready .election_runtime .get() .is_some_and(|rt| rt.is_quarantined()) { "quarantined (divergent suffix); reseeds on next boot" - } else if state.membership().self_role() == Some(tidaldb::wal::format::MemberRole::Removed) + } else if unready.membership().self_role() + == Some(tidaldb::wal::format::MemberRole::Removed) { "removed from the cluster (decommissioned)" } else { @@ -3419,6 +3873,34 @@ pub struct LocalStatusResponse { /// or `removed`. Distinct from `role` (the election role); a learner is a /// `follower` here. membership_role: String, + /// Per-shard-group status for every group THIS node hosts (m11p6). For the + /// legacy single group this is one row mirroring the flat fields above; with + /// sharding it carries one row per hosted group, so an operator (and the + /// kill-node exit gate) can see which shard-leaderships this node holds and + /// confirm only a dead node's leaderships move. + #[serde(default)] + shards: Vec, +} + +/// One hosted shard group's status within a [`LocalStatusResponse`] (m11p6). +#[derive(Clone, Serialize, ToSchema)] +pub struct ShardStatusRow { + /// The data-shard group id. + shard: u16, + /// Whether this node currently leads this group. + is_leader: bool, + /// The region this node believes leads this group (`null` mid-election). + leader: Option, + /// This group's current election term. + term: u64, + /// This node's election role in this group (`leader`/`follower`/…). + role: String, + /// Replication events this node has applied for this group. + applied_events: u64, + /// Events this node lags this group's leader by (0 when leading). + lag_events: u64, + /// This group's quorum commit index (meaningful when leading). + commit_index: u64, } /// Local replication / leadership status for THIS region. @@ -3431,10 +3913,19 @@ pub struct LocalStatusResponse { (status = 503, description = "Server shutting down"), ), )] +#[allow(clippy::significant_drop_tightening)] pub async fn status_local( - State(state): State>, + State(node): State>, ) -> std::result::Result, ClusterAppError> { - Ok(Json(state.local_status().map_err(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. + let mut status = node + .replica_for(None)? + .local_status() + .map_err(ClusterAppError)?; + status.shards = node.shard_status_rows(); + Ok(Json(status)) } // ── Aggregated cluster status ───────────────────────────────────────────────── @@ -3442,14 +3933,25 @@ pub async fn status_local( /// `GET /cluster/status` aggregated response body. #[derive(Serialize, ToSchema)] pub struct AggregatedStatusResponse { - /// This node's view of the current leader. + /// This node's view of the current leader. **For `S=1` this is THE cluster + /// leader; for `S>1` it is the DEFAULT (lowest-id hosted) group's leader** — + /// a sharded cluster has one leader per group, so read `shards` for the full + /// picture. The flat `leader`/`relay_log_len`/`regions` block is the + /// default group's cross-region view, kept verbatim for `S=1` back-compat. leader: String, - /// The leader's relay seqno (high-water-mark): the count every region should - /// converge to. Read from the leader's local status (or directly if this node - /// leads). + /// The default group's leader relay seqno (high-water-mark): the count every + /// region of that group should converge to. relay_log_len: u64, - /// Per-region replication status, in topology id order. + /// Per-region replication status for the default group, in topology id order. regions: Vec, + /// Per-hosted-group leadership rows from THIS node (m11p6): one row per shard + /// group this node replicates, so an operator (and the kill-node exit gate) + /// can see which shard-leaderships this node holds across all its groups — + /// the multi-leader truth the flat `leader` field cannot express. For `S=1` + /// this is a single row mirroring the flat fields. NB: it is THIS node's view + /// of its hosted groups; a cross-cluster per-shard merge is the L4 surface. + #[serde(default)] + shards: Vec, } /// One region's aggregated status within an [`AggregatedStatusResponse`]. @@ -3471,7 +3973,7 @@ pub struct AggregatedRegionStatus { reachable: bool, } -/// Aggregated replication status across EVERY region. +/// Aggregated replication status across EVERY region of the default shard group. /// /// Queries every region's `/cluster/status/local` (own region in-process, peers /// over HTTP) concurrently with a 500ms per-peer budget, then assembles the @@ -3479,6 +3981,11 @@ pub struct AggregatedRegionStatus { /// region's `lag_events = leader_last_seq.saturating_sub(applied)`. An /// unreachable peer is honestly reported as `reachable: false`, `partitioned: /// true`, `applied 0`, `lag = leader_last_seq` (worst-case). +/// +/// m11p6: the flat fields describe the DEFAULT (lowest-id hosted) group only — +/// exact for `S=1`. `shards` carries this node's per-group leadership so a +/// multi-shard operator/the kill-node gate is not blind to the other groups; a +/// full cross-cluster per-shard aggregation is the L4 surface. #[utoipa::path( get, path = "/cluster/status", @@ -3487,9 +3994,11 @@ pub struct AggregatedRegionStatus { (status = 200, description = "Cluster-wide replication status", body = AggregatedStatusResponse), ), )] +#[allow(clippy::significant_drop_tightening)] pub async fn cluster_status( - State(state): State>, + State(node): State>, ) -> std::result::Result, ClusterAppError> { + let state = node.replica_for(None)?; let leader_name = state.leader_name(); let regions = state.all_regions_for_status(); @@ -3601,6 +4110,9 @@ pub async fn cluster_status( leader: leader_name, relay_log_len: leader_last_seq, regions: region_rows, + // Per-hosted-group leadership for THIS node — the multi-shard view the + // flat `leader` cannot express (S=1 ⇒ one row mirroring the flat fields). + shards: node.shard_status_rows(), })) } @@ -3614,7 +4126,7 @@ pub async fn cluster_status( /// per-row flag (copied from each region's own local status, plus the /// unreachable override) stands. fn apply_leader_partition_view( - state: &Arc, + state: &Arc, mut rows: Vec, ) -> Vec { if !state.is_leader() { @@ -3678,11 +4190,13 @@ pub struct RegionRequest { // One linear protocol pass (marked leg -> fenced transfer -> takeover wait -> // legacy fallback); splitting it would scatter the transfer's ordering rules. #[allow(clippy::too_many_lines)] +#[allow(clippy::significant_drop_tightening)] pub async fn cluster_promote( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result, ClusterAppError> { + let state = node.replica_for(None)?; 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 @@ -3756,7 +4270,7 @@ pub async fn cluster_promote( // pull and frontier reports flow regardless of this leader's // outbound breaker state. let target_shard = shard_of_region(target); - let target_mark = |state: &RegionClusterState| { + let target_mark = |state: &ShardReplica| { state .commit .peer_marks() @@ -3955,11 +4469,13 @@ pub struct CatchupRequest { ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn cluster_catchup( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + let state = node.replica_for(None)?; if !is_internal(&headers) { return Err(ClusterAppError(ServerError::BadRequest( "/cluster/catchup is internal; the x-tidal-internal marker is required \ @@ -3994,9 +4510,11 @@ pub async fn cluster_catchup( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn cluster_reseed( - State(state): State>, + State(node): State>, ) -> std::result::Result { + let state = node.replica_for(None)?; // The resume seqno: this node's applied frontier (against the current // leader's shard) + 1 — the first seqno past what it has durably applied. let from_seqno = { @@ -4029,7 +4547,7 @@ pub async fn cluster_reseed( .into_response()) } -/// The outcome of [`RegionClusterState::handle_remove`]. +/// The outcome of [`ShardReplica::handle_remove`]. enum RemoveOutcome { /// The member was removed; the `Removed` record committed at `version`. Removed { version: u64 }, @@ -4079,9 +4597,11 @@ pub struct MembersResponse { ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn cluster_members( - State(state): State>, + State(node): State>, ) -> std::result::Result { + let state = node.replica_for(None)?; let roster = state.membership().roster(); let members = roster .members @@ -4130,11 +4650,13 @@ pub async fn cluster_members( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn cluster_member_remove( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + let state = node.replica_for(None)?; // Non-leader: forward to the leader (the conf-change must run there). if !is_internal(&headers) && !state.is_leader() { return forward_write(&state, "/cluster/members/remove", &req, &headers).await; @@ -4177,11 +4699,13 @@ pub async fn cluster_member_remove( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn cluster_join( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + let state = node.replica_for(None)?; if !is_internal(&headers) && !state.is_leader() { return forward_write(&state, "/cluster/join", &req, &headers).await; } @@ -4251,11 +4775,13 @@ pub struct JoinHttpRequest { ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn cluster_partition( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + let state = node.replica_for(None)?; if !is_internal(&headers) && !state.is_leader() { return forward_write(&state, "/cluster/partition", &req, &headers).await; } @@ -4289,11 +4815,13 @@ pub async fn cluster_partition( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn cluster_heal( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + let state = node.replica_for(None)?; if !is_internal(&headers) && !state.is_leader() { return forward_write(&state, "/cluster/heal", &req, &headers).await; } @@ -4341,11 +4869,22 @@ pub async fn cluster_heal( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn create_item( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + // m11p6: hash-route to the replica that owns this entity's shard group. + // Continue ⇒ the existing leader/forward path below; Break ⇒ this node hosts + // no replica of the group, so the relayed cross-node response is returned. + let state = match node + .route_or_forward(req.entity_id, "/items", &req, &headers) + .await + { + std::ops::ControlFlow::Continue(replica) => replica, + std::ops::ControlFlow::Break(resp) => return resp, + }; // Non-leader external request → forward to the leader and relay verbatim. if !is_internal(&headers) && !state.is_leader() { return forward_write(&state, "/items", &req, &headers).await; @@ -4360,8 +4899,7 @@ pub async fn create_item( let entity = EntityId::new(req.entity_id); let metadata = req.metadata.clone(); let seq = - offload_region_read(move || RegionClusterState::apply_item_local(&db, entity, &metadata)) - .await?; + offload_region_read(move || ShardReplica::apply_item_local(&db, entity, &metadata)).await?; if ack == AckMode::Quorum && let Some(seq) = seq { @@ -4385,11 +4923,19 @@ pub async fn create_item( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn write_embedding( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + let state = match node + .route_or_forward(req.entity_id, "/embeddings", &req, &headers) + .await + { + std::ops::ControlFlow::Continue(replica) => replica, + std::ops::ControlFlow::Break(resp) => return resp, + }; if !is_internal(&headers) && !state.is_leader() { return forward_write(&state, "/embeddings", &req, &headers).await; } @@ -4400,10 +4946,9 @@ pub async fn write_embedding( let db = state.db_arc().map_err(ClusterAppError)?; let entity = EntityId::new(req.entity_id); let values = req.values.clone(); - let seq = offload_region_read(move || { - RegionClusterState::apply_embedding_local(&db, entity, &values) - }) - .await?; + let seq = + offload_region_read(move || ShardReplica::apply_embedding_local(&db, entity, &values)) + .await?; if ack == AckMode::Quorum && let Some(seq) = seq { @@ -4438,11 +4983,20 @@ pub async fn write_embedding( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn write_signal( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + // m11p6: hash-route the signal to the replica owning its entity's group. + let state = match node + .route_or_forward(req.entity_id, "/signals", &req, &headers) + .await + { + std::ops::ControlFlow::Continue(replica) => replica, + std::ops::ControlFlow::Break(resp) => return resp, + }; // A non-leader external request forwards to the leader. A marked request // (forwarded here) must be on the leader, or the cluster has a stale view — // fall through to the leader check, which surfaces NotLeader honestly. @@ -4523,11 +5077,21 @@ pub struct HardNegRequest { ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn write_hardneg( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + // m11p6: a hard-negative co-locates with the ITEM it hides (entity-sharded), + // so the item's shard ranks-and-filters it; route by item_id. + let state = match node + .route_or_forward(req.item_id, "/hardnegs", &req, &headers) + .await + { + std::ops::ControlFlow::Continue(replica) => replica, + std::ops::ControlFlow::Break(resp) => return resp, + }; if !is_internal(&headers) && !state.is_leader() { return forward_write(&state, "/hardnegs", &req, &headers).await; } @@ -4558,7 +5122,7 @@ pub async fn write_hardneg( /// runbook §8). A leadership change mid-wait (epoch bump / deactivation) /// fails with `NotLeader`: a demoted leader must never claim quorum. async fn await_quorum( - state: &Arc, + state: &Arc, seq: u64, ) -> std::result::Result { if state.commit.needed_peers() == 0 { @@ -4650,16 +5214,16 @@ impl tidal_net::sources::AppliedSink for CommitIndexSink { /// Node-side consumer of a typed `snapshot-required` catch-up refusal (m11p5 /// §2.4): latch the durable reseed marker so the reseed runs on the next boot. /// -/// Holds a `Weak` (set once the node is in its final Arc), +/// Holds a `Weak` (set once the node is in its final Arc), /// so the transport's catch-up path can never keep the node alive past /// shutdown. The retry timer keeps its standing wake-up regardless (the m11p4 /// re-arm-on-skip liveness fix) — this only latches the marker; it does not /// touch the running engine. /// The leader-side `JoinCluster` adapter (m11p5 §3.3): bridges the gRPC -/// `JoinHooks` trait to [`RegionClusterState::handle_join`]. A `Weak` so the +/// `JoinHooks` trait to [`ShardReplica::handle_join`]. A `Weak` so the /// runtime cannot leak the node. struct NodeJoinHooks { - node: Weak, + node: Weak, } impl tidal_net::JoinHooks for NodeJoinHooks { @@ -4684,7 +5248,7 @@ impl tidal_net::JoinHooks for NodeJoinHooks { } struct NodeSnapshotRequiredSink { - node: Weak, + node: Weak, } impl tidal_net::sources::SnapshotRequiredSink for NodeSnapshotRequiredSink { @@ -4728,7 +5292,7 @@ fn with_seq_header(status: StatusCode, seq: Option) -> Response { } async fn forward_write( - state: &Arc, + state: &Arc, path: &str, body: &B, headers: &HeaderMap, @@ -4742,11 +5306,7 @@ async fn forward_write( let auth = forwarded_auth(headers); // The caller's ack-mode override travels WITH the write (m11p3): the // leader honors the caller's choice, not this gateway's default. - let passthrough: Vec<(&'static str, String)> = headers - .get(forward::ACK_HEADER) - .and_then(|v| v.to_str().ok()) - .map(|v| vec![(forward::ACK_HEADER, v.to_owned())]) - .unwrap_or_default(); + let passthrough = forward::ack_passthrough(headers); match forward_json_with_headers( &state.client, &url, @@ -4757,26 +5317,10 @@ async fn forward_write( ) .await { - Ok(resp) => { - // Relay the leader's seq/dedup headers so the original caller - // sees the write's replicated-log verdict through the forward. - let mut response = (resp.status, Json(resp.body)).into_response(); - if let Some(seq) = resp.seq - && let Ok(value) = axum::http::HeaderValue::from_str(&seq) - { - response.headers_mut().insert( - axum::http::HeaderName::from_static(forward::SEQ_HEADER), - value, - ); - } - if resp.deduplicated { - response.headers_mut().insert( - axum::http::HeaderName::from_static(forward::DEDUP_HEADER), - axum::http::HeaderValue::from_static(forward::DEDUP_HEADER_VALUE), - ); - } - Ok(response) - } + // Relay the leader's seq/dedup headers so the original caller sees the + // write's replicated-log verdict through the forward (shared with the + // cross-shard gateway hop via `forward::relay_forwarded`). + Ok(resp) => Ok(forward::relay_forwarded(resp)), Err(e) => { // Leader unreachable: the typed 503 names the leader, its address, // and the connect error (single body shape via ClusterAppError). @@ -4794,7 +5338,7 @@ async fn forward_write( /// Best-effort broadcast of an item/embedding write to every peer with the /// internal marker set. Returns the per-peer success/failure outcome. async fn broadcast_to_peers( - state: &Arc, + state: &Arc, path: &str, body: &B, headers: &HeaderMap, @@ -4812,6 +5356,46 @@ async fn broadcast_to_peers( .await } +/// Scatter a corpus-wide read over the hosted shard groups and merge: run +/// `per_db` on each group's [`TidalDb`], **SUM** `total_candidates` (entity- +/// sharded groups own disjoint key subsets — no dedup-by-max would undercount), +/// then score-sort descending and truncate to `limit`. The ONE place the +/// `/feed`//`/search` merge contract lives, so the two surfaces cannot drift. +/// +/// `S=1` short-circuits to the single group's result, which the engine already +/// returns score-sorted and limited — no redundant re-sort, byte-for-byte the +/// pre-m11p6 single read. The cross-group merge keeps each group's own diversity +/// pass but does NOT re-diversify across groups (a cross-shard re-rank is the L4 +/// follow-up; disjoint groups make the score-merge sound for cardinality). +fn scatter_merge( + dbs: &[Arc], + limit: usize, + score: impl Fn(&T) -> f64, + per_db: F, +) -> std::result::Result<(Vec, usize), ServerError> +where + F: Fn(&TidalDb) -> std::result::Result<(Vec, usize), ServerError>, +{ + // S=1: the one group's result is already ranked + limited by the engine. + if let [only] = dbs { + return per_db(only); + } + let mut merged: Vec = Vec::new(); + let mut total = 0usize; + for db in dbs { + let (items, candidates) = per_db(db)?; + total = total.saturating_add(candidates); + merged.extend(items); + } + merged.sort_by(|a, b| { + score(b) + .partial_cmp(&score(a)) + .unwrap_or(std::cmp::Ordering::Equal) + }); + merged.truncate(limit); + Ok((merged, total)) +} + /// Ranked feed. Default read region is LOCAL; a `?region=` that names a DIFFERENT /// region is forwarded to that region's process (region-aware reads). An internal /// (marked) request always serves locally, so a forwarded read never loops. @@ -4828,12 +5412,14 @@ async fn broadcast_to_peers( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn feed( - State(state): State>, + State(node): State>, headers: HeaderMap, RawQuery(raw_query): RawQuery, Query(query): Query, ) -> std::result::Result { + let state = node.replica_for(None)?; // Region-aware read: forward a foreign `?region=` to its owner unless this is // already an internal (forwarded) request, which serves locally. if is_internal(&headers) { @@ -4853,9 +5439,8 @@ pub async fn feed( return Ok(resp); } - let mut builder = Retrieve::builder() - .profile(&query.profile) - .limit(query.clamped_limit() as usize); + let limit = query.clamped_limit() as usize; + let mut builder = Retrieve::builder().profile(&query.profile).limit(limit); if let Some(user_id) = query.user_id { builder = builder.for_user(user_id); } @@ -4863,13 +5448,27 @@ pub async fn feed( .build() .map_err(|e| ClusterAppError(ServerError::Tidal(e.into())))?; - let db = state.db_arc().map_err(ClusterAppError)?; - let result = - offload_region_read(move || db.retrieve(&retrieve).map_err(ServerError::Tidal)).await?; + // m11p6: scatter the corpus-wide read over the hosted shard groups and merge + // (see `scatter_merge`). NB: complete only when this node hosts a replica of + // EVERY group (S=1 and the RF=N exit-gate shape); cross-node read fan-out for + // a partial placement is the L4 follow-up (use `/sharded/*` until then). + let dbs = node.hosted_dbs(); + let (items, total_candidates) = offload_region_read(move || { + scatter_merge( + &dbs, + limit, + |it: &tidaldb::query::RetrieveResult| it.score, + |db| { + let r = db.retrieve(&retrieve).map_err(ServerError::Tidal)?; + Ok((r.items, r.total_candidates)) + }, + ) + }) + .await?; Ok(Json(FeedResponse { - items: feed_items(&result.items), - total_candidates: result.total_candidates, + items: feed_items(&items), + total_candidates, region: query.region, }) .into_response()) @@ -4890,12 +5489,14 @@ pub async fn feed( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn search( - State(state): State>, + State(node): State>, headers: HeaderMap, RawQuery(raw_query): RawQuery, Query(query): Query, ) -> std::result::Result { + let state = node.replica_for(None)?; if is_internal(&headers) { state .read_region(query.region.as_deref()) @@ -4912,9 +5513,8 @@ pub async fn search( return Ok(resp); } - let mut builder = Search::builder() - .query(&query.query) - .limit(query.clamped_limit()); + let limit = query.clamped_limit(); + let mut builder = Search::builder().query(&query.query).limit(limit); if let Some(user_id) = query.user_id { builder = builder.for_user(user_id); } @@ -4922,16 +5522,26 @@ pub async fn search( .build() .map_err(|e| ClusterAppError(ServerError::Tidal(e.into())))?; - let db = state.db_arc().map_err(ClusterAppError)?; - let result = offload_region_read(move || { - db.reload_text_index().map_err(ServerError::Tidal)?; - db.search(&search_query).map_err(ServerError::Tidal) + // m11p6: scatter the search over the hosted shard groups and merge (see + // `scatter_merge`; same full-placement caveat as `feed`). + let dbs = node.hosted_dbs(); + let (items, total_candidates) = offload_region_read(move || { + scatter_merge( + &dbs, + limit as usize, + |it: &tidaldb::query::SearchResultItem| it.score, + |db| { + db.reload_text_index().map_err(ServerError::Tidal)?; + let r = db.search(&search_query).map_err(ServerError::Tidal)?; + Ok((r.items, r.total_candidates)) + }, + ) }) .await?; Ok(Json(SearchResponse { - items: search_items(&result.items), - total_candidates: result.total_candidates, + items: search_items(&items), + total_candidates, region: query.region, }) .into_response()) @@ -4946,7 +5556,7 @@ pub async fn search( /// `raw_query` is the originating request's verbatim query string (without the /// `?`), so the owner runs the IDENTICAL query (`profile`/`limit`/`user_id`/`region`). async fn maybe_forward_region_read( - state: &Arc, + state: &Arc, region: Option<&str>, path: &str, raw_query: Option<&str>, @@ -5033,11 +5643,13 @@ pub struct ReconcileSnapshotResponse { ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn cluster_reconcile_snapshot( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(remote): Json, ) -> std::result::Result, ClusterAppError> { + let state = node.replica_for(None)?; if !is_internal(&headers) { return Err(ClusterAppError(ServerError::BadRequest( "/cluster/reconcile/snapshot is internal; the x-tidal-internal marker is required \ @@ -5097,11 +5709,13 @@ pub struct ReconcileResponse { ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn cluster_reconcile( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + let state = node.replica_for(None)?; let id = state.resolve_region(&req.region).map_err(ClusterAppError)?; if id == state.region { return Err(ClusterAppError(ServerError::BadRequest( @@ -5176,20 +5790,20 @@ pub async fn cluster_reconcile( /// Build the engine `ShardRouter` shard list (every region, in id order) used /// for `/sharded/*` hash-partitioning and read fan-out. -fn sharded_region_ids(state: &Arc) -> Vec { +fn sharded_region_ids(state: &Arc) -> Vec { let mut ids: Vec = state.id_to_name.keys().copied().collect(); ids.sort_by_key(|r| r.0); ids } /// Region id → name map for the scatter-gather metadata. -fn sharded_region_names(state: &Arc) -> HashMap { +fn sharded_region_names(state: &Arc) -> HashMap { state.id_to_name.clone() } /// Build the HTTP scatter-gather context for this gateway. fn http_shard_context( - state: &Arc, + state: &Arc, auth: Option, ) -> Result> { let db = state.db_arc()?; @@ -5206,7 +5820,7 @@ fn http_shard_context( /// not the owner; else apply locally. Returns the relayed response, or the local /// status on a local apply. async fn sharded_write_route( - state: &Arc, + state: &Arc, headers: &HeaderMap, entity_id: u64, path: &str, @@ -5253,11 +5867,13 @@ async fn sharded_write_route( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn sharded_create_item( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + let state = node.replica_for(None)?; let db = state.db_arc().map_err(ClusterAppError)?; let entity = EntityId::new(req.entity_id); let metadata = req.metadata.clone(); @@ -5267,7 +5883,7 @@ pub async fn sharded_create_item( req.entity_id, "/sharded/items", &req, - move || RegionClusterState::apply_item_local(&db, entity, &metadata).map(|_seq| ()), + move || ShardReplica::apply_item_local(&db, entity, &metadata).map(|_seq| ()), StatusCode::CREATED, ) .await @@ -5287,11 +5903,13 @@ pub async fn sharded_create_item( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn sharded_write_embedding( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + let state = node.replica_for(None)?; let db = state.db_arc().map_err(ClusterAppError)?; let entity = EntityId::new(req.entity_id); let values = req.values.clone(); @@ -5301,7 +5919,7 @@ pub async fn sharded_write_embedding( req.entity_id, "/sharded/embeddings", &req, - move || RegionClusterState::apply_embedding_local(&db, entity, &values).map(|_seq| ()), + move || ShardReplica::apply_embedding_local(&db, entity, &values).map(|_seq| ()), StatusCode::NO_CONTENT, ) .await @@ -5323,11 +5941,13 @@ pub async fn sharded_write_embedding( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn sharded_write_signal( - State(state): State>, + State(node): State>, headers: HeaderMap, Json(req): Json, ) -> std::result::Result { + let state = node.replica_for(None)?; let db = state.db_arc().map_err(ClusterAppError)?; let entity = EntityId::new(req.entity_id); let signal = req.signal.clone(); @@ -5362,11 +5982,13 @@ pub async fn sharded_write_signal( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn sharded_feed( - State(state): State>, + State(node): State>, headers: HeaderMap, Query(query): Query, ) -> std::result::Result, ClusterAppError> { + let state = node.replica_for(None)?; let mut builder = Retrieve::builder() .profile(query.profile()) .limit(query.clamped_limit()); @@ -5408,11 +6030,13 @@ pub async fn sharded_feed( ), security(("bearerAuth" = [])), )] +#[allow(clippy::significant_drop_tightening)] pub async fn sharded_search( - State(state): State>, + State(node): State>, headers: HeaderMap, Query(query): Query, ) -> std::result::Result, ClusterAppError> { + let state = node.replica_for(None)?; let mut builder = Search::builder() .query(query.query_text()) .limit(query.clamped_limit_u32()); diff --git a/tidal-server/src/cluster/reseed.rs b/tidal-server/src/cluster/reseed.rs index d97bac6..06518f0 100644 --- a/tidal-server/src/cluster/reseed.rs +++ b/tidal-server/src/cluster/reseed.rs @@ -2,7 +2,7 @@ //! //! When a node's data dir carries a durable `reseed_required` marker, it does //! NOT open the existing engine and serve degraded forever — instead, BEFORE -//! constructing [`RegionClusterState`], it: +//! constructing [`ShardReplica`], it: //! //! 1. **Recovers any interrupted prior swap** (§2.3) — this runs FIRST, before //! `ElectionStore` classification, because classifying first re-opens restart @@ -25,7 +25,7 @@ //! voting enabled (§2.2; the reseed needs a leader and the leader may need this //! node's vote, so blocking boot forever would deadlock the cluster). The //! post-open seed (§2.6) is driven by the install sentinel inside -//! [`RegionClusterState::new`], not here. +//! [`ShardReplica::new`], not here. //! //! # The swap protocol (§2.3, review-corrected ordering) //! @@ -67,7 +67,7 @@ const COMPLETE_SENTINEL: &str = "COMPLETE"; /// The install sentinel copied INTO staging BEFORE the swap (§2.6). /// /// Its presence in the canonical data dir AFTER open tells -/// `RegionClusterState::new` to run the post-open seed exactly once +/// `ShardReplica::new` to run the post-open seed exactly once /// (crash-idempotent: a crash before the seed's sentinel-delete re-runs the same /// seed against the unchanged WAL). pub const INSTALL_PENDING_SENTINEL: &str = "reseed-install-pending"; @@ -370,7 +370,7 @@ struct Candidate { /// The outcome of a boot-time install attempt. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum InstallOutcome { - /// A snapshot was installed (the swap completed). `RegionClusterState::new` + /// A snapshot was installed (the swap completed). `ShardReplica::new` /// will find the install sentinel and run the §2.6 post-open seed. Installed, /// The leader reported `needed=false` (the live WAL still serves our @@ -385,7 +385,7 @@ pub enum InstallOutcome { /// Run the boot-time snapshot install for a marker boot (§2.2–§2.7). /// -/// Returns the outcome; `RegionClusterState::new` is constructed afterward in +/// Returns the outcome; `ShardReplica::new` is constructed afterward in /// every case (a fresh dir after `Installed`, the existing dir otherwise). /// /// # Errors @@ -532,7 +532,7 @@ pub fn run_boot_install_with( /// Returns: /// - `Ok(Some(snapshot_seq))` — a snapshot was streamed, verified, and swapped /// in; the new data dir carries the install sentinel (recording -/// `leader_region`) so `RegionClusterState::new` runs the §2.6 post-open seed. +/// `leader_region`) so `ShardReplica::new` runs the §2.6 post-open seed. /// - `Ok(None)` — the leader reported `needed=false` (the live WAL still serves /// `from_seqno`): nothing was installed, proceed empty (the small-cluster case /// where the stream serves from 1). @@ -780,7 +780,7 @@ fn attempt_install( snapshot_seq, term = leader.term, "reseed boot: snapshot installed and swapped in; the post-open seed runs once \ - RegionClusterState opens the new data dir (§2.6)" + ShardReplica opens the new data dir (§2.6)" ); Ok(AttemptResult::Installed) } @@ -999,7 +999,7 @@ fn copy_identity_into_staging(data_dir: &Path, staging: &Path) -> Result<()> { /// The install sentinel's body (m11p5 §2.6). /// /// It carries the artifact's recovered seq plus the DISCOVERED leader's region. -/// Both drive the post-open path: `RegionClusterState::new` seeds the frontier +/// Both drive the post-open path: `ShardReplica::new` seeds the frontier /// and issues the post-install catch-up pull against the discovered leader's /// shard (NOT the boot topology leader, which for a reseeded ex-leader is the /// node itself). @@ -1058,7 +1058,7 @@ fn write_complete_sentinel(staging: &Path) -> Result<()> { /// Read the install sentinel's recorded body from `data_dir`, if present. /// -/// Called by `RegionClusterState::new` after `open_region_db` to drive the §2.6 +/// Called by `ShardReplica::new` after `open_region_db` to drive the §2.6 /// post-open seed AND the post-install catch-up pull. `None` ⇒ this was not an /// install boot. A 10-byte sentinel carries the discovered leader's region; an /// 8-byte one (an interrupted older install) is read with `leader_region: None` diff --git a/tidal-server/src/cluster/topology.rs b/tidal-server/src/cluster/topology.rs index 3c21abc..8e7aeb7 100644 --- a/tidal-server/src/cluster/topology.rs +++ b/tidal-server/src/cluster/topology.rs @@ -24,7 +24,7 @@ use crate::{ pub struct TopologySpec { /// All regions in declaration order. The 0-based index of a region in this /// list IS its `RegionId`, so every process that parses the same file agrees - /// on the region → id mapping (see `ClusterState::new` / `RegionClusterState::new`). + /// on the region → id mapping (see `ClusterState::new` / `ShardReplica::new`). pub regions: Vec, /// Name of the region that initially leads (must name a declared region). pub leader: String, @@ -56,6 +56,77 @@ pub struct TopologySpec { /// leader lease, auto-election on). #[serde(default)] pub election: ElectionSpec, + /// Optional shard-group declaration (m11p6). When present, the keyspace is + /// hash-split into `shards.len()` groups, each a replication group at RF = + /// `replicas.len()` with its own WAL/relay/commit-index and its own elected + /// leader; a node hosts one [`ShardReplica`](super::node) per group it + /// appears in. When ABSENT, the legacy "1 shard × RF=all-regions" group is + /// synthesized (id 0, every region a replica with its address verbatim, + /// `leader` = [`Self::leader`]) — byte-for-byte today's single-log cluster. + /// See [`Self::resolve_shard_groups`]. + #[serde(default)] + pub shards: Option>, +} + +/// One shard group's declaration (the optional `shards:` YAML block, m11p6). +#[derive(Debug, Deserialize)] +pub struct ShardSpec { + /// Dense, unique group id in `[0, shards.len())`. The gateway hash-routes an + /// entity to `ShardId(id)` via [`tidaldb::replication::ShardRouter`]; the id + /// also names the group's data subdir and gRPC port offset. + pub id: u16, + /// Term-0 / preferred leader (must name a declared region that is also one of + /// this group's `replicas`). Defaults to the first replica when omitted. + #[serde(default)] + pub leader: Option, + /// The RF replica nodes hosting this group. + pub replicas: Vec, +} + +/// One replica placement within a [`ShardSpec`] (m11p6). +#[derive(Debug, Deserialize)] +pub struct ShardReplicaSpec { + /// The hosting node — must name a declared region. + pub node: String, + /// This replica's advertised gRPC address (`host:port`, may be DNS). When + /// omitted, derived as the node's `grpc_addr` with `id` added to its port + /// (so id 0 = the node's address verbatim — the legacy port). + #[serde(default)] + pub grpc_addr: Option, + /// This replica's local gRPC bind (`SocketAddr`). When omitted, derived from + /// the node's `grpc_bind` (or `grpc_addr`) with `id` added to its port. + #[serde(default)] + pub grpc_bind: Option, +} + +/// A fully-resolved shard group: ids resolved to [`RegionId`]s, addresses +/// resolved (explicit or port-derived). Produced by +/// [`TopologySpec::resolve_shard_groups`] — the single seam that replaces the +/// 1:1 [`shard_of_region`] assumption. +#[derive(Debug, Clone)] +pub struct ResolvedShardGroup { + /// The data-shard group id (gateway hash output; map key; subdir/port name). + pub shard: ShardId, + /// The group's term-0 / preferred leader node. + pub leader: RegionId, + /// `leader`'s region name (display). + pub leader_name: String, + /// The group's RF replica placements. + pub replicas: Vec, +} + +/// One resolved replica placement within a [`ResolvedShardGroup`] (m11p6). +#[derive(Debug, Clone)] +pub struct ResolvedReplica { + /// The hosting node's region id. + pub region: RegionId, + /// The hosting node's region name. + pub name: String, + /// The replica's advertised gRPC address (explicit or port-derived). + pub grpc_addr: String, + /// The replica's local gRPC bind, when explicit/derived (else `None` ⇒ the + /// node derives it from `grpc_addr` at boot per `resolve_grpc_bind_addr`). + pub grpc_bind: Option, } /// Election / failure-detector tuning (the optional `election:` YAML block, @@ -257,6 +328,13 @@ pub struct RegionSpec { /// metrics listener before m11p1 and the roadmap calls that out as a gap. #[serde(default)] pub metrics_addr: Option, + /// Optional placement label (m11p6), e.g. an availability zone (`"az-a"`), + /// demoting "region" from a shard identity to a placement hint. **Parsed and + /// accepted but NOT yet honored:** the read-affinity router that would prefer + /// a replica in the reader's zone is a later layer (L4/p8), so setting this + /// today has no runtime effect. Carried now so the schema is forward-stable. + #[serde(default)] + pub zone: Option, } /// TLS material for one region's gRPC transport (the optional `grpc_tls:` @@ -440,13 +518,14 @@ fn validate_spec_values(spec: &TopologySpec) -> Result<()> { ))); } } + validate_shards(spec)?; Ok(()) } impl TopologySpec { /// Resolve the cluster-write-pool config from `write_workers`. /// - /// Shared by both `ClusterState::new` and `RegionClusterState::new` so the + /// Shared by both `ClusterState::new` and `ShardReplica::new` so the /// two cluster modes size their write pool the same way from the same field. #[must_use] pub(crate) fn write_pool_config(&self) -> ClusterWritePoolConfig { @@ -493,6 +572,236 @@ impl TopologySpec { .find(|r| r.name == region_name) .and_then(|r| r.metrics_addr.clone()) } + + /// The number of shard groups (m11p6): `shards.len()`, or 1 when `shards:` + /// is absent (the legacy single replicated log). Drives the gateway's + /// [`tidaldb::replication::ShardRouter`] (`Hash(n)` for `n > 1`, `Single` + /// for `n == 1`). + #[must_use] + pub fn shard_count(&self) -> usize { + self.shards.as_ref().map_or(1, Vec::len) + } + + /// Resolve the shard-group assignment (m11p6) — the single seam that + /// generalizes the 1:1 [`shard_of_region`] mapping. With `shards:` present, + /// resolves each replica's node name to a [`RegionId`] and its gRPC address + /// (explicit, or the node's address with the shard id added to its port). + /// Absent ⇒ one group spanning every region (id 0, the topology `leader`, + /// addresses verbatim) — byte-for-byte today's single-log cluster (id 0 ⇒ + /// port offset 0 ⇒ the declared address). + /// + /// # Errors + /// + /// Returns [`ServerError::SchemaConfig`] if a replica/leader names an + /// undeclared region, a group has no replicas, ids are not dense/unique, a + /// derived port overflows, or two shards on one node collide on a gRPC + /// address. + #[allow(clippy::too_many_lines)] + pub fn resolve_shard_groups(&self) -> Result> { + // The positional region→id map, overflow-guarded ONCE here (a RegionId is + // a u16). Every id below reuses this map rather than re-deriving the cast + // with a different overflow policy — one source of truth for the seam + // every durable id depends on. + let mut name_to_id: std::collections::HashMap<&str, RegionId> = + std::collections::HashMap::with_capacity(self.regions.len()); + for (i, r) in self.regions.iter().enumerate() { + let id = RegionId(u16::try_from(i).map_err(|_| { + ServerError::SchemaConfig("topology declares more than 65535 regions".into()) + })?); + name_to_id.insert(r.name.as_str(), id); + } + let region_of = |name: &str| -> Result { + name_to_id.get(name).copied().ok_or_else(|| { + ServerError::SchemaConfig(format!( + "shards: replica/leader names undeclared region '{name}'" + )) + }) + }; + + let Some(shards) = self.shards.as_ref() else { + // Legacy synthesis: one group, RF = all regions, addresses verbatim. + let leader = region_of(&self.leader)?; + let replicas = self + .regions + .iter() + .map(|r| ResolvedReplica { + region: name_to_id[r.name.as_str()], + name: r.name.clone(), + grpc_addr: r.grpc_addr.clone().unwrap_or_default(), + grpc_bind: r.grpc_bind.clone(), + }) + .collect(); + return Ok(vec![ResolvedShardGroup { + shard: ShardId(0), + leader, + leader_name: self.leader.clone(), + replicas, + }]); + }; + + let mut groups = Vec::with_capacity(shards.len()); + for spec in shards { + if spec.replicas.is_empty() { + return Err(ServerError::SchemaConfig(format!( + "shard {} declares no replicas", + spec.id + ))); + } + let mut replicas = Vec::with_capacity(spec.replicas.len()); + let mut seen_nodes = std::collections::HashSet::new(); + for r in &spec.replicas { + let region = region_of(&r.node)?; + if !seen_nodes.insert(region) { + return Err(ServerError::SchemaConfig(format!( + "shard {} lists node '{}' twice", + spec.id, r.node + ))); + } + let node = &self.regions[usize::from(region.0)]; + let grpc_addr = match &r.grpc_addr { + Some(a) => a.clone(), + None => offset_host_port( + node.grpc_addr.as_deref().ok_or_else(|| { + ServerError::SchemaConfig(format!( + "shard {} replica '{}' has no grpc_addr and node declares none \ + to derive from", + spec.id, r.node + )) + })?, + spec.id, + )?, + }; + let grpc_bind = match &r.grpc_bind { + Some(b) => Some(b.clone()), + None => node + .grpc_bind + .as_deref() + .map(|b| offset_host_port(b, spec.id)) + .transpose()?, + }; + replicas.push(ResolvedReplica { + region, + name: r.node.clone(), + grpc_addr, + grpc_bind, + }); + } + let leader_name = spec + .leader + .clone() + .unwrap_or_else(|| replicas[0].name.clone()); + let leader = region_of(&leader_name)?; + if !replicas.iter().any(|r| r.region == leader) { + return Err(ServerError::SchemaConfig(format!( + "shard {} leader '{leader_name}' is not one of its replicas", + spec.id + ))); + } + groups.push(ResolvedShardGroup { + shard: ShardId(spec.id), + leader, + leader_name, + replicas, + }); + } + validate_no_port_collisions(&groups)?; + Ok(groups) + } +} + +/// Add `offset` to the port of a `host:port` address; `offset == 0` returns the +/// input verbatim (the legacy-shard guarantee — no reformatting). +fn offset_host_port(addr: &str, offset: u16) -> Result { + if offset == 0 { + return Ok(addr.to_string()); + } + let (host, port) = addr.rsplit_once(':').ok_or_else(|| { + ServerError::SchemaConfig(format!( + "cannot derive a shard port from '{addr}' (no ':port' to offset)" + )) + })?; + let base: u16 = port.parse().map_err(|_| { + ServerError::SchemaConfig(format!("'{addr}' port is not a number to offset")) + })?; + let derived = base.checked_add(offset).ok_or_else(|| { + ServerError::SchemaConfig(format!( + "deriving shard port from '{addr}' + {offset} overflows u16 — set an explicit \ + grpc_addr for this replica" + )) + })?; + Ok(format!("{host}:{derived}")) +} + +/// Reject two shard replicas on the SAME node that resolve to the same gRPC +/// address (a derived/explicit port clash would make two of the node's shard +/// transports fight for one socket). +fn validate_no_port_collisions(groups: &[ResolvedShardGroup]) -> Result<()> { + let mut seen: std::collections::HashMap<(RegionId, &str), ShardId> = + std::collections::HashMap::new(); + for g in groups { + for r in &g.replicas { + if let Some(prev) = seen.insert((r.region, r.grpc_addr.as_str()), g.shard) { + return Err(ServerError::SchemaConfig(format!( + "node '{}' resolves shards {} and {} to the same grpc_addr '{}' — set \ + distinct ports (derived ports add the shard id to the node port)", + r.name, prev.0, g.shard.0, r.grpc_addr + ))); + } + } + } + Ok(()) +} + +/// Structural validation of the optional `shards:` block (mode-independent; +/// address resolution is deferred to [`TopologySpec::resolve_shard_groups`]). +/// Checks dense unique ids in `[0, n)`, non-empty replicas, and that named +/// nodes/leaders exist — so a typo fails at load, not at routing time. +fn validate_shards(spec: &TopologySpec) -> Result<()> { + let Some(shards) = spec.shards.as_ref() else { + return Ok(()); + }; + if shards.is_empty() { + return Err(ServerError::SchemaConfig( + "shards: declared but empty (omit the key for the legacy single group)".into(), + )); + } + let names: std::collections::HashSet<&str> = + spec.regions.iter().map(|r| r.name.as_str()).collect(); + let mut ids: Vec = shards.iter().map(|s| s.id).collect(); + ids.sort_unstable(); + for (want, got) in ids.iter().enumerate() { + if usize::from(*got) != want { + return Err(ServerError::SchemaConfig(format!( + "shard ids must be dense and unique in [0, {}); got {ids:?}", + shards.len() + ))); + } + } + for s in shards { + if s.replicas.is_empty() { + return Err(ServerError::SchemaConfig(format!( + "shard {} declares no replicas (RF must be >= 1)", + s.id + ))); + } + for r in &s.replicas { + if !names.contains(r.node.as_str()) { + return Err(ServerError::SchemaConfig(format!( + "shard {} replica node '{}' is not a declared region", + s.id, r.node + ))); + } + } + if let Some(l) = s.leader.as_deref() + && !s.replicas.iter().any(|r| r.node == l) + { + return Err(ServerError::SchemaConfig(format!( + "shard {} leader '{l}' must be one of its replicas", + s.id + ))); + } + } + Ok(()) } /// Validate a topology for **multi-process** (`--region`) mode. @@ -628,6 +937,7 @@ mod tests { http_addr: http.map(str::to_string), grpc_tls: None, metrics_addr: None, + zone: None, } } @@ -643,6 +953,7 @@ mod tests { replication: ReplicationSpec::default(), wal: WalSpec::default(), election: ElectionSpec::default(), + shards: None, } } @@ -837,4 +1148,120 @@ leader: us-east assert_eq!(tls.ca_cert, PathBuf::from("/etc/tidal/ca.pem")); assert!(tls.client_cert.is_none(), "mTLS pair defaults to off"); } + + // ── m11p6 shard-group schema ───────────────────────────────────────────── + + #[test] + fn absent_shards_synthesizes_one_group_verbatim() { + // The legacy guarantee: no `shards:` ⇒ one group, RF = all regions, + // leader = topology.leader, addresses byte-for-byte (offset 0). + let t = full_topology(); + assert_eq!(t.shard_count(), 1); + let groups = t.resolve_shard_groups().expect("legacy synthesis"); + assert_eq!(groups.len(), 1); + let g = &groups[0]; + assert_eq!(g.shard, ShardId(0)); + assert_eq!(g.leader, RegionId(0)); + assert_eq!(g.leader_name, "us-east"); + assert_eq!(g.replicas.len(), 2); + assert_eq!(g.replicas[0].region, RegionId(0)); + assert_eq!(g.replicas[0].grpc_addr, "127.0.0.1:9601"); // verbatim + assert_eq!(g.replicas[1].grpc_addr, "127.0.0.1:9602"); // verbatim + } + + fn three_node_yaml(shards: &str) -> TopologySpec { + let yaml = format!( + "regions:\n - {{ name: us-east, grpc_addr: \"127.0.0.1:9601\", \ + http_addr: \"127.0.0.1:9501\", zone: az-a }}\n - {{ name: eu-west, \ + grpc_addr: \"127.0.0.1:9611\", http_addr: \"127.0.0.1:9511\", zone: az-b }}\n \ + - {{ name: ap-south, grpc_addr: \"127.0.0.1:9621\", http_addr: \"127.0.0.1:9521\" }}\n\ + leader: us-east\n{shards}" + ); + load_topology_from_str(&yaml).expect("topology parses + validates") + } + + fn load_topology_from_str(raw: &str) -> Result { + let spec: TopologySpec = serde_yml::from_str(raw) + .map_err(|e| ServerError::SchemaConfig(format!("parse: {e}")))?; + validate_spec_values(&spec)?; + Ok(spec) + } + + #[test] + fn explicit_shards_resolve_with_derived_and_explicit_ports() { + let t = three_node_yaml( + "shards:\n - id: 0\n leader: us-east\n replicas: [{node: us-east}, \ + {node: eu-west}, {node: ap-south}]\n - id: 1\n leader: eu-west\n \ + replicas: [{node: us-east, grpc_addr: \"127.0.0.1:7000\"}, {node: eu-west}, \ + {node: ap-south}]\n", + ); + assert_eq!(t.shard_count(), 2); + // `zone:` parses onto the RegionSpec (accepted, not yet honored). + assert_eq!(t.regions[0].zone.as_deref(), Some("az-a")); + let groups = t.resolve_shard_groups().expect("resolve"); + assert_eq!(groups.len(), 2); + // Shard 0: derived = node port + 0 (verbatim). + assert_eq!(groups[0].shard, ShardId(0)); + assert_eq!(groups[0].leader, RegionId(0)); + assert_eq!(groups[0].replicas[0].grpc_addr, "127.0.0.1:9601"); + assert_eq!(groups[0].replicas[1].grpc_addr, "127.0.0.1:9611"); + // Shard 1: us-east explicit 7000; eu-west/ap-south derived = port + 1. + assert_eq!(groups[1].shard, ShardId(1)); + assert_eq!(groups[1].leader, RegionId(1)); + assert_eq!(groups[1].replicas[0].grpc_addr, "127.0.0.1:7000"); + assert_eq!(groups[1].replicas[1].grpc_addr, "127.0.0.1:9612"); + assert_eq!(groups[1].replicas[2].grpc_addr, "127.0.0.1:9622"); + } + + #[test] + fn shards_reject_non_dense_ids() { + let yaml = "shards:\n - {id: 0, replicas: [{node: us-east}]}\n \ + - {id: 2, replicas: [{node: eu-west}]}\n"; + let err = load_topology_from_str(&format!( + "regions:\n - {{name: us-east, grpc_addr: \"1:1\", http_addr: \"1:1\"}}\n \ + - {{name: eu-west, grpc_addr: \"1:2\", http_addr: \"1:2\"}}\nleader: us-east\n{yaml}" + )) + .expect_err("non-dense ids rejected"); + assert!(err.to_string().contains("dense"), "got {err}"); + } + + #[test] + fn shards_reject_unknown_node_and_bad_leader() { + let bad_node = load_topology_from_str( + "regions:\n - {name: us-east, grpc_addr: \"1:1\", http_addr: \"1:1\"}\nleader: us-east\n\ + shards:\n - {id: 0, replicas: [{node: ghost}]}\n", + ) + .expect_err("unknown node"); + assert!(bad_node.to_string().contains("ghost"), "got {bad_node}"); + + let bad_leader = load_topology_from_str( + "regions:\n - {name: us-east, grpc_addr: \"1:1\", http_addr: \"1:1\"}\n \ + - {name: eu-west, grpc_addr: \"1:2\", http_addr: \"1:2\"}\nleader: us-east\n\ + shards:\n - {id: 0, leader: eu-west, replicas: [{node: us-east}]}\n", + ) + .expect_err("leader not a replica"); + assert!( + bad_leader + .to_string() + .contains("must be one of its replicas"), + "got {bad_leader}" + ); + } + + #[test] + fn shards_reject_same_node_port_collision() { + // us-east hosts shards 0 and 1 but both resolve to the same explicit addr. + let t = load_topology_from_str( + "regions:\n - {name: us-east, grpc_addr: \"127.0.0.1:9601\", http_addr: \"1:1\"}\n \ + - {name: eu-west, grpc_addr: \"127.0.0.1:9611\", http_addr: \"1:2\"}\nleader: us-east\n\ + shards:\n - {id: 0, replicas: [{node: us-east, grpc_addr: \"127.0.0.1:5000\"}, \ + {node: eu-west}]}\n - {id: 1, leader: eu-west, replicas: [{node: us-east, \ + grpc_addr: \"127.0.0.1:5000\"}, {node: eu-west}]}\n", + ) + .expect("structural validation passes"); + let err = t + .resolve_shard_groups() + .expect_err("port collision rejected"); + assert!(err.to_string().contains("same grpc_addr"), "got {err}"); + } } diff --git a/tidal-server/src/main.rs b/tidal-server/src/main.rs index 27c301a..80d64f1 100644 --- a/tidal-server/src/main.rs +++ b/tidal-server/src/main.rs @@ -3,7 +3,7 @@ use std::{net::SocketAddr, path::PathBuf, sync::Arc}; use clap::{Args, Parser, Subcommand}; use tidal_server::{ cluster::{ - ClusterMode, ClusterState, RegionClusterState, build_cluster_router, build_region_router, + ClusterMode, ClusterNode, ClusterState, build_cluster_router, build_region_router, ensure_experimental_enabled, join_boot, load_topology, }, config::{CONFIG_DIR_SCHEMA_FILE, CONFIG_DIR_TOPOLOGY_FILE, load_schema, resolve_config_path}, @@ -270,16 +270,16 @@ async fn run_region_cluster(args: ClusterArgs, region: String) -> Result<()> { let data_dir = args.data_dir.clone(); let api_key = read_api_key(); - // `RegionClusterState::new` builds the GrpcTransport via `GrpcTransport::new`, + // `ShardReplica::new` builds the GrpcTransport via `GrpcTransport::new`, // which blocks on its own tokio runtime — must run off this reactor. The // m11p5 boot-time snapshot install (§2.2–§2.7) ALSO blocks (a leader- // discovery loop + a synchronous snapshot fetch), so both run on the same // dedicated thread, install FIRST: swap-recovery and the snapshot swap must - // complete BEFORE `RegionClusterState::new` opens the engine (a marker boot + // complete BEFORE `ShardReplica::new` opens the engine (a marker boot // opens the NEW data dir; a fallback opens the existing one degraded). let state = std::thread::Builder::new() .name("region-build".into()) - .spawn(move || -> Result { + .spawn(move || -> Result { // §2.2: the install pre-open step runs ONLY when a reseed marker is // present (plain topology boots never contact a peer before serving). // `run_boot_install` returns immediately (NotNeeded) when no marker @@ -289,7 +289,10 @@ async fn run_region_cluster(args: ClusterArgs, region: String) -> Result<()> { tidal_server::cluster::reseed::run_boot_install(&topology, ®ion, dir)?; tracing::info!(?outcome, region = %region, "boot-time reseed install evaluated"); } - RegionClusterState::new( + // m11p6: `ClusterNode::new` resolves the shard-group assignment and + // opens one `ShardReplica` per group this node hosts (S=1 is the + // legacy group spanning every region, the node data dir verbatim). + ClusterNode::new( &topology, ®ion, schema, @@ -307,7 +310,7 @@ async fn run_region_cluster(args: ClusterArgs, region: String) -> Result<()> { /// Seed-join boot (m11p5 §3.4–§3.6): this node is NOT declared in the local /// topology file. It learns its roster, assigned id, and term from a `--seed`, -/// installs a snapshot when needed, then constructs `RegionClusterState` from +/// installs a snapshot when needed, then constructs `ShardReplica` from /// the LEARNED roster (not the topology's `regions:` list). /// /// §3.5 knob-source rule: the local topology/config file is STILL required for @@ -350,7 +353,7 @@ async fn run_seed_join_cluster(args: ClusterArgs, region: String) -> Result<()> // We do NOT call `validate_multiproc` on this file's `regions:` (it would // demand the joiner be declared there); the runtime roster is the join // response, and the synthesized topology IS validated by `validate_multiproc` - // inside `RegionClusterState::new`. + // inside `ShardReplica::new`. let knobs = load_topology(topology_path.as_deref())?; let advertise_grpc = args.advertise_grpc.clone().ok_or_else(|| { @@ -396,12 +399,12 @@ async fn run_seed_join_cluster(args: ClusterArgs, region: String) -> Result<()> let metrics = args.metrics.clone(); let seeds = args.seed.clone(); // The §2.7 join loop + snapshot install BLOCK (leader discovery, a - // synchronous JoinCluster, a snapshot fetch), and `RegionClusterState::new` + // synchronous JoinCluster, a snapshot fetch), and `ShardReplica::new` // builds the GrpcTransport on its own runtime — both must run off this // reactor, on the same dedicated thread, install FIRST. let state = std::thread::Builder::new() .name("region-seed-join-build".into()) - .spawn(move || -> Result { + .spawn(move || -> Result { let join_boot::SeedJoinBoot { topology: synth, region: synth_region, @@ -415,7 +418,9 @@ async fn run_seed_join_cluster(args: ClusterArgs, region: String) -> Result<()> data_dir: &data_dir, api_key: api_key_for_join, })?; - RegionClusterState::new( + // m11p6: a seed-join synthesizes the legacy single group (the joiner + // learns its shard assignment from the membership log, not the file). + ClusterNode::new( &synth, &synth_region, schema, @@ -475,10 +480,11 @@ impl ServeState for ClusterState { } } -impl ServeState for RegionClusterState { +impl ServeState for ClusterNode { const WHAT: &'static str = "region"; fn started(self: &Arc) { - self.start_election_driver(); + // m11p6: start every hosted shard group's election driver. + self.start_all_elections(); } fn set_shutting_down(&self) { Self::set_shutting_down(self); // the inherent method, as above diff --git a/tidal-server/tests/cluster_grpc.rs b/tidal-server/tests/cluster_grpc.rs index 973550d..e3188e2 100644 --- a/tidal-server/tests/cluster_grpc.rs +++ b/tidal-server/tests/cluster_grpc.rs @@ -30,6 +30,7 @@ fn three_region_topology() -> TopologySpec { http_addr: None, grpc_tls: None, metrics_addr: None, + zone: None, }) .into_iter() .collect(), @@ -39,6 +40,7 @@ fn three_region_topology() -> TopologySpec { replication: ReplicationSpec::default(), wal: WalSpec::default(), election: ElectionSpec::default(), + shards: None, } } diff --git a/tidal-server/tests/cluster_region.rs b/tidal-server/tests/cluster_region.rs index 16565f3..b968945 100644 --- a/tidal-server/tests/cluster_region.rs +++ b/tidal-server/tests/cluster_region.rs @@ -1,6 +1,6 @@ //! m8p10 in-process multi-process-cluster tests. //! -//! Each test builds TWO `RegionClusterState`s in ONE test process — distinct +//! Each test builds TWO `ShardReplica`s in ONE test process — distinct //! topologies pointing at each other's REAL loopback gRPC addresses — and drives //! them over real HTTP (axum on loopback) + real `GrpcTransport` replication. //! Unlike `cluster_e2e.rs` (tier-3, spawns OS processes), these run in the @@ -24,10 +24,11 @@ use std::{ }; use tidal_server::cluster::{ - ElectionSpec, RegionClusterState, RegionSpec, ReplicationSpec, TimeoutsSpec, TopologySpec, - WalSpec, build_region_router, + ClusterNode, ElectionSpec, RegionSpec, ReplicationSpec, ShardReplicaSpec, ShardSpec, + TimeoutsSpec, TopologySpec, WalSpec, build_region_router, }; -use tidaldb::schema::{DecaySpec, EntityKind, Schema, SchemaBuilder, Window}; +use tidaldb::replication::shard::ShardRouter; +use tidaldb::schema::{DecaySpec, EntityId, EntityKind, Schema, SchemaBuilder, Window}; use tidaldb::wal::format::{MemberEntry, MemberRole, MembershipRecord}; /// A single-`view`-signal schema with a `hide` hard-negative signal (so the @@ -95,6 +96,7 @@ impl Pair { http_addr: Some(self.leader_http.to_string()), grpc_tls: None, metrics_addr: None, + zone: None, }, RegionSpec { name: self.follower_name.clone(), @@ -103,6 +105,7 @@ impl Pair { http_addr: Some(self.follower_http.to_string()), grpc_tls: None, metrics_addr: None, + zone: None, }, ], leader: self.leader_name.clone(), @@ -111,6 +114,7 @@ impl Pair { replication: ReplicationSpec::default(), wal: WalSpec::default(), election: ElectionSpec::default(), + shards: None, } } } @@ -131,15 +135,11 @@ fn region_dir() -> tempfile::TempDir { /// Build one region node off the reactor (GrpcTransport::new blocks on its own /// runtime, so it must run on a plain thread). `dir` is the node's data dir; /// see [`region_dir`] for the declaration-order contract. -fn build_region( - topology: TopologySpec, - region: &str, - dir: &tempfile::TempDir, -) -> RegionClusterState { +fn build_region(topology: TopologySpec, region: &str, dir: &tempfile::TempDir) -> ClusterNode { let region = region.to_string(); let data_dir = dir.path().to_path_buf(); std::thread::spawn(move || { - RegionClusterState::new( + ClusterNode::new( &topology, ®ion, region_schema(), @@ -310,6 +310,330 @@ fn region_node_replicates_over_grpc() { rt.shutdown_timeout(Duration::from_secs(2)); } +/// m11p6 sharding × replication (in-process): 2 nodes × 2 shards × RF=2, with +/// shard 0 led by node A and shard 1 led by node B (balanced leaders). Every +/// node hosts a replica of BOTH groups. A write routes by entity hash to its +/// shard's leader (A applies its shard-0 writes locally and FORWARDS its +/// shard-1 writes to B, and vice versa — the unified path, no /sharded needed); +/// each shard replicates to its follower on the other node; and a /feed read on +/// either node scatters over both local groups and merges, returning items from +/// BOTH shards. This is the exit-gate shape at 2×2 — proof the +/// replicated-XOR-sharded split is gone. +#[test] +fn region_sharded_writes_route_per_shard_and_reads_scatter() { + // 2 nodes, 4 gRPC ports (one per (node, shard)), 2 HTTP ports. + let names = ["node-a".to_string(), "node-b".to_string()]; + let http = [free_addr(), free_addr()]; + // grpc[node][shard] + let grpc = [[free_addr(), free_addr()], [free_addr(), free_addr()]]; + + let topology = || -> TopologySpec { + let regions = (0..2) + .map(|i| RegionSpec { + name: names[i].clone(), + // The node's base grpc_addr is shard 0's; shard 1 sets its own. + grpc_addr: Some(grpc[i][0].to_string()), + grpc_bind: None, + http_addr: Some(http[i].to_string()), + grpc_tls: None, + metrics_addr: None, + zone: None, + }) + .collect(); + let shard = |id: u16, leader: usize| ShardSpec { + id, + leader: Some(names[leader].clone()), + replicas: (0..2) + .map(|n| ShardReplicaSpec { + node: names[n].clone(), + grpc_addr: Some(grpc[n][id as usize].to_string()), + grpc_bind: None, + }) + .collect(), + }; + TopologySpec { + regions, + leader: names[0].clone(), // legacy field unused once shards: is set + write_workers: None, + timeouts: TimeoutsSpec::default(), + replication: ReplicationSpec::default(), + wal: WalSpec::default(), + election: ElectionSpec::default(), + shards: Some(vec![shard(0, 0), shard(1, 1)]), + } + }; + + let dirs: Vec = (0..2).map(|_| region_dir()).collect(); + let node_a = Arc::new(build_region(topology(), &names[0], &dirs[0])); + let node_b = Arc::new(build_region(topology(), &names[1], &dirs[1])); + + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build() + .unwrap(); + serve(&rt, build_region_router(Arc::clone(&node_a), None), http[0]); + serve(&rt, build_region_router(Arc::clone(&node_b), None), http[1]); + + let client = reqwest::blocking::Client::new(); + let base_a = format!("http://{}", http[0]); + let base_b = format!("http://{}", http[1]); + + // Partition entity ids by their shard so we can assert BOTH shards receive + // writes (the same FNV-1a router the gateway uses). + let router = ShardRouter::hash(2).unwrap(); + let mut shard0_ids = Vec::new(); + let mut shard1_ids = Vec::new(); + for i in 1..=24u64 { + match router.route(EntityId::new(i)).0 { + 0 => shard0_ids.push(i), + _ => shard1_ids.push(i), + } + } + assert!( + !shard0_ids.is_empty() && !shard1_ids.is_empty(), + "the hash must spread 24 ids across both shards (got {} / {})", + shard0_ids.len(), + shard1_ids.len() + ); + + // Write every item + a view signal through node A ONLY. A leads shard 0 + // (applies locally) and follows shard 1 (forwards to B) — one client, one + // endpoint, both shards. + for i in 1..=24u64 { + let resp = client + .post(format!("{base_a}/items")) + .json(&serde_json::json!({ + "entity_id": i, "metadata": { "title": format!("item {i}") } + })) + .send() + .unwrap(); + assert!( + resp.status().is_success(), + "POST /items id={i}: {}", + resp.status() + ); + let resp = client + .post(format!("{base_a}/signals")) + .json(&serde_json::json!({ "entity_id": i, "signal": "view", "weight": 1.0 })) + .send() + .unwrap(); + assert!( + resp.status().is_success(), + "POST /signals id={i}: {}", + resp.status() + ); + } + + // Both nodes' feeds must, after convergence, return EVERY item — proof that + // (1) A forwarded shard-1 writes to B, (2) each shard replicated to its + // follower, and (3) the read scatters over both local groups and merges. + let feed_ids = |base: &str| -> std::collections::HashSet { + let feed: serde_json::Value = client + .get(format!("{base}/feed?profile=trending&limit=50")) + .send() + .unwrap() + .json() + .unwrap(); + feed["items"] + .as_array() + .unwrap() + .iter() + .filter_map(|it| it["entity_id"].as_u64()) + .collect() + }; + + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let a = feed_ids(&base_a); + let b = feed_ids(&base_b); + let all: std::collections::HashSet = (1..=24u64).collect(); + if a == all && b == all { + // Sanity: both shards are actually represented (not one big shard). + assert!( + shard0_ids.iter().all(|i| a.contains(i)) + && shard1_ids.iter().all(|i| a.contains(i)), + "node A feed must merge BOTH shards" + ); + break; + } + assert!( + Instant::now() <= deadline, + "feeds did not converge to all 24 items across both shards within 10s \ + (A={}, B={})", + a.len(), + b.len() + ); + std::thread::sleep(Duration::from_millis(50)); + } + + rt.shutdown_timeout(Duration::from_secs(2)); +} + +/// m11p6 PARTIAL placement (the `EntityRoute::Remote` path + group-local reads): +/// 3 nodes, 2 shards, RF=2, placed so shard 0 = {node-a (leader), node-b} and +/// shard 1 = {node-b, node-c (leader)}. So node-a hosts ONLY shard 0, node-c +/// ONLY shard 1, node-b BOTH. Writing every entity through node-a forces its +/// shard-1 writes down the not-hosted (`Remote`) path — forwarded leader-first to +/// a shard-1 replica node, which applies. node-b (full placement) then merges the +/// whole corpus; node-a / node-c (partial) feed only their own group — the +/// documented L2 read scope, asserted so it is intentional, not a silent gap. +#[test] +fn region_sharded_subset_placement_forwards_and_reads_are_group_scoped() { + let names = [ + "node-a".to_string(), + "node-b".to_string(), + "node-c".to_string(), + ]; + let http = [free_addr(), free_addr(), free_addr()]; + let (a_s0, b_s0, b_s1, c_s1) = (free_addr(), free_addr(), free_addr(), free_addr()); + + let topology = || -> TopologySpec { + // Each node's RegionSpec base grpc_addr (a hosted addr); replicas are + // explicit per (node, shard), so the base is only the declared advertise. + let bases = [a_s0, b_s0, c_s1]; + let regions = (0..3) + .map(|i| RegionSpec { + name: names[i].clone(), + grpc_addr: Some(bases[i].to_string()), + grpc_bind: None, + http_addr: Some(http[i].to_string()), + grpc_tls: None, + metrics_addr: None, + zone: None, + }) + .collect(); + let replica = |node: usize, addr: SocketAddr| ShardReplicaSpec { + node: names[node].clone(), + grpc_addr: Some(addr.to_string()), + grpc_bind: None, + }; + TopologySpec { + regions, + leader: names[0].clone(), + write_workers: None, + timeouts: TimeoutsSpec::default(), + replication: ReplicationSpec::default(), + wal: WalSpec::default(), + election: ElectionSpec::default(), + shards: Some(vec![ + ShardSpec { + id: 0, + leader: Some(names[0].clone()), + replicas: vec![replica(0, a_s0), replica(1, b_s0)], + }, + ShardSpec { + id: 1, + leader: Some(names[1].clone()), + replicas: vec![replica(1, b_s1), replica(2, c_s1)], + }, + ]), + } + }; + + let dirs: Vec = (0..3).map(|_| region_dir()).collect(); + let node_a = Arc::new(build_region(topology(), &names[0], &dirs[0])); + let node_b = Arc::new(build_region(topology(), &names[1], &dirs[1])); + let node_c = Arc::new(build_region(topology(), &names[2], &dirs[2])); + + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build() + .unwrap(); + serve(&rt, build_region_router(Arc::clone(&node_a), None), http[0]); + serve(&rt, build_region_router(Arc::clone(&node_b), None), http[1]); + serve(&rt, build_region_router(Arc::clone(&node_c), None), http[2]); + + let client = reqwest::blocking::Client::new(); + let base_a = format!("http://{}", http[0]); + let base_b = format!("http://{}", http[1]); + let base_c = format!("http://{}", http[2]); + + let router = ShardRouter::hash(2).unwrap(); + let mut shard0_ids = std::collections::HashSet::new(); + let mut shard1_ids = std::collections::HashSet::new(); + for i in 1..=24u64 { + if router.route(EntityId::new(i)).0 == 0 { + shard0_ids.insert(i); + } else { + shard1_ids.insert(i); + } + } + assert!( + !shard0_ids.is_empty() && !shard1_ids.is_empty(), + "the hash must spread 24 ids across both shards" + ); + + // Write every item + a view signal through node-A ONLY. Shard-0 ids apply + // locally (A leads shard 0); shard-1 ids take EntityRoute::Remote (A hosts no + // shard-1 replica) and forward to a shard-1 replica node — proof the + // not-hosted gateway path works for both items and signals. + for i in 1..=24u64 { + let item = client + .post(format!("{base_a}/items")) + .json(&serde_json::json!({ "entity_id": i, "metadata": { "title": format!("item {i}") } })) + .send() + .unwrap(); + assert!( + item.status().is_success(), + "POST /items id={i} via node-a: {}", + item.status() + ); + let sig = client + .post(format!("{base_a}/signals")) + .json(&serde_json::json!({ "entity_id": i, "signal": "view", "weight": 1.0 })) + .send() + .unwrap(); + assert!( + sig.status().is_success(), + "POST /signals id={i} via node-a: {}", + sig.status() + ); + } + + let feed_ids = |base: &str| -> std::collections::HashSet { + let feed: serde_json::Value = client + .get(format!("{base}/feed?profile=trending&limit=50")) + .send() + .unwrap() + .json() + .unwrap(); + feed["items"] + .as_array() + .unwrap() + .iter() + .filter_map(|it| it["entity_id"].as_u64()) + .collect() + }; + let all: std::collections::HashSet = (1..=24u64).collect(); + + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let a = feed_ids(&base_a); + let b = feed_ids(&base_b); + let c = feed_ids(&base_c); + // node-b hosts BOTH groups ⇒ the whole corpus once replication converges; + // node-a / node-c host ONE group ⇒ exactly that group's ids (group-local + // read by design). + if b == all && a == shard0_ids && c == shard1_ids { + break; + } + assert!( + Instant::now() <= deadline, + "subset-placement feeds did not converge (a={}/{}, b={}/24, c={}/{})", + a.len(), + shard0_ids.len(), + b.len(), + c.len(), + shard1_ids.len() + ); + std::thread::sleep(Duration::from_millis(50)); + } + + rt.shutdown_timeout(Duration::from_secs(2)); +} + /// A write to the FOLLOWER (a non-leader node) is FORWARDED to the leader; with /// the leader process not running, the forward fails and degrades to a 503 whose /// body names the (unreachable) leader — the task-03 leader-unreachable contract. @@ -673,6 +997,7 @@ impl Trio { http_addr: Some(self.http[i].to_string()), grpc_tls: None, metrics_addr: None, + zone: None, }) .collect(), leader: self.names[0].clone(), @@ -681,6 +1006,7 @@ impl Trio { replication: ReplicationSpec::default(), wal: WalSpec::default(), election: ElectionSpec::default(), + shards: None, } } } @@ -724,7 +1050,7 @@ fn region_node_heal_backfills_missed_items() { // DOWN during the leader's item/signal broadcast. The leader's best-effort // HTTP broadcast to it will fail (connection refused) and land in `failed`, // exactly as it would for a crashed/restarting region. The follower's gRPC - // receiver IS running (started in RegionClusterState::new), so once the + // receiver IS running (started in ShardReplica::new), so once the // leader heals it, the relay re-ships signals — but the item broadcast that // failed during downtime is what heal must backfill. @@ -850,7 +1176,7 @@ fn region_node_lag_honest_across_promote() { // Three nodes: us-east (leader, shard 0), eu-west (shard 1), ap-south // (shard 2). Dirs FIRST (see `region_dir` for the drop-order contract). let dirs: Vec = (0..3).map(|_| region_dir()).collect(); - let nodes: Vec> = (0..3) + let nodes: Vec> = (0..3) .map(|i| Arc::new(build_region(trio.topology(), &trio.names[i], &dirs[i]))) .collect(); // Build the runtime AFTER the nodes (and their data-dir guards): locals diff --git a/tidal-server/tests/cluster_routes.rs b/tidal-server/tests/cluster_routes.rs index d8e2949..079af47 100644 --- a/tidal-server/tests/cluster_routes.rs +++ b/tidal-server/tests/cluster_routes.rs @@ -1,6 +1,6 @@ //! m8p10 task 03: cross-process route tests. //! -//! Builds 3-node in-process clusters — distinct `RegionClusterState`s pointing at +//! Builds 3-node in-process clusters — distinct `ShardReplica`s pointing at //! each other's REAL loopback gRPC + HTTP addresses — and drives them over real //! HTTP (axum on loopback) + real `GrpcTransport` replication + real reqwest //! forwarding. No OS processes (tier-3 OS-process coverage is tasks 04/05); these @@ -26,8 +26,8 @@ use std::{ }; use tidal_server::cluster::{ - ElectionSpec, RegionClusterState, RegionSpec, ReplicationSpec, TimeoutsSpec, TopologySpec, - WalSpec, build_region_router, + ClusterNode, ElectionSpec, RegionSpec, ReplicationSpec, TimeoutsSpec, TopologySpec, WalSpec, + build_region_router, }; use tidaldb::schema::{DecaySpec, EntityKind, Schema, SchemaBuilder, Window}; @@ -87,6 +87,7 @@ impl Cluster3 { http_addr: Some(self.http[i].to_string()), grpc_tls: None, metrics_addr: None, + zone: None, }) .collect(), leader: self.names[0].clone(), @@ -95,6 +96,7 @@ impl Cluster3 { replication: ReplicationSpec::default(), wal: WalSpec::default(), election: ElectionSpec::default(), + shards: None, } } @@ -113,15 +115,11 @@ fn region_dir() -> tempfile::TempDir { /// Build one region node off the reactor (GrpcTransport::new blocks on its own /// runtime, so it must run on a plain thread). -fn build_region( - topology: TopologySpec, - region: &str, - dir: &tempfile::TempDir, -) -> RegionClusterState { +fn build_region(topology: TopologySpec, region: &str, dir: &tempfile::TempDir) -> ClusterNode { let region = region.to_string(); let data_dir = dir.path().to_path_buf(); std::thread::spawn(move || { - RegionClusterState::new( + ClusterNode::new( &topology, ®ion, region_schema(), diff --git a/tidal-server/tests/support/multiproc.rs b/tidal-server/tests/support/multiproc.rs index 11bbc5e..abaa035 100644 --- a/tidal-server/tests/support/multiproc.rs +++ b/tidal-server/tests/support/multiproc.rs @@ -7,7 +7,7 @@ //! write to `nodes[0]` replicated inside node 0's own fabric and never crossed //! into `nodes[1]`'s process. That harness's own module docs flag this as the //! gap. `MultiProcCluster` closes it: each process runs `tidal-server cluster -//! --region ` and owns exactly ONE region (a `RegionClusterState`), peering +//! --region ` and owns exactly ONE region (a `ShardReplica`), peering //! with siblings over real gRPC and forwarding over real HTTP. Convergence is //! therefore verified against EVERY follower process's own `/cluster/status/local`, //! not a single node's internal view.