From 8aa1fbb4140ac720b360465073388fe5ad0f84a7 Mon Sep 17 00:00:00 2001 From: jordan Date: Sun, 30 Aug 2026 13:57:36 -0600 Subject: [PATCH] vector search: normalize the query, instrument the blob path, expose per-group vector counts Three real defects, plus a retracted fourth that was a probe artifact. P3 (fixed) - query/stored normalization asymmetry. The write path L2-normalized every stored vector; the read path passed the caller's raw query straight to the index, so the two sides lived in different spaces. With unit v, d = |q|^2 - 2q.v + 1, so a non-unit query shifted and scaled every distance by |q|^2. Measured live: 591-1174 against a documented [0,4], and an exact match scoring |q|^2 - 1 instead of ~0. vector_search_items now normalizes with the canonical l2_normalize; a zero-norm query (no direction, so nearest-by-cosine is undefined) is rejected with 400. Ranking is unchanged - |q|^2 and 1 are constant across candidates - which is why it went unnoticed; what broke was every absolute use of the number. WIRE-VISIBLE, recorded in CHANGELOG. P2 (fixed) - the blob path had zero instrumentation. Added per-kind tidaldb_cluster_blobs_originated/applied/apply_failed totals. Label cardinality is fixed at 4 by construction via a new BlobKind enum, and BlobRecord::blob_kind is now the ONE exhaustive match over the variants (kind() derives from it), so a new variant is a compile error in one place instead of a silent zero in three. Only the live apply path is counted - boot replay would inflate applied past originated on every restart. Coverage gap (fixed) - tidaldb_usearch_vector_count rendered only the metrics owner's shard group, so on a 3-group node two thirds of the corpus had no vector-count series at all. Co-located groups now render shard="N"; the owner stays unlabeled for wire compatibility, so an alert grouped by (shard) buckets each replica set separately without double-counting. P1 (RETRACTED) - the "replica-divergent vector index" does not exist. Every probe wrote through the /sharded/ surface, which hash-partitions and applies to the owning region's local store with no WAL append, and therefore does not replicate BY DESIGN (cluster/node.rs:8828-8829). A controlled A/B settled it: on /items plus /embeddings all 6 entities reach all 3 replicas; on the sharded surface four of six reach exactly one node. Both are now pinned by tests. See tmp/vector-search-correctness/diagnosis.md and the k3s-fleet cluster-state.yaml entry RETRACTED_blob_replication_rf1_2026_08_30. Pre-work: usearch_index.rs 872 to 503 lines by extracting its tests to a sibling (the project's existing path-attribute convention), and the three hand-rolled l2_normalize copies collapsed to one. The two entity copies used a zero threshold about 2900x looser than the canonical one; normalize_centroid now names the centroid zero-tolerance policy once, and a test pins the tightened behavior. Tests: 2107 lib (+5), 8 vector_search e2e (+4, three of which fail without the P3 fix), 4 cluster_sharding e2e (+2). The heavy multiproc tests in cluster_sharding are now serialized - four concurrent 3-node clusters made the pre-existing failover test miss its 10s budget. --- CHANGELOG.md | 45 +++ CODING_GUIDELINES.md | 17 +- docs/runbooks/cluster.md | 118 ++++++ tidal-server/tests/cluster_sharding.rs | 265 ++++++++++++ tidal-server/tests/vector_search.rs | 140 +++++++ tidal/src/db/items.rs | 126 +++++- tidal/src/db/metrics/cluster.rs | 196 +++++++++ tidal/src/db/metrics/mod.rs | 47 +++ tidal/src/db/metrics/tests.rs | 42 ++ tidal/src/db/mod.rs | 15 +- tidal/src/db/query_ops.rs | 27 +- tidal/src/entities/multi_preference.rs | 27 +- tidal/src/entities/preference.rs | 84 +++- tidal/src/storage/vector/lifecycle/mod.rs | 2 +- .../src/storage/vector/lifecycle/normalize.rs | 57 ++- tidal/src/storage/vector/mod.rs | 14 +- tidal/src/storage/vector/usearch_index.rs | 378 +----------------- .../src/storage/vector/usearch_index_tests.rs | 367 +++++++++++++++++ tidal/src/wal/format/batch.rs | 90 ++++- 19 files changed, 1630 insertions(+), 427 deletions(-) create mode 100644 tidal/src/storage/vector/usearch_index_tests.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index a4cf644..ceacf2d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,51 @@ change to a public API or a persisted format ships with a documented migration path in this file. This supersedes the `0.1.0` "no stability guarantees" note below. +### Fixed + +**`vector_search` distances now honor the documented `[0.0, 4.0]` contract (wire-visible)** + +The read path passed the caller's raw query vector straight to the index while the +write path L2-normalized every stored vector, so the two sides lived in different +spaces. With unit `v`, `d = |q|² − 2q·v + 1`: a non-unit query shifted and scaled +every distance by `|q|²`. Measured on a live RF3 cluster, `/vector_search` returned +distances of **591–1174** against a contract promising `≤ 4`, and an exact match +scored `|q|² − 1` instead of ~0. + +`db::query_ops::vector_search_items` now normalizes the query with the canonical +`storage::vector::l2_normalize` before searching. A zero-norm query — which has no +direction, so "nearest by cosine" is undefined — is rejected with **400** instead of +returning an arbitrary ranking. + +**Ranking is unchanged.** `|q|²` and `1` are constant across candidates, so the +ordering was already correct cosine order; that is why this went unnoticed. Absolute +uses of the number were the casualties: distance thresholding, dedup-by-distance, +cross-query comparison, and recall measurement. + +*Migration:* any client comparing distances against a hard-coded threshold derived +from the old unnormalized values must re-derive it. An exact match now scores ~0 +(`2.4e-7` measured; the f16 quantization floor is ~`1e-3`, so compare with a +tolerance, never equality) and all distances fall in `[0.0, 4.0]`. + +**Blob replication is now observable (`tidaldb_cluster_blobs_*`)** + +The blob path — item metadata, embeddings, term markers, cluster membership — had +**zero** instrumentation: `lag_events` counts WAL apply, so a blob that never shipped +or never applied moved no number anywhere. Added per-kind counters +`tidaldb_cluster_blobs_originated_total`, `_applied_total` and `_apply_failed_total`, +labelled by record kind with cardinality fixed at 4 by construction. Comparing +`originated` on the writing node against `applied` on each peer localises a +replication gap to enqueue, ship, or apply in one query. + +**Every co-located shard group now reports its vector count** + +`tidaldb_usearch_vector_count` was rendered only for the node's metrics-owner group, +so on a 3-group node two thirds of the corpus had no vector-count series and could +diverge unobserved. Co-located groups now render +`tidaldb_usearch_vector_count{shard="N"}`; the owner's series stays unlabeled for wire +compatibility, so an alert grouped `by (shard)` buckets each replica set separately +without double-counting. + ### Added **Multi-vector user preference modeling + ANN candidate-gen (M12) — a warm user is many interests, not one averaged vector: per-user preference clusters drive a top-M ANN fan-out in `for_you`** diff --git a/CODING_GUIDELINES.md b/CODING_GUIDELINES.md index 5207d80..71614be 100644 --- a/CODING_GUIDELINES.md +++ b/CODING_GUIDELINES.md @@ -122,7 +122,22 @@ Do not build HNSW from scratch. USearch provides 126K+ QPS, predicate callbacks ### Normalize embeddings at insertion time -For cosine similarity, normalize vectors to unit length and use L2 distance (equivalent for unit vectors, more SIMD-friendly). Store normalized vectors — never re-normalize at query time. +For cosine similarity, normalize vectors to unit length and use L2 distance (equivalent for unit vectors, more SIMD-friendly). + +Both sides of the comparison must live in the same space, and each is normalized exactly once: + +- **Stored vectors** are normalized at insertion (`storage::vector::lifecycle::ops`) and **never re-normalized afterwards** — not on read, not on rebuild, not per query. Re-deriving a norm you already have, across millions of stored vectors, is pure waste; that is what this rule forbids. +- **Query vectors** are normalized on the read path (`db::query_ops::vector_search_items`), because a caller sends raw model output. This is not "re-normalizing" — the query has never been normalized before. + +Normalizing the query is what makes the distance contract in `storage/vector/mod.rs:49` true. With unit `v`, `d = |q|² − 2q·v + 1`, so a non-unit query shifts and scales every distance by `|q|²`: the documented `[0.0, 4.0]` range becomes unsatisfiable and an exact match scores `|q|² − 1` instead of ~0. Measured on the RF3 cluster 2026-08-30, a non-unit query reported distances of 591–1174 against a contract promising `≤ 4`; a unit query on the same corpus scored an exact match at `2.4e-7`. + +Ranking is unaffected by the omission (`|q|²` and `1` are constant across candidates, so ordering is still cosine order), which is exactly why it went unnoticed for so long. What breaks is every *absolute* use of the number: thresholding, dedup-by-distance, cross-query comparison, and recall measurement. + +Under f16 quantization an exact match lands near `1e-3`, not exactly `0` — assert a tolerance, never equality. + +A zero-norm query has no direction, so "nearest by cosine" is undefined; reject it as caller error (400) rather than returning an arbitrary ranking. + +There is exactly **one** implementation of this arithmetic — `storage::vector::l2_normalize` and its in-place twin `l2_normalize_in_place`. Do not hand-roll another: three copies with three different zero thresholds are how the two directions drifted apart in the first place. ### Adaptive filtered search diff --git a/docs/runbooks/cluster.md b/docs/runbooks/cluster.md index a55b560..2ccfd56 100644 --- a/docs/runbooks/cluster.md +++ b/docs/runbooks/cluster.md @@ -1543,6 +1543,124 @@ kill-points, then a local gated soak. It is not the Ref-A calendar gate unless The guarantee→test map is [docs/planning/milestone-11/guarantee-traceability.md](../planning/milestone-11/guarantee-traceability.md). +## 16. Vector index divergence + +Target of the `runbook_url` on `TidalDBClusterVectorIndexDiverged`, +`TidalDBClusterBlobApplyFailing`, and `TidalDBClusterPeerShipFailing`. + +**Symptom.** Replicas of one shard group answer the same vector query differently: an +entity is the top hit on one replica and absent on the others. Health surfaces stay +green throughout — `lag_events` counts WAL apply, not blob apply, so a blob that never +lands moves no lag number. + +### 16.0 First: rule out the two benign causes + +Both look exactly like a replication defect. On 2026-08-30 the first one consumed several +rounds of investigation and produced a retracted durability incident report. + +1. **The write used `/sharded/*`.** That surface hash-partitions and applies to the + owning region's **local store with no WAL append**, so it **does not replicate — by + design** (`tidal-server/src/cluster/node.rs:8828-8829`; + `sharded_write_embedding` → `ShardReplica::apply_embedding_local`). An entity written + that way living on exactly one node is the contract, not a fault. Only the + **non-sharded** surface (`/items`, `/embeddings`, `/signals`) rides the leader WAL + relay and replicates. + *Check:* re-probe with `/embeddings` and see whether parity holds. It settles in + seconds what measuring the sharded path more precisely never will. +2. **A rolling deploy is in flight.** A restarted pod rebuilds its index from store and + legitimately reads a different count until it finishes. This is why the alert carries + `for: 15m`. + +Both are covered by tests: `mp_embedding_is_searchable_on_every_replica_without_restart` +and `mp_sharded_surface_writes_are_local_to_the_owner` +(`tidal-server/tests/cluster_sharding.rs`). + +### DO NOT restart a pod first + +`rebuild_from_store` runs at open and re-derives the whole index from the durable +store, which **converges the counts and destroys the evidence**. Capture every number +below *before* touching a pod. The restart is a diagnostic step with a specific +meaning (§16.3), not a remedy. + +### 16.1 Read the counts per shard group + +`tidaldb_usearch_vector_count` is per shard **group**, not per node: the node's +metrics-owner group renders unlabeled and each co-located group renders `shard="N"`. +The owner is deterministically the same group on every node, so compare within a +label set, never across: + +```bash +kubectl -n observability port-forward svc/vmsingle 18428:8428 & +curl -s --get --data-urlencode \ + 'query=tidaldb_usearch_vector_count{namespace="tidaldb-cluster"}' \ + localhost:18428/prometheus/api/v1/query | jq -r \ + '.data.result[] | "\(.metric.pod) shard=\(.metric.shard // "0(owner)") \(.value[1])"' +``` + +Replicas of one group MUST agree. A spread that persists past a rolling deploy is real. + +### 16.2 Localise it with the blob ledger + +```bash +# on the node that ACCEPTED the write: +# tidaldb_cluster_blobs_originated_total{kind="embedding"} +# on every peer: +# tidaldb_cluster_blobs_applied_total{kind="embedding"} +# tidaldb_cluster_blobs_apply_failed_total{kind="embedding"} +``` + +With RF3 a healthy cluster shows `applied ≈ (RF-1) × originated` in aggregate — one +apply per follower. Read it per (pod, kind), not as a fleet-wide subtraction. + +| reading | meaning | next step | +| --- | --- | --- | +| `apply_failed` > 0 | the receiver rejected a record and **halted** that stream rather than skipping it | read that pod's logs for the apply error; the stream is stalled, not silently lossy | +| peer's `applied` never advances while the writer's `originated` does | the record is not arriving — enqueue or ship | §16.4 | +| `applied` advances but that group's `usearch_vector_count` lags | it arrived and was not indexed | index-gap; the store is intact, a restart recovers it | + +### 16.3 The decisive test (only after 16.1 and 16.2 are captured) + +Confirm `kubectl -n tidaldb-cluster get pdb tidaldb -o jsonpath='{.status.disruptionsAllowed}'` +is `1`, run a quorum-write probe (`POST /sharded/items` → expect 201), then restart +**one follower** that lacks the entity — never the leader. + +- entity becomes retrievable after the rebuild ⇒ its **durable store had it**, only the + live index was missing it. Data was safe. +- entity still absent ⇒ its **store never had it**. Before calling this a durability + event, re-check §16.0: for a `/sharded/*` write this is the intended outcome and the + store was never supposed to hold it. If the write went through the **non-sharded** + surface and the entity is still absent after a rebuild, that *is* a durability event — + the entity exists on fewer replicas than the replication factor claims. Record the + affected count and escalate. + +Re-probe the quorum write afterwards and confirm 201 again. + +### 16.4 Check the ship path + +```bash +kubectl -n tidaldb-cluster logs tidaldb- -c tidaldb | grep 'ship sender' +``` + +`batch ship failing; retrying every 100ms … consecutive_failures=N` means the leader +cannot reach that peer. **`tidaldb_cluster_peer_breaker_state` is not a reliable +signal here** — on 2026-08-30 it read `0` (closed/healthy) through 2697 consecutive +failures. Trust `tidaldb_cluster_peer_ship_failures_total`'s rate, which +`TidalDBClusterPeerShipFailing` now alerts on. + +Also note `GET /cluster/status`'s `regions[]` block can report peers +`partitioned: true, reachable: false` with full `lag_events` while the same response's +`shards[]` and `peer_acked_seqno` show complete convergence. **The `shards[]` block is +authoritative**; `regions[]` is a stale legacy view. + +### 16.5 Expected convergence window + +A write is not instantly visible on every replica, and that is not this bug. Ship and +apply normally complete in well under a second on an in-cluster (low-RTT) deployment. +"Still converging" vs "diverged" is decided by whether `applied` on the lagging peer is +**advancing**: if it is, wait; if it is flat while the writer's `originated` climbs, it +is the failure in §16.2. The alert's `for: 15m` exists so a rolling deploy's legitimate +rebuild window never pages. + ## Cross-references - **Kubernetes deployment** — [docs/runbooks/kubernetes.md](kubernetes.md) diff --git a/tidal-server/tests/cluster_sharding.rs b/tidal-server/tests/cluster_sharding.rs index c365222..ef24b8c 100644 --- a/tidal-server/tests/cluster_sharding.rs +++ b/tidal-server/tests/cluster_sharding.rs @@ -70,6 +70,25 @@ fn killpoints() -> usize { .unwrap_or(3) } +/// Serializes the heavy multi-process tests in THIS target. +/// +/// Each test here spawns 3 OS processes hosting 3 shard groups each. The harness's +/// `spawn_lock` only serializes the spawn itself and is released as soon as +/// `start_sharded` returns, so without this every test in the file can have a live +/// cluster simultaneously — 4 clusters, 12 processes, all electing and shipping at +/// once. That contention makes `mp_sharded_kill_node_moves_only_its_leaderships_zero_loss` +/// miss its 10s failover budget: it passes alone and fails in the full target. +/// +/// Held for the whole test body, so exactly one cluster is alive at a time. Poison is +/// recovered rather than propagated: one failing test must not cascade into "the rest +/// panicked on a poisoned lock", which hides the original failure. +fn heavy_test_guard() -> std::sync::MutexGuard<'static, ()> { + static LOCK: std::sync::OnceLock> = std::sync::OnceLock::new(); + LOCK.get_or_init(|| std::sync::Mutex::new(())) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + /// Find the node index whose region name matches `name`. fn idx_of(cluster: &MultiProcCluster, name: &str) -> usize { (0..cluster.len()) @@ -82,6 +101,7 @@ fn idx_of(cluster: &MultiProcCluster, name: &str) -> usize { /// never stop, and zero acknowledged writes are lost — across several kill points. #[test] fn mp_sharded_kill_node_moves_only_its_leaderships_zero_loss() { + let _heavy = heavy_test_guard(); let mut cluster = MultiProcCluster::start_sharded(NODES, SHARDS, Some(FAST_ELECTION_YAML)); // The gateway's entity→shard hash (the same FNV-1a router every node routes by). let router = ShardRouter::hash(SHARDS as u16).expect("build shard router"); @@ -305,6 +325,7 @@ fn mp_sharded_kill_node_moves_only_its_leaderships_zero_loss() { /// and the per-group reuse of the m11p4/m11p5 machinery end to end. #[test] fn mp_sharded_rebalance_verbs_move_one_group() { + let _heavy = heavy_test_guard(); let cluster = MultiProcCluster::start_sharded(NODES, SHARDS, Some(FAST_ELECTION_YAML)); let before = cluster.wait_shard_leaders_agreed(convergence_budget()); for s in 0..SHARDS as u16 { @@ -454,3 +475,247 @@ fn wait_until(budget: Duration, mut cond: impl FnMut() -> bool) -> bool { std::thread::sleep(Duration::from_millis(100)); } } + +/// An embedding written through ANY node must be searchable on EVERY replica of +/// its shard group, WITHOUT a restart. +/// +/// Guards the replication contract of the NON-sharded write surface (`/items`, +/// `/embeddings`): those ride the leader WAL relay, so every replica of the entity's +/// group must end up able to answer for it. +/// +/// Uses `/items` + `/embeddings` DELIBERATELY, not `/sharded/*`. The `/sharded/*` +/// surface hash-partitions and applies to the owner's LOCAL store without a WAL +/// append (`node.rs:8828-8829`, `ShardReplica::apply_embedding_local`), so it is +/// single-copy BY DESIGN and a parity assertion against it fails correctly — see +/// `mp_sharded_surface_writes_are_local_to_the_owner` below, which pins that +/// intended behavior. Writing a parity test against the wrong surface is exactly +/// the mistake that made a 2026-08-30 live probe look like a durability incident. +/// +/// **The assertion is deliberately PRE-restart.** `rebuild_from_store` at open +/// re-derives the whole index from the durable store, so a test that restarts before +/// asserting converges the replicas regardless of whether the live path works — it +/// would have PASSED against the bug it exists to catch. That is exactly how the +/// live divergence stayed hidden across a rolling deploy. +/// +/// Full placement (3 groups on every node) means `/vector_search` scatters over all +/// locally hosted groups and is corpus-complete per node, so "absent here" is a real +/// absence and not a fan-out artifact. +#[test] +fn mp_embedding_is_searchable_on_every_replica_without_restart() { + let _heavy = heavy_test_guard(); + let cluster = MultiProcCluster::start_sharded(NODES, SHARDS, Some(FAST_ELECTION_YAML)); + let leaders = cluster.wait_shard_leaders_agreed(convergence_budget()); + println!("[parity] group leaders: {leaders:?}"); + + // Entity ids are hash-routed across groups, so a spread covers all three and + // the test does not depend on which group any single id lands in. + const ENTITIES: [u64; 6] = [911_001, 911_002, 911_003, 911_004, 911_005, 911_006]; + + // Write each entity through a DIFFERENT node, round-robin. The live failure was + // sensitive to which node accepted the write (it never landed on the acceptor), + // so exercising every entry point is the point. + for (n, entity) in ENTITIES.iter().enumerate() { + let via = n % NODES; + let resp = cluster.post( + via, + "/items", + &serde_json::json!({ "entity_id": entity, "metadata": { "title": "parity" } }), + ); + assert_eq!( + resp.status().as_u16(), + 201, + "entity {entity} via node {via}: /sharded/items must 201" + ); + let resp = cluster.post( + via, + "/embeddings", + &serde_json::json!({ "entity_id": entity, "values": embedding_for(*entity) }), + ); + assert_eq!( + resp.status().as_u16(), + 204, + "entity {entity} via node {via}: /sharded/embeddings must 204" + ); + } + + cluster.wait_converged_all(convergence_budget()); + + // Poll every node for every entity, bounded by the convergence budget. Replication + // is asynchronous, so a brief absence is legitimate; a PERSISTENT one is the bug. + let deadline = Instant::now() + convergence_budget(); + let mut missing: Vec<(usize, u64)> = Vec::new(); + loop { + missing.clear(); + for entity in ENTITIES { + for node in 0..NODES { + if !vector_search_finds(&cluster, node, entity) { + missing.push((node, entity)); + } + } + } + if missing.is_empty() || Instant::now() >= deadline { + break; + } + std::thread::sleep(Duration::from_millis(250)); + } + + if !missing.is_empty() { + // Dump the blob ledger before failing: `originated` on the writer vs + // `applied`/`apply_failed` per peer localises the gap to enqueue, ship, or + // apply. Diagnosing from the failure output beats re-running by hand. + for node in 0..NODES { + println!( + "[parity] node {node} ({}) blob ledger:\n{}", + cluster.region_name(node), + blob_ledger(&cluster, node) + ); + } + } + assert!( + missing.is_empty(), + "an embedding must be searchable on EVERY replica without a restart, but \ + (node, entity) pairs are still missing after the convergence budget: {missing:?}. \ + The non-sharded surface rides the leader WAL relay, so this means a blob left \ + the leader and never landed - check the blob ledger dumped above." + ); +} + +/// This node's `tidaldb_cluster_blobs_*` and per-group vector-count lines. +fn blob_ledger(cluster: &MultiProcCluster, node: usize) -> String { + let resp = cluster.get(node, "/metrics"); + if resp.status().as_u16() != 200 { + return format!(" ", resp.status()); + } + resp.text().map_or_else( + |e| format!(" "), + |body| { + body.lines() + .filter(|l| { + (l.starts_with("tidaldb_cluster_blobs_") + || l.starts_with("tidaldb_usearch_vector_count")) + && !l.ends_with(" 0") + }) + .map(|l| format!(" {l}")) + .collect::>() + .join("\n") + }, + ) +} + +/// A deterministic, entity-varying 4-dim embedding. +/// +/// NOT a constant vector: every constant vector normalizes to the same unit vector, +/// so all of them are equidistant and a presence test over them proves nothing. +fn embedding_for(entity: u64) -> Vec { + #[allow(clippy::cast_precision_loss)] + let v = (entity % 997) as f32; + vec![v, v + 1.0, v + 2.0, v + 3.0] +} + +/// Does `node`'s own `/vector_search` return `entity` for its exact stored vector? +/// +/// Queries the node's OWN address, never a shared/load-balanced endpoint — routing a +/// probe through a gateway hides precisely this class of bug (it answers from an +/// arbitrary replica). +fn vector_search_finds(cluster: &MultiProcCluster, node: usize, entity: u64) -> bool { + let resp = cluster.post( + node, + "/vector_search", + &serde_json::json!({ "vector": embedding_for(entity), "k": 5 }), + ); + if resp.status().as_u16() != 200 { + return false; + } + let body: serde_json::Value = match resp.json() { + Ok(b) => b, + Err(_) => return false, + }; + body["items"].as_array().is_some_and(|items| { + items + .iter() + .any(|it| it["entity_id"].as_u64() == Some(entity)) + }) +} + +/// The `/sharded/*` write surface is single-copy BY DESIGN — pin it. +/// +/// `node.rs:8828-8829` states it: the `/sharded/*` surface hash-partitions and applies +/// to the owning region's LOCAL store, and does NOT ride the leader WAL relay (that is +/// the non-sharded surface). `sharded_write_embedding` therefore calls +/// `ShardReplica::apply_embedding_local`, which performs no WAL append and so ships +/// nothing to peers. +/// +/// This test exists because that property is easy to mistake for a replication defect: +/// a 2026-08-30 live probe wrote embeddings through `/sharded/embeddings`, found each +/// one on exactly one of three nodes, and was initially recorded as a durability +/// incident. It is not one — but a caller who assumes `/sharded/*` writes are +/// replicated is building on sand, so the semantics deserve an executable statement +/// rather than a comment. +/// +/// If a future change makes `/sharded/*` replicate, this test SHOULD fail: that is a +/// deliberate contract change, and the failure is the prompt to update the docs, the +/// runbook, and any durability claim that depends on it. +#[test] +fn mp_sharded_surface_writes_are_local_to_the_owner() { + let _heavy = heavy_test_guard(); + let cluster = MultiProcCluster::start_sharded(NODES, SHARDS, Some(FAST_ELECTION_YAML)); + let _leaders = cluster.wait_shard_leaders_agreed(convergence_budget()); + + // Six ids spread across the hash space, each written through a different node. + const ENTITIES: [u64; 6] = [922_001, 922_002, 922_003, 922_004, 922_005, 922_006]; + for (n, entity) in ENTITIES.iter().enumerate() { + let via = n % NODES; + assert_eq!( + cluster + .post( + via, + "/sharded/items", + &serde_json::json!({ "entity_id": entity, "metadata": {} }) + ) + .status() + .as_u16(), + 201 + ); + assert_eq!( + cluster + .post( + via, + "/sharded/embeddings", + &serde_json::json!({ "entity_id": entity, "values": embedding_for(*entity) }) + ) + .status() + .as_u16(), + 204 + ); + } + cluster.wait_converged_all(convergence_budget()); + // Generous settle: the claim is "never replicates", so give replication every + // chance to happen before asserting that it did not. + std::thread::sleep(Duration::from_secs(5)); + + // At least one entity must be visible on strictly fewer than all replicas. + // Asserting "exactly 1 node" for EVERY entity would be over-fitting: reads scatter + // over all locally hosted groups, so an owner that co-locates the querying group + // can legitimately answer for more than one id. + let spread: Vec<(u64, usize)> = ENTITIES + .iter() + .map(|&e| { + ( + e, + (0..NODES) + .filter(|&n| vector_search_finds(&cluster, n, e)) + .count(), + ) + }) + .collect(); + println!("[sharded-semantics] visible-replica counts: {spread:?}"); + assert!( + spread.iter().any(|&(_, n)| n < NODES), + "the /sharded/* surface must NOT replicate (node.rs:8828-8829); every entity \ + reached all {NODES} replicas, so the contract changed: {spread:?}" + ); + assert!( + spread.iter().all(|&(_, n)| n >= 1), + "every /sharded/* write must still be durable on its owner: {spread:?}" + ); +} diff --git a/tidal-server/tests/vector_search.rs b/tidal-server/tests/vector_search.rs index 9c57c4a..384d87f 100644 --- a/tidal-server/tests/vector_search.rs +++ b/tidal-server/tests/vector_search.rs @@ -45,6 +45,18 @@ fn axis_vector(axis: usize, mag: f32) -> Vec { v } +/// `v` scaled by `s` — a TRUE scalar multiple, i.e. the same direction at a +/// different magnitude. +/// +/// Necessary because `axis_vector(axis, mag)` pins a fixed `0.01` on the second +/// axis regardless of `mag`, so varying `mag` there changes the DIRECTION once the +/// vector is normalized, not just its length. A magnitude-independence test built +/// on `axis_vector` would be testing the wrong thing (and would fail for a correct +/// implementation). +fn scaled(v: &[f32], s: f32) -> Vec { + v.iter().map(|x| x * s).collect() +} + async fn post_json(app: &axum::Router, uri: &str, body: serde_json::Value) -> StatusCode { app.clone() .oneshot( @@ -186,3 +198,131 @@ async fn vector_search_dimension_mismatch_is_400() { .await; assert_eq!(status, StatusCode::BAD_REQUEST); } + +/// An exact-match query must score ~0, not `|q|² − 1`. +/// +/// The write path L2-normalizes every stored vector; until 2026-08-30 the read path +/// passed the caller's raw query straight to the index, so the two sides lived in +/// different spaces. With unit `v`, `d = |q|² − 2q·v + 1`. Measured live, a non-unit +/// query returned 591–1174 against a documented `[0.0, 4.0]`. +#[tokio::test] +async fn vector_search_exact_match_scores_zero() { + let app = make_app(); + seed(&app).await; + + // Item 1's exact direction, scaled x3 so the query is NOT unit-length: the + // point is that the server owns the normalization. A unit query would pass + // even against the bug. Must be a true scalar multiple (see `scaled`), or it + // is a different direction and not an exact match at all. + let query = scaled(&axis_vector(0, 1.0), 3.0); + let (status, body) = post_json_full( + &app, + "/vector_search", + serde_json::json!({ "vector": query, "k": 1 }), + ) + .await; + assert_eq!(status, StatusCode::OK, "body: {body}"); + + let d = body["items"][0]["distance"].as_f64().expect("distance"); + // f16 quantization puts an exact match near 1e-3, not at 0 — assert a tolerance. + // Equality here would be a flaky test, not a stricter one. + assert!( + d < 1e-2, + "an exact-direction match must score ~0, got {d} (pre-fix this was |q|^2 - 1 ~= 8)" + ); +} + +/// Every distance must fall inside the range documented at +/// `tidal/src/storage/vector/mod.rs:49`. +/// +/// This is the assertion whose absence let 591–1174 ship unnoticed: the existing +/// ordering test proves distances ASCEND, which stayed true the whole time. +#[tokio::test] +async fn vector_search_distances_within_documented_range() { + let app = make_app(); + seed(&app).await; + + // A range of query magnitudes, since the bug scaled distances by |q|^2 — one + // magnitude could coincidentally land in range. + for mag in [0.5_f32, 1.0, 3.0, 10.0] { + let (status, body) = post_json_full( + &app, + "/vector_search", + serde_json::json!({ "vector": scaled(&axis_vector(0, 1.0), mag), "k": 3 }), + ) + .await; + assert_eq!(status, StatusCode::OK, "mag {mag}, body: {body}"); + + let dists: Vec = body["items"] + .as_array() + .expect("items") + .iter() + .map(|it| it["distance"].as_f64().expect("distance")) + .collect(); + assert!( + dists.iter().all(|d| (0.0..=4.0).contains(d)), + "mag {mag}: outside documented [0.0, 4.0] (storage/vector/mod.rs:49): {dists:?}" + ); + } +} + +/// A zero-norm query has no direction, so "nearest by cosine" is undefined for it. +/// Reject it as caller error rather than returning an arbitrary ranking. +#[tokio::test] +async fn vector_search_rejects_zero_query() { + let app = make_app(); + seed(&app).await; + + let (status, body) = post_json_full( + &app, + "/vector_search", + serde_json::json!({ "vector": vec![0.0_f32; DIM], "k": 3 }), + ) + .await; + assert_eq!( + status, + StatusCode::BAD_REQUEST, + "a zero query must be rejected, not answered with an arbitrary ranking; body: {body}" + ); +} + +/// Normalizing the query must NOT change ranking — it is a contract fix, not a +/// behavioral one. +/// +/// `|q|²` and `1` are constant across candidates, so ordering was already correct +/// cosine order. This is the regression guard: if a future change to the read path +/// alters ranking, this fails even though the range assertions still pass. +#[tokio::test] +async fn vector_search_ordering_unchanged_by_normalization() { + let app = make_app(); + seed(&app).await; + + // The same direction at four magnitudes must produce the IDENTICAL id order. + // Pre-fix this held too (the bug was scale-only), so this test passes before + // AND after — which is exactly what makes it a guard rather than a symptom test. + let mut orders = Vec::new(); + for mag in [0.5_f32, 1.0, 3.0, 10.0] { + let (status, body) = post_json_full( + &app, + "/vector_search", + serde_json::json!({ "vector": scaled(&axis_vector(0, 1.0), mag), "k": 3 }), + ) + .await; + assert_eq!(status, StatusCode::OK, "mag {mag}"); + orders.push( + body["items"] + .as_array() + .expect("items") + .iter() + .map(|it| it["entity_id"].as_u64().expect("entity_id")) + .collect::>(), + ); + } + assert_eq!(orders[0], vec![1, 3, 2], "golden order for an axis-0 query"); + for (i, o) in orders.iter().enumerate() { + assert_eq!( + o, &orders[0], + "query magnitude must not affect ranking (index {i}): {orders:?}" + ); + } +} diff --git a/tidal/src/db/items.rs b/tidal/src/db/items.rs index 103f66a..71f43d5 100644 --- a/tidal/src/db/items.rs +++ b/tidal/src/db/items.rs @@ -46,19 +46,28 @@ impl TidalDb { // to serve catch-up from); their items are memory-only anyway. return Ok(None); }; - let pending = sender - .append_blob_staged(std::sync::Arc::new(record())) - .map_err(|e| { - TidalError::Durability(crate::schema::DurabilityError { - message: format!("WAL blob append staging failed: {e}"), - }) - })?; + let record = std::sync::Arc::new(record()); + // Kind captured before the Arc moves into the append. Cluster-only path: + // `replicate_blobs` gated the early return above, so this never fires on a + // standalone node and cannot activate the cluster series there. + #[cfg(feature = "metrics")] + let kind = record.blob_kind(); + let pending = sender.append_blob_staged(record).map_err(|e| { + TidalError::Durability(crate::schema::DurabilityError { + message: format!("WAL blob append staging failed: {e}"), + }) + })?; let seq = pending.wait().map_err(|e| { TidalError::Durability(crate::schema::DurabilityError { message: format!("WAL blob append failed: {e}"), }) })?; super::wal_bridge::bump_last_seq_atomic(&self.last_wal_seq, seq); + // Counted only after the append is durable: this counter's contract is + // "entered the outbound stream", so a staging or fsync failure must not + // inflate it and make a peer look like it lost a record that never shipped. + #[cfg(feature = "metrics")] + self.metrics.cluster.observe_blobs_originated(kind, 1); Ok(Some(seq)) } /// Write (or overwrite) item metadata and update in-memory indexes. @@ -162,10 +171,52 @@ impl TidalDb { /// mid-batch halt re-applies safely). On a durability error every append /// staged by this call has still been WAITED — no staged blob is left /// unresolved behind the halt; the first error is returned. + /// + /// # Metrics + /// + /// Increments `tidaldb_cluster_blobs_applied_total` per kind on success and + /// `..._apply_failed_total` on failure. This is the **live** apply path only; + /// boot-time replay of already-counted records goes through + /// [`replay_recovered_blobs`](Self::replay_recovered_blobs) and is deliberately + /// NOT counted here — counting it would inflate `applied` past the writer's + /// `originated` on every restart and make the comparison worthless. + pub(crate) fn apply_replicated_blobs(&self, records: Vec) -> crate::Result<()> { + // Tally before the records move into the apply; kinds are needed on both + // the success and the failure edge. + #[cfg(feature = "metrics")] + let tally = { + let mut t = [0u64; crate::wal::format::batch::BlobKind::COUNT]; + for r in &records { + t[r.blob_kind().index()] += 1; + } + t + }; + let outcome = self.apply_replicated_blobs_inner(records); + #[cfg(feature = "metrics")] + { + let cluster = &self.metrics.cluster; + for kind in crate::wal::format::batch::BlobKind::ALL { + let n = tally[kind.index()]; + if n == 0 { + continue; + } + if outcome.is_ok() { + cluster.observe_blobs_applied(kind, n); + } else { + // One failure per kind present in the halted round. The round is + // all-or-nothing from the receiver's perspective, and the error + // is returned unchanged below — counted, never absorbed. + cluster.observe_blob_apply_failed(kind); + } + } + } + outcome + } + // One linear three-phase pass (validate -> journal -> apply); splitting it // would scatter the WAL-first ordering invariants across helpers. #[allow(clippy::too_many_lines)] - pub(crate) fn apply_replicated_blobs(&self, records: Vec) -> crate::Result<()> { + fn apply_replicated_blobs_inner(&self, records: Vec) -> crate::Result<()> { /// One validated record, parsed exactly once in Phase 1 and applied /// in Phase 3 (item metadata is deserialized here, never re-parsed). enum BlobApply<'a> { @@ -991,6 +1042,65 @@ mod tests { b.build().expect("schema must be valid") } + /// The apply-side ledger counts both edges, and counting a failure must not + /// turn it into a success. + /// + /// `apply_failed` exists to make a lost blob visible; if incrementing it also + /// swallowed the `Err`, the receiver would advance past a record it never + /// applied and the counter would document a silent data loss instead of + /// preventing one (`CODING_GUIDELINES` :193-195). + #[test] + fn blob_apply_counts_both_edges_and_still_propagates_the_error() { + use crate::wal::format::batch::{BlobRecord, EmbeddingRecord, TermMarkerRecord}; + + let db = TidalDb::builder() + .ephemeral() + .with_schema(minimal_schema()) + .open() + .unwrap(); + + // Success edge: a valid, finite, non-zero-norm embedding. + db.apply_replicated_blobs(vec![BlobRecord::Embedding(EmbeddingRecord { + entity_id: 1, + values: vec![1.0, 0.0, 0.0], + })]) + .expect("a valid embedding blob must apply"); + + // Failure edge: term 0 is invalid by construction (no topology era ever + // journals one), so Phase 1 rejects it deterministically. + let err = db + .apply_replicated_blobs(vec![BlobRecord::TermMarker(TermMarkerRecord { + term: 0, + leader_region: 0, + })]) + .expect_err("term 0 must be rejected, not counted-and-swallowed"); + assert!( + format!("{err}").contains("term 0"), + "the original error must reach the caller unchanged, got: {err}" + ); + + let mut out = String::new(); + db.metrics.cluster.render_into(&mut out, 0); + assert!( + out.contains( + r#"tidaldb_cluster_blobs_applied_total{kind="embedding",partition_id="0"} 1"# + ), + "successful apply must be counted:\n{out}" + ); + assert!( + out.contains( + r#"tidaldb_cluster_blobs_apply_failed_total{kind="term_marker",partition_id="0"} 1"# + ), + "failed apply must be counted under its own kind:\n{out}" + ); + assert!( + out.contains( + r#"tidaldb_cluster_blobs_applied_total{kind="term_marker",partition_id="0"} 0"# + ), + "a failed apply must NOT also count as applied:\n{out}" + ); + } + #[test] fn write_item_rejects_oversized_metadata_value() { let db = TidalDb::builder().ephemeral().open().unwrap(); diff --git a/tidal/src/db/metrics/cluster.rs b/tidal/src/db/metrics/cluster.rs index f754d2c..264e92b 100644 --- a/tidal/src/db/metrics/cluster.rs +++ b/tidal/src/db/metrics/cluster.rs @@ -13,6 +13,7 @@ //! the [`crate::wal::WalConfig::sync_observer`] hook the open path wires; //! - `tidal-server`'s cluster write pool reports queue depth and 429s. +use crate::wal::format::batch::BlobKind; use std::sync::{ Arc, RwLock, atomic::{AtomicBool, AtomicU64, Ordering}, @@ -57,6 +58,34 @@ impl PeerShipMetrics { } } +/// Per-[`BlobKind`] counters for the blob replication path. +/// +/// Kind-indexed arrays, **not** a `HashMap`: the four kinds are a +/// closed set fixed at compile time, so the `kind` label's cardinality is 4 and +/// cannot be driven by anything arriving on the wire. A string-keyed map here +/// would let a malformed record mint an unbounded series. +#[derive(Debug)] +struct BlobCounters { + /// Records this node created from a LOCAL client write and journaled into + /// its own WAL — and therefore into its outbound replication stream. + originated: [AtomicU64; BlobKind::COUNT], + /// Records the applier accepted (`Ok`) on this node (follower side). + applied: [AtomicU64; BlobKind::COUNT], + /// Records whose apply returned `Err`. The error is still propagated — this + /// counts failures, it does not absorb them. + apply_failed: [AtomicU64; BlobKind::COUNT], +} + +impl BlobCounters { + fn new() -> Self { + Self { + originated: std::array::from_fn(|_| AtomicU64::new(0)), + applied: std::array::from_fn(|_| AtomicU64::new(0)), + apply_failed: std::array::from_fn(|_| AtomicU64::new(0)), + } + } +} + /// Cluster-mode replication metrics (`tidaldb_cluster_*`). /// /// Cheap atomics/histograms; safe to update from the ship sender threads, the @@ -182,6 +211,20 @@ pub struct ClusterMetrics { /// `scatter_degraded_total` it separates "one flaky peer" from "this node /// can reach nobody". scatter_shard_unavailable_total: AtomicU64, + /// Blob-path ledger: originated / applied / apply-failed, per [`BlobKind`]. + /// + /// The blob path carries embeddings, item metadata, term markers and + /// membership. Before this existed, nothing counted any of them: `lag_events` + /// tracks WAL apply, so a blob that never shipped or never applied moved no + /// number anywhere. A persistent cross-replica vector-index divergence + /// (measured 2026-08-30: 33335/33334/33334) therefore sat with replication lag + /// reading 0 and every health surface green. + /// + /// `originated` on the writing node vs `applied` on each peer is the whole + /// diagnosis: originated outrunning applied means the record never landed + /// (store-gap); the two matching while that replica's `usearch_vector_count` + /// lags means it landed and was never indexed (index-gap). + blobs: BlobCounters, } impl ClusterMetrics { @@ -217,6 +260,7 @@ impl ClusterMetrics { healing_peers: AtomicU64::new(0), scatter_degraded_total: AtomicU64::new(0), scatter_shard_unavailable_total: AtomicU64::new(0), + blobs: BlobCounters::new(), } } @@ -415,6 +459,39 @@ impl ClusterMetrics { self.group_commit_events.observe(batch_events as u64); } + /// Count `n` blob records of `kind` originated by a local client write on this + /// node and journaled into its WAL (hence into its outbound stream). + /// + /// Leader side of the blob ledger. Pairs with + /// [`observe_blobs_applied`](Self::observe_blobs_applied) on the followers: the + /// two diverging is the signal that a record left here and never landed there. + /// + /// Deliberately counted at ORIGIN rather than at ship time. Blobs travel inside + /// WAL segments, so a true per-kind ship counter would mean decoding every + /// shipped segment on the ship hot path purely to label a metric. At origin the + /// kind is already known statically and the count answers the same question. + pub fn observe_blobs_originated(&self, kind: BlobKind, n: u64) { + self.blobs.originated[kind.index()].fetch_add(n, Ordering::Relaxed); + } + + /// Count `n` blob records of `kind` that this node's applier accepted. + /// + /// Follower side of the ledger. Count only the **live** apply path — boot-time + /// replay (`replay_recovered_blobs`) re-applies records that were already + /// counted when they first arrived, so counting it too would inflate `applied` + /// past `shipped` on every restart and make the comparison useless. + pub fn observe_blobs_applied(&self, kind: BlobKind, n: u64) { + self.blobs.applied[kind.index()].fetch_add(n, Ordering::Relaxed); + } + + /// Count one blob apply of `kind` that returned `Err`. + /// + /// Counting a failure must never convert it into a success: callers increment + /// this **and** propagate the error (`CODING_GUIDELINES` :193-195). + pub fn observe_blob_apply_failed(&self, kind: BlobKind) { + self.blobs.apply_failed[kind.index()].fetch_add(1, Ordering::Relaxed); + } + /// Update the stream gauges: the flushed high-water mark and the quorum /// commit index (m11p3). pub fn set_relay_frontiers(&self, last_seq: u64, durable_seq: u64) { @@ -686,6 +763,40 @@ impl ClusterMetrics { &extra, ); + // Blob-path ledger, one line per kind. Emitted BEFORE the per-peer block + // below on purpose: that block returns early when no peers are registered, + // and `applied` is incremented on followers — which may have no peer cells + // at all. Rendering after the early return would hide exactly the number + // this ledger exists to expose. + let kind_prefix = shard.map_or(String::new(), |s| format!("shard=\"{s}\",")); + for (metric, help, arr) in [ + ( + "tidaldb_cluster_blobs_originated_total", + "Blob records originated by a local write on this node and journaled into its outbound stream, by record kind", + &self.blobs.originated, + ), + ( + "tidaldb_cluster_blobs_applied_total", + "Blob records this node's applier accepted, by record kind (live path only, excludes boot replay)", + &self.blobs.applied, + ), + ( + "tidaldb_cluster_blobs_apply_failed_total", + "Blob applies that returned an error, by record kind (the error is propagated, not absorbed)", + &self.blobs.apply_failed, + ), + ] { + let _ = writeln!(out, "\n# HELP {metric} {help}\n# TYPE {metric} counter"); + for kind in BlobKind::ALL { + let _ = writeln!( + out, + "{metric}{{{kind_prefix}kind=\"{}\",partition_id=\"{partition_id}\"}} {}", + kind.label(), + arr[kind.index()].load(Ordering::Relaxed) + ); + } + } + // Per-peer series, labeled by peer shard id + this node's partition (and // this group's shard when co-located). Snapshot the cells under the read // lock, then render lock-free (the ship sender threads update these cells @@ -764,6 +875,91 @@ fn emit_scalar(out: &mut String, name: &str, help: &str, type_str: &str, value: mod tests { use super::*; + /// The blob ledger must render every kind of every family, with a `kind` + /// label whose cardinality is fixed at compile time. + /// + /// Guards step 3 of the counter-add pattern (`metrics/mod.rs:7-10`): a counter + /// that increments but never renders is invisible, which is the exact failure + /// this ledger was built to end. + #[test] + fn blob_ledger_renders_every_kind_with_bounded_cardinality() { + let m = ClusterMetrics::new(); + m.mark_active(); + m.observe_blobs_originated(BlobKind::Embedding, 3); + m.observe_blobs_applied(BlobKind::Embedding, 2); + m.observe_blob_apply_failed(BlobKind::Embedding); + m.observe_blobs_originated(BlobKind::ItemMetadata, 5); + + let mut out = String::new(); + m.render_into(&mut out, 7); + + assert!( + out.contains( + r#"tidaldb_cluster_blobs_originated_total{kind="embedding",partition_id="7"} 3"# + ), + "originated series missing or mislabeled:\n{out}" + ); + assert!( + out.contains( + r#"tidaldb_cluster_blobs_applied_total{kind="embedding",partition_id="7"} 2"# + ), + "applied series missing or mislabeled:\n{out}" + ); + assert!( + out.contains( + r#"tidaldb_cluster_blobs_apply_failed_total{kind="embedding",partition_id="7"} 1"# + ), + "apply_failed series missing or mislabeled:\n{out}" + ); + assert!( + out.contains( + r#"tidaldb_cluster_blobs_originated_total{kind="item_metadata",partition_id="7"} 5"# + ), + "a second kind must be counted independently:\n{out}" + ); + + // Exactly 3 families x 4 kinds. Asserting the total pins the label domain: + // a string-keyed map would let wire input mint extra series here. + let series = out + .lines() + .filter(|l| l.starts_with("tidaldb_cluster_blobs_")) + .count(); + assert_eq!( + series, + 3 * BlobKind::COUNT, + "blob ledger must emit exactly 3 families x {} kinds:\n{out}", + BlobKind::COUNT + ); + } + + /// A node with NO registered peers must still render the ledger. + /// + /// The per-peer block in `render_labeled` returns early when the peer map is + /// empty, so anything emitted after it vanishes on such a node. `applied` is + /// incremented on FOLLOWERS, which is precisely where the peer map can be + /// empty — rendering after that return would hide the one number the ledger + /// exists to expose. This test fails if the block is ever moved below it. + #[test] + fn blob_ledger_renders_on_a_node_with_no_peers() { + let m = ClusterMetrics::new(); + m.mark_active(); + m.observe_blobs_applied(BlobKind::Embedding, 11); + + let mut out = String::new(); + m.render_into(&mut out, 0); + + assert!( + out.contains( + r#"tidaldb_cluster_blobs_applied_total{kind="embedding",partition_id="0"} 11"# + ), + "ledger must survive the empty-peer early return:\n{out}" + ); + assert!( + !out.contains("tidaldb_cluster_peer_"), + "no peers were registered, so no per-peer series should appear:\n{out}" + ); + } + #[test] fn inactive_until_marked_and_renders_peer_series() { let m = ClusterMetrics::new(); diff --git a/tidal/src/db/metrics/mod.rs b/tidal/src/db/metrics/mod.rs index a6f650a..7f2c78f 100644 --- a/tidal/src/db/metrics/mod.rs +++ b/tidal/src/db/metrics/mod.rs @@ -284,6 +284,15 @@ pub struct MetricsState { /// shipped S=1 topology), so its output is byte-identical to pre-m11p8. #[cfg(feature = "metrics")] cluster_siblings: std::sync::RwLock)>>, + /// Co-located groups' NODE-level metrics, for the per-group series that + /// `cluster_siblings` does not carry (currently the vector count). + /// + /// Without this, `tidaldb_usearch_vector_count` exposes only the metrics + /// OWNER's group, so on a 3-group node two thirds of the corpus has no + /// vector-count series at all and can diverge silently. Measured on the RF3 + /// cluster 2026-08-30. + #[cfg(feature = "metrics")] + node_siblings: std::sync::RwLock)>>, /// Extra Prometheus series contributed by the embedding application, appended /// verbatim by [`render_prometheus`](Self::render_prometheus). @@ -358,6 +367,8 @@ impl MetricsState { #[cfg(feature = "metrics")] cluster_siblings: std::sync::RwLock::new(Vec::new()), #[cfg(feature = "metrics")] + node_siblings: std::sync::RwLock::new(Vec::new()), + #[cfg(feature = "metrics")] extra_renderer: std::sync::OnceLock::new(), } } @@ -395,6 +406,21 @@ impl MetricsState { .push((shard, metrics)); } + /// Register a co-located shard group's NODE-level metrics so this node's one + /// `/metrics` listener also exposes that group's per-group node series + /// (currently `tidaldb_usearch_vector_count{shard=""}`). + /// + /// Separate from [`register_cluster_sibling`](Self::register_cluster_sibling) + /// because the two carry different state: that one shares + /// `Arc`, this one the whole sibling `MetricsState`. + #[cfg(feature = "metrics")] + pub fn register_node_sibling(&self, shard: u16, metrics: std::sync::Arc) { + self.node_siblings + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push((shard, metrics)); + } + /// Maximum age (nanoseconds) a checkpoint may reach before health reports /// degraded. The periodic checkpoint thread runs every 30s; a checkpoint /// older than 5 minutes means the thread is stuck, dead, or failing — all @@ -614,6 +640,27 @@ impl MetricsState { "gauge", self.usearch_vector_count.load(Ordering::Relaxed) as f64, ); + // Co-located groups' vector counts, each stamped `shard="N"`. The + // owner's line above stays UNLABELED for wire compatibility, and the + // owner is deterministically the same group on every node, so a + // cross-node comparison grouped `by (shard)` puts each replica set in + // its own bucket (owner -> the empty-label bucket) and never mixes + // unlike groups or double-counts a node. + { + use std::fmt::Write; + for (shard, sib) in self + .node_siblings + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .iter() + { + let _ = writeln!( + &mut out, + "tidaldb_usearch_vector_count{{shard=\"{shard}\"}} {}", + sib.usearch_vector_count.load(Ordering::Relaxed) + ); + } + } write_metric_line( &mut out, "tidaldb_bitmap_index_cardinality", diff --git a/tidal/src/db/metrics/tests.rs b/tidal/src/db/metrics/tests.rs index a4cf053..413d965 100644 --- a/tidal/src/db/metrics/tests.rs +++ b/tidal/src/db/metrics/tests.rs @@ -409,3 +409,45 @@ fn metrics_feature_counters_exist() { state.signal_writes_total.fetch_add(1, Ordering::Relaxed); assert_eq!(state.signal_writes_total.load(Ordering::Relaxed), 1); } + +/// Every co-located shard group must get its own `tidaldb_usearch_vector_count`. +/// +/// Before 2026-08-30 only the metrics OWNER's group had this series, so on the +/// 3-group RF3 cluster two thirds of each node's corpus had no vector count at +/// all and a divergence there was unobservable. The owner stays UNLABELED for +/// wire compatibility, so a cross-node alert grouped `by (shard)` puts each +/// replica set in its own bucket without mixing groups or double-counting. +#[test] +fn every_colocated_group_exposes_its_own_vector_count() { + let owner = MetricsState::new(); + owner.usearch_vector_count.store(33335, Ordering::Relaxed); + + for (shard, count) in [(1u16, 111u64), (2u16, 222u64)] { + let sib = std::sync::Arc::new(MetricsState::new()); + sib.usearch_vector_count.store(count, Ordering::Relaxed); + owner.register_node_sibling(shard, sib); + } + + let out = owner.render_prometheus(); + assert!( + out.contains("\ntidaldb_usearch_vector_count 33335\n"), + "owner series must stay unlabeled for wire compatibility:\n{out}" + ); + assert!( + out.contains("tidaldb_usearch_vector_count{shard=\"1\"} 111"), + "group 1 count missing:\n{out}" + ); + assert!( + out.contains("tidaldb_usearch_vector_count{shard=\"2\"} 222"), + "group 2 count missing:\n{out}" + ); + // Exactly one series per hosted group: 1 owner + 2 siblings. More would mean + // a node double-counts itself and any sum()/max() over it is wrong. + assert_eq!( + out.lines() + .filter(|l| l.starts_with("tidaldb_usearch_vector_count")) + .count(), + 3, + "one vector-count series per hosted group, no duplicates:\n{out}" + ); +} diff --git a/tidal/src/db/mod.rs b/tidal/src/db/mod.rs index dea1092..68482f6 100644 --- a/tidal/src/db/mod.rs +++ b/tidal/src/db/mod.rs @@ -606,18 +606,29 @@ impl TidalDb { Arc::clone(&self.metrics.cluster) } - /// Expose `sibling`'s `tidaldb_cluster_*` series through THIS db's `/metrics` + /// Expose `sibling`'s co-located-group series through THIS db's `/metrics` /// listener, stamped with `shard=""` (m11p8). /// /// When several shard groups co-locate in one process, only one binds the /// engine's `/metrics` server (the metrics owner); the others register here /// so a single per-node scrape target still exposes every co-located group's - /// replication series, collision-free. A no-op-equivalent on the S=1 topology + /// series, collision-free. A no-op-equivalent on the S=1 topology /// (one shard per node), so its `/metrics` output is unchanged. + /// + /// Registers BOTH families: + /// * the sibling's `tidaldb_cluster_*` replication series, and + /// * its node-level `tidaldb_usearch_vector_count`. + /// + /// The vector count was missing until 2026-08-30: on the RF3 cluster each pod + /// hosts three groups but exposed only the owner's count, so two thirds of + /// every node's corpus had no vector-count series and could diverge with + /// nothing to compare. #[cfg(feature = "metrics")] pub fn register_metrics_sibling(&self, shard: u16, sibling: &Self) { self.metrics .register_cluster_sibling(shard, Arc::clone(&sibling.metrics.cluster)); + self.metrics + .register_node_sibling(shard, Arc::clone(&sibling.metrics)); } } diff --git a/tidal/src/db/query_ops.rs b/tidal/src/db/query_ops.rs index 9800054..fa4a34c 100644 --- a/tidal/src/db/query_ops.rs +++ b/tidal/src/db/query_ops.rs @@ -582,6 +582,10 @@ impl TidalDb { // The registry read guard is deliberately held across the slot lookup AND the // index search: `slot` borrows from the guard, so it cannot be dropped earlier. #[allow(clippy::significant_drop_tightening)] + // `skip_all` + explicit fields: the query vector is 1536 f32 in production, so a + // default `instrument` would Debug-log the whole buffer on every search. Record + // its shape, never its contents — matching `retrieve` and `search` above. + #[tracing::instrument(skip_all, fields(dims = query_vector.len(), k, ef_search))] pub fn vector_search_items( &self, query_vector: &[f32], @@ -604,7 +608,28 @@ impl TidalDb { )) })?; let ef = ef_search.unwrap_or(slot.params.ef_search); - slot.index.search(query_vector, k, ef).map_err(|e| { + // Stored vectors are unit-norm (`storage::vector::lifecycle::ops` L2-normalizes + // every embedding at insertion). Normalizing the QUERY is what makes the + // documented `[0.0, 4.0]` range in `storage/vector/mod.rs:49` actually true: + // with unit `v`, `d = |q|^2 - 2q.v + 1`, so a non-unit query shifts and scales + // every distance by `|q|^2`. Measured on the RF3 cluster 2026-08-30: a + // non-unit query reported 591-1174 where the contract promises `<= 4`, and an + // exact match scored `|q|^2 - 1` instead of ~0. Ranking was unaffected + // (`|q|^2` and `1` are constant across candidates, so the ordering is still + // cosine order) — what broke was every ABSOLUTE use of the number: + // thresholding, dedup-by-distance, cross-query comparison, recall measurement. + // + // `CODING_GUIDELINES` §4's "never re-normalize at query time" governs the + // already-normalized STORED vectors; normalizing a caller's raw query is not + // re-normalizing. §4 now says so explicitly so the two docs cannot drift apart + // again. + // + // A zero-norm query has no direction, so "nearest by cosine" is undefined for + // it: reject it as caller error rather than return an arbitrary ranking. + let query = crate::storage::vector::l2_normalize(query_vector).map_err(|e| { + TidalError::invalid_input(format!("vector_search query vector rejected: {e}")) + })?; + slot.index.search(&query, k, ef).map_err(|e| { // A dimension-mismatched query vector is a CALLER error (→ 400 at the // HTTP boundary), not an engine fault — keep it distinct from a real // backend failure (→ 500). The latter is genuinely internal. diff --git a/tidal/src/entities/multi_preference.rs b/tidal/src/entities/multi_preference.rs index 845b576..6f94aba 100644 --- a/tidal/src/entities/multi_preference.rs +++ b/tidal/src/entities/multi_preference.rs @@ -62,6 +62,9 @@ use dashmap::DashMap; use crate::signals::decay::forward_decay_step; pub use crate::entities::preference::PreferenceVectors; +// The single named home of the centroid zero-tolerance policy; the arithmetic +// itself lives in `storage::vector::l2_normalize_in_place`. +use crate::entities::preference::normalize_centroid; /// Cold-start interaction threshold N. /// @@ -383,7 +386,7 @@ impl MultiPreferenceVectors { // cold-start seed BEFORE taking the cluster guard so we never hold the // `clusters` entry guard across another DashMap access. let mut normalized = interaction_embedding.to_vec(); - l2_normalize(&mut normalized); + normalize_centroid(&mut normalized); let seed = (!self.clusters.contains_key(&user_id)) .then(|| { self.cold_start @@ -442,7 +445,7 @@ impl MultiPreferenceVectors { } if let Some(mut clusters) = self.clusters.get_mut(&user_id) { let mut normalized = interaction_embedding.to_vec(); - l2_normalize(&mut normalized); + normalize_centroid(&mut normalized); if let Some(idx) = nearest_cluster(clusters.value(), &normalized) { blend_into(&mut clusters.value_mut()[idx].centroid, &normalized, lr); } @@ -839,7 +842,7 @@ fn decode_multi_value(value: &[u8], expected_dim: usize) -> Option> // cosine scoring downstream (mirrors the single-vector restore). centroid.push(if f.is_nan() { 0.0 } else { f }); } - l2_normalize(&mut centroid); + normalize_centroid(&mut centroid); clusters.push(Cluster { centroid, update_count, @@ -904,7 +907,7 @@ fn blend_into(pref: &mut [f32], interaction: &[f32], lr: f32) { for (p, &i) in pref.iter_mut().zip(interaction.iter()) { *p = (1.0 - lr).mul_add(*p, lr * i); } - l2_normalize(pref); + normalize_centroid(pref); } /// Index + cosine of the nearest centroid to `embedding` (vectors assumed @@ -960,16 +963,6 @@ fn centroid_cmp(a: &[f32], b: &[f32]) -> std::cmp::Ordering { .unwrap_or_else(|| a.len().cmp(&b.len())) } -/// L2-normalize in place; an all-zero vector is left untouched. -fn l2_normalize(vec: &mut [f32]) { - let norm: f32 = vec.iter().map(|x| x * x).sum::().sqrt(); - if norm > f32::EPSILON { - for v in vec.iter_mut() { - *v /= norm; - } - } -} - /// Wall-clock nanoseconds, via the engine's clock-anomaly-safe `Timestamp`. fn now_ns() -> u64 { crate::schema::Timestamp::now().as_nanos() @@ -992,7 +985,7 @@ mod tests { let mut v = vec![0.0f32; dim]; v[a] = wa; v[b] = wb; - l2_normalize(&mut v); + normalize_centroid(&mut v); v } @@ -1310,7 +1303,7 @@ mod tests { for v in &vec { value.extend_from_slice(&v.to_le_bytes()); } - l2_normalize(&mut vec); + normalize_centroid(&mut vec); let key = encode_key(EntityId::new(0), Tag::Preference, &42u64.to_be_bytes()); storage.put(&key, &value).unwrap(); @@ -1465,7 +1458,7 @@ mod tests { for count in [2u64, 258, 514] { let storage = InMemoryBackend::new(); let mut vec = axis_vec(dim, 3); - l2_normalize(&mut vec); + normalize_centroid(&mut vec); let value = crate::entities::preference::encode_legacy_row(count, &vec); assert_eq!( value[0], FORMAT_VERSION, diff --git a/tidal/src/entities/preference.rs b/tidal/src/entities/preference.rs index cf82ff8..80c9075 100644 --- a/tidal/src/entities/preference.rs +++ b/tidal/src/entities/preference.rs @@ -15,6 +15,8 @@ use dashmap::DashMap; +use crate::storage::vector::{VectorError, l2_normalize_in_place}; + /// Per-user preference vector, L2-normalized. /// /// The vector is updated via exponential moving average: each new interaction @@ -84,7 +86,7 @@ impl PreferenceVectors { if vec.len() != self.dim { return false; } - l2_normalize(&mut vec); + normalize_centroid(&mut vec); self.inner.insert(user_id, vec); true } @@ -171,11 +173,11 @@ impl PreferenceVectors { for (p, &i) in pref.iter_mut().zip(interaction_embedding.iter()) { *p = (1.0 - lr).mul_add(*p, lr * i); } - l2_normalize(pref); + normalize_centroid(pref); } Entry::Vacant(vac) => { let mut v = interaction_embedding.to_vec(); - l2_normalize(&mut v); + normalize_centroid(&mut v); vac.insert(v); } } @@ -408,13 +410,35 @@ impl PreferenceVectors { } } -/// L2-normalize a vector in-place. If the vector has zero magnitude, it remains -/// as-is (all zeros). -fn l2_normalize(vec: &mut [f32]) { - let norm: f32 = vec.iter().map(|x| x * x).sum::().sqrt(); - if norm > f32::EPSILON { - for v in vec.iter_mut() { - *v /= norm; +/// L2-normalize an accumulated preference centroid in place, tolerating a cold +/// (all-zero) vector. +/// +/// This is the **one** place the centroid zero-tolerance policy is named. The math +/// itself lives in [`crate::storage::vector::l2_normalize_in_place`]; this wrapper +/// only decides what a zero norm *means* here: +/// +/// A preference centroid is legitimately all-zero until the first signal folds in, +/// so "no direction yet" is an ordinary state, not an error — the vector is left at +/// zero and will normalize on the next fold. Embedding writes deliberately do **not** +/// get this leniency (`storage::vector::lifecycle::ops` propagates the error) because +/// a zero embedding entering the HNSW index is unrecoverable garbage. +/// +/// Both `preference` and `multi_preference` call this; before, each carried its own +/// copy of the arithmetic with a zero threshold ~2900x looser than the canonical one, +/// which is how the two directions of normalization drifted apart in the first place. +/// Vectors with `0 < ||v|| < 3.45e-4` are now left alone rather than amplified, since +/// dividing by a norm that small yields a direction made of rounding error. +pub(crate) fn normalize_centroid(vec: &mut [f32]) { + if let Err(err) = l2_normalize_in_place(vec) { + // A cold or numerically-zero centroid is the ONLY error this can return + // (see `l2_norm_checked`), and it is an ordinary state here: leave the + // vector exactly as it was. + // + // Anything else is unreachable today. Leaving the centroid untouched stays + // the safe default if normalization ever grows a second failure mode, but + // say so out loud rather than discarding it silently. + if !matches!(err, VectorError::ZeroNormVector) { + tracing::warn!(error = %err, "unexpected error normalizing preference centroid"); } } } @@ -460,7 +484,7 @@ pub(crate) fn decode_legacy_row(value: &[u8], expected_dim: usize) -> Option<(u6 let f = f32::from_le_bytes(value[off..off + 4].try_into().ok()?); vec.push(if f.is_nan() { 0.0 } else { f }); } - l2_normalize(&mut vec); + normalize_centroid(&mut vec); Some((update_count, vec)) } @@ -528,10 +552,46 @@ mod tests { #[test] fn l2_normalize_zero_vec() { let mut v = vec![0.0f32, 0.0, 0.0]; - l2_normalize(&mut v); + normalize_centroid(&mut v); assert!(v.iter().all(|&x| x == 0.0)); } + /// Pins the ONE behavior change from collapsing the three `l2_normalize` + /// copies into `storage::vector::l2_normalize_in_place`. + /// + /// The deleted in-place copies rejected only `||v|| <= f32::EPSILON` (1.2e-7); + /// the canonical implementation rejects `||v||^2 < f32::EPSILON`, i.e. + /// `||v|| < 3.45e-4` — about 2900x stricter. A centroid in that gap is now left + /// alone instead of being scaled to unit length. + /// + /// That is deliberate: dividing by a norm that small amplifies float noise by + /// more than 2900x, so the "direction" produced is mostly rounding error. + /// Leaving the vector as-is is the honest answer, and it matches what embeddings + /// have always done. Asserted rather than merely documented so the choice cannot + /// be un-made silently. + #[test] + fn normalize_centroid_leaves_numerically_zero_vector_untouched() { + // ||v|| = 1e-5 * sqrt(3) = 1.73e-5, inside the gap between the two thresholds. + let mut v = vec![1e-5f32, 1e-5, 1e-5]; + let before = v.clone(); + normalize_centroid(&mut v); + assert_eq!( + v, before, + "a centroid with norm below the canonical threshold must be left as-is, \ + not amplified into a direction made of rounding error" + ); + + // Just above the threshold it must still normalize, so the guard is a floor + // and not a silent no-op for real data. + let mut real = vec![1.0f32, 2.0, 3.0]; + normalize_centroid(&mut real); + let norm: f32 = real.iter().map(|x| x * x).sum::().sqrt(); + assert!( + (1.0 - norm).abs() < 1e-5, + "real centroid must reach unit norm, got {norm}" + ); + } + #[test] fn update_with_custom_rate_blends_at_given_lr() { let pv = PreferenceVectors::new(2); diff --git a/tidal/src/storage/vector/lifecycle/mod.rs b/tidal/src/storage/vector/lifecycle/mod.rs index f3adb08..14df036 100644 --- a/tidal/src/storage/vector/lifecycle/mod.rs +++ b/tidal/src/storage/vector/lifecycle/mod.rs @@ -13,6 +13,6 @@ pub mod normalize; pub mod ops; pub mod serde; -pub use normalize::l2_normalize; +pub use normalize::{l2_normalize, l2_normalize_in_place}; pub use ops::{delete_embedding, insert_embedding, update_embedding}; pub use serde::{deserialize_embedding, embedding_store_key, serialize_embedding}; diff --git a/tidal/src/storage/vector/lifecycle/normalize.rs b/tidal/src/storage/vector/lifecycle/normalize.rs index 2b15a8c..633a132 100644 --- a/tidal/src/storage/vector/lifecycle/normalize.rs +++ b/tidal/src/storage/vector/lifecycle/normalize.rs @@ -23,11 +23,7 @@ use super::super::VectorError; /// /// The returned vector has L2 norm within `1e-5` of 1.0. pub fn l2_normalize(v: &[f32]) -> Result, VectorError> { - let norm_sq: f32 = v.iter().map(|x| x * x).sum(); - if norm_sq < f32::EPSILON { - return Err(VectorError::ZeroNormVector); - } - let norm = norm_sq.sqrt(); + let norm = l2_norm_checked(v)?; let result: Vec = v.iter().map(|x| x / norm).collect(); // Post-condition: verify normalization. @@ -39,6 +35,57 @@ pub fn l2_normalize(v: &[f32]) -> Result, VectorError> { Ok(result) } +/// L2-normalize a vector to unit length **in place**. +/// +/// Identical semantics to [`l2_normalize`], including the zero-norm rejection; +/// it exists so callers on the hot signal-fold path can normalize an accumulated +/// centroid without allocating a second `Vec` per call (`CODING_GUIDELINES` §1). +/// +/// # Errors +/// +/// Returns [`VectorError::ZeroNormVector`] if the vector has zero norm. Callers +/// for whom a zero vector is legitimate (a cold preference centroid, say) must +/// say so at the call site — see `entities::preference::normalize_centroid`. +/// This function never silently leaves a vector unnormalized. +/// +/// # Post-conditions +/// +/// On `Ok`, the vector has L2 norm within `1e-5` of 1.0. On `Err`, the vector is +/// left exactly as it was. +pub fn l2_normalize_in_place(v: &mut [f32]) -> Result<(), VectorError> { + let norm = l2_norm_checked(v)?; + for x in v.iter_mut() { + *x /= norm; + } + + // Post-condition: verify normalization. + debug_assert!({ + let result_norm: f32 = v.iter().map(|x| x * x).sum::().sqrt(); + (1.0 - result_norm).abs() < 1e-5 + }); + + Ok(()) +} + +/// The L2 norm of `v`, rejecting vectors that are numerically indistinguishable +/// from zero. +/// +/// Shared by both normalization directions so the zero threshold can never drift +/// between them — the reason three hand-rolled copies of this math existed before +/// (two of them with a threshold ~2900x looser than this one). +/// +/// The test is on the *squared* norm against `f32::EPSILON`, i.e. it rejects +/// `||v|| < 3.45e-4`. Dividing by a norm that small amplifies float noise by more +/// than 2900x, so the resulting "direction" is mostly rounding error; refusing is +/// more correct than manufacturing one. +fn l2_norm_checked(v: &[f32]) -> Result { + let norm_sq: f32 = v.iter().map(|x| x * x).sum(); + if norm_sq < f32::EPSILON { + return Err(VectorError::ZeroNormVector); + } + Ok(norm_sq.sqrt()) +} + // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- diff --git a/tidal/src/storage/vector/mod.rs b/tidal/src/storage/vector/mod.rs index 8e326b6..721f57b 100644 --- a/tidal/src/storage/vector/mod.rs +++ b/tidal/src/storage/vector/mod.rs @@ -31,7 +31,7 @@ use std::path::Path; pub use brute::BruteForceIndex; pub use lifecycle::{ delete_embedding, deserialize_embedding, embedding_store_key, insert_embedding, l2_normalize, - serialize_embedding, update_embedding, + l2_normalize_in_place, serialize_embedding, update_embedding, }; pub use mock::{MockVectorIndex, VectorIndexCall}; pub use registry::{EmbeddingSlotRegistry, EmbeddingSlotState, EmbeddingSource, HnswParams}; @@ -46,7 +46,17 @@ pub struct VectorSearchResult { /// The ID of the matched vector. pub id: VectorId, /// L2 squared distance from the query vector. Lower = more similar. - /// For L2-normalized vectors, the range is `[0.0, 4.0]` where `0.0` = identical. + /// + /// The range is `[0.0, 4.0]`, where `0.0` = identical. This holds because + /// BOTH sides are unit-norm: stored vectors are normalized at insertion + /// (`lifecycle::ops`) and the query is normalized on the read path + /// (`db::query_ops::vector_search_items`), so a caller cannot observe an + /// out-of-range distance by sending a non-unit query. + /// + /// Under the default `QuantizationLevel::F16` an exact match lands near + /// `1e-3` rather than exactly `0.0` — compare against a tolerance, never + /// equality. (Measured live: `2.4e-7` for an exact match on a 1536-dim + /// corpus, well inside that bound.) pub distance: f32, } diff --git a/tidal/src/storage/vector/usearch_index.rs b/tidal/src/storage/vector/usearch_index.rs index 227f3d3..f9cdcb8 100644 --- a/tidal/src/storage/vector/usearch_index.rs +++ b/tidal/src/storage/vector/usearch_index.rs @@ -498,375 +498,9 @@ impl VectorIndex for UsearchIndex { } #[cfg(test)] -#[allow(clippy::unwrap_used)] -mod tests { - use super::*; - - fn default_config(dimensions: usize) -> VectorIndexConfig { - VectorIndexConfig { - dimensions, - metric: DistanceMetric::L2, - quantization: QuantizationLevel::F16, - connectivity: 16, - ef_construction: 200, - ef_search: 200, - } - } - - /// A full HNSW graph must keep accepting NEW keys. - /// - /// Regression for the production defect measured 2026-08-30 on the RF3 - /// cluster: `add` cannot grow the graph, and nothing on the write path - /// reserved, so once `size()` reached the reserved capacity every new - /// embedding failed with "Reserve capacity ahead of insertions!" and - /// returned HTTP 500. Items, signals, searches and re-embeds of EXISTING - /// entities all still worked, so the store had silently become - /// read/update-only for vectors. This asserts inserts continue past the - /// reservation, and that the vectors are actually retrievable afterwards. - #[test] - fn usearch_insert_grows_past_reserved_capacity() { - let index = UsearchIndex::new(default_config(4)).unwrap(); - index.reserve(4).unwrap(); - // Drive off the REPORTED capacity, not the requested one: USearch rounds a - // reservation up (reserve(4) reported 64 here), so a hardcoded insert count - // can sit entirely inside the reservation and never exercise growth at all. - let reserved = index.inner.capacity(); - let target = reserved + 100; - - for i in 0..target as u64 { - let f = i as f32; - index - .insert(i, &[f, f + 1.0, f + 2.0, f + 3.0]) - .unwrap_or_else(|e| { - panic!("insert {i} failed past reserved capacity {reserved}: {e}") - }); - } - assert_eq!(index.len_live(), target, "every new key must be indexed"); - assert!( - index.inner.capacity() > reserved, - "capacity should have grown beyond the original {reserved}" - ); - - // The grown graph must still answer, not merely accept writes. - let hits = index.search(&[10.0, 11.0, 12.0, 13.0], 3, 0).unwrap(); - assert!( - !hits.is_empty(), - "search over the grown index returned nothing" - ); - assert_eq!( - hits[0].id, 10, - "nearest neighbour should be the exact match" - ); - - // Re-embedding an existing key must still work after growth. - index.insert(10, &[0.0, 0.0, 0.0, 0.0]).unwrap(); - assert_eq!( - index.len_live(), - target, - "upsert must not change the live count" - ); - } - - #[test] - fn usearch_new_is_empty() { - let index = UsearchIndex::new(default_config(128)).unwrap(); - assert!(index.is_empty()); - assert_eq!(index.len(), 0); - assert_eq!(index.len_live(), 0); - } - - #[test] - fn usearch_insert_and_len() { - let config = default_config(4); - let index = UsearchIndex::new(config).unwrap(); - index.reserve(10).unwrap(); - index.insert(1, &[1.0, 2.0, 3.0, 4.0]).unwrap(); - index.insert(2, &[5.0, 6.0, 7.0, 8.0]).unwrap(); - assert_eq!(index.len(), 2); - assert_eq!(index.len_live(), 2); - assert!(!index.is_empty()); - } - - #[test] - fn usearch_insert_replaces_existing_key() { - // Regression (m12 reseed wedge): `insert` is upsert per the trait - // contract. USearch is `multi: false`, so re-inserting a key must - // REPLACE — not fail with "Duplicate keys not allowed in high-level - // wrappers". This is exactly the reseed path where post-snapshot WAL - // replay re-applies an embedding the snapshot already loaded. - let config = default_config(4); - let index = UsearchIndex::new(config).unwrap(); - index.reserve(10).unwrap(); - - index.insert(1, &[1.0, 0.0, 0.0, 0.0]).unwrap(); - // Re-insert the SAME key with a different vector — must not error. - index.insert(1, &[0.0, 1.0, 0.0, 0.0]).unwrap(); - - // Replacement, not a second slot: exactly one LIVE key. - assert_eq!(index.len_live(), 1); - - // Search reflects the REPLACEMENT vector (distance ~0 to the new one). - let results = index.search(&[0.0, 1.0, 0.0, 0.0], 1, 200).unwrap(); - assert_eq!(results.len(), 1); - assert_eq!(results[0].id, 1); - assert!( - results[0].distance < 0.01, - "replacement vector should match query, got distance {}", - results[0].distance - ); - } - - #[test] - fn usearch_dimension_mismatch_insert() { - let config = default_config(4); - let index = UsearchIndex::new(config).unwrap(); - let result = index.insert(1, &[1.0, 2.0]); // 2D into 4D index - assert!(result.is_err()); - match result.unwrap_err() { - VectorError::DimensionMismatch { expected, got } => { - assert_eq!(expected, 4); - assert_eq!(got, 2); - } - other => panic!("expected DimensionMismatch, got {other:?}"), - } - } - - #[test] - fn usearch_dimension_mismatch_search() { - let config = default_config(4); - let index = UsearchIndex::new(config).unwrap(); - index.reserve(2).unwrap(); - index.insert(1, &[1.0, 2.0, 3.0, 4.0]).unwrap(); - let result = index.search(&[1.0, 2.0], 1, 200); // 2D query on 4D index - assert!(result.is_err()); - assert!(matches!( - result.unwrap_err(), - VectorError::DimensionMismatch { - expected: 4, - got: 2 - } - )); - } - - #[test] - fn usearch_delete_not_found() { - let config = default_config(4); - let index = UsearchIndex::new(config).unwrap(); - let result = index.delete(999); - assert!(result.is_err()); - assert!(matches!( - result.unwrap_err(), - VectorError::NotFound { id: 999 } - )); - } - - #[test] - fn usearch_search_empty_index() { - let config = default_config(4); - let index = UsearchIndex::new(config).unwrap(); - let results = index.search(&[1.0, 2.0, 3.0, 4.0], 5, 200).unwrap(); - assert!(results.is_empty()); - } - - #[test] - fn usearch_basic_search() { - let config = default_config(4); - let index = UsearchIndex::new(config).unwrap(); - index.reserve(10).unwrap(); - - index.insert(1, &[1.0, 0.0, 0.0, 0.0]).unwrap(); - index.insert(2, &[0.0, 1.0, 0.0, 0.0]).unwrap(); - index.insert(3, &[0.0, 0.0, 1.0, 0.0]).unwrap(); - - let results = index.search(&[1.0, 0.0, 0.0, 0.0], 3, 200).unwrap(); - assert_eq!(results.len(), 3); - // Closest should be vector 1 (identical to query) - assert_eq!(results[0].id, 1); - } - - #[test] - fn usearch_delete_and_search() { - let config = default_config(4); - let index = UsearchIndex::new(config).unwrap(); - index.reserve(10).unwrap(); - - index.insert(1, &[1.0, 0.0, 0.0, 0.0]).unwrap(); - index.insert(2, &[0.0, 1.0, 0.0, 0.0]).unwrap(); - index.insert(3, &[0.0, 0.0, 1.0, 0.0]).unwrap(); - - index.delete(2).unwrap(); - assert_eq!(index.len_live(), 2); - - let results = index.search(&[0.0, 1.0, 0.0, 0.0], 3, 200).unwrap(); - // Vector 2 must not appear in results - assert!(results.iter().all(|r| r.id != 2)); - } - - #[test] - fn usearch_is_send_and_sync() { - fn assert_send_sync() {} - assert_send_sync::(); - } - - #[test] - fn usearch_save_load_roundtrip() { - let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("test_usearch.idx"); - - let config = default_config(4); - let index = UsearchIndex::new(config.clone()).unwrap(); - index.reserve(10).unwrap(); - index.insert(1, &[1.0, 0.0, 0.0, 0.0]).unwrap(); - index.insert(2, &[0.0, 1.0, 0.0, 0.0]).unwrap(); - - index.save(&path).unwrap(); - - let loaded = UsearchIndex::load(&path, &config).unwrap(); - assert_eq!(loaded.len(), 2); - assert_eq!(loaded.len_live(), 2); - - let results = loaded.search(&[1.0, 0.0, 0.0, 0.0], 1, 200).unwrap(); - assert_eq!(results[0].id, 1); - } - - #[test] - fn usearch_view_readonly() { - let dir = tempfile::tempdir().unwrap(); - let path = dir.path().join("test_usearch_view.idx"); - - let config = default_config(4); - let index = UsearchIndex::new(config.clone()).unwrap(); - index.reserve(10).unwrap(); - index.insert(1, &[1.0, 0.0, 0.0, 0.0]).unwrap(); - index.insert(2, &[0.0, 1.0, 0.0, 0.0]).unwrap(); - - index.save(&path).unwrap(); - - let viewed = UsearchIndex::view(&path, &config).unwrap(); - assert_eq!(viewed.len(), 2); - - let results = viewed.search(&[1.0, 0.0, 0.0, 0.0], 1, 200).unwrap(); - assert_eq!(results[0].id, 1); - } - - #[test] - fn usearch_filtered_search() { - let config = default_config(4); - let index = UsearchIndex::new(config).unwrap(); - index.reserve(10).unwrap(); - - for i in 0..10_u64 { - #[allow(clippy::cast_precision_loss)] - let v = [i as f32 * 0.1, 0.0, 0.0, 0.0]; - index.insert(i, &v).unwrap(); - } - - // Only allow even IDs - let results = index - .filtered_search(&[0.5, 0.0, 0.0, 0.0], 5, 200, &|id| id % 2 == 0) - .unwrap(); - - for r in &results { - assert!( - r.id % 2 == 0, - "odd ID {} in even-only filtered search", - r.id - ); - } - } - - #[test] - fn usearch_reserve() { - let config = default_config(4); - let index = UsearchIndex::new(config).unwrap(); - index.reserve(1000).unwrap(); - assert!(index.is_empty()); // no side effects - } - - /// m12p3: concurrent searches that request DIFFERENT `ef_search` values must - /// each return correct results — the epoch guard must serialize only the - /// beam-width change, never corrupt a neighbour set. We hammer the index from - /// many threads alternating between two `ef` values and assert every result is - /// a valid, in-range, descending-distance neighbour list. - #[test] - fn usearch_concurrent_mixed_ef_search_is_correct() { - use std::sync::Arc; - - let dim = 64; - let n = 2_000_u64; - let config = VectorIndexConfig { - dimensions: dim, - metric: DistanceMetric::L2, - quantization: QuantizationLevel::F16, - connectivity: 16, - ef_construction: 200, - ef_search: 100, - }; - let index = Arc::new(UsearchIndex::new(config).unwrap()); - index.reserve(n as usize).unwrap(); - for id in 0..n { - #[allow(clippy::cast_precision_loss)] - let mut v = vec![0.0_f32; dim]; - v[(id as usize) % dim] = 1.0; - v[(id as usize / dim) % dim] += 0.5; - index.insert(id, &v).unwrap(); - } - - let query: Vec = { - let mut v = vec![0.0_f32; dim]; - v[0] = 1.0; - v - }; - - let mut handles = Vec::new(); - for t in 0..8_u64 { - let index = Arc::clone(&index); - let query = query.clone(); - handles.push(std::thread::spawn(move || { - for i in 0..200_u64 { - // Alternate beam widths so threads contend on the epoch guard. - let ef = if (t + i) % 2 == 0 { 16 } else { 256 }; - let results = index.search(&query, 10, ef).unwrap(); - assert!(results.len() <= 10); - // Distances are ascending (closest first) and in the L2 range. - let mut prev = f32::NEG_INFINITY; - for r in &results { - assert!(r.id < n, "id {} out of range", r.id); - assert!(r.distance >= -1e-3, "negative distance {}", r.distance); - assert!( - r.distance + 1e-3 >= prev, - "results not ascending by distance" - ); - prev = r.distance; - } - } - })); - } - for h in handles { - h.join().unwrap(); - } - } - - #[test] - fn usearch_tombstone_ratio() { - let config = default_config(4); - let index = UsearchIndex::new(config).unwrap(); - index.reserve(10).unwrap(); - - index.insert(1, &[1.0, 0.0, 0.0, 0.0]).unwrap(); - index.insert(2, &[0.0, 1.0, 0.0, 0.0]).unwrap(); - - // No tombstones yet. - assert_eq!(index.len(), 2); - assert_eq!(index.len_live(), 2); - assert!((index.tombstone_ratio() - 0.0).abs() < f64::EPSILON); - - index.delete(1).unwrap(); - // len() still shows 2 (total_slots includes the tombstoned slot). - // len_live() shows 1 (USearch's size() excludes removed entries). - // tombstone_ratio = (2 - 1) / 2 = 0.5 - assert_eq!(index.len(), 2); - assert_eq!(index.len_live(), 1); - assert!((index.tombstone_ratio() - 0.5).abs() < f64::EPSILON); - } -} +// The moved block keeps its original `unwrap_used` allowance; the casts are test +// fixtures building f32 vectors from loop indices, exactly as the sibling test +// modules in this subsystem do. +#[allow(clippy::unwrap_used, clippy::cast_precision_loss)] +#[path = "usearch_index_tests.rs"] +mod tests; diff --git a/tidal/src/storage/vector/usearch_index_tests.rs b/tidal/src/storage/vector/usearch_index_tests.rs new file mode 100644 index 0000000..883e6e9 --- /dev/null +++ b/tidal/src/storage/vector/usearch_index_tests.rs @@ -0,0 +1,367 @@ +use super::*; + +fn default_config(dimensions: usize) -> VectorIndexConfig { + VectorIndexConfig { + dimensions, + metric: DistanceMetric::L2, + quantization: QuantizationLevel::F16, + connectivity: 16, + ef_construction: 200, + ef_search: 200, + } +} + +/// A full HNSW graph must keep accepting NEW keys. +/// +/// Regression for the production defect measured 2026-08-30 on the RF3 +/// cluster: `add` cannot grow the graph, and nothing on the write path +/// reserved, so once `size()` reached the reserved capacity every new +/// embedding failed with "Reserve capacity ahead of insertions!" and +/// returned HTTP 500. Items, signals, searches and re-embeds of EXISTING +/// entities all still worked, so the store had silently become +/// read/update-only for vectors. This asserts inserts continue past the +/// reservation, and that the vectors are actually retrievable afterwards. +#[test] +fn usearch_insert_grows_past_reserved_capacity() { + let index = UsearchIndex::new(default_config(4)).unwrap(); + index.reserve(4).unwrap(); + // Drive off the REPORTED capacity, not the requested one: USearch rounds a + // reservation up (reserve(4) reported 64 here), so a hardcoded insert count + // can sit entirely inside the reservation and never exercise growth at all. + let reserved = index.inner.capacity(); + let target = reserved + 100; + + for i in 0..target as u64 { + let f = i as f32; + index + .insert(i, &[f, f + 1.0, f + 2.0, f + 3.0]) + .unwrap_or_else(|e| panic!("insert {i} failed past reserved capacity {reserved}: {e}")); + } + assert_eq!(index.len_live(), target, "every new key must be indexed"); + assert!( + index.inner.capacity() > reserved, + "capacity should have grown beyond the original {reserved}" + ); + + // The grown graph must still answer, not merely accept writes. + let hits = index.search(&[10.0, 11.0, 12.0, 13.0], 3, 0).unwrap(); + assert!( + !hits.is_empty(), + "search over the grown index returned nothing" + ); + assert_eq!( + hits[0].id, 10, + "nearest neighbour should be the exact match" + ); + + // Re-embedding an existing key must still work after growth. + index.insert(10, &[0.0, 0.0, 0.0, 0.0]).unwrap(); + assert_eq!( + index.len_live(), + target, + "upsert must not change the live count" + ); +} + +#[test] +fn usearch_new_is_empty() { + let index = UsearchIndex::new(default_config(128)).unwrap(); + assert!(index.is_empty()); + assert_eq!(index.len(), 0); + assert_eq!(index.len_live(), 0); +} + +#[test] +fn usearch_insert_and_len() { + let config = default_config(4); + let index = UsearchIndex::new(config).unwrap(); + index.reserve(10).unwrap(); + index.insert(1, &[1.0, 2.0, 3.0, 4.0]).unwrap(); + index.insert(2, &[5.0, 6.0, 7.0, 8.0]).unwrap(); + assert_eq!(index.len(), 2); + assert_eq!(index.len_live(), 2); + assert!(!index.is_empty()); +} + +#[test] +fn usearch_insert_replaces_existing_key() { + // Regression (m12 reseed wedge): `insert` is upsert per the trait + // contract. USearch is `multi: false`, so re-inserting a key must + // REPLACE — not fail with "Duplicate keys not allowed in high-level + // wrappers". This is exactly the reseed path where post-snapshot WAL + // replay re-applies an embedding the snapshot already loaded. + let config = default_config(4); + let index = UsearchIndex::new(config).unwrap(); + index.reserve(10).unwrap(); + + index.insert(1, &[1.0, 0.0, 0.0, 0.0]).unwrap(); + // Re-insert the SAME key with a different vector — must not error. + index.insert(1, &[0.0, 1.0, 0.0, 0.0]).unwrap(); + + // Replacement, not a second slot: exactly one LIVE key. + assert_eq!(index.len_live(), 1); + + // Search reflects the REPLACEMENT vector (distance ~0 to the new one). + let results = index.search(&[0.0, 1.0, 0.0, 0.0], 1, 200).unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].id, 1); + assert!( + results[0].distance < 0.01, + "replacement vector should match query, got distance {}", + results[0].distance + ); +} + +#[test] +fn usearch_dimension_mismatch_insert() { + let config = default_config(4); + let index = UsearchIndex::new(config).unwrap(); + let result = index.insert(1, &[1.0, 2.0]); // 2D into 4D index + assert!(result.is_err()); + match result.unwrap_err() { + VectorError::DimensionMismatch { expected, got } => { + assert_eq!(expected, 4); + assert_eq!(got, 2); + } + other => panic!("expected DimensionMismatch, got {other:?}"), + } +} + +#[test] +fn usearch_dimension_mismatch_search() { + let config = default_config(4); + let index = UsearchIndex::new(config).unwrap(); + index.reserve(2).unwrap(); + index.insert(1, &[1.0, 2.0, 3.0, 4.0]).unwrap(); + let result = index.search(&[1.0, 2.0], 1, 200); // 2D query on 4D index + assert!(result.is_err()); + assert!(matches!( + result.unwrap_err(), + VectorError::DimensionMismatch { + expected: 4, + got: 2 + } + )); +} + +#[test] +fn usearch_delete_not_found() { + let config = default_config(4); + let index = UsearchIndex::new(config).unwrap(); + let result = index.delete(999); + assert!(result.is_err()); + assert!(matches!( + result.unwrap_err(), + VectorError::NotFound { id: 999 } + )); +} + +#[test] +fn usearch_search_empty_index() { + let config = default_config(4); + let index = UsearchIndex::new(config).unwrap(); + let results = index.search(&[1.0, 2.0, 3.0, 4.0], 5, 200).unwrap(); + assert!(results.is_empty()); +} + +#[test] +fn usearch_basic_search() { + let config = default_config(4); + let index = UsearchIndex::new(config).unwrap(); + index.reserve(10).unwrap(); + + index.insert(1, &[1.0, 0.0, 0.0, 0.0]).unwrap(); + index.insert(2, &[0.0, 1.0, 0.0, 0.0]).unwrap(); + index.insert(3, &[0.0, 0.0, 1.0, 0.0]).unwrap(); + + let results = index.search(&[1.0, 0.0, 0.0, 0.0], 3, 200).unwrap(); + assert_eq!(results.len(), 3); + // Closest should be vector 1 (identical to query) + assert_eq!(results[0].id, 1); +} + +#[test] +fn usearch_delete_and_search() { + let config = default_config(4); + let index = UsearchIndex::new(config).unwrap(); + index.reserve(10).unwrap(); + + index.insert(1, &[1.0, 0.0, 0.0, 0.0]).unwrap(); + index.insert(2, &[0.0, 1.0, 0.0, 0.0]).unwrap(); + index.insert(3, &[0.0, 0.0, 1.0, 0.0]).unwrap(); + + index.delete(2).unwrap(); + assert_eq!(index.len_live(), 2); + + let results = index.search(&[0.0, 1.0, 0.0, 0.0], 3, 200).unwrap(); + // Vector 2 must not appear in results + assert!(results.iter().all(|r| r.id != 2)); +} + +#[test] +fn usearch_is_send_and_sync() { + fn assert_send_sync() {} + assert_send_sync::(); +} + +#[test] +fn usearch_save_load_roundtrip() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("test_usearch.idx"); + + let config = default_config(4); + let index = UsearchIndex::new(config.clone()).unwrap(); + index.reserve(10).unwrap(); + index.insert(1, &[1.0, 0.0, 0.0, 0.0]).unwrap(); + index.insert(2, &[0.0, 1.0, 0.0, 0.0]).unwrap(); + + index.save(&path).unwrap(); + + let loaded = UsearchIndex::load(&path, &config).unwrap(); + assert_eq!(loaded.len(), 2); + assert_eq!(loaded.len_live(), 2); + + let results = loaded.search(&[1.0, 0.0, 0.0, 0.0], 1, 200).unwrap(); + assert_eq!(results[0].id, 1); +} + +#[test] +fn usearch_view_readonly() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("test_usearch_view.idx"); + + let config = default_config(4); + let index = UsearchIndex::new(config.clone()).unwrap(); + index.reserve(10).unwrap(); + index.insert(1, &[1.0, 0.0, 0.0, 0.0]).unwrap(); + index.insert(2, &[0.0, 1.0, 0.0, 0.0]).unwrap(); + + index.save(&path).unwrap(); + + let viewed = UsearchIndex::view(&path, &config).unwrap(); + assert_eq!(viewed.len(), 2); + + let results = viewed.search(&[1.0, 0.0, 0.0, 0.0], 1, 200).unwrap(); + assert_eq!(results[0].id, 1); +} + +#[test] +fn usearch_filtered_search() { + let config = default_config(4); + let index = UsearchIndex::new(config).unwrap(); + index.reserve(10).unwrap(); + + for i in 0..10_u64 { + #[allow(clippy::cast_precision_loss)] + let v = [i as f32 * 0.1, 0.0, 0.0, 0.0]; + index.insert(i, &v).unwrap(); + } + + // Only allow even IDs + let results = index + .filtered_search(&[0.5, 0.0, 0.0, 0.0], 5, 200, &|id| id % 2 == 0) + .unwrap(); + + for r in &results { + assert!( + r.id % 2 == 0, + "odd ID {} in even-only filtered search", + r.id + ); + } +} + +#[test] +fn usearch_reserve() { + let config = default_config(4); + let index = UsearchIndex::new(config).unwrap(); + index.reserve(1000).unwrap(); + assert!(index.is_empty()); // no side effects +} + +/// m12p3: concurrent searches that request DIFFERENT `ef_search` values must +/// each return correct results — the epoch guard must serialize only the +/// beam-width change, never corrupt a neighbour set. We hammer the index from +/// many threads alternating between two `ef` values and assert every result is +/// a valid, in-range, descending-distance neighbour list. +#[test] +fn usearch_concurrent_mixed_ef_search_is_correct() { + use std::sync::Arc; + + let dim = 64; + let n = 2_000_u64; + let config = VectorIndexConfig { + dimensions: dim, + metric: DistanceMetric::L2, + quantization: QuantizationLevel::F16, + connectivity: 16, + ef_construction: 200, + ef_search: 100, + }; + let index = Arc::new(UsearchIndex::new(config).unwrap()); + index.reserve(n as usize).unwrap(); + for id in 0..n { + #[allow(clippy::cast_precision_loss)] + let mut v = vec![0.0_f32; dim]; + v[(id as usize) % dim] = 1.0; + v[(id as usize / dim) % dim] += 0.5; + index.insert(id, &v).unwrap(); + } + + let query: Vec = { + let mut v = vec![0.0_f32; dim]; + v[0] = 1.0; + v + }; + + let mut handles = Vec::new(); + for t in 0..8_u64 { + let index = Arc::clone(&index); + let query = query.clone(); + handles.push(std::thread::spawn(move || { + for i in 0..200_u64 { + // Alternate beam widths so threads contend on the epoch guard. + let ef = if (t + i) % 2 == 0 { 16 } else { 256 }; + let results = index.search(&query, 10, ef).unwrap(); + assert!(results.len() <= 10); + // Distances are ascending (closest first) and in the L2 range. + let mut prev = f32::NEG_INFINITY; + for r in &results { + assert!(r.id < n, "id {} out of range", r.id); + assert!(r.distance >= -1e-3, "negative distance {}", r.distance); + assert!( + r.distance + 1e-3 >= prev, + "results not ascending by distance" + ); + prev = r.distance; + } + } + })); + } + for h in handles { + h.join().unwrap(); + } +} + +#[test] +fn usearch_tombstone_ratio() { + let config = default_config(4); + let index = UsearchIndex::new(config).unwrap(); + index.reserve(10).unwrap(); + + index.insert(1, &[1.0, 0.0, 0.0, 0.0]).unwrap(); + index.insert(2, &[0.0, 1.0, 0.0, 0.0]).unwrap(); + + // No tombstones yet. + assert_eq!(index.len(), 2); + assert_eq!(index.len_live(), 2); + assert!((index.tombstone_ratio() - 0.0).abs() < f64::EPSILON); + + index.delete(1).unwrap(); + // len() still shows 2 (total_slots includes the tombstoned slot). + // len_live() shows 1 (USearch's size() excludes removed entries). + // tombstone_ratio = (2 - 1) / 2 = 0.5 + assert_eq!(index.len(), 2); + assert_eq!(index.len_live(), 1); + assert!((index.tombstone_ratio() - 0.5).abs() < f64::EPSILON); +} diff --git a/tidal/src/wal/format/batch.rs b/tidal/src/wal/format/batch.rs index 996ff21..d0d1363 100644 --- a/tidal/src/wal/format/batch.rs +++ b/tidal/src/wal/format/batch.rs @@ -645,6 +645,73 @@ pub enum BatchPayload { Membership(MembershipRecord), } +/// The four blob record kinds, as a dense enum. +/// +/// [`BlobRecord::kind`] already gave the on-disk header byte, but a metrics label +/// and a counter-array index both want a *dense* domain (`0..COUNT`), and the +/// header bytes are 1-based with no compile-time guarantee they stay contiguous. +/// This enum supplies that domain, and [`BlobRecord::blob_kind`] is the single +/// exhaustive match over the variants — `kind()` is derived from it, so adding a +/// `BlobRecord` variant is a compile error in exactly one place rather than a +/// silently-zero counter in three. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +#[repr(u8)] +pub enum BlobKind { + /// Item-metadata mutation, batch kind [`BATCH_KIND_ITEM_METADATA`]. + ItemMetadata = 0, + /// Item-embedding mutation, batch kind [`BATCH_KIND_EMBEDDING`]. + Embedding = 1, + /// Term marker, batch kind [`BATCH_KIND_TERM_MARKER`]. + TermMarker = 2, + /// Cluster-membership record, batch kind [`BATCH_KIND_MEMBERSHIP`]. + Membership = 3, +} + +impl BlobKind { + /// Number of kinds. The bound for any per-kind array; also the exact + /// cardinality of the `kind` metrics label, which is why the label can never + /// blow up regardless of what arrives on the wire. + pub const COUNT: usize = 4; + + /// Every kind, in index order. Lets a renderer walk the closed set without + /// hand-listing variants at the call site. + pub const ALL: [Self; Self::COUNT] = [ + Self::ItemMetadata, + Self::Embedding, + Self::TermMarker, + Self::Membership, + ]; + + /// Dense array index in `0..COUNT`. + #[must_use] + pub const fn index(self) -> usize { + self as usize + } + + /// Stable Prometheus label value. Changing one of these renames a live series, + /// so treat them as part of the metrics wire contract. + #[must_use] + pub const fn label(self) -> &'static str { + match self { + Self::ItemMetadata => "item_metadata", + Self::Embedding => "embedding", + Self::TermMarker => "term_marker", + Self::Membership => "membership", + } + } + + /// The on-disk batch-header byte for this kind. + #[must_use] + pub const fn batch_kind(self) -> u8 { + match self { + Self::ItemMetadata => BATCH_KIND_ITEM_METADATA, + Self::Embedding => BATCH_KIND_EMBEDDING, + Self::TermMarker => BATCH_KIND_TERM_MARKER, + Self::Membership => BATCH_KIND_MEMBERSHIP, + } + } +} + /// One blob (kind-1/2) record submitted for a WAL append. /// /// The writer-side counterpart of the blob arms of [`BatchPayload`]: a blob @@ -664,15 +731,26 @@ pub enum BlobRecord { } impl BlobRecord { + /// This record's [`BlobKind`]. + /// + /// **The one exhaustive match over `BlobRecord`'s variants.** Everything that + /// needs to discriminate a blob — the header byte, the metrics label, the + /// counter index — routes through here, so a new variant fails to compile in a + /// single place instead of silently miscounting at each call site. + #[must_use] + pub const fn blob_kind(&self) -> BlobKind { + match self { + Self::ItemMetadata(_) => BlobKind::ItemMetadata, + Self::Embedding(_) => BlobKind::Embedding, + Self::TermMarker(_) => BlobKind::TermMarker, + Self::Membership(_) => BlobKind::Membership, + } + } + /// The batch kind this record encodes as (header `flags` byte). #[must_use] pub const fn kind(&self) -> u8 { - match self { - Self::ItemMetadata(_) => BATCH_KIND_ITEM_METADATA, - Self::Embedding(_) => BATCH_KIND_EMBEDDING, - Self::TermMarker(_) => BATCH_KIND_TERM_MARKER, - Self::Membership(_) => BATCH_KIND_MEMBERSHIP, - } + self.blob_kind().batch_kind() } /// The entity this mutation targets (0 for control records — a term marker