From 31ee612f27506fdc79fe48c3151000d32900b927 Mon Sep 17 00:00:00 2001 From: jx12n Date: Sun, 14 Jun 2026 15:17:35 -0600 Subject: [PATCH] =?UTF-8?q?feat(m12p4):=20sharded=20ingestion=20=E2=80=94?= =?UTF-8?q?=20scatter-gather=20pool=20+=20cross-shard=20unified=20reads=20?= =?UTF-8?q?(L4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scale write throughput across data-shard groups while keeping a single unified read surface: - scatter_gather.rs: pooled fan-out across shard groups (replaces per-request client construction); cross-shard query results merged on one node - cluster/node.rs: cross-shard read routing — a read on any node gathers from every shard group's leader and unions results - cluster/forward.rs: fix h2 204 forward-relay bug (relay_forwarded skips body for 1xx/204/304 — synthesized JSON body on a 204 triggered HTTP/2 RST_STREAM on the real mTLS plane) - dto.rs: cross-shard query/result DTOs - k8s/cluster/: enable 3-group `shards:` topology (statefulset, service-peers, topology-configmap) - k8s/cluster-local-kind/: local-kind overlay to run the T5 gate without Ref-A - tidal-stress/k8s/stress-job-t5.yaml: 2-generator sharded throughput job - tests: cluster_cross_shard_reads.rs + multiproc support; ran real on kind - docs/profiling/m12p4-t5-sharded-throughput.md: T5 throughput findings --- docs/profiling/m12p4-t5-sharded-throughput.md | 112 +++ k8s/cluster-local-kind/kustomization.yaml | 43 + k8s/cluster/service-peers.yaml | 12 + k8s/cluster/statefulset.yaml | 13 + k8s/cluster/topology-configmap.yaml | 34 +- tidal-server/src/cluster/forward.rs | 91 +- tidal-server/src/cluster/node.rs | 875 +++++++++++++++++- tidal-server/src/cluster/routes.rs | 2 + tidal-server/src/dto.rs | 45 +- tidal-server/src/router.rs | 3 + tidal-server/src/scatter_gather.rs | 457 +++++---- .../tests/cluster_cross_shard_reads.rs | 264 ++++++ tidal-server/tests/cluster_region.rs | 18 +- tidal-server/tests/support/multiproc.rs | 235 +++++ tidal-stress/k8s/stress-job-t5.yaml | 145 +++ 15 files changed, 2092 insertions(+), 257 deletions(-) create mode 100644 docs/profiling/m12p4-t5-sharded-throughput.md create mode 100644 k8s/cluster-local-kind/kustomization.yaml create mode 100644 tidal-server/tests/cluster_cross_shard_reads.rs create mode 100644 tidal-stress/k8s/stress-job-t5.yaml diff --git a/docs/profiling/m12p4-t5-sharded-throughput.md b/docs/profiling/m12p4-t5-sharded-throughput.md new file mode 100644 index 0000000..72cfd3e --- /dev/null +++ b/docs/profiling/m12p4-t5-sharded-throughput.md @@ -0,0 +1,112 @@ +# m12p4 / T5 — Sharded ingestion at scale (real-cluster run, 2026-06-14) + +**What ran.** A real 3-shard-group × RF=3 tidalDB cluster on a local `kind` +cluster (single node, 24 vCPU / 50 GB), full mTLS (cert-manager-issued certs), +the m12p4 server image, driven by **two** in-cluster `tidal-stress` generator +pods (the "second stress generator" m12p4 calls for). Not a simulation — real +binaries, real WAL + quorum, real gRPC replication, real inter-node mTLS HTTP. + +This is the honest record of what the run **proved**, what it **found**, and why +the headline gate (`≥2.5× scaling AND ≥5000 quorum writes/s`) remains +**Ref-A/k3s-pending** — now with data and a sharper reason, not just a hardware +caveat. + +## What the run proved (real, reproducible) + +- **The 3-group topology deploys and serves on real k8s.** `shards:` enabled + (`k8s/cluster/topology-configmap.yaml`), every pod binds a derived gRPC port + per group (9601/9602/9603), `/cluster/status` shows **balanced leaders** — + shard 0→tidaldb-0, 1→tidaldb-1, 2→tidaldb-2, all term 0. Fresh parallel boot + converges in **~30–48 s**. (Reproduce: `k8s/cluster-local-kind/` overlay.) +- **Cross-group quorum writes work end-to-end** — a write for an entity owned by + another group's leader is hash-routed + forwarded over the inter-node mTLS + plane and quorum-acked, `x-tidal-seq` verdict header riding back. +- **Graceful behavior under overload** — at offered loads past the engine knee, + the cluster sheds with 503 backpressure / request timeouts; **0 pod crashes, + 0 acked loss** (a timed-out write is never acked, so it is not acked loss). + +## The blocker this run uncovered and fixed (HTTP/2 forward relay) + +The FIRST cross-group write returned curl **exit 92 (HTTP/2 stream error)** / +HTTP 000; the identical write with `--http1.1` returned 204. Root cause: the +inter-node/client plane serves h2 (ALPN `h2`+`http/1.1`), `/signals`//`/embeddings` +return **204 No Content**, and the forward relay (`forward::relay_forwarded`) +attached a **synthesized JSON body to a 204** (`forward_json_with_headers` fills +an empty peer body). HTTP/2 RST_STREAMs a 204 that carries a DATA frame; HTTP/1.1 +tolerated it, which masked the bug until the sharded forward ran over the **real +h2 plane** (never exercised before — was itself Ref-A/k3s-pending). Fixed in +`relay_forwarded` (skip the body for 1xx/204/304 via `status_forbids_body`) + +3 unit tests; the one direct relay site now routes through the helper too. After +the fix, all cross-group writes return 204 over h2. + +This is a genuine correctness fix that only a real-h2-plane run surfaces — +in-process / plaintext multiproc e2e (reqwest→plaintext = h1.1) cannot catch it. + +## Throughput numbers (measured) + +Open-loop, coordinated-omission-corrected, `--ack quorum`, `--mix writes`, +`--embedding-dim 1536` (schema width). "Sustained" = highest ramp stage at +≤1% error; aggregate = sum of the two generator pods. + +| Config (per-pod CPU) | Sustained quorum writes/s (≤1% err) | Behavior past the knee | +|----------------------|-------------------------------------|------------------------| +| **1 group, RF=3** (2) | **~5,500** (2×~2,775) | latency rises to p99 ~280 ms, 503s — single leader saturates | +| **3 groups, RF=3** (2) | **~4,000–5,000** | collapses earlier (stage 2: 78% shed, 40 s timeouts) | + +Both generators are **identical** and cap at **~2,300–3,900 rps each** — the knee +is the generator's in-flight×latency limit (`client-shed`), not engine errors, at +the lower stages. Two generators offer **at most ~7,800 rps aggregate**. + +## The finding: full-placement sharding does not scale writes at fixed per-pod CPU + +The roadmap premise — "3 groups → ~3× writes" — assumes the bottleneck is the +**single-leader funnel** (one leader serializing WAL append + quorum), which +sharding parallelizes across S leaders. That holds **only when per-pod CPU is not +the binding constraint.** + +With **full placement** (every pod replicates EVERY group — the shape chosen for +failover simplicity), each pod runs **all S groups**: leader for 1 + follower for +S−1. So a pod's total replication work scales with **S**, while its CPU is fixed. +At 2 vCPU/pod the 3-group pods are doing 3 groups' WAL/fsync/ship/apply in the +same budget as the 1-group pods do 1 group's — so the 3-group cluster saturates +at a **comparable or lower** aggregate than single-group, not 3× higher. Measured: +3-group ≈ single-group (~4–5k), then 3-group collapses *first* under overload. + +**Implication.** Full-placement sharding buys **failover** (any pod loss keeps +every group's quorum) and **leader-funnel relief**, but it does **not** raise +aggregate write throughput at a fixed cluster CPU budget — the per-pod replication +overhead grows with S. Real write *scaling* needs **partitioned placement** (each +pod hosts a SUBSET of groups, so adding pods adds both groups and CPU) and/or +per-pod CPU headroom so the leader funnel — not per-pod CPU — is the bottleneck. + +## Why the gate stays Ref-A/k3s-pending (now with data) + +To demonstrate `≥2.5× scaling AND ≥5,000/s` you need ALL of: +1. **Per-pod CPU headroom** so the single-leader funnel (not per-pod CPU) binds — + i.e. enough cores that a pod running S groups isn't CPU-saturated. +2. **Generators that can offer ≥14k rps** — 2.5× over the ~5.5k single-group + ceiling. Two TLS generators cap at ~7.8k aggregate; that's ~1.4× at most. +3. **Or partitioned placement** across more nodes (the architecturally correct + write-scaling shape), which a single kind node cannot host meaningfully. + +A single shared 24-vCPU laptop cannot satisfy (1)+(2) simultaneously — the engine +and the TLS generators contend for the same cores. This is precisely the +hardware dependency the roadmap names; the run confirms it with numbers and adds +the full-placement insight above. + +## Artifacts (committed) + +- `k8s/cluster/topology-configmap.yaml` — 3-group `shards:` block enabled. +- `k8s/cluster/statefulset.yaml`, `service-peers.yaml` — derived per-shard ports. +- `k8s/cluster-local-kind/` — local-kind overlay (image + storageclass override). +- `tidal-stress/k8s/stress-job-t5.yaml` — the 2-generator T5 Job (production shape). +- `tidal-server/src/cluster/forward.rs` — the h2 204-relay fix + tests. + +## Recommended follow-up (for the visionary) + +The T5 gate as written assumes full-placement scales writes; the measured reality +says it scales failover, not write throughput, at fixed CPU. Re-scope T5 to either +(a) **partitioned placement** on multi-node k3s/Ref-A (the real write-scaling +proof), or (b) restate the gate as "leader-funnel relief at CPU headroom" and size +the pods accordingly. Either way the generator fleet must be ≥4 pods to offer the +load a scaled cluster can absorb. diff --git a/k8s/cluster-local-kind/kustomization.yaml b/k8s/cluster-local-kind/kustomization.yaml new file mode 100644 index 0000000..e89b3b3 --- /dev/null +++ b/k8s/cluster-local-kind/kustomization.yaml @@ -0,0 +1,43 @@ +# Local-kind overlay for the cluster set — used to run the m12p4 T5 sharded +# throughput gate on a local `kind` cluster (no private registry, no Ref-A). +# +# It reuses the canonical k8s/cluster/ base verbatim (including the 3-group +# `shards:` topology and cert-manager certs) and only swaps the registry-pinned +# server image for a locally-built tag that `kind load docker-image` puts on the +# node. cert-manager must be present in the kind cluster (the shared `canopy` +# cluster already has it); the credentials Secret is still created out-of-band. +# +# Run: +# docker build -f docker/deploy/Dockerfile -t tidaldb-server:m12p4-local . +# kind load docker-image tidaldb-server:m12p4-local --name canopy +# kubectl create namespace tidaldb-cluster +# kubectl -n tidaldb-cluster create secret generic tidaldb-credentials \ +# --from-literal=TIDAL_API_KEY="$(openssl rand -hex 32)" \ +# --from-literal=TIDAL_CLUSTER_KEY="$(openssl rand -hex 32)" +# kubectl apply -k k8s/cluster-local-kind/ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization + +namespace: tidaldb-cluster + +resources: + - ../cluster + +# Swap the registry-pinned (digest) server image for the locally-loaded tag. +# Kustomize matches on the name part before the `@sha256:` digest. +images: + - name: registry.threesix.ai/tidal/server + newName: tidaldb-server + newTag: m12p4-local + +# kind ships a default StorageClass named `standard` (rancher.io/local-path), +# not `local-path` as the canonical StatefulSet names. Repoint the PVC template +# at the kind default so the claims bind instead of hanging Pending. +patches: + - target: + kind: StatefulSet + name: tidaldb + patch: |- + - op: replace + path: /spec/volumeClaimTemplates/0/spec/storageClassName + value: standard diff --git a/k8s/cluster/service-peers.yaml b/k8s/cluster/service-peers.yaml index 0fd36d7..7c1ffe3 100644 --- a/k8s/cluster/service-peers.yaml +++ b/k8s/cluster/service-peers.yaml @@ -33,9 +33,21 @@ spec: - name: http port: 9500 targetPort: http + # One gRPC port per hosted shard group (m11p6/m12p4 3-group topology). A + # headless Service routes by pod DNS → pod IP, so peers dial the derived + # per-group ports (9601/9602/9603) on the pod directly regardless of this + # list; the entries are declared so the named ports stay discoverable and + # match statefulset.yaml. With the legacy single group only `grpc` (9601) + # is bound. - name: grpc port: 9601 targetPort: grpc + - name: grpc-1 + port: 9602 + targetPort: grpc-1 + - name: grpc-2 + port: 9603 + targetPort: grpc-2 - name: metrics port: 9091 targetPort: metrics diff --git a/k8s/cluster/statefulset.yaml b/k8s/cluster/statefulset.yaml index ffaa76d..d016b66 100644 --- a/k8s/cluster/statefulset.yaml +++ b/k8s/cluster/statefulset.yaml @@ -153,8 +153,21 @@ spec: ports: - name: http containerPort: 9500 + # One gRPC port per hosted shard group (m11p6/m12p4). With the + # 3-group `shards:` block enabled in the topology ConfigMap, every + # pod replicates all three groups and binds a derived port per + # group: shard 0 → 9601, shard 1 → 9602, shard 2 → 9603 + # (`node base port + shard id`; see topology-configmap.yaml). The + # headless Service reaches each by pod DNS, so these are declared + # for clarity/NetworkPolicy; the bind itself is driven by the + # topology. Collapse back to a single `grpc` port if `shards:` is + # removed (legacy single group). - name: grpc containerPort: 9601 + - name: grpc-1 + containerPort: 9602 + - name: grpc-2 + containerPort: 9603 - name: metrics containerPort: 9091 # Three probes map to the three health endpoints. The readinessProbe is diff --git a/k8s/cluster/topology-configmap.yaml b/k8s/cluster/topology-configmap.yaml index 7c9dfd8..fc9a67f 100644 --- a/k8s/cluster/topology-configmap.yaml +++ b/k8s/cluster/topology-configmap.yaml @@ -78,19 +78,29 @@ data: # Term-0 bootstrap leader only — post-election this field is dead config # (durable election_state governs; a restart always boots a follower). leader: tidaldb-0 - # ── Optional sharding × replication (m11p6) ────────────────────────────── - # Absent `shards:` ⇒ ONE group, RF = all pods (what this file ships). To - # scale WRITES horizontally, split the entity space into S groups, each a - # replication group at RF with its own elected leader, leaders balanced. A - # pod hosting several groups binds one gRPC port per group (omit - # replicas[].grpc_addr to derive `pod base port + shard id`); each group's - # data lives under /shard-{id:05}/. Full placement (every pod - # replicates every group) is the simplest shape. Operators rebalance with + # ── Sharding × replication (m11p6 / m12p4 T5) ──────────────────────────── + # Absent `shards:` ⇒ ONE group, RF = all pods. We ENABLE 3 groups × RF=3 so + # the write path scales horizontally: each group is an independent + # replication group with its OWN elected leader, and the three leaders are + # balanced one-per-pod (tidaldb-0 leads shard 0, tidaldb-1 shard 1, + # tidaldb-2 shard 2). The entity space is hash-partitioned across the three + # groups, so three leaders absorb writes in parallel instead of one — the + # ~S× write-throughput claim m12p4/T5 proves. + # + # Full placement: every pod replicates EVERY group, so any pod can serve a + # corpus-wide read locally (no cross-node read fan-out needed at this shape) + # and a single pod loss never loses a group's quorum (2 of 3 survive per + # group). A pod hosting all three groups binds one gRPC port per group: + # `replicas[].grpc_addr` is omitted, so each is DERIVED as + # `node base port + shard id` (tidaldb-N binds 9601 for shard 0, 9602 for + # shard 1, 9603 for shard 2 — matching the extra containerPorts in + # statefulset.yaml). Each group's data lives under + # /shard-{id:05}/ in the one PVC. Operators rebalance with # `POST /cluster/shards/{id}/transfer` and `/replicas` (see runbook §6a). - # shards: - # - { id: 0, leader: tidaldb-0, replicas: [ {node: tidaldb-0}, {node: tidaldb-1}, {node: tidaldb-2} ] } - # - { id: 1, leader: tidaldb-1, replicas: [ {node: tidaldb-0}, {node: tidaldb-1}, {node: tidaldb-2} ] } - # - { id: 2, leader: tidaldb-2, replicas: [ {node: tidaldb-0}, {node: tidaldb-1}, {node: tidaldb-2} ] } + shards: + - { id: 0, leader: tidaldb-0, replicas: [ {node: tidaldb-0}, {node: tidaldb-1}, {node: tidaldb-2} ] } + - { id: 1, leader: tidaldb-1, replicas: [ {node: tidaldb-0}, {node: tidaldb-1}, {node: tidaldb-2} ] } + - { id: 2, leader: tidaldb-2, replicas: [ {node: tidaldb-0}, {node: tidaldb-1}, {node: tidaldb-2} ] } replication: # ack=quorum: a write succeeds once a MAJORITY of the replica set durably # holds it (m11p3). Callers can still override per-request with x-tidal-ack. diff --git a/tidal-server/src/cluster/forward.rs b/tidal-server/src/cluster/forward.rs index aace55a..c0519f8 100644 --- a/tidal-server/src/cluster/forward.rs +++ b/tidal-server/src/cluster/forward.rs @@ -221,15 +221,40 @@ pub struct ForwardedResponse { pub body: serde_json::Value, } +/// Whether `status` is one for which HTTP forbids a response body — so relaying +/// a (possibly synthesized) JSON body would produce a frame an HTTP/2 client +/// rejects with `RST_STREAM`. Covers 1xx informational, 204 No Content, and 304 +/// Not Modified (the statuses RFC 9110 §6.4.1 / RFC 9113 give no message body). +#[must_use] +pub fn status_forbids_body(status: StatusCode) -> bool { + status.is_informational() + || status == StatusCode::NO_CONTENT + || status == StatusCode::NOT_MODIFIED +} + /// 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. +/// cannot drift on the relay contract. A bodyless status (204/304/1xx) is +/// relayed WITHOUT a body so the HTTP/2 client never sees an illegal DATA frame. #[must_use] pub fn relay_forwarded(resp: ForwardedResponse) -> Response { - let mut response = (resp.status, Json(resp.body)).into_response(); + // A bodyless status (204 No Content, 304 Not Modified, 1xx) MUST NOT carry a + // response body. HTTP/2 enforces this: a DATA frame on such a response is a + // protocol error and the client RST_STREAMs it (HTTP/1.1 silently tolerates + // it, which masked this until the sharded write FORWARD path ran over the h2 + // inter-node/client plane). A peer's `/signals`//`/embeddings` write returns + // 204, and [`forward_json_with_headers`] synthesizes a JSON object for an + // empty peer body — so relaying it verbatim would attach a body to a 204. + // Relay the verdict (status + the seq/dedup headers below) WITHOUT a body + // when the status forbids one. + let mut response = if status_forbids_body(resp.status) { + resp.status.into_response() + } else { + (resp.status, Json(resp.body)).into_response() + }; if let Some(seq) = resp.seq && let Ok(value) = HeaderValue::from_str(&seq) { @@ -511,4 +536,66 @@ mod tests { let out = futures_util::future::join_all(futs).await; assert_eq!(out, vec![1, 2, 3]); } + + #[test] + fn status_forbids_body_covers_bodyless_statuses() { + assert!(status_forbids_body(StatusCode::NO_CONTENT)); + assert!(status_forbids_body(StatusCode::NOT_MODIFIED)); + assert!(status_forbids_body(StatusCode::CONTINUE)); // 1xx informational + assert!(!status_forbids_body(StatusCode::OK)); + assert!(!status_forbids_body(StatusCode::CREATED)); + assert!(!status_forbids_body(StatusCode::INTERNAL_SERVER_ERROR)); + } + + #[tokio::test] + async fn relay_forwarded_omits_body_for_no_content() { + // A 204 from a forwarded signal/embedding write MUST relay without a + // body: an HTTP/2 client RST_STREAMs a 204 that carries a DATA frame + // (the bug the sharded write FORWARD path hit over the h2 inter-node / + // client plane — HTTP/1.1 silently tolerated it). The replicated-log + // seqno verdict header still rides back. + let resp = ForwardedResponse { + seq: Some("42".to_string()), + deduplicated: false, + status: StatusCode::NO_CONTENT, + // forward_json_with_headers synthesizes a JSON object for an empty + // peer body — exactly what must NOT become a 204 DATA frame. + body: serde_json::json!({ "forwarded": true }), + }; + let response = relay_forwarded(resp); + assert_eq!(response.status(), StatusCode::NO_CONTENT); + assert_eq!( + response + .headers() + .get(SEQ_HEADER) + .and_then(|v| v.to_str().ok()), + Some("42"), + "the replicated-log seqno must still ride back on a 204" + ); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("collect relayed body"); + assert!( + body.is_empty(), + "a 204 response must carry NO body (HTTP/2 rejects a 204+DATA frame), got {} bytes", + body.len() + ); + } + + #[tokio::test] + async fn relay_forwarded_keeps_body_for_ok() { + // A body-bearing status (200) relays the peer's JSON body verbatim. + let resp = ForwardedResponse { + seq: None, + deduplicated: false, + status: StatusCode::OK, + body: serde_json::json!({ "items": [1, 2, 3] }), + }; + let response = relay_forwarded(resp); + assert_eq!(response.status(), StatusCode::OK); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .expect("collect relayed body"); + assert!(!body.is_empty(), "a 200 relay must keep the JSON body"); + } } diff --git a/tidal-server/src/cluster/node.rs b/tidal-server/src/cluster/node.rs index 3b733ea..4daf4ff 100644 --- a/tidal-server/src/cluster/node.rs +++ b/tidal-server/src/cluster/node.rs @@ -6145,6 +6145,234 @@ where Ok((merged, total)) } +// ── m12p4: cross-shard unified reads (the m11p6 L4 follow-up) ────────────────── + +/// One remote shard group's contribution to a cross-shard read: its parsed items +/// (already in the engine merge type `T`) and the per-group `total_candidates`. +struct RemoteGroupSlice { + items: Vec, + total_candidates: usize, +} + +/// The merged outcome of a corpus-wide cluster read assembled from the local +/// scatter PLUS every missing group's remote slice. +struct CrossShardRead { + /// Local + remote items, score-sorted descending and truncated to `limit`. + items: Vec, + /// `total_candidates` SUMMED across the disjoint sources (local groups + each + /// remote group), mirroring [`scatter_merge`]'s entity-sharded rule — the + /// groups own disjoint key subsets, so the candidate universes add up. + total_candidates: usize, + /// Names of the missing groups whose every forward target was unreachable + /// (the honest degraded contract — never a silent truncation). Empty ⇒ a + /// fully-covered corpus read. + unavailable_shards: Vec, +} + +impl CrossShardRead { + /// Whether any missing group could not be reached on this read. + const fn degraded(&self) -> bool { + !self.unavailable_shards.is_empty() + } +} + +/// Score-merge the local scatter result with every missing group's remote slice, +/// SUM the disjoint `total_candidates`, sort descending, and truncate to `limit`. +/// +/// `T` is the engine merge type (`RetrieveResult` / `SearchResultItem`); `score` +/// reads its ranking score. The disjoint-source sum matches [`scatter_merge`]'s +/// entity-sharded rule — each group owns a distinct key subset, so the candidate +/// universes add rather than dedup-by-max. +fn merge_cross_shard( + local_items: Vec, + local_total: usize, + remote: Vec>, + unavailable_shards: Vec, + limit: usize, + score: impl Fn(&T) -> f64, + set_rank: impl Fn(&mut T, usize), +) -> CrossShardRead { + let mut merged = local_items; + let mut total = local_total; + for slice in remote { + total = total.saturating_add(slice.total_candidates); + merged.extend(slice.items); + } + merged.sort_by(|a, b| { + score(b) + .partial_cmp(&score(a)) + .unwrap_or(std::cmp::Ordering::Equal) + }); + merged.truncate(limit); + // Re-stamp the 1-based page rank over the MERGED order. Each group ranks its + // own slice locally (and remote slices arrive with rank 0), so without this + // the wire `rank` would be incoherent across the merge — local items keep + // their per-group 1,2,3… while every remote item reports 0. `set_rank` is a + // no-op for result types that carry no rank field (e.g. vector matches). + for (i, item) in merged.iter_mut().enumerate() { + set_rank(item, i + 1); + } + CrossShardRead { + items: merged, + total_candidates: total, + unavailable_shards, + } +} + +/// The data-shard groups in the cluster this node does NOT host a replica of +/// (`placement.keys() − groups.keys()`), in ascending id order. Empty ⇒ full +/// placement (or `S=1`): the local scatter already covers the whole corpus and +/// the gateway returns it UNCHANGED — the byte-for-byte short-circuit. +impl ClusterNode { + fn missing_groups(&self) -> Vec { + self.placement + .keys() + .copied() + .filter(|s| !self.groups.contains_key(s)) + .collect() + } + + /// Build the internal per-group cross-shard read hop for group `shard`: + /// `GET {target}{path}?{base_query}&shard={shard}` carrying the internal + /// marker, this node's per-node token, and the forwarded bearer auth. Tries + /// `candidates` (leader-first) in order, failing over on a CONNECT-class error + /// so one dead replica does not strand a group; a real HTTP response (2xx or + /// otherwise) ends the walk. Returns the parsed `{items, total_candidates}` + /// slice, or `None` when EVERY target was unreachable (caller marks the group + /// degraded — never silently dropped). + /// + /// `parse_item` maps one `serde_json::Value` element of the response `items` + /// array into the engine merge type `T`, so RETRIEVE/SEARCH/vector each keep + /// their own wire field set (`score` vs. `distance`). + async fn fetch_remote_group( + &self, + shard: ShardId, + path: &str, + base_query: &str, + headers: &HeaderMap, + parse_item: impl Fn(&serde_json::Value) -> Option, + ) -> Option> { + let candidates = self.forward_candidates(shard); + if candidates.is_empty() { + tracing::warn!( + shard = shard.0, + "cross-shard read: missing group has no reachable forward target" + ); + return None; + } + let auth = forwarded_auth(headers); + // Mint a fresh per-node token off this node's local replica so the remote + // node's marker guard sees a verified sibling and serves the internal hop. + let node_token = self + .replica_for(None) + .ok() + .and_then(|r| r.mint_node_token()); + + let sep = if base_query.is_empty() { "" } else { "&" }; + for http_addr in &candidates { + let url = format!( + "{}?{base_query}{sep}shard={}", + peer_url(http_addr, path), + shard.0 + ); + let mut req = self + .client + .get(&url) + .header(forward::INTERNAL_MARKER, forward::INTERNAL_MARKER_VALUE); + if let Some(auth) = &auth { + req = req.header(axum::http::header::AUTHORIZATION, auth); + } + if let Some(token) = &node_token { + req = req.header(crate::cluster::security::NODE_TOKEN_HEADER, token.clone()); + } + match req.send().await { + Ok(resp) if resp.status().is_success() => { + let body: serde_json::Value = + resp.json().await.unwrap_or(serde_json::Value::Null); + return Some(parse_group_slice(&body, &parse_item)); + } + Ok(resp) => { + // A real verdict from a live target (e.g. 400/500): the group + // is reachable but errored. Do NOT fail over to another + // replica on a non-connect status — surface it as degraded. + tracing::warn!( + shard = shard.0, %url, status = %resp.status(), + "cross-shard read: remote group returned non-success" + ); + return None; + } + Err(e) => { + // Connect-class failure: try the next replica before degrading. + tracing::warn!(shard = shard.0, %url, error = %e, "cross-shard read: target unreachable, trying next"); + } + } + } + None + } +} + +/// Parse a remote group's `{items:[...], total_candidates}` JSON into a typed +/// slice. `total_candidates` falls back to the item count when absent, so a +/// partial-but-present response never under-reports below what it returned. +fn parse_group_slice( + body: &serde_json::Value, + parse_item: impl Fn(&serde_json::Value) -> Option, +) -> RemoteGroupSlice { + let items: Vec = body + .get("items") + .and_then(|v| v.as_array()) + .map(|arr| arr.iter().filter_map(&parse_item).collect()) + .unwrap_or_default(); + let total = body + .get("total_candidates") + .and_then(serde_json::Value::as_u64) + .map_or(items.len(), |t| t as usize); + RemoteGroupSlice { + items, + total_candidates: total, + } +} + +/// Fan out a cross-shard read to every `missing` group CONCURRENTLY and gather +/// the slices, marking any group whose every target was unreachable as degraded. +/// The merge itself is the caller's ([`merge_cross_shard`]); this owns only the +/// scatter + honest-degraded accounting. +async fn fetch_missing_groups( + node: &Arc, + missing: &[ShardId], + path: &str, + base_query: &str, + headers: &HeaderMap, + parse_item: impl Fn(&serde_json::Value) -> Option + Clone, +) -> (Vec>, Vec) { + let futures: Vec<_> = missing + .iter() + .map(|&shard| { + let node = Arc::clone(node); + let path = path.to_string(); + let base_query = base_query.to_string(); + let headers = headers.clone(); + let parse_item = parse_item.clone(); + async move { + let slice = node + .fetch_remote_group(shard, &path, &base_query, &headers, parse_item) + .await; + (shard, slice) + } + }) + .collect(); + + let mut slices = Vec::with_capacity(missing.len()); + let mut unavailable = Vec::new(); + for (shard, slice) in futures_util::future::join_all(futures).await { + match slice { + Some(s) => slices.push(s), + None => unavailable.push(format!("s{}", shard.0)), + } + } + (slices, unavailable) +} + /// 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. @@ -6162,6 +6390,10 @@ where security(("bearerAuth" = [])), )] #[allow(clippy::significant_drop_tightening)] +// m12p4: region-aware forward + cross-shard internal branch + local scatter + +// missing-group fan-out + merge is one linear read assembly; splitting it would +// scatter the byte-for-byte full-placement short-circuit from the partial path. +#[allow(clippy::too_many_lines)] pub async fn feed( State(node): State>, headers: HeaderMap, @@ -6189,24 +6421,61 @@ pub async fn feed( } 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); - } - // m12p2: "more like this" seed for `profile=related` ANN candidate-gen. - if let Some(seed) = query.similar_to { - builder = builder.similar_to(EntityId::new(seed)); - } - let retrieve = builder - .build() - .map_err(|e| ClusterAppError(ServerError::Tidal(e.into())))?; + let build_retrieve = || { + let mut builder = Retrieve::builder().profile(&query.profile).limit(limit); + if let Some(user_id) = query.user_id { + builder = builder.for_user(user_id); + } + // m12p2: "more like this" seed for `profile=related` ANN candidate-gen. + if let Some(seed) = query.similar_to { + builder = builder.similar_to(EntityId::new(seed)); + } + builder + .build() + .map_err(|e| ClusterAppError(ServerError::Tidal(e.into()))) + }; - // 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). + // m12p4 cross-shard internal hop: an internal request with a `?shard=g` + // selector reads ONLY group `g` (the gateway's per-group fan-out leg) and + // serves it locally — single-db, NO further `hosted_dbs` scatter and NO + // re-fan-out, so the gateway's remote hop never loops. + // + // Security: gating on `is_internal` is sufficient and deliberate. The m11p7 + // marker-pinning middleware (`cluster_auth_middleware`, a blanket layer on + // every protected route incl. this one) 403s ANY request carrying the + // internal marker WITHOUT a valid node token whenever a cluster key is + // configured — so an external caller can NEVER reach this single-group branch + // in a secured deployment. With NO cluster key the marker is hint-only by the + // documented trusted-network model, the SAME model the sibling fan-out leg + // relies on (it cannot present a token either), so a stricter `Principal::Node` + // gate here would break no-cluster-key partial-placement fan-out, not harden it. + if is_internal(&headers) + && let Some(shard) = query.shard.map(ShardId) + { + let db = node + .replica_for(Some(shard))? + .db_arc() + .map_err(ClusterAppError)?; + let retrieve = build_retrieve()?; + let (items, total_candidates) = offload_region_read(move || { + let r = db.retrieve(&retrieve).map_err(ServerError::Tidal)?; + Ok((r.items, r.total_candidates)) + }) + .await?; + return Ok(Json(FeedResponse { + items: feed_items(&items), + total_candidates, + region: query.region, + unavailable_shards: None, // single-region serve: complete + }) + .into_response()); + } + + // m11p6: scatter the corpus-wide read over the LOCALLY hosted shard groups + // and merge (see `scatter_merge`). + let retrieve = build_retrieve()?; let dbs = node.hosted_dbs(); - let (items, total_candidates) = offload_region_read(move || { + let (local_items, local_total) = offload_region_read(move || { scatter_merge( &dbs, limit, @@ -6219,14 +6488,106 @@ pub async fn feed( }) .await?; + // m12p4: full placement / `S=1` → the local scatter already covers the whole + // corpus; return it UNCHANGED (byte-for-byte the pre-m12p4 response). Under + // PARTIAL placement, fan out to the groups this node does not host and merge. + let missing = node.missing_groups(); + if missing.is_empty() { + return Ok(Json(FeedResponse { + items: feed_items(&local_items), + total_candidates: local_total, + region: query.region, + unavailable_shards: None, // full placement / S=1: corpus-complete locally + }) + .into_response()); + } + + let base_query = feed_base_query(&query); + let (remote, unavailable) = + fetch_missing_groups(&node, &missing, "/feed", &base_query, &headers, |v| { + parse_scored_item(v).map(|(entity_id, score)| tidaldb::query::RetrieveResult { + entity_id, + score, + rank: 0, + signals: Vec::new(), + }) + }) + .await; + let merged = merge_cross_shard( + local_items, + local_total, + remote, + unavailable, + limit, + |it: &tidaldb::query::RetrieveResult| it.score, + |it: &mut tidaldb::query::RetrieveResult, rank| it.rank = rank, + ); + let unavailable_shards = merged.degraded().then(|| merged.unavailable_shards.clone()); + if let Some(shards) = &unavailable_shards { + tracing::warn!( + unavailable = ?shards, + "cross-shard /feed served degraded (some groups unreachable)" + ); + } + Ok(Json(FeedResponse { - items: feed_items(&items), - total_candidates, + items: feed_items(&merged.items), + total_candidates: merged.total_candidates, region: query.region, + unavailable_shards, }) .into_response()) } +/// Build the internal cross-shard `/feed` query string for the per-group hop: +/// `profile`/`limit`/`user_id`/`similar_to` only — NOT `region` (the hop is +/// region-agnostic) and NOT `shard` (the fan-out appends the per-group selector +/// itself). +fn feed_base_query(query: &FeedQuery) -> String { + use std::fmt::Write as _; + let mut q = format!( + "profile={}&limit={}", + cross_shard_urlencode(&query.profile), + query.clamped_limit() + ); + if let Some(uid) = query.user_id { + let _ = write!(q, "&user_id={uid}"); + } + if let Some(seed) = query.similar_to { + let _ = write!(q, "&similar_to={seed}"); + } + q +} + +/// Parse one `{entity_id, score}` element of a remote read's `items` array into +/// an `(EntityId, f64)` pair. A malformed element is skipped (`None`), never a +/// hard error — the merge proceeds with what parsed. +fn parse_scored_item(v: &serde_json::Value) -> Option<(EntityId, f64)> { + let entity_id = v.get("entity_id").and_then(serde_json::Value::as_u64)?; + let score = v.get("score").and_then(serde_json::Value::as_f64)?; + Some((EntityId::new(entity_id), score)) +} + +/// Percent-encode a cross-shard read's query-string value (space + the reserved +/// set), enough for the `profile` / `query` params the per-group hop carries. +/// The hop URL is `?{base_query}&shard={g}`, so an unencoded `&`/`=`/space in a +/// `profile`/`query` value would corrupt the receiver's parse. +fn cross_shard_urlencode(s: &str) -> String { + use std::fmt::Write as _; + let mut out = String::with_capacity(s.len()); + for b in s.bytes() { + match b { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { + out.push(b as char); + } + _ => { + let _ = write!(out, "%{b:02X}"); + } + } + } + out +} + /// Ranked search. Region-aware reads as [`feed`]: a foreign `?region=` forwards /// to its owner unless the request is internal (marked), which serves locally. #[utoipa::path( @@ -6243,6 +6604,9 @@ pub async fn feed( security(("bearerAuth" = [])), )] #[allow(clippy::significant_drop_tightening)] +// m12p4: same linear read assembly as `feed` (region forward + cross-shard +// internal branch + local scatter + missing-group fan-out + merge). +#[allow(clippy::too_many_lines)] pub async fn search( State(node): State>, headers: HeaderMap, @@ -6267,18 +6631,46 @@ pub async fn search( } 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); - } - let search_query = builder - .build() - .map_err(|e| ClusterAppError(ServerError::Tidal(e.into())))?; + let build_search = || { + let mut builder = Search::builder().query(&query.query).limit(limit); + if let Some(user_id) = query.user_id { + builder = builder.for_user(user_id); + } + builder + .build() + .map_err(|e| ClusterAppError(ServerError::Tidal(e.into()))) + }; - // m11p6: scatter the search over the hosted shard groups and merge (see - // `scatter_merge`; same full-placement caveat as `feed`). + // m12p4 cross-shard internal hop: an internal `?shard=g` request searches + // ONLY group `g` (the gateway's per-group fan-out leg) and serves it locally, + // single-db, NO `hosted_dbs` scatter and NO re-fan-out. + if is_internal(&headers) + && let Some(shard) = query.shard.map(ShardId) + { + let db = node + .replica_for(Some(shard))? + .db_arc() + .map_err(ClusterAppError)?; + let search_query = build_search()?; + let (items, total_candidates) = offload_region_read(move || { + 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?; + return Ok(Json(SearchResponse { + items: search_items(&items), + total_candidates, + region: query.region, + unavailable_shards: None, // single-region serve: complete + }) + .into_response()); + } + + // m11p6: scatter the search over the LOCALLY hosted shard groups and merge. + let search_query = build_search()?; let dbs = node.hosted_dbs(); - let (items, total_candidates) = offload_region_read(move || { + let (local_items, local_total) = offload_region_read(move || { scatter_merge( &dbs, limit as usize, @@ -6292,14 +6684,76 @@ pub async fn search( }) .await?; + // m12p4: full placement / `S=1` → unchanged. Partial placement → fan out. + let missing = node.missing_groups(); + if missing.is_empty() { + return Ok(Json(SearchResponse { + items: search_items(&local_items), + total_candidates: local_total, + region: query.region, + unavailable_shards: None, // full placement / S=1: corpus-complete locally + }) + .into_response()); + } + + let base_query = search_base_query(&query); + let (remote, unavailable) = + fetch_missing_groups(&node, &missing, "/search", &base_query, &headers, |v| { + parse_scored_item(v).map(|(entity_id, score)| { + tidaldb::query::search::SearchResultItem { + entity_id, + score, + rank: 0, + bm25_score: None, + semantic_score: None, + signals: Vec::new(), + metadata: None, + } + }) + }) + .await; + let merged = merge_cross_shard( + local_items, + local_total, + remote, + unavailable, + limit as usize, + |it: &tidaldb::query::search::SearchResultItem| it.score, + |it: &mut tidaldb::query::search::SearchResultItem, rank| it.rank = rank, + ); + let unavailable_shards = merged.degraded().then(|| merged.unavailable_shards.clone()); + if let Some(shards) = &unavailable_shards { + tracing::warn!( + unavailable = ?shards, + "cross-shard /search served degraded (some groups unreachable)" + ); + } + Ok(Json(SearchResponse { - items: search_items(&items), - total_candidates, + items: search_items(&merged.items), + total_candidates: merged.total_candidates, region: query.region, + unavailable_shards, }) .into_response()) } +/// Build the internal cross-shard `/search` query string for the per-group hop: +/// `query`/`limit`/`user_id` only — NOT `region` and NOT `shard` (the fan-out +/// appends the per-group selector itself). +fn search_base_query(query: &SearchQueryParams) -> String { + use std::fmt::Write as _; + let mut q = format!( + "query={}&limit={}", + cross_shard_urlencode(&query.query), + query.clamped_limit() + ); + if let Some(uid) = query.user_id { + let _ = write!(q, "&user_id={uid}"); + } + q +} + /// Pure k-NN vector search (the m12p1 recall probe). Serves LOCALLY from this /// node's hosted shard groups — no `?region=` forwarding: it is a measurement /// surface, and at the S=1 exit-gate shape every region replica holds the full @@ -6320,6 +6774,7 @@ pub async fn search( )] pub async fn vector_search( State(node): State>, + headers: HeaderMap, Json(req): Json, ) -> std::result::Result { if req.vector.is_empty() { @@ -6329,10 +6784,35 @@ pub async fn vector_search( } let k = req.clamped_k(); let ef_search = req.ef_search(); - let vector = req.vector; + // m12p4 cross-shard internal hop: an internal request carrying `shard=g` + // probes ONLY group `g` (the gateway's per-group fan-out leg) and serves it + // locally, single-db, NO `hosted_dbs` scatter and NO re-fan-out. + if is_internal(&headers) + && let Some(shard) = req.shard.map(ShardId) + { + let db = node + .replica_for(Some(shard))? + .db_arc() + .map_err(ClusterAppError)?; + let vector = req.vector; + let items = offload_region_read(move || { + db.vector_search_items(&vector, k, ef_search) + .map_err(ServerError::Tidal) + }) + .await?; + return Ok(Json(VectorSearchResponse { + items: vector_matches(&items), + region: None, + unavailable_shards: None, // single-region serve: complete + }) + .into_response()); + } + + // m11p6: scatter the probe over the LOCALLY hosted shard groups and merge. + let vector = req.vector.clone(); let dbs = node.hosted_dbs(); - let (items, _total) = offload_region_read(move || { + let (local_items, local_total) = offload_region_read(move || { scatter_merge( &dbs, k, @@ -6350,13 +6830,177 @@ pub async fn vector_search( }) .await?; + // m12p4: full placement / `S=1` → unchanged. Partial placement → fan out the + // probe to each missing group (POST body carrying `shard=g`) and merge by + // ascending distance so whole-corpus recall is no longer local-shard-only. + let missing = node.missing_groups(); + if missing.is_empty() { + return Ok(Json(VectorSearchResponse { + items: vector_matches(&local_items), + region: None, + unavailable_shards: None, // full placement / S=1: corpus-complete locally + }) + .into_response()); + } + + let (remote, unavailable) = node + .fetch_missing_groups_vector(&missing, &req, k, ef_search, &headers) + .await; + let merged = merge_cross_shard( + local_items, + local_total, + remote, + unavailable, + k, + // Closest-first: negate distance so the shared descending merge keeps the + // nearest neighbours. + |r: &tidaldb::storage::vector::VectorSearchResult| -f64::from(r.distance), + // Vector matches carry no rank field (ordered by distance on the wire). + |_r: &mut tidaldb::storage::vector::VectorSearchResult, _rank| {}, + ); + let unavailable_shards = merged.degraded().then(|| merged.unavailable_shards.clone()); + if let Some(shards) = &unavailable_shards { + tracing::warn!( + unavailable = ?shards, + "cross-shard /vector_search served degraded (some groups unreachable)" + ); + } + Ok(Json(VectorSearchResponse { - items: vector_matches(&items), + items: vector_matches(&merged.items), region: None, + unavailable_shards, }) .into_response()) } +impl ClusterNode { + /// Fan out the vector probe to every missing group CONCURRENTLY (POST body + /// carrying the per-group `shard` selector), gather each group's nearest + /// slice, and mark any group whose every target was unreachable as degraded + /// — the POST analogue of [`fetch_missing_groups`] for the body-carried + /// query vector. + async fn fetch_missing_groups_vector( + self: &Arc, + missing: &[ShardId], + req: &VectorSearchRequest, + k: usize, + ef_search: Option, + headers: &HeaderMap, + ) -> ( + Vec>, + Vec, + ) { + let auth = forwarded_auth(headers); + let node_token = self + .replica_for(None) + .ok() + .and_then(|r| r.mint_node_token()); + let futures: Vec<_> = missing + .iter() + .map(|&shard| { + let node = Arc::clone(self); + let auth = auth.clone(); + let node_token = node_token.clone(); + // The per-group hop body: the same vector + knobs, with `shard` + // set so the remote serves ONLY this group and never re-fans-out. + let body = VectorSearchRequest { + vector: req.vector.clone(), + k: u32::try_from(k).unwrap_or(u32::MAX), + ef_search: ef_search.map(|e| u32::try_from(e).unwrap_or(u32::MAX)), + shard: Some(shard.0), + }; + async move { + let slice = node + .fetch_remote_group_vector( + shard, + &body, + auth.as_deref(), + node_token.as_deref(), + ) + .await; + (shard, slice) + } + }) + .collect(); + + let mut slices = Vec::with_capacity(missing.len()); + let mut unavailable = Vec::new(); + for (shard, slice) in futures_util::future::join_all(futures).await { + match slice { + Some(s) => slices.push(s), + None => unavailable.push(format!("s{}", shard.0)), + } + } + (slices, unavailable) + } + + /// POST the vector probe to one missing group's forward targets (leader-first, + /// failing over on a connect error). Returns the parsed `{items:[{entity_id, + /// distance}], …}` slice, or `None` when every target was unreachable. + async fn fetch_remote_group_vector( + &self, + shard: ShardId, + body: &VectorSearchRequest, + auth: Option<&str>, + node_token: Option<&str>, + ) -> Option> { + let candidates = self.forward_candidates(shard); + if candidates.is_empty() { + tracing::warn!( + shard = shard.0, + "cross-shard vector probe: missing group has no reachable forward target" + ); + return None; + } + for http_addr in &candidates { + let url = peer_url(http_addr, "/vector_search"); + let mut req = self + .client + .post(&url) + .header(forward::INTERNAL_MARKER, forward::INTERNAL_MARKER_VALUE) + .json(body); + if let Some(auth) = auth { + req = req.header(axum::http::header::AUTHORIZATION, auth); + } + if let Some(token) = node_token { + req = req.header(crate::cluster::security::NODE_TOKEN_HEADER, token); + } + match req.send().await { + Ok(resp) if resp.status().is_success() => { + let json: serde_json::Value = + resp.json().await.unwrap_or(serde_json::Value::Null); + return Some(parse_group_slice(&json, parse_vector_match)); + } + Ok(resp) => { + tracing::warn!( + shard = shard.0, %url, status = %resp.status(), + "cross-shard vector probe: remote group returned non-success" + ); + return None; + } + Err(e) => { + tracing::warn!(shard = shard.0, %url, error = %e, "cross-shard vector probe: target unreachable, trying next"); + } + } + } + None + } +} + +/// Parse one `{entity_id, distance}` element of a `/vector_search` response into +/// an engine `VectorSearchResult`. A malformed element is skipped (`None`). +fn parse_vector_match( + v: &serde_json::Value, +) -> Option { + let id = v.get("entity_id").and_then(serde_json::Value::as_u64)?; + let distance = v.get("distance").and_then(serde_json::Value::as_f64)?; + Some(tidaldb::storage::vector::VectorSearchResult { + id, + distance: distance as f32, + }) +} + /// If `region` names a DIFFERENT region than this node owns, forward the read /// (verbatim query string, marker set, auth passed through) to that region's /// process and relay its response. Returns `Ok(None)` when the read should be @@ -6406,6 +7050,15 @@ async fn maybe_forward_region_read( Ok(resp) => { let status = resp.status(); let bytes = resp.bytes().await.unwrap_or_default(); + // Honor the SAME bodyless-status contract as the write-relay helper + // (`forward::relay_forwarded`): a 204/304/1xx must carry NO body, or + // an HTTP/2 client RST_STREAMs it. These read forwards return 200 or + // a body-bearing error today, but routing through the shared guard + // keeps the read- and write-relay paths from drifting if a forwarded + // read ever learns to return a bodyless status (e.g. a 304 ETag). + if forward::status_forbids_body(status) { + return Ok(Some(status.into_response())); + } let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null); Ok(Some((status, Json(body)).into_response())) @@ -6678,7 +7331,10 @@ async fn sharded_write_route( ) .await { - Ok(resp) => Ok((resp.status, Json(resp.body)).into_response()), + // Relay through the shared helper so a bodyless status (204 from a + // signal/embedding write) carries NO body — an HTTP/2 client RST_STREAMs + // a 204+body — and the peer's x-tidal-seq/dedup verdict headers ride back. + Ok(resp) => Ok(forward::relay_forwarded(resp)), Err(e) => Err(ClusterAppError(ServerError::RegionUnreachable { region: state.region_name_of(owner).to_string(), cause: e, @@ -7029,3 +7685,154 @@ mod auth_middleware_tests { } } } + +/// m12p4 cross-shard unified reads — the merge + wire-parse logic that the three +/// read handlers share. These prove the disjoint-source contract (SUM totals, +/// score-merge, truncate) and the honest-degraded accounting WITHOUT a live +/// cluster; the multi-process partial-placement fan-out is covered end-to-end by +/// `tests/cluster_cross_shard_reads.rs` (the `cluster-e2e` integration suite). +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::float_cmp)] +mod cross_shard_tests { + use super::*; + + fn rr(entity: u64, score: f64) -> tidaldb::query::RetrieveResult { + tidaldb::query::RetrieveResult { + entity_id: EntityId::new(entity), + score, + rank: 0, + signals: Vec::new(), + } + } + + #[test] + fn merge_sums_disjoint_totals_and_score_sorts() { + // Local group returned 2 items / saw 10 candidates; one remote group + // returned 2 items / saw 7 candidates. The groups own disjoint keys, so + // the merged total SUMS (17), and the page is score-sorted descending. + let local = vec![rr(1, 0.9), rr(2, 0.4)]; + let remote = vec![RemoteGroupSlice { + items: vec![rr(3, 0.7), rr(4, 0.2)], + total_candidates: 7, + }]; + let merged = merge_cross_shard( + local, + 10, + remote, + Vec::new(), + 3, + |it| it.score, + |it, r| { + it.rank = r; + }, + ); + assert_eq!(merged.total_candidates, 17, "disjoint groups SUM totals"); + assert!(!merged.degraded(), "all groups reachable ⇒ not degraded"); + let ids: Vec = merged.items.iter().map(|i| i.entity_id.as_u64()).collect(); + assert_eq!(ids, vec![1, 3, 2], "score-sorted then truncated to limit=3"); + // Rank is re-stamped over the MERGED order (1-based), so a remote item + // (id 3, which arrived with rank 0) gets a coherent global page rank, not 0. + let ranks: Vec = merged.items.iter().map(|i| i.rank).collect(); + assert_eq!( + ranks, + vec![1, 2, 3], + "merged page rank is global 1-based, no rank-0 hole" + ); + } + + #[test] + fn merge_marks_unreachable_group_degraded_never_hard_fails() { + // One missing group came back (1 item), another was unreachable. The read + // returns the partial page AND surfaces the unreachable group by name — + // never a silent truncation, never an error. + let local = vec![rr(1, 0.5)]; + let remote = vec![RemoteGroupSlice { + items: vec![rr(2, 0.8)], + total_candidates: 4, + }]; + let merged = merge_cross_shard( + local, + 3, + remote, + vec!["s2".to_string()], + 10, + |it| it.score, + |it, r| { + it.rank = r; + }, + ); + assert!(merged.degraded(), "an unreachable group degrades the read"); + assert_eq!(merged.unavailable_shards, vec!["s2".to_string()]); + // The reachable items still merge and rank (degraded ≠ empty). + let ids: Vec = merged.items.iter().map(|i| i.entity_id.as_u64()).collect(); + assert_eq!(ids, vec![2, 1]); + assert_eq!(merged.total_candidates, 7, "local 3 + remote 4"); + } + + #[test] + fn parse_group_slice_reads_items_and_total_with_fallback() { + let body = serde_json::json!({ + "items": [{"entity_id": 7, "score": 0.6}, {"entity_id": 8, "score": 0.3}], + "total_candidates": 42 + }); + let slice = parse_group_slice(&body, |v| { + parse_scored_item(v).map(|(e, s)| rr(e.as_u64(), s)) + }); + assert_eq!(slice.total_candidates, 42); + assert_eq!(slice.items.len(), 2); + assert_eq!(slice.items[0].entity_id.as_u64(), 7); + + // Missing total_candidates falls back to the item count, never below it. + let body2 = serde_json::json!({ "items": [{"entity_id": 1, "score": 0.1}] }); + let slice2 = parse_group_slice(&body2, |v| { + parse_scored_item(v).map(|(e, s)| rr(e.as_u64(), s)) + }); + assert_eq!(slice2.total_candidates, 1, "fallback = item count"); + + // A malformed element is skipped, never a panic / hard error. + let body3 = serde_json::json!({ + "items": [{"entity_id": 1, "score": 0.1}, {"oops": true}], + "total_candidates": 5 + }); + let slice3 = parse_group_slice(&body3, |v| { + parse_scored_item(v).map(|(e, s)| rr(e.as_u64(), s)) + }); + assert_eq!(slice3.items.len(), 1, "the bad element is dropped"); + } + + #[test] + fn parse_vector_match_reads_distance_field() { + // The vector wire shape carries `distance`, not `score` — the probe merges + // by ascending distance (negated in the handler's merge key). + let v = serde_json::json!({ "entity_id": 9, "distance": 0.25 }); + let m = parse_vector_match(&v).unwrap(); + assert_eq!(m.id, 9); + assert_eq!(m.distance, 0.25_f32); + assert!(parse_vector_match(&serde_json::json!({"entity_id": 9})).is_none()); + } + + #[test] + fn vector_merge_keeps_nearest_first_across_groups() { + // Two groups' nearest sets merge so the globally-closest (smallest + // distance) come first after the negated-distance descending sort. + fn vr(id: u64, distance: f32) -> tidaldb::storage::vector::VectorSearchResult { + tidaldb::storage::vector::VectorSearchResult { id, distance } + } + let local = vec![vr(1, 0.5), vr(2, 1.5)]; + let remote = vec![RemoteGroupSlice { + items: vec![vr(3, 0.2), vr(4, 2.0)], + total_candidates: 2, + }]; + let merged = merge_cross_shard( + local, + 2, + remote, + Vec::new(), + 3, + |r| -f64::from(r.distance), + |_r, _rank| {}, + ); + let ids: Vec = merged.items.iter().map(|r| r.id).collect(); + assert_eq!(ids, vec![3, 1, 2], "ascending distance, truncated to k=3"); + } +} diff --git a/tidal-server/src/cluster/routes.rs b/tidal-server/src/cluster/routes.rs index 65000ec..1597e4d 100644 --- a/tidal-server/src/cluster/routes.rs +++ b/tidal-server/src/cluster/routes.rs @@ -471,6 +471,7 @@ pub async fn feed( items: feed_items(&result.items), total_candidates: result.total_candidates, region: query.region, + unavailable_shards: None, // single-process region serve: complete })) } @@ -528,6 +529,7 @@ pub async fn search( items: search_items(&result.items), total_candidates: result.total_candidates, region: query.region, + unavailable_shards: None, // single-process region serve: complete })) } diff --git a/tidal-server/src/dto.rs b/tidal-server/src/dto.rs index d9bad37..994bcf2 100644 --- a/tidal-server/src/dto.rs +++ b/tidal-server/src/dto.rs @@ -91,6 +91,13 @@ pub struct VectorSearchRequest { /// recall/latency knob). Omitted = the slot's configured default. #[serde(default)] pub ef_search: Option, + /// Internal cross-shard read selector (m12p4). When the cluster gateway fans + /// a corpus-wide probe out to a node that hosts only some shard groups, it + /// sets this on the per-group internal hop so the remote serves ONLY group + /// `shard` and never re-fans-out. Absent on every external request — clients + /// never set it; the gateway merges the per-group slices itself. + #[serde(default)] + pub shard: Option, } impl VectorSearchRequest { @@ -158,6 +165,14 @@ pub struct FeedQuery { #[serde(default)] #[param(example = 42)] pub similar_to: Option, + /// Internal cross-shard read selector (m12p4). The cluster gateway sets this + /// on the per-group internal hop when it fans a corpus-wide `/feed` out to a + /// node hosting a strict subset of shard groups, so the remote reads ONLY + /// group `shard` and never re-fans-out. Absent on every external request — + /// clients never set it; the gateway merges the per-group slices itself. + #[serde(default)] + #[param(example = 0)] + pub shard: Option, } impl FeedQuery { @@ -193,6 +208,14 @@ pub struct SearchQueryParams { /// Target region (cluster mode only; rejected with 400 standalone). #[serde(default)] pub region: Option, + /// Internal cross-shard read selector (m12p4). The cluster gateway sets this + /// on the per-group internal hop when it fans a corpus-wide `/search` out to + /// a node hosting a strict subset of shard groups, so the remote searches + /// ONLY group `shard` and never re-fans-out. Absent on every external + /// request — clients never set it; the gateway merges the per-group slices. + #[serde(default)] + #[param(example = 0)] + pub shard: Option, } impl SearchQueryParams { @@ -234,6 +257,13 @@ pub struct FeedResponse { pub total_candidates: usize, /// Region the feed was served from (cluster mode); `null` standalone. pub region: Option, + /// Names of shard groups that could not be reached for this cross-shard read + /// (m12p4). Present ONLY when the result is degraded — a complete read omits + /// it. Its presence tells the client the page is PARTIAL (fewer items and a + /// lower `total_candidates` than a complete read), so a degraded read is never + /// silently indistinguishable from a complete one. + #[serde(skip_serializing_if = "Option::is_none")] + pub unavailable_shards: Option>, } /// One ranked item in a feed response. @@ -268,6 +298,11 @@ pub struct SearchResponse { pub total_candidates: usize, /// Region the search was served from (cluster mode); `null` standalone. pub region: Option, + /// Shard groups unreachable for this cross-shard read (m12p4); present ONLY + /// when the result is degraded (partial page), omitted on a complete read. + /// See [`FeedResponse::unavailable_shards`]. + #[serde(skip_serializing_if = "Option::is_none")] + pub unavailable_shards: Option>, } /// One ranked item in a search response. @@ -294,9 +329,13 @@ pub struct VectorSearchResponse { pub items: Vec, /// Always `null`: the recall probe serves locally (standalone, or — in /// cluster mode — merged across the node's hosted shard groups), so there is - /// no single serving region to report. Cross-region routing for this probe is - /// the m12p4 cross-shard read follow-up. + /// no single serving region to report. pub region: Option, + /// Shard groups unreachable for this cross-shard probe (m12p4); present ONLY + /// when the result is degraded (partial nearest-set), omitted when complete. + /// See [`FeedResponse::unavailable_shards`]. + #[serde(skip_serializing_if = "Option::is_none")] + pub unavailable_shards: Option>, } /// One nearest-neighbor match: the entity and its distance from the query. @@ -389,6 +428,7 @@ mod tests { limit, region: None, similar_to: None, + shard: None, } } @@ -398,6 +438,7 @@ mod tests { user_id: None, limit, region: None, + shard: None, } } diff --git a/tidal-server/src/router.rs b/tidal-server/src/router.rs index ec0aa0e..9a346ba 100644 --- a/tidal-server/src/router.rs +++ b/tidal-server/src/router.rs @@ -384,6 +384,7 @@ pub(crate) async fn feed( items: feed_items(&result.items), total_candidates: result.total_candidates, region: query.region, + unavailable_shards: None, // standalone: no shards, never degraded })) } @@ -430,6 +431,7 @@ pub(crate) async fn search( items: search_items(&result.items), total_candidates: result.total_candidates, region: query.region, + unavailable_shards: None, // standalone: no shards, never degraded })) } @@ -473,6 +475,7 @@ pub(crate) async fn vector_search( Ok(Json(VectorSearchResponse { items: vector_matches(&result), region: None, + unavailable_shards: None, // standalone: no shards, never degraded })) } diff --git a/tidal-server/src/scatter_gather.rs b/tidal-server/src/scatter_gather.rs index 2511711..9513e89 100644 --- a/tidal-server/src/scatter_gather.rs +++ b/tidal-server/src/scatter_gather.rs @@ -22,7 +22,7 @@ use std::{ collections::{HashMap, HashSet}, - sync::{Arc, Condvar, Mutex, OnceLock, mpsc}, + sync::{Arc, OnceLock, mpsc}, time::{Duration, Instant}, }; @@ -143,120 +143,132 @@ fn clamp_deadline_ms(requested: Option) -> u64 { } } -// ── Global scatter-gather worker bound ─────────────────────────────────────── +// ── Global scatter-gather worker pool ──────────────────────────────────────── -/// Floor on the total concurrent shard-worker threads the process may run. -/// Even on a single-core host the fan-out gets meaningful parallelism. +/// Floor on the pool's persistent shard-worker threads. Even on a single-core +/// host the fan-out gets meaningful parallelism. const MIN_SHARD_WORKERS: usize = 8; -/// Multiplier applied to available parallelism to size the cap. Shard workers +/// Multiplier applied to available parallelism to size the pool. Shard workers /// are query/IO-bound (a blocking `TidalDb` read), not purely CPU-bound, so a /// modest oversubscription keeps cores busy without unbounded growth. const SHARD_WORKERS_PER_CORE: usize = 8; -/// Hard ceiling on the total concurrent shard-worker threads, independent of -/// core count, so a many-core host still cannot spawn an unbounded thread set -/// under a query storm. +/// Hard ceiling on the pool's worker threads, independent of core count, so a +/// many-core host still keeps a bounded thread set under a query storm. const MAX_SHARD_WORKERS: usize = 256; +/// Queued (not-yet-started) shard jobs permitted per worker before a further +/// submission is refused and the shard is reported degraded. A small multiple +/// lets a fan-out burst queue briefly rather than shed, while a sustained storm +/// still sheds promptly — the queue can never grow without bound. +const SHARD_QUEUE_DEPTH_PER_WORKER: usize = 8; -/// A process-wide counting semaphore bounding the number of scatter-gather -/// shard-worker threads that may execute a blocking shard query at once. +/// A boxed shard-query job run on a pool worker. /// -/// The router-level [`ConcurrencyLimitLayer`](tower::limit::ConcurrencyLimitLayer) -/// caps in-flight HTTP requests, but each sharded request still fans out one -/// detached thread per live shard. Without a fan-out cap, `requests × shards` -/// detached OS threads can pile up under a burst. This semaphore caps the -/// AGGREGATE concurrent shard workers regardless of how many requests fan out, -/// so the node sheds load cleanly instead of exhausting OS threads. +/// It owns everything it needs (an `Arc` clone, the cloned per-shard query +/// closure, the per-request result `Sender`) and sends its outcome back over +/// that channel when it finishes — or early-returns if the request budget +/// already elapsed before a worker picked it up. +type ShardJob = Box; + +/// A fixed, process-global pool of persistent OS threads that run scatter-gather +/// shard queries. /// -/// A worker that cannot acquire a permit before the request deadline exits -/// WITHOUT running its query; the coordinator then reports that shard as -/// degraded (never silently dropped). That is the correct behavior under -/// overload: the shard genuinely could not be serviced within budget, and the -/// would-be worker thread retires immediately rather than parking indefinitely. -struct ShardWorkerSemaphore { - /// Available permits. Guarded by the mutex; waiters block on the condvar. - permits: Mutex, - available: Condvar, +/// Replaces the previous design of spawning one detached `std::thread` PER SHARD +/// PER QUERY (plus a separate counting semaphore to cap the aggregate). At high +/// read QPS across many shards, per-query thread creation/teardown dominated; a +/// pool of reused threads removes that churn entirely. The pool's fixed worker +/// count IS the aggregate concurrency cap — no separate semaphore is needed: at +/// most `workers` shard queries run at once and excess jobs queue. +/// +/// The detach-on-deadline contract is preserved EXACTLY. The coordinator submits +/// one job per live shard, each carrying a clone of the per-request result +/// `Sender`, then drains the matching `Receiver` with `recv_timeout` against the +/// remaining budget. A job that finishes after the coordinator has already +/// returned sends into a `Receiver` that has been dropped; the send fails +/// harmlessly and the worker moves straight on to the next job. The coordinator +/// never joins a worker, so one slow shard's blocking query can never delay the +/// partial result — identical to the old detached-thread behavior, minus the +/// per-query spawn. +/// +/// Submissions beyond the bounded queue are refused (`try_send` → `Full`); the +/// coordinator marks that shard degraded immediately rather than blocking or +/// growing the queue without bound — the pool analogue of the old "permit not +/// acquired before deadline → degrade" path. +struct ShardReadPool { + sender: crossbeam::channel::Sender, } -impl ShardWorkerSemaphore { - fn new(permits: usize) -> Self { - Self { - permits: Mutex::new(permits.max(1)), - available: Condvar::new(), +impl ShardReadPool { + /// Build the pool and start `workers` persistent threads draining a bounded + /// queue of depth `workers * SHARD_QUEUE_DEPTH_PER_WORKER`. + /// + /// # Panics + /// + /// Panics if a worker OS thread cannot be spawned. The pool is built once, + /// lazily, before any shard query runs, so a thread-exhausted host fails + /// loudly at first use rather than turning the same failure into a + /// per-request degrade on the hot path (the very hazard the pool removes). + fn new(workers: usize) -> Self { + let workers = workers.max(1); + let queue_depth = workers.saturating_mul(SHARD_QUEUE_DEPTH_PER_WORKER).max(1); + // Bounded so a sustained storm sheds (degrades shards) instead of growing + // the queue without limit. crossbeam's MPMC channel lets every worker + // pull from the same queue with no shared `Mutex`. + let (sender, receiver) = crossbeam::channel::bounded::(queue_depth); + for i in 0..workers { + let rx: crossbeam::channel::Receiver = receiver.clone(); + std::thread::Builder::new() + .name(format!("scatter-shard-pool-{i}")) + .spawn(move || { + // `recv` blocks until a job arrives and only errors once every + // `Sender` is dropped — which never happens for the + // process-global pool, so these threads live for the process + // lifetime and are reused across every request. + while let Ok(job) = rx.recv() { + // Isolate a panicking shard query so it cannot retire this + // persistent worker. The old per-query design spawned a + // fresh thread each time, so a panic killed an ephemeral + // thread and the next query got a new one; a fixed pool has + // no such self-heal, so an unguarded panic would PERMANENTLY + // shrink the pool toward an all-shards-degraded wedge. The + // job sends its outcome before returning, so a caught panic + // just means that shard never reported, and the coordinator + // degrades it — the correct, honest signal. + if std::panic::catch_unwind(std::panic::AssertUnwindSafe(job)).is_err() { + tracing::error!( + worker = i, + "scatter-gather shard job panicked; worker survives, shard degraded" + ); + } + } + }) + .expect("spawn scatter-gather shard-pool worker thread"); } + Self { sender } } - /// Acquire one permit, waiting at most `timeout`. Returns a guard that - /// releases the permit on drop, or `None` if no permit became available in - /// time (the caller should degrade rather than block further). - fn acquire_timeout(&self, timeout: Duration) -> Option> { - let deadline = Instant::now() + timeout; - // Poison is benign here: the only critical section is the integer - // permit count and a notify; a panic mid-update cannot leave a torn - // value, so recover the guard and continue (house pattern). - let mut permits = self - .permits - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - loop { - if *permits > 0 { - *permits -= 1; - return Some(ShardWorkerPermit { sem: self }); - } - let remaining = deadline.saturating_duration_since(Instant::now()); - if remaining.is_zero() { - return None; - } - let (next, timed_out) = self - .available - .wait_timeout(permits, remaining) - .unwrap_or_else(std::sync::PoisonError::into_inner); - permits = next; - if timed_out.timed_out() && *permits == 0 { - return None; - } - } - } - - fn release(&self) { - { - let mut permits = self - .permits - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - *permits += 1; - } - // One released permit wakes at most one waiter. Notify after dropping the - // lock so the woken waiter does not immediately re-block on a held guard. - self.available.notify_one(); + /// Submit a shard job. Returns `false` if the bounded queue is full (the + /// caller degrades that shard); never blocks. + fn submit(&self, job: ShardJob) -> bool { + // `Ok` ⇒ enqueued. `Full` ⇒ queue saturated → degrade. `Disconnected` + // cannot occur (the pool holds the receiver for the process lifetime), + // but treating any non-`Ok` as "not accepted" is the safe default. + self.sender.try_send(job).is_ok() } } -/// RAII permit: releases its slot back to the semaphore on drop, even if the -/// shard query panics. -struct ShardWorkerPermit<'a> { - sem: &'a ShardWorkerSemaphore, -} +/// The process-global shard-read pool, sized once from available parallelism on +/// first use. +static SHARD_READ_POOL: OnceLock = OnceLock::new(); -impl Drop for ShardWorkerPermit<'_> { - fn drop(&mut self) { - self.sem.release(); - } -} - -/// The process-global shard-worker semaphore, sized once from available -/// parallelism on first use. -static SHARD_WORKER_SEMAPHORE: OnceLock = OnceLock::new(); - -/// Resolve the global shard-worker semaphore, initializing it on first use. -fn shard_worker_semaphore() -> &'static ShardWorkerSemaphore { - SHARD_WORKER_SEMAPHORE.get_or_init(|| { +/// Resolve the global shard-read pool, initializing it on first use. +fn shard_read_pool() -> &'static ShardReadPool { + SHARD_READ_POOL.get_or_init(|| { let cores = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get); - let permits = cores + let workers = cores .saturating_mul(SHARD_WORKERS_PER_CORE) .clamp(MIN_SHARD_WORKERS, MAX_SHARD_WORKERS); - tracing::info!(permits, "scatter-gather shard-worker cap initialized"); - ShardWorkerSemaphore::new(permits) + tracing::info!(workers, "scatter-gather shard-read pool initialized"); + ShardReadPool::new(workers) }) } @@ -425,34 +437,32 @@ struct GatherState { /// Fan out a per-shard query CONCURRENTLY and gather the results under a hard /// total-time budget. /// -/// Each non-partitioned shard runs `query_one` on its own **detached** OS -/// thread holding an owned `Arc` clone. The underlying +/// Each non-partitioned shard's `query_one` runs as a job on the process-global +/// [`shard_read_pool`], holding an owned `Arc` clone. The underlying /// `TidalDb` query is a *blocking* call, so true thread-level concurrency (not /// cooperative async) is required for one slow shard not to serialize behind -/// the others — and the threads must be detached, not scoped, so the -/// coordinator can return its partial result the instant the budget expires -/// without joining a shard whose blocking query is still in flight. (A scoped -/// join would re-introduce the very hang this fix removes.) The coordinator -/// drains the result channel with [`mpsc::Receiver::recv_timeout`] against the -/// remaining budget. +/// the others — and the coordinator must NOT join a worker, so it can return its +/// partial result the instant the budget expires without waiting on a shard +/// whose blocking query is still in flight. The coordinator drains the result +/// channel with [`mpsc::Receiver::recv_timeout`] against the remaining budget. /// /// Shards that error, or that fail to report by the deadline, are recorded in -/// `unavailable_shards` (degraded) — they are NEVER silently dropped. A -/// detached worker that finishes after the deadline simply sends into a -/// receiver that has already been dropped; the send fails harmlessly and the -/// worker exits. Because every worker only *reads* the shared cluster, leaking -/// a still-running worker past the request is sound. +/// `unavailable_shards` (degraded) — they are NEVER silently dropped. A pool +/// worker that finishes after the deadline simply sends into a receiver that has +/// already been dropped; the send fails harmlessly and the worker moves on to +/// the next job. Because every worker only *reads* the shared context, a +/// still-running worker outliving the request is sound. /// /// # Fan-out cap /// -/// Each worker must acquire a permit from the process-global -/// [`shard_worker_semaphore`] before running its blocking query, so the -/// AGGREGATE number of concurrently-executing shard workers is bounded -/// regardless of how many sharded requests fan out at once. A worker that -/// cannot get a permit within the remaining budget retires immediately and is -/// reported degraded — never silently dropped, and never left parked -/// indefinitely. Together with the router's request-concurrency limit this caps -/// total detached threads under a query storm instead of growing them without +/// The pool's fixed worker count bounds the AGGREGATE number of concurrently- +/// executing shard queries regardless of how many sharded requests fan out at +/// once — no per-query thread spawn, no separate semaphore. A shard whose job +/// cannot be enqueued (the bounded queue is saturated) is reported degraded +/// immediately — never silently dropped, and never left parked. A queued job a +/// worker only reaches after the budget elapsed skips its query and reports +/// degraded too. Together with the router's request-concurrency limit this caps +/// total worker threads under a query storm instead of growing them without /// bound. fn dispatch_shards( ctx: &Arc, @@ -506,45 +516,43 @@ where // is marked degraded immediately and excluded from the wait set so we never // burn the whole deadline waiting on a result that can never arrive. let mut dispatched: Vec = Vec::with_capacity(live_shards.len()); - let sem = shard_worker_semaphore(); + let pool = shard_read_pool(); + // Absolute instant the request budget expires; captured into each job so a + // job a worker only reaches after the deadline can skip its now-pointless + // query instead of piling abandoned work onto the pool under a storm. + let job_deadline = start + deadline; for &shard in &live_shards { let tx = tx.clone(); let ctx = Arc::clone(ctx); let query_one = query_one.clone(); - let spawned = std::thread::Builder::new() - .name(format!("scatter-shard-{}", shard.0)) - .spawn(move || { - // Bound the AGGREGATE concurrent shard workers process-wide: a - // worker that cannot get a permit before the budget expires - // retires immediately and reports degraded, rather than parking - // a thread or running an out-of-budget query. The permit is held - // for exactly the blocking query and released on drop (even on - // panic). See [`shard_worker_semaphore`]. - let remaining = deadline.saturating_sub(start.elapsed()); - // `_permit` (RAII) is held for exactly the blocking query and - // released when the closure returns; `None` means the cap was hit - // before the deadline, so this worker retires as degraded. - let outcome = sem.acquire_timeout(remaining).map_or_else( - || { - Err(ServerError::Unavailable( - "scatter-gather worker cap reached before deadline".into(), - )) - }, - |_permit| query_one(&ctx, shard), - ); - // The receiver may already have moved on after the deadline; a - // closed channel is expected and benign, so the error is dropped. - let _ = tx.send((shard, outcome)); - }); - match spawned { - Ok(_handle) => dispatched.push(shard), - Err(e) => { - // OS thread exhaustion: degrade this shard rather than hang or - // panic. The other shards still race normally. - let name = shard_name(region_names, shard); - tracing::error!(shard = shard.0, region = %name, error = %e, "failed to spawn scatter-gather worker; marking degraded"); - state.unavailable_shards.push(name); + let job: ShardJob = Box::new(move || { + // If the budget elapsed before a worker picked this job up, skip the + // blocking query and report degraded. The coordinator has very + // likely already returned, so this send lands in a dropped receiver + // and fails harmlessly. + if Instant::now() >= job_deadline { + let _ = tx.send(( + shard, + Err(ServerError::Unavailable( + "scatter-gather job started after deadline".into(), + )), + )); + return; } + let outcome = query_one(&ctx, shard); + // The receiver may already have moved on after the deadline; a + // closed channel is expected and benign, so the error is dropped. + let _ = tx.send((shard, outcome)); + }); + if pool.submit(job) { + dispatched.push(shard); + } else { + // Bounded pool queue saturated under load: degrade this shard + // immediately rather than block or grow the queue without bound. The + // other shards still race normally. + let name = shard_name(region_names, shard); + tracing::warn!(shard = shard.0, region = %name, "scatter-gather pool queue full; marking shard degraded"); + state.unavailable_shards.push(name); } } // Drop the coordinator's own sender so the channel closes once every worker @@ -2358,77 +2366,128 @@ mod tests { ); } - /// C16: the shard-worker semaphore must cap the AGGREGATE concurrent shard - /// workers. With 2 permits, no more than 2 workers may hold a permit at - /// once even when 8 contend, and every worker eventually completes (no - /// deadlock, no lost permit on release). + /// C16: the shard-read pool caps the AGGREGATE concurrent shard workers at + /// its fixed worker count. With 2 workers, no more than 2 jobs run at once + /// even when 8 are submitted, and every job eventually completes (no + /// deadlock, threads reused across the burst). #[test] - fn shard_worker_semaphore_caps_concurrency() { + fn shard_read_pool_caps_concurrency() { use std::sync::atomic::{AtomicUsize, Ordering}; - let sem = Arc::new(ShardWorkerSemaphore::new(2)); + let pool = ShardReadPool::new(2); let live = Arc::new(AtomicUsize::new(0)); let peak = Arc::new(AtomicUsize::new(0)); let completed = Arc::new(AtomicUsize::new(0)); - let handles: Vec<_> = (0..8) - .map(|_| { - let sem = Arc::clone(&sem); - let live = Arc::clone(&live); - let peak = Arc::clone(&peak); - let completed = Arc::clone(&completed); - std::thread::spawn(move || { - // Generous timeout: every worker should get a permit - // eventually (the holders release quickly). - let permit = sem - .acquire_timeout(Duration::from_secs(5)) - .expect("permit must become available within timeout"); - let now = live.fetch_add(1, Ordering::SeqCst) + 1; - peak.fetch_max(now, Ordering::SeqCst); - // Hold the permit briefly so contention is real. - std::thread::sleep(Duration::from_millis(5)); - live.fetch_sub(1, Ordering::SeqCst); - completed.fetch_add(1, Ordering::SeqCst); - drop(permit); - }) - }) - .collect(); - for h in handles { - h.join().expect("worker thread must not panic"); + for _ in 0..8 { + let live = Arc::clone(&live); + let peak = Arc::clone(&peak); + let completed = Arc::clone(&completed); + let job: ShardJob = Box::new(move || { + let now = live.fetch_add(1, Ordering::SeqCst) + 1; + peak.fetch_max(now, Ordering::SeqCst); + // Hold the worker briefly so contention is real. + std::thread::sleep(Duration::from_millis(5)); + live.fetch_sub(1, Ordering::SeqCst); + completed.fetch_add(1, Ordering::SeqCst); + }); + assert!(pool.submit(job), "queue (depth 16) accepts all 8 jobs"); + } + + // Spin-wait for all jobs to finish (generous ceiling; no fixed sleep). + let start = Instant::now(); + while completed.load(Ordering::SeqCst) < 8 { + assert!( + start.elapsed() < Duration::from_secs(5), + "every job must complete (no deadlock); saw {}", + completed.load(Ordering::SeqCst) + ); + std::thread::yield_now(); } assert!( peak.load(Ordering::SeqCst) <= 2, - "no more than 2 workers may hold a permit at once, saw {}", + "no more than 2 jobs may run at once, saw {}", peak.load(Ordering::SeqCst) ); assert_eq!( completed.load(Ordering::SeqCst), 8, - "every worker must complete (no deadlock, permits all returned)" - ); - // All permits returned: a fresh acquire succeeds immediately. - assert!( - sem.acquire_timeout(Duration::from_millis(1)).is_some(), - "all permits should be back after every worker finished" + "every job must complete (no deadlock, workers reused)" ); } - /// C16: a worker that cannot get a permit before its deadline retires - /// instead of parking forever. With zero spare permits and a tiny timeout, - /// `acquire_timeout` returns `None` so the caller degrades the shard. + /// C16: a saturated pool refuses further submissions (so the coordinator + /// degrades that shard) instead of blocking or growing the queue. One + /// worker, queue depth 8: occupy the worker plus fill the 8 queue slots, + /// then prove the next submit is refused. #[test] - fn shard_worker_semaphore_times_out_when_saturated() { - let sem = ShardWorkerSemaphore::new(1); - let _held = sem - .acquire_timeout(Duration::from_millis(1)) - .expect("first acquire takes the only permit"); - // No permit left; a short-deadline acquire must give up (degrade), - // not block indefinitely. - let denied = sem.acquire_timeout(Duration::from_millis(10)); + fn shard_read_pool_refuses_when_saturated() { + let pool = ShardReadPool::new(1); + + // Park the sole worker until released, so nothing drains the queue. + let (started_tx, started_rx) = std::sync::mpsc::channel::<()>(); + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let park: ShardJob = Box::new(move || { + started_tx.send(()).expect("signal worker started"); + release_rx.recv().expect("wait for release"); + }); + assert!(pool.submit(park), "worker accepts the parking job"); + started_rx + .recv_timeout(Duration::from_secs(5)) + .expect("worker started the parking job"); + + // Fill the 8 queue slots (the worker is busy on `park`). + for _ in 0..SHARD_QUEUE_DEPTH_PER_WORKER { + let filler: ShardJob = Box::new(|| {}); + assert!(pool.submit(filler), "queue slot accepts a filler job"); + } + + // Queue full: the next submit is refused so the caller degrades. + let overflow: ShardJob = Box::new(|| {}); assert!( - denied.is_none(), - "saturated semaphore must time out so the worker retires and degrades" + !pool.submit(overflow), + "saturated pool must refuse so the shard is degraded, not block" ); + + // Release the worker so the pool drains cleanly. + release_tx.send(()).expect("release the worker"); + } + + /// A panicking shard job must NOT retire its worker — the persistent pool has + /// no per-query self-heal, so an unguarded panic would permanently shrink it. + /// One worker: panic a job, then prove a later job still runs (worker alive). + #[test] + fn shard_read_pool_survives_panicking_job() { + let pool = ShardReadPool::new(1); + + // Silence the default panic hook for the duration so the deliberately + // panicking job does not spam the test output with a backtrace. + let prev = std::panic::take_hook(); + std::panic::set_hook(Box::new(|_| {})); + + let (panicked_tx, panicked_rx) = std::sync::mpsc::channel::<()>(); + let panic_job: ShardJob = Box::new(move || { + // Signal BEFORE panicking so the test knows the worker reached it. + panicked_tx.send(()).expect("signal pre-panic"); + panic!("deliberate shard-job panic"); + }); + assert!(pool.submit(panic_job), "pool accepts the panicking job"); + panicked_rx + .recv_timeout(Duration::from_secs(5)) + .expect("worker ran the panicking job"); + + std::panic::set_hook(prev); + + // The SAME (sole) worker must still process the next job — if the panic + // had killed it, this would never run and the recv would time out. + let (done_tx, done_rx) = std::sync::mpsc::channel::<()>(); + let after: ShardJob = Box::new(move || { + done_tx.send(()).expect("signal post-panic job ran"); + }); + assert!(pool.submit(after), "pool accepts the post-panic job"); + done_rx + .recv_timeout(Duration::from_secs(5)) + .expect("worker survived the panic and ran the next job"); } } diff --git a/tidal-server/tests/cluster_cross_shard_reads.rs b/tidal-server/tests/cluster_cross_shard_reads.rs new file mode 100644 index 0000000..2f08a2a --- /dev/null +++ b/tidal-server/tests/cluster_cross_shard_reads.rs @@ -0,0 +1,264 @@ +//! Tier-3 cross-shard unified reads suite (m12p4, the m11p6 L4 follow-up; REAL +//! multi-process cluster with PARTIAL shard placement). +//! +//! The m11p6 read fan-out (`scatter_merge`) is corpus-complete ONLY when a node +//! hosts a replica of EVERY shard group (full placement / `S=1`). Under PARTIAL +//! placement a node misses the groups it does not host, so a `/feed` on it would +//! be local-shard-only. m12p4 closes that: the gateway runs its LOCAL scatter, +//! then fans out to the groups it does NOT host (`forward_candidates`, internal +//! `?shard=g` hop) and merges. Two pillars: +//! +//! 1. **Cross-node fan-out makes a partial-node read corpus-complete** — items +//! written across all groups are ALL returned by a `/feed` served from a node +//! that hosts a strict subset of those groups (it reaches the missing groups +//! over HTTP). The same `/feed` on a node hosting a different subset returns +//! the SAME corpus — coverage is placement-independent. +//! +//! 2. **An unreachable missing group degrades, never hard-fails** — kill the +//! sole node hosting a group, and a partial node's `/feed` still returns the +//! items from the reachable groups (a partial, non-empty page) instead of a +//! 5xx — the honest-degraded contract. +//! +//! Run: `cargo test -p tidal-server --features cluster-e2e --test cluster_cross_shard_reads -- --nocapture` + +#![cfg(feature = "cluster-e2e")] +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::cast_possible_truncation, + clippy::cast_precision_loss, + clippy::too_many_lines +)] + +mod support; + +use std::collections::BTreeSet; +use std::time::Duration; + +use support::multiproc::{MultiProcCluster, convergence_budget}; +use tidaldb::{replication::shard::ShardRouter, schema::EntityId}; + +/// Fast election so a single-replica group self-elects promptly and a killed +/// node's failover (pillar 2's reachable groups) completes inside the budget. +const FAST_ELECTION_YAML: &str = "election:\n heartbeat_interval_ms: 100\n election_timeout_min_ms: 500\n election_timeout_max_ms: 1000\n leader_lease_ms: 350"; + +const NODES: usize = 3; +const SHARDS: usize = 3; +const ITEMS: u64 = 60; + +/// Partial placement (m12p4): a "ring" where each group has RF=2 (so the engine +/// builds its WAL ship feed — a single-replica group has no peer and is rejected) +/// but each NODE hosts a STRICT SUBSET (2 of the 3 groups). Every node therefore +/// misses exactly one group and MUST cross-shard fan out to be corpus-complete. +/// +/// * group 0: nodes [0, 1] (leader 0) — node 2 misses it +/// * group 1: nodes [1, 2] (leader 1) — node 0 misses it +/// * group 2: nodes [2, 0] (leader 2) — node 1 misses it +/// +/// So node 0 hosts {0, 2} and misses group 1; node 1 hosts {0, 1} and misses +/// group 2; node 2 hosts {1, 2} and misses group 0. +fn partial_placement() -> Vec> { + vec![vec![0, 1], vec![1, 2], vec![2, 0]] +} + +/// Write `1..=ITEMS` items (each with a `view` signal so it ranks in `for_you`) +/// through `gateway`'s `/items` + `/signals` — the gateway hash-routes each to +/// its owning group's leader (forwarding when the gateway does not host it). All +/// items land on SOME group; the set spans all `SHARDS` groups by the router hash. +fn seed_corpus(cluster: &MultiProcCluster, gateway: usize) -> BTreeSet { + let mut written = BTreeSet::new(); + for e in 1..=ITEMS { + let item = cluster.post( + gateway, + "/items", + &serde_json::json!({ "entity_id": e, "metadata": { "title": format!("item {e}") } }), + ); + assert!( + item.status().is_success(), + "POST /items for entity {e} should route + apply (status {})", + item.status() + ); + let sig = cluster.post( + gateway, + "/signals", + &serde_json::json!({ "entity_id": e, "signal": "view", "weight": 1.0 }), + ); + assert!( + sig.status().is_success(), + "POST /signals for entity {e} should route + apply (status {})", + sig.status() + ); + written.insert(e); + } + written +} + +/// The set of entity ids a `/feed` on `gateway` returned (best-effort: returns +/// the parsed item `entity_id`s; panics if the read itself failed). +fn feed_entities(cluster: &MultiProcCluster, gateway: usize, limit: u32) -> BTreeSet { + let body = cluster.get_json(gateway, &format!("/feed?profile=for_you&limit={limit}")); + body["items"] + .as_array() + .unwrap_or(&Vec::new()) + .iter() + .filter_map(|it| it["entity_id"].as_u64()) + .collect() +} + +/// Which groups the written corpus actually spans (by the gateway router hash) — +/// the test only asserts cross-shard completeness if the corpus genuinely touches +/// a group some partial node does not host. +fn groups_touched(written: &BTreeSet) -> BTreeSet { + let router = ShardRouter::hash(SHARDS as u16).expect("build shard router"); + written + .iter() + .map(|&e| router.route(EntityId::new(e)).0) + .collect() +} + +/// Pillar 1: a `/feed` served from a node hosting a STRICT SUBSET of the groups +/// returns items spanning ALL groups (the cross-node fan-out), and the same read +/// on a differently-placed node returns the SAME corpus. +#[test] +fn mp_partial_placement_feed_spans_all_groups() { + let cluster = MultiProcCluster::start_sharded_partial( + NODES, + &partial_placement(), + Some(FAST_ELECTION_YAML), + ); + let _ = cluster.wait_shard_leaders_agreed_partial(convergence_budget()); + + // Seed the whole corpus through node 0's gateway (it forwards each write to + // the owning group's leader — including group 1, which node 0 does not host). + let written = seed_corpus(&cluster, 0); + let touched = groups_touched(&written); + assert_eq!( + touched, + (0..SHARDS as u16).collect(), + "test corpus must touch EVERY group so each partial node misses real data — got {touched:?}" + ); + + // Let the cross-group write forwards + signal applies settle on every leader. + std::thread::sleep(Duration::from_millis(800)); + + let router = ShardRouter::hash(SHARDS as u16).expect("build shard router"); + + // Node 0 hosts {0, 2} and MISSES group 1 — its group-1 items are reachable + // ONLY by the cross-shard fan-out to a node hosting group 1. + let on_node0 = feed_entities(&cluster, 0, ITEMS as u32); + let group1_items: BTreeSet = written + .iter() + .copied() + .filter(|&e| router.route(EntityId::new(e)).0 == 1) + .collect(); + assert!( + !group1_items.is_empty(), + "precondition: some written items hash to group 1 (node 0's missing group)" + ); + let returned_group1: BTreeSet = on_node0.intersection(&group1_items).copied().collect(); + assert_eq!( + returned_group1, group1_items, + "node 0's /feed must return EVERY group-1 item via the cross-shard fan-out \ + (missing items ⇒ the read was local-shard-only): returned {returned_group1:?} of {group1_items:?}" + ); + // The full corpus is covered (the page holds them all). + assert_eq!( + on_node0, written, + "node 0's cross-shard /feed must cover the WHOLE corpus" + ); + + // Placement-independence: node 1 hosts {0, 1} and MISSES group 2, yet returns + // the SAME whole corpus (its missing group differs from node 0's). + let on_node1 = feed_entities(&cluster, 1, ITEMS as u32); + assert_eq!( + on_node1, written, + "node 1's cross-shard /feed (different missing group) must cover the same whole corpus" + ); +} + +/// Pillar 2: killing the sole node hosting a group degrades a partial node's +/// `/feed` to a partial (non-empty) page — never a hard failure. +#[test] +fn mp_partial_placement_feed_degrades_when_group_unreachable() { + let mut cluster = MultiProcCluster::start_sharded_partial( + NODES, + &partial_placement(), + Some(FAST_ELECTION_YAML), + ); + let _ = cluster.wait_shard_leaders_agreed_partial(convergence_budget()); + + let written = seed_corpus(&cluster, 0); + std::thread::sleep(Duration::from_millis(800)); + + // Sanity: before the kill, node 0's cross-shard feed covers the whole corpus. + let before = feed_entities(&cluster, 0, ITEMS as u32); + assert_eq!(before, written, "pre-kill /feed must be corpus-complete"); + + // Group 1 lives on nodes {1, 2}; node 0 hosts groups {0, 2}. Kill BOTH of + // group 1's replicas → group 1 is unreachable from node 0's fan-out, while + // node 0's LOCAL replicas of groups 0 and 2 still serve reads (a read needs no + // leader). Node 0 is the lone survivor for groups 0 and 2. + cluster.kill_hard(1); + cluster.kill_hard(2); + // Let the connect-fail / breaker surface (the fan-out hop to group 1 errors). + std::thread::sleep(Duration::from_secs(2)); + + // The degraded read MUST still succeed (HTTP 200) and return the reachable + // groups' items — never a 5xx, never an empty page. + let resp = cluster.get(0, &format!("/feed?profile=for_you&limit={ITEMS}")); + assert!( + resp.status().is_success(), + "a missing-group outage must DEGRADE the read, not fail it (status {})", + resp.status() + ); + let body: serde_json::Value = resp.json().expect("feed body is JSON"); + let returned: BTreeSet = body["items"] + .as_array() + .expect("items array") + .iter() + .filter_map(|it| it["entity_id"].as_u64()) + .collect(); + + let router = ShardRouter::hash(SHARDS as u16).expect("build shard router"); + // Groups 0 and 2 are hosted locally on node 0 → reachable. Group 1 → gone. + let reachable: BTreeSet = written + .iter() + .copied() + .filter(|&e| { + let g = router.route(EntityId::new(e)).0; + g == 0 || g == 2 + }) + .collect(); + let group1: BTreeSet = written + .iter() + .copied() + .filter(|&e| router.route(EntityId::new(e)).0 == 1) + .collect(); + + assert!( + !returned.is_empty(), + "degraded /feed must still return the reachable groups' items, not an empty page" + ); + // Every reachable-group item is still present (node 0's local scatter is fine). + assert!( + reachable.is_subset(&returned), + "degraded /feed must still cover groups 0+2 (locally hosted): missing {:?}", + reachable.difference(&returned).collect::>() + ); + // The unreachable group's items are gone — degraded, not magically present. + assert!( + returned.is_disjoint(&group1), + "group 1 is unreachable (both replicas killed), so its items cannot appear" + ); + // m12p4 review fix (CRITICAL): the degradation must be VISIBLE on the wire, + // not server-log-only — the client must be able to tell a partial page from a + // complete one. A degraded read carries a non-empty `unavailable_shards`. + let unavailable = body["unavailable_shards"] + .as_array() + .expect("degraded /feed must surface `unavailable_shards` on the wire (not silent)"); + assert!( + !unavailable.is_empty(), + "a degraded read must NAME the unreachable group(s) so the client knows the page is partial" + ); +} diff --git a/tidal-server/tests/cluster_region.rs b/tidal-server/tests/cluster_region.rs index d887254..2e97cf7 100644 --- a/tidal-server/tests/cluster_region.rs +++ b/tidal-server/tests/cluster_region.rs @@ -781,20 +781,22 @@ fn region_sharded_subset_placement_forwards_and_reads_are_group_scoped() { 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 { + // m12p4 cross-shard unified reads: EVERY node now returns the WHOLE corpus. + // node-b hosts both groups (local scatter covers it); node-a hosts only + // group 0 and node-c only group 1, but each cross-shard fans out to the + // group it does not host, so all three converge to the full 24-id set. + // (Pre-m12p4, node-a returned only `shard0_ids` and node-c only + // `shard1_ids` — the local-shard-only limitation the L4 fan-out closes.) + if a == all && b == all && c == all { break; } assert!( Instant::now() <= deadline, - "subset-placement feeds did not converge (a={}/{}, b={}/24, c={}/{})", + "subset-placement cross-shard feeds did not converge to the whole corpus \ + (a={}/24, b={}/24, c={}/24)", a.len(), - shard0_ids.len(), b.len(), - c.len(), - shard1_ids.len() + c.len() ); std::thread::sleep(Duration::from_millis(50)); } diff --git a/tidal-server/tests/support/multiproc.rs b/tidal-server/tests/support/multiproc.rs index 277f11e..81088a4 100644 --- a/tidal-server/tests/support/multiproc.rs +++ b/tidal-server/tests/support/multiproc.rs @@ -292,6 +292,14 @@ pub struct MultiProcCluster { /// absent); `S` for a [`start_sharded`](MultiProcCluster::start_sharded) /// cluster. Used by the per-shard status helpers. shards: usize, + /// m12p4 partial placement: `placement[g]` = node indices hosting group `g`. + /// `None` for full placement (every node hosts every group). Lets the + /// per-shard leader wait know which node OWNS which group. + partial_placement: Option>>, + /// m12p4: how many shard groups each node hosts (index = node). Empty for the + /// full-placement harnesses (every node hosts `shards`). The partial-aware + /// leader wait uses this to size each node's expected `shards[]` row count. + host_group_counts: Vec, /// Held so the tempdir (configs + per-node data dirs) outlives every process. _tmp: tempfile::TempDir, } @@ -371,6 +379,8 @@ impl MultiProcCluster { .build() .expect("build blocking client"), shards: 1, + partial_placement: None, + host_group_counts: Vec::new(), _tmp: tmp, }; @@ -459,6 +469,8 @@ impl MultiProcCluster { .build() .expect("build blocking client"), shards, + partial_placement: None, + host_group_counts: Vec::new(), _tmp: tmp, }; @@ -481,6 +493,115 @@ impl MultiProcCluster { harness } + /// Spawn an `nodes`-process cluster of `placement.len()` shard groups at the + /// EXPLICIT, possibly PARTIAL placement `placement` (m12p4 cross-shard read + /// gate): `placement[g]` is the list of node indices that replicate group `g`, + /// with `placement[g][0]` its term-0 preferred leader. A node that appears in + /// some-but-not-all groups hosts a STRICT SUBSET of the corpus — exactly the + /// shape the L4 cross-shard read fan-out exists for. + /// + /// Unlike [`start_sharded`](Self::start_sharded) (full placement, every node a + /// replica of every group), here each node binds a gRPC port ONLY for the + /// groups it actually hosts. The gateway on a partial node must fan a + /// corpus-wide `/feed` out to the groups it does not host to be complete. + /// + /// # Panics + /// + /// Panics if `placement` is empty, any group lists no replica, a group's + /// leader is not its first replica's owner, an index is out of range, files + /// cannot be written, or any process is not healthy within [`boot_budget`]. + #[must_use] + pub fn start_sharded_partial( + nodes: usize, + placement: &[Vec], + topology_extra: Option<&str>, + ) -> Self { + assert!(nodes >= 2, "need at least 2 nodes for a cluster"); + assert!(!placement.is_empty(), "need at least 1 shard group"); + let shards = placement.len(); + for (g, replicas) in placement.iter().enumerate() { + assert!(!replicas.is_empty(), "group {g} lists no replica"); + for &n in replicas { + assert!(n < nodes, "group {g} names node {n} but only {nodes} nodes"); + } + } + let spawn_guard = spawn_lock() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + + let tmp = tempfile::tempdir().expect("create temp dir"); + let bin = tidal_server_bin(); + + // Each node allocates one gRPC port PER GROUP IT HOSTS (not per group in + // the cluster). `shard_grpc[g]` is bound only for hosted groups; a group + // the node does not host gets a placeholder it never binds (its port is + // never written into a `replicas:` entry, so nothing dials it). + let host_groups: Vec> = (0..nodes) + .map(|n| (0..shards).filter(|g| placement[*g].contains(&n)).collect()) + .collect(); + let plans: Vec = (0..nodes) + .map(|i| { + let data_dir = tmp.path().join(format!("region-{i}")); + std::fs::create_dir_all(&data_dir).expect("create per-node data dir"); + // One real port per group; unhosted slots stay allocated but + // unused (keeps `shard_grpc[g]` index-stable across all groups). + let shard_grpc: Vec = (0..shards).map(|_| free_addr()).collect(); + RegionPlan { + name: region_name(i), + grpc: shard_grpc[0], + http: free_addr(), + data_dir, + shard_grpc, + } + }) + .collect(); + + let schema_path = write_schema(tmp.path()); + let topology_paths: Vec = (0..nodes) + .map(|i| { + write_partial_sharded_topology_for(tmp.path(), &plans, i, placement, topology_extra) + }) + .collect(); + + let harness = Self { + plans, + nodes: Vec::new(), + schema_path, + topology_paths, + rewrite: identity_rewrite(), + log: "warn".into(), + extra_env: HashMap::new(), + client: reqwest::blocking::Client::builder() + .build() + .expect("build blocking client"), + shards, + partial_placement: Some(placement.to_vec()), + // Per-node hosted-group counts, so the partial-aware leader wait knows + // how many `shards[]` rows each node should report. + host_group_counts: host_groups.iter().map(Vec::len).collect(), + _tmp: tmp, + }; + let mut harness = harness; + + for i in 0..harness.plans.len() { + let child = harness.spawn_process(&bin, i, &[]); + let plan = &harness.plans[i]; + harness.nodes.push(NodeHandle { + name: plan.name.clone(), + http: plan.http, + process: Some(child), + }); + } + + std::thread::sleep(SETTLE); + let deadline = Instant::now() + boot_budget(); + for i in 0..harness.nodes.len() { + harness.wait_health_inner(i, deadline); + } + drop(spawn_guard); + harness + } + /// Spawn one region's process. `extra` is appended to the per-node env /// (used by `restart` to inject overrides). The binary, schema, and this /// region's topology file are stable across restarts. @@ -1135,6 +1256,73 @@ impl MultiProcCluster { } } + /// m12p4 PARTIAL-placement analogue of [`agreed_shard_leaders`]. Each node + /// reports only the `shards[]` rows for the groups IT hosts (a node-local + /// count from `host_group_counts`), and a group is agreed once EVERY node that + /// hosts it reports the SAME (non-null) leader. A node hosting a strict subset + /// no longer trips the full-placement `rows.len() == self.shards` guard. + /// + /// # Panics + /// + /// Panics if called on a non-partial harness (use [`agreed_shard_leaders`]). + #[must_use] + fn agreed_shard_leaders_partial(&self) -> Option> { + let placement = self + .partial_placement + .as_ref() + .expect("agreed_shard_leaders_partial needs a partial-placement harness"); + let live: Vec = self.live_indices(); + if live.is_empty() { + return None; + } + let mut per_shard: HashMap = HashMap::new(); + for &idx in &live { + let st = self.local_status(idx)?; + let rows = st["shards"].as_array()?; + // The node must have opened EVERY group it hosts (else it is still + // booting — wait, do not mis-read as agreed). + if rows.len() != self.host_group_counts[idx] { + return None; + } + for row in rows { + let shard = u16::try_from(row["shard"].as_u64()?).ok()?; + let leader = row["leader"].as_str()?.to_string(); + match per_shard.get(&shard) { + Some(seen) if seen != &leader => return None, + Some(_) => {} + None => { + per_shard.insert(shard, leader); + } + } + } + } + // Agreed only when every group has a reported leader (some node hosts it). + (per_shard.len() == placement.len()).then_some(per_shard) + } + + /// Block until every group's hosting nodes agree on a leader, under PARTIAL + /// placement. The m12p4 analogue of [`wait_shard_leaders_agreed`]. + /// + /// # Panics + /// + /// Panics on timeout (dumping the last partial view) or on a non-partial + /// harness. + #[must_use] + pub fn wait_shard_leaders_agreed_partial(&self, timeout: Duration) -> HashMap { + let deadline = Instant::now() + timeout; + loop { + if let Some(map) = self.agreed_shard_leaders_partial() { + return map; + } + assert!( + Instant::now() <= deadline, + "partial shard leaders did not converge within {timeout:?}; last view: {:?}", + self.agreed_shard_leaders_partial() + ); + std::thread::sleep(POLL_INTERVAL); + } + } + /// Best-effort hint for a failing node, to make boot failures diagnosable in /// the panic message. Node logs go to /dev/null by default (re-run with /// `TIDAL_TEST_NODE_LOGS=inherit` to see them), so we surface the bind @@ -1310,6 +1498,53 @@ fn write_sharded_topology_for( path } +/// Write the PARTIAL-placement sharded topology file process `idx` loads (m12p4): +/// a `regions:` list (every node) plus a `shards:` block where each group `g` +/// declares ONLY the replica entries in `placement[g]` (a subset of nodes), led +/// (term-0) by `placement[g][0]`. A node absent from `placement[g]` never appears +/// in group `g`'s `replicas:`, so it does not host the group — its gateway must +/// cross-shard fan out to reach `g`. Identity addressing, so every process's file +/// is byte-identical (SIGKILL not TCP-proxy, no per-observer rewrite). +fn write_partial_sharded_topology_for( + dir: &Path, + plans: &[RegionPlan], + idx: usize, + placement: &[Vec], + extra: Option<&str>, +) -> PathBuf { + let path = dir.join(format!("topology-{idx}.yaml")); + let mut body = String::from("regions:\n"); + for plan in plans { + let _ = writeln!(body, " - name: {}", plan.name); + let _ = writeln!(body, " grpc_addr: \"{}\"", plan.shard_grpc[0]); + let _ = writeln!(body, " http_addr: \"{}\"", plan.http); + } + let _ = writeln!(body, "shards:"); + for (g, replicas) in placement.iter().enumerate() { + assert!(!replicas.is_empty(), "group {g} lists no replica"); + let _ = writeln!(body, " - id: {g}"); + // Term-0 leader = the group's first listed replica node. + let _ = writeln!(body, " leader: {}", plans[replicas[0]].name); + let _ = writeln!(body, " replicas:"); + for &node in replicas { + let _ = writeln!( + body, + " - {{ node: {}, grpc_addr: \"{}\" }}", + plans[node].name, plans[node].shard_grpc[g] + ); + } + } + // Legacy field — required by the loader, ignored once `shards:` is present. + let _ = writeln!(body, "leader: {}", plans[0].name); + if let Some(extra) = extra { + let _ = writeln!(body, "{extra}"); + } + let mut f = std::fs::File::create(&path).expect("create partial sharded topology file"); + f.write_all(body.as_bytes()) + .expect("write partial sharded topology file"); + path +} + /// Write the shared schema file. Matches the signal set the in-process route /// tests use (`view` decayed, `like` decayed, `hide` permanent for `/hardnegs`) /// so the harness asserts against identical engine behavior, plus a `title` text diff --git a/tidal-stress/k8s/stress-job-t5.yaml b/tidal-stress/k8s/stress-job-t5.yaml new file mode 100644 index 0000000..b537dc0 --- /dev/null +++ b/tidal-stress/k8s/stress-job-t5.yaml @@ -0,0 +1,145 @@ +# T5 — sharded quorum-write throughput at scale (m12p4 / G3). +# +# Goal: prove the 3-group × RF=3 topology scales WRITES ~S× over a single group. +# With `shards:` enabled (k8s/cluster/topology-configmap.yaml) the regular +# `/signals`//`/items`//`/embeddings` surface hash-routes each entity's write to +# its OWNING group's leader, so the three balanced leaders (tidaldb-0/1/2 lead +# shards 0/1/2) absorb writes in parallel instead of one leader funnelling all. +# +# TWO generators. The m11p6 knee was the GENERATOR's in-flight cap (~4k rps), +# not the engine (~30% CPU at 3k/s). A single generator cannot saturate three +# parallel leaders, so this Job runs `parallelism: 2` (completionMode: Indexed): +# two generator pods drive the ramp concurrently and their throughput sums. +# Both spread writes across all three `--target`s (no `--leader-url` pin), so +# every pod gateways AND leads one group — no single forwarding bottleneck. +# +# Aggregate the two pods' `ok_per_sec` (each logs its own JSON summary) for the +# cluster total. Gate: aggregate ≥ 5,000 quorum signal writes/s AND ≥ 2.5× the +# single-group baseline (stress-job-m11p6-baseline.yaml), zero acked loss per +# group, 0% error. The single-generator absolute on constrained hardware may sit +# below 5k; the SCALING RATIO (3-group vs single-group) is the architecture +# proof and is hardware-independent. +# +# TLS: the :9500 plane serves a private-CA cert (tidaldb-cluster-tls). Each +# generator trusts it via the mounted ca.crt (--ca-cert) — verified TLS, never +# --insecure. Targets are the pod-DNS SANs (https://), never pod IPs. +# +# Apply: kubectl apply -f tidal-stress/k8s/stress-job-t5.yaml +# Watch: kubectl logs -f job/tidal-stress-t5 -n tidaldb-cluster --all-containers --prefix +# Sum: grep '"ok_per_sec"' across both pods' --json-summary outputs +# Rearm: kubectl delete job tidal-stress-t5 -n tidaldb-cluster +apiVersion: batch/v1 +kind: Job +metadata: + name: tidal-stress-t5 + namespace: tidaldb-cluster + labels: + app.kubernetes.io/name: tidal-stress + app.kubernetes.io/part-of: tidaldb +spec: + backoffLimit: 0 + ttlSecondsAfterFinished: 7200 + # Two generator pods, both running to completion — the "second stress + # generator" m12p4 calls for. Indexed so each pod has a stable + # JOB_COMPLETION_INDEX (0/1) in its logs for per-generator attribution. + completions: 2 + parallelism: 2 + completionMode: Indexed + template: + metadata: + labels: + app.kubernetes.io/name: tidal-stress + app.kubernetes.io/part-of: tidaldb + spec: + restartPolicy: Never + automountServiceAccountToken: false + # Spread the two generators onto distinct nodes so neither generator pod + # competes with the other for CPU — each must be free to drive its full + # in-flight budget (the m11p6 knee was generator CPU/in-flight, not engine). + topologySpreadConstraints: + - maxSkew: 1 + topologyKey: kubernetes.io/hostname + whenUnsatisfiable: ScheduleAnyway + labelSelector: + matchLabels: + app.kubernetes.io/name: tidal-stress + securityContext: + runAsNonRoot: true + runAsUser: 1000 + runAsGroup: 1000 + seccompProfile: + type: RuntimeDefault + containers: + - name: stress + image: registry.threesix.ai/tidal/stress@sha256:e130aa871f5df17a14a9e13e7df606c602b95a03d8eba49490ff7481e6e2b2b3 # m11-44b768b (TLS-aware) + imagePullPolicy: IfNotPresent + args: + # Spread writes across all three pods (each gateways + leads one + # group); NO --leader-url pin, so the forwarding load is balanced. + - --target + - https://tidaldb-0.tidaldb-peers.tidaldb-cluster.svc.cluster.local:9500 + - --target + - https://tidaldb-1.tidaldb-peers.tidaldb-cluster.svc.cluster.local:9500 + - --target + - https://tidaldb-2.tidaldb-peers.tidaldb-cluster.svc.cluster.local:9500 + - --ca-cert + - /etc/tidaldb/tls/ca.crt + - --ack + - quorum + - --ramp + - peach-100k + - --stage-secs + - "120" + - --mix + - writes + - --corpus + - "20000" + - --users + - "100000" + # Each generator drives its own in-flight budget; two pods double the + # aggregate offered load over a single generator. + - --max-inflight + - "5000" + - --poll-status + # m11p9 machine-readable gates: a per-generator JSON summary plus + # hard ceilings. 0% error + no knee are the per-group zero-acked-loss + # / SLO bar; aggregate the two pods' ok_per_sec for the 5k/2.5x gate. + - --json-summary + - /tmp/t5-summary.json + - --max-error-pct + - "0" + env: + - name: TIDAL_API_KEY + valueFrom: + secretKeyRef: + name: tidaldb-credentials + key: TIDAL_API_KEY + - name: TIDAL_STRESS_LOG + value: warn + resources: + requests: + cpu: 500m + memory: 256Mi + limits: + cpu: "3" + memory: 1Gi + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + capabilities: + drop: ["ALL"] + volumeMounts: + - name: cluster-tls + mountPath: /etc/tidaldb/tls + readOnly: true + - name: tmp + mountPath: /tmp + volumes: + - name: cluster-tls + secret: + secretName: tidaldb-cluster-tls + items: + - key: ca.crt + path: ca.crt + - name: tmp + emptyDir: {}