diff --git a/.env.example b/.env.example index ed47929..364a10c 100644 --- a/.env.example +++ b/.env.example @@ -77,9 +77,24 @@ RUST_LOG=compass=info # past the budget; detached collections re-attach on demand). 0 = unbounded. # COMPASS_MAX_ATTACHED=0 -# ── Telemetry (anonymous; opt out) ────────────────────────────────────────── +# ── Serve-from-storage (cold reads) ───────────────────────────────────────── +# Serve-from-storage: semantic queries on UNATTACHED collections are answered +# directly from object storage (a few range reads, ~100s of ms) instead of +# waiting for a full index rebuild. Implies COMPASS_LAZY_ATTACH. Cloud only. +# COMPASS_COLD_SERVE=true +# Cold hits on a namespace before a background attach warms it (0 = never). +# COMPASS_WARM_AFTER=3 +# IVF clusters probed per segment per cold query (recall/latency knob). +# Default 8 reaches warm-parity recall on clustered embedding spaces; raise +# it for unstructured vector data (see docs/search-quality.md). +# COMPASS_COLD_NPROBE=8 + # Global in-flight request cap (backpressure). Unset = effectively unlimited. # COMPASS_MAX_CONCURRENCY=1024 -# COMPASS_TELEMETRY=off -# DO_NOT_TRACK=1 +# ── Telemetry (anonymous; OPT-IN, off by default) ─────────────────────────── +# Compass never phones home unless you set this. When on, it sends a startup +# event + daily heartbeat (random instance id, version, OS/arch, collection +# and vector counts — never document content or queries). DO_NOT_TRACK=1 is +# honored even when opted in. +# COMPASS_TELEMETRY=on diff --git a/.github/ISSUE_TEMPLATE/bug_report.md b/.github/ISSUE_TEMPLATE/bug_report.md deleted file mode 100644 index 5aae13c..0000000 --- a/.github/ISSUE_TEMPLATE/bug_report.md +++ /dev/null @@ -1,51 +0,0 @@ ---- -name: Bug Report -about: Report a bug or unexpected behavior -title: "[BUG] " -labels: bug -assignees: '' - ---- - -## Describe the bug - -A clear and concise description of what the bug is. - -## Steps to reproduce - -1. ... -2. ... -3. ... - -## Expected behavior - -What should happen. - -## Actual behavior - -What actually happens instead. - -## Environment - -- **Compass version** (or git SHA): -- **OS and architecture**: -- **Rust version** (`rustc --version`): -- **Data directory size**: -- **Collection size** (approximate): - -## Logs - -If applicable, reproduce with debug logging: -```bash -RUST_LOG=compass=debug ./compass -``` - -Then paste relevant log output here: - -``` -[paste logs] -``` - -## Additional context - -Any other context that might be helpful. diff --git a/.github/ISSUE_TEMPLATE/config.yml b/.github/ISSUE_TEMPLATE/config.yml index f6dbf1f..91c1c2f 100644 --- a/.github/ISSUE_TEMPLATE/config.yml +++ b/.github/ISSUE_TEMPLATE/config.yml @@ -1,5 +1,5 @@ blank_issues_enabled: true contact_links: - name: Security vulnerability - url: mailto:security@runcaptain.com + url: mailto:support@runcaptain.com about: Report security issues privately via email — do not open a public issue. diff --git a/.github/ISSUE_TEMPLATE/feature_request.md b/.github/ISSUE_TEMPLATE/feature_request.md deleted file mode 100644 index 4cf58dc..0000000 --- a/.github/ISSUE_TEMPLATE/feature_request.md +++ /dev/null @@ -1,28 +0,0 @@ ---- -name: Feature Request -about: Suggest an idea for Compass -title: "[FEATURE] " -labels: enhancement -assignees: '' - ---- - -## Description - -A clear and concise description of what you'd like to see. - -## Motivation - -Why should this feature exist? What problem does it solve? - -## Proposed solution - -Describe how you'd like the feature to work. - -## Alternatives - -Have you considered any alternative approaches? - -## Additional context - -Any other context or examples (e.g., similar features in other projects). diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 79e5f95..d8ff15f 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -27,7 +27,9 @@ jobs: - name: Verify tag matches Cargo.toml run: | - CARGO_VERSION=$(grep "^version" crates/compass/Cargo.toml | head -1 | sed 's/.*"\([^"]*\)".*/\1/') + # The version lives in [workspace.package] in the ROOT manifest; + # crate manifests say `version.workspace = true`. + CARGO_VERSION=$(grep "^version" Cargo.toml | head -1 | sed 's/.*"\([^"]*\)".*/\1/') if [ "${{ steps.version.outputs.version }}" != "$CARGO_VERSION" ]; then echo "Tag version ${{ steps.version.outputs.version }} does not match Cargo.toml version $CARGO_VERSION" exit 1 diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index ff7cb09..bb54bed 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -43,23 +43,27 @@ crates/compass/src/ mod.rs SearchMode enum + re-exports. backend.rs VectorIndex trait shim. UsearchHnswIndex (CPU) lives here. vector.rs USearch HNSW build + search + persistence (CPU primitives). - tantivy_fts.rs Full-text search via Tantivy (BM25). + tantivy_fts.rs Full-text search via Tantivy (BM25) + facet treemaps. hybrid.rs Reciprocal Rank Fusion (RRF, k=60) over FTS + semantic. + ivf.rs IVF clustering built at compaction (cold-read layout). + cold.rs Serve-from-storage query path (range reads, no attach). + filter_index.rs Roaring-treemap metadata filter index (warm pushdown). + chunk_store.rs / chunk_cache.rs redb chunk store + bounded LRU cache. + collections/ + partitions.rs Tenant-partition routing helpers. + cloud.rs Segment codec (CSEG0003), materialize, bucket config. + storage/ Storage trait + local disk + object-store backends + LSM. + metrics.rs /metrics counters. telemetry.rs: opt-in usage pings. ``` -## Vector backend abstraction +## Vector backends -All vector backends implement `compass_index_api::VectorIndex`. The default backend is `UsearchHnswIndex` (CPU, mmap-backed, disk-persistent). The opt-in GPU backend is `compass_vector_gpu::CuvsHnswIndex` (CAGRA build on GPU, HNSW search on CPU). - -Selection happens at startup in `search::backend::build_backend`, driven by the `COMPASS_BACKEND` environment variable: - -| Value | Behavior | -|-------|----------| -| `cpu` (default) | USearch on CPU. Always available. | -| `gpu` | cuVS on GPU. Requires the `gpu` feature and a CUDA-capable device. Falls back to CPU with a warning if either is missing. | -| `auto` | Probe for GPU, fall back to CPU silently if unavailable. | - -The trait is intentionally narrow: `build`, `add`, `search`, `len`, `dims`, `save`, `backend_name`. New backends should fit through this surface or extend it via a follow-up trait, not by branching on a concrete type. +The engine uses USearch HNSW directly (CPU, mmap-backed, disk-persistent) +for warm serving, plus an IVF layout inside segments (`search/ivf.rs`) for +serve-from-storage cold reads. `compass-index-api` (a narrow `VectorIndex` +trait) and `compass-vector-gpu` (cuVS) exist as standalone crates for a +future GPU integration but are NOT wired into the engine — there is no +`COMPASS_BACKEND` knob and no `gpu` feature on the `compass` crate today. ## Storage layout @@ -80,8 +84,8 @@ data// In cloud mode the object-storage bucket additionally holds, per collection: `collection.json` (bucket config), `manifest` (LSM manifest, CAS-committed), -`wal/{uuid}.frag` (WAL fragments), `segments/{uuid}` (CSEG0002 sectioned -segments), and `id-alloc` (CAS-leased chunk-id blocks). +`wal/{uuid}.frag` (WAL fragments), `segments/{uuid}` (CSEG0003 sectioned +segments: row-addressable metadata + IVF-clustered vectors; v2 readable), and `id-alloc` (CAS-leased chunk-id blocks). The disk format is the contract. Bumping it requires a migration path documented in CHANGELOG.md. @@ -118,7 +122,8 @@ cuVS CAGRA build on an A10G runs ~12x faster than USearch CPU build at the same 1. Create a new crate `crates/compass-vector-/`. 2. Depend on `compass-index-api` (workspace dep) and your backend library. 3. Implement `VectorIndex` (and `LoadableIndex` if loading from disk makes sense). -4. Add a `#[cfg(feature = "")]`-gated branch in `search::backend::build_backend`. +4. Wire it into the engine (there is currently no runtime backend selector — + proposing that wiring is part of such a PR; open an issue first). 5. Document the build prerequisites in `ARCHITECTURE.md` (this file). 6. Add a smoke binary under `src/bin/` that builds, queries, and prints latency. diff --git a/CHANGELOG.md b/CHANGELOG.md index 2a577d4..80965bb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ## [Unreleased] -### Added — tenant-partitioned collections (Phase 6) +### Added — serve-from-storage ("true serverless") + +- **`COMPASS_COLD_SERVE=true`**: semantic queries on collections (and tenant partitions) that are NOT attached are answered directly from object storage — a manifest read, cached centroid/TOC artifacts, and a handful of range-GETs — instead of triggering a full index rebuild. Compaction now writes IVF-clustered vector sections (`cent:`/`clu:`, k-means, unit-normalized) plus a row-addressable metadata index (`meta2`/`metaidx`) into segments (format CSEG0003; v2 segments remain readable, pre-v0.4 readers fail loudly on v3). Cold reads see the full committed state including the WAL tail and tombstones, so read-your-writes holds by construction; metadata filters apply; FTS on a cold namespace returns a clear error (inverted indexes still need an attach). Repeated cold hits (`COMPASS_WARM_AFTER`, default 3) promote a background attach so hot namespaces migrate to the fast path on their own. RAM per cold namespace is megabytes (centroids + directories), independent of collection size. + +### Added — tenant-partitioned collections - **`config.partition_by`**: create a collection partitioned by a metadata field (e.g. `tenant_id`) and every chunk routes to an internal per-tenant partition — a full engine namespace (own LSM, indexes, attach/evict lifecycle) behind one collection API. Searches and deletes filter by the partition field (exact → one partition; `{"in": [...]}` fans out up to 16, merged by score); chunk ids are collection-unique via the parent's CAS id allocator; partitions auto-create on first ingest (writer role included), attach on demand, are hidden from listings, and cascade-delete with the parent. This moves the scale envelope from per-collection to per-tenant: RAM and refresh cost track the HOT tenant set, so one collection can hold billions of vectors across tenants while serving on bounded memory. Not yet routed on partitioned collections (clear errors): relations, facets, TAMS lookup, vector-space CRUD. @@ -31,9 +35,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), - Pork audit (three independent review passes): −1,200 lines of dead weight removed — the unwired VectorIndex/GPU backend plumbing (`COMPASS_BACKEND` did nothing), a third never-called filter evaluator, never-wired filter-index persistence codecs, the legacy vector writer, the `rayon` dependency, and assorted dead fields/params. `delete_by_filter` now resolves ids through the same roaring filter-index pushdown as search (one filter semantics, not three). The `dead_code` lint is enabled again crate-wide. `collections/mod.rs` shrank from 7,100 to 3,700 lines (test modules extracted to files). +### Changed (behavior) + +- **Telemetry is now opt-in** (`COMPASS_TELEMETRY=on`); previously it defaulted on. An engine whose promise is "data never leaves your machine" should not phone home by default. + ### Scope & limitations (honest) -- Warm, not cold: attach cost is proportional to collection size until the sectioned segment format + serve-from-storage indexes land (roadmap Phases 5–6). Cross-node convergence is periodic (refresh interval), not synchronous — use `min_seq` when you need read-your-writes. +- Cold serving is semantic-only: full-text (and hybrid-with-text) queries on a cold namespace return a clear error until it warms — BM25 still needs local indexes. Cold recall depends on embedding-space structure (see docs/search-quality.md); the default nprobe reaches warm parity on clustered embeddings. Cross-node convergence is periodic (refresh interval), not synchronous — use `min_seq` when you need read-your-writes (cold reads have it by construction). ## [0.3.0] - 2026-07-03 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 253ef23..a8dc67f 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -15,15 +15,24 @@ cargo run --release # serves on http://localhost:4001 Compass is a Cargo workspace. Useful invocations: +Prerequisites on Linux: `cmake`, `pkg-config`, `libssl-dev` (what CI +installs). On Windows there is a known linker clash between `esaxx-rs` and +`cxx` — use `cargo check` locally and run builds/tests in Docker or WSL. + ```bash -cargo build # builds the default member (`compass`) -cargo build -p compass-index-api # builds just the trait crate -cargo build --features gpu # adds the GPU backend (Linux + CUDA only) -cargo test --workspace # runs all tests in all crates -cargo clippy --workspace -- -D warnings # lint check (CI requires zero warnings) -cargo fmt --all --check # format check (CI requires clean diff) +cargo build # default member (`compass`) +cargo test --workspace --exclude compass-vector-gpu # all tests CI runs +cargo test -p compass --features object-storage # + S3/GCS backend tests +cargo clippy --workspace --exclude compass-vector-gpu --all-targets -- -D warnings +cargo fmt --all --check # CI requires a clean diff ``` +`compass-vector-gpu` is a standalone experimental crate (cuVS; needs CUDA +12+, CMake, a long first build) that is NOT wired into the engine yet — +every CI job excludes it, and so should you unless you're working on it. +Tests against real object storage skip cleanly unless `COMPASS_TEST_S3_BUCKET` +is set (CI runs them against MinIO). + See [`ARCHITECTURE.md`](ARCHITECTURE.md) for the module map and where to put new code. ## What we accept @@ -35,9 +44,8 @@ See [`ARCHITECTURE.md`](ARCHITECTURE.md) for the module map and where to put new - Documentation improvements, especially examples. **Out of scope (for now):** -- Storage backends other than the local filesystem. - New embedding model integrations (we plug into HuggingFace TEI / vLLM via the `embed_endpoint` config; pull requests adding new in-process embedders need a strong motivation). -- Cluster / replication features (Compass is single-node by design; horizontal scaling is via sharding behind a load balancer). +- Consensus/quorum replication. Compass scales out via object storage as the source of truth (stateless writers + serving nodes + serve-from-storage cold reads); PRs should build on that model, not introduce node-to-node coordination. If you're not sure, open an issue first and ask. @@ -45,7 +53,7 @@ If you're not sure, open an issue first and ask. - [ ] `cargo fmt --all` clean. - [ ] `cargo clippy --workspace -- -D warnings` clean. -- [ ] `cargo test --workspace` green. +- [ ] `cargo test --workspace --exclude compass-vector-gpu` green. - [ ] `CHANGELOG.md` updated under the `[Unreleased]` section. - [ ] Public API changes have rustdoc comments. - [ ] Behavior changes have a test that would have caught the regression. @@ -75,7 +83,7 @@ Open an issue with: ## Reporting security issues -Don't open a public issue. Email `founders@runcaptain.com` with the details. We'll acknowledge within two business days. +Don't open a public issue. Email `support@runcaptain.com` with the details. We'll acknowledge within two business days. ## Code of conduct diff --git a/Cargo.toml b/Cargo.toml index 3e43044..241cf58 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,8 @@ default-members = ["crates/compass"] [workspace.package] version = "0.3.0" edition = "2021" -authors = ["Captain Technologies "] +authors = ["Captain Technologies "] +license = "Apache-2.0" repository = "https://github.com/runcaptain/compass" homepage = "https://runcaptain.com" rust-version = "1.88" diff --git a/README.md b/README.md index 0318bc5..8d5041a 100644 --- a/README.md +++ b/README.md @@ -437,7 +437,18 @@ Supported: `s3://bucket[/prefix]` (AWS S3, MinIO, Cloudflare R2 — set `COMPASS In this mode the bucket is the **source of truth**: every write lands durably in object storage first (an LSM of immutable WAL fragments + a CAS-committed manifest), and a node that boots with an empty disk discovers its collections from the bucket and rebuilds all local indexes — chunks, hierarchy, and typed relations included. Compaction (automatic past a WAL threshold, or via `POST /compact`) folds fragments into segments and physically reclaims deleted data. -Scope, honestly: reads are served from the locally rebuilt indexes (durable-via-cloud, fast-via-local) — this is not stateless multi-node serving, and cold-start recovery materializes the live set in RAM. See [CHANGELOG](CHANGELOG.md) for details. +### Serverless topologies + +With the bucket as the source of truth, nodes become disposable roles you mix per workload (full reference: [docs/deployment.md](docs/deployment.md)): + +- **Serving node** (default): full local indexes, fast reads; converges on other nodes' writes via a background manifest refresher (`COMPASS_REFRESH_INTERVAL`, default 5s). Writes return a `seq`; pass it back as `min_seq` for read-your-writes. +- **Writer node** (`COMPASS_ROLE=writer`): stateless, append-only, boots in milliseconds, refuses reads. Durable immediately; searchable on serving nodes within the refresh interval. +- **Cold serving** (`COMPASS_COLD_SERVE=true`): answers *semantic* queries on collections it has never attached, straight from object-storage range reads — first query in ~tens of ms instead of a minutes-long index rebuild. Repeated hits promote a background attach (`COMPASS_WARM_AFTER`). Full-text on a cold collection returns a clear error until it warms. Recall characteristics: [docs/search-quality.md](docs/search-quality.md). +- **Lazy attach + LRU** (`COMPASS_LAZY_ATTACH`, `COMPASS_MAX_ATTACHED`): boot registers namespaces without loading them; RAM tracks the hot set. + +### Multi-tenant partitions + +Create a collection with `"config": {"partition_by": "tenant_id"}` and every chunk routes to an internal per-tenant partition — its own indexes and attach/evict lifecycle behind one collection API. Searches and deletes filter by the partition field (exact match, or `{"in": [...]}` to fan out across up to 16 tenants); chunk ids stay collection-unique; partitions auto-create on first ingest and cascade-delete with the parent. Serving RAM tracks the hot-tenant set, not the tenant count — this also works fully offline in local mode. For local development against MinIO: @@ -569,16 +580,20 @@ PUT /collections/:name/default-vector-space Switch default space GET /collections/:name/segments/at Temporal segment lookup (TAMS) GET /health Health check -GET /metrics Prometheus-text metrics +GET /metrics Prometheus-text metrics (unauthenticated by design, like /health — exposes collection names + counts; firewall it if that matters) ``` ## Contributing See [CONTRIBUTING.md](CONTRIBUTING.md) for development setup, PR guidelines, and commit conventions. +## Telemetry + +Telemetry is **off by default** — Compass never phones home unless you set `COMPASS_TELEMETRY=on`. When opted in, it sends a startup event and a daily heartbeat to PostHog (random instance id, version, OS/arch, collection and vector counts — never document content, queries, or metadata). `DO_NOT_TRACK=1` is honored even when opted in. + ## Security -To report a vulnerability, email **security@runcaptain.com**. See [SECURITY.md](SECURITY.md) for details. +To report a vulnerability, email **support@runcaptain.com**. See [SECURITY.md](SECURITY.md) for details. ## License diff --git a/SECURITY.md b/SECURITY.md index 673ef1c..3510741 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -2,7 +2,7 @@ ## Reporting a Vulnerability -We take the security of Compass seriously. If you discover a security vulnerability, please email founders@runcaptain.com with the following information: +We take the security of Compass seriously. If you discover a security vulnerability, please email support@runcaptain.com with the following information: 1. **Description** of the vulnerability 2. **Steps to reproduce** (if applicable) @@ -28,7 +28,7 @@ We will acknowledge your report within **two business days** and work with you t ### Model Weights -- Compass downloads model weights on first run (e.g., BGE-small via Hugging Face Hub). +Compass never downloads anything at runtime. Model weights are fetched only if you run `scripts/download-models.sh` (or `huggingface-cli`) yourself. - Verify downloaded files match expected checksums when possible. - For air-gapped deployments, pre-download and verify model weights before use. diff --git a/crates/compass-index-api/Cargo.toml b/crates/compass-index-api/Cargo.toml index 89f299d..8afefa0 100644 --- a/crates/compass-index-api/Cargo.toml +++ b/crates/compass-index-api/Cargo.toml @@ -3,6 +3,7 @@ name = "compass-index-api" version.workspace = true edition.workspace = true authors.workspace = true +license.workspace = true repository.workspace = true homepage.workspace = true rust-version.workspace = true diff --git a/crates/compass-vector-gpu/Cargo.toml b/crates/compass-vector-gpu/Cargo.toml index 3ab0b38..5b3b5d1 100644 --- a/crates/compass-vector-gpu/Cargo.toml +++ b/crates/compass-vector-gpu/Cargo.toml @@ -3,6 +3,7 @@ name = "compass-vector-gpu" version.workspace = true edition.workspace = true authors.workspace = true +license.workspace = true repository.workspace = true homepage.workspace = true rust-version.workspace = true diff --git a/crates/compass/Cargo.toml b/crates/compass/Cargo.toml index 088c101..2e89457 100644 --- a/crates/compass/Cargo.toml +++ b/crates/compass/Cargo.toml @@ -3,6 +3,7 @@ name = "compass" version.workspace = true edition.workspace = true authors.workspace = true +license.workspace = true repository.workspace = true homepage.workspace = true rust-version.workspace = true @@ -14,10 +15,10 @@ path = "src/main.rs" [features] default = [] -# Requires CUDA 12+ and a Linux host. See ARCHITECTURE.md for build details. -# Opt-in object-storage backend (S3/GCS/Azure/MinIO) via the object_store crate. -# Off by default — local-first deployments pull in zero extra dependencies. -object-storage = ["dep:object_store", "dep:futures"] +# Opt-in object-storage backend (S3/GCS/Azure/MinIO) via the object_store +# crate. Off by default; local-first builds skip the object_store dependency +# tree (`futures` is unconditional — the cold read path uses it). +object-storage = ["dep:object_store"] [dependencies] # Internal trait crate — defines VectorIndex, IndexParams, IndexError. @@ -50,7 +51,7 @@ thiserror = { workspace = true } lru = { workspace = true } # Object-storage backend deps (optional, enabled by the `object-storage` feature). object_store = { workspace = true, optional = true } -futures = { workspace = true, optional = true } +futures = { workspace = true } reqwest = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } diff --git a/crates/compass/src/api/mod.rs b/crates/compass/src/api/mod.rs index 6eb0096..7c45a40 100644 --- a/crates/compass/src/api/mod.rs +++ b/crates/compass/src/api/mod.rs @@ -4,7 +4,9 @@ // vector space CRUD, rebuild triggers, and status checks. // // Bearer-token auth middleware is applied to all routes -// except /health. See `AuthConfig` and `auth_middleware` below. +// except /health and /metrics (both unauthenticated by design; /metrics +// exposes collection names + counts — firewall it if that matters). +// See `AuthConfig` and `auth_middleware` below. pub mod collections; pub mod delete; diff --git a/crates/compass/src/collections/cloud.rs b/crates/compass/src/collections/cloud.rs index b9b0c19..bfa8191 100644 --- a/crates/compass/src/collections/cloud.rs +++ b/crates/compass/src/collections/cloud.rs @@ -164,48 +164,77 @@ pub struct Segment { } /// v2 binary segment magic. v1 segments are JSON (decoded via fallback). -const SEG_MAGIC_V2: [u8; 8] = *b"CSEG0002"; - -/// Encode a segment in the v2 sectioned binary layout: +pub(crate) const SEG_MAGIC_V2: [u8; 8] = *b"CSEG0002"; +/// v3 adds serve-from-storage sections: row-addressable chunk metadata +/// (`meta2` + `metaidx`) and IVF-clustered vectors (`cent:`/`clu:` replace +/// `emb:` for spaces past the clustering threshold). v3 readers decode v2; +/// v2 readers FAIL LOUDLY on v3 (magic mismatch) rather than silently +/// dropping sections — do not mix pre-v0.5 readers with v0.5 writers. +pub(crate) const SEG_MAGIC_V3: [u8; 8] = *b"CSEG0003"; + +/// Encode a segment in the v3 sectioned binary layout: /// `[magic][u64 max_id][u32 toc_len][toc JSON][sections...]` -/// Sections: `meta` (JSON chunks with embeddings STRIPPED), `emb:` -/// (`[u32 dims][u64 n][n × (u64 id + dims×f32 LE)]`), `rels` (JSON), -/// `tombs` (u64 LE array), `rtombs` (JSON ids). Embeddings dominate segment -/// size; storing them as raw f32 instead of JSON decimals is ~10× smaller and -/// range-readable by section. -pub fn encode_segment_v2(seg: &Segment) -> Result, StorageError> { - let err = |e: String| StorageError::Io(format!("segment v2 encode: {e}")); +/// Sections: +/// `meta2` — concatenated per-chunk JSON rows (embeddings stripped); +/// each row independently parseable so cold reads can fetch +/// single chunks by byte range +/// `metaidx` — `[u64 n][n × (u64 id, u64 off, u32 len)]` sorted by id, +/// offsets into `meta2` +/// `emb:` — flat `[u32 dims][u64 n][n × (u64 id + dims×f32)]`, +/// only for spaces below the clustering threshold +/// `cent:` / `clu:` — IVF centroids + clustered vectors +/// (see search/ivf.rs) for spaces at/above the threshold; +/// vectors are stored L2-normalized +/// `rels` (JSON), `tombs` (u64 LE array), `rtombs` (JSON ids) +pub fn encode_segment_v3(seg: &Segment) -> Result, StorageError> { + let err = |e: String| StorageError::Io(format!("segment v3 encode: {e}")); let mut sections: Vec<(String, Vec)> = Vec::new(); - let mut meta_chunks: Vec = Vec::with_capacity(seg.chunks.len()); + // Row-addressable metadata + index (sorted by id for range lookups). + let mut sorted: Vec<&DocumentChunk> = seg.chunks.iter().collect(); + sorted.sort_by_key(|c| c.id); + let mut meta2 = Vec::new(); + let mut metaidx = Vec::with_capacity(8 + sorted.len() * 20); + metaidx.extend_from_slice(&(sorted.len() as u64).to_le_bytes()); let mut by_space: std::collections::BTreeMap)>> = std::collections::BTreeMap::new(); - for c in &seg.chunks { + for c in sorted { let mut m = c.clone(); for (space, emb) in std::mem::take(&mut m.embeddings) { by_space.entry(space).or_default().push((c.id, emb)); } - meta_chunks.push(m); + let row = serde_json::to_vec(&m).map_err(|e| err(e.to_string()))?; + metaidx.extend_from_slice(&c.id.to_le_bytes()); + metaidx.extend_from_slice(&(meta2.len() as u64).to_le_bytes()); + metaidx.extend_from_slice(&(row.len() as u32).to_le_bytes()); + meta2.extend_from_slice(&row); } - sections.push(( - "meta".into(), - serde_json::to_vec(&meta_chunks).map_err(|e| err(e.to_string()))?, - )); + sections.push(("meta2".into(), meta2)); + sections.push(("metaidx".into(), metaidx)); + for (space, rows) in by_space { - let dims = rows.first().map(|(_, v)| v.len()).unwrap_or(0) as u32; - let mut buf = Vec::with_capacity(12 + rows.len() * (8 + dims as usize * 4)); - buf.extend_from_slice(&dims.to_le_bytes()); - buf.extend_from_slice(&(rows.len() as u64).to_le_bytes()); - for (id, v) in &rows { - if v.len() as u32 != dims { + let dims = rows.first().map(|(_, v)| v.len()).unwrap_or(0); + for (_, v) in &rows { + if v.len() != dims { return Err(err(format!("ragged dims in space '{space}'"))); } - buf.extend_from_slice(&id.to_le_bytes()); - for x in v { - buf.extend_from_slice(&x.to_le_bytes()); + } + if rows.len() >= crate::search::ivf::CLUSTER_MIN_ROWS && dims > 0 { + let (cent, clu) = crate::search::ivf::build_sections(rows, dims); + sections.push((format!("cent:{space}"), cent)); + sections.push((format!("clu:{space}"), clu)); + } else { + let mut buf = Vec::with_capacity(12 + rows.len() * (8 + dims * 4)); + buf.extend_from_slice(&(dims as u32).to_le_bytes()); + buf.extend_from_slice(&(rows.len() as u64).to_le_bytes()); + for (id, v) in &rows { + buf.extend_from_slice(&id.to_le_bytes()); + for x in v { + buf.extend_from_slice(&x.to_le_bytes()); + } } + sections.push((format!("emb:{space}"), buf)); } - sections.push((format!("emb:{space}"), buf)); } sections.push(( "rels".into(), @@ -227,7 +256,7 @@ pub fn encode_segment_v2(seg: &Segment) -> Result, StorageError> { .collect(); let toc_bytes = serde_json::to_vec(&toc).map_err(|e| err(e.to_string()))?; let mut out = Vec::new(); - out.extend_from_slice(&SEG_MAGIC_V2); + out.extend_from_slice(&SEG_MAGIC_V3); out.extend_from_slice(&seg.max_id.to_le_bytes()); out.extend_from_slice(&(toc_bytes.len() as u32).to_le_bytes()); out.extend_from_slice(&toc_bytes); @@ -237,7 +266,7 @@ pub fn encode_segment_v2(seg: &Segment) -> Result, StorageError> { Ok(out) } -fn decode_segment_v2(bytes: &[u8]) -> Result { +fn decode_segment_sectioned(bytes: &[u8]) -> Result { let err = |e: String| StorageError::Io(format!("segment v2 decode: {e}")); let need = |n: usize, have: usize| -> Result<(), StorageError> { if have < n { @@ -259,6 +288,7 @@ fn decode_segment_v2(bytes: &[u8]) -> Result { ..Default::default() }; let mut embs: HashMap>> = HashMap::new(); + let mut cent_dims: HashMap = HashMap::new(); for (name, len) in toc { let len = len as usize; need(pos + len, bytes.len())?; @@ -266,6 +296,28 @@ fn decode_segment_v2(bytes: &[u8]) -> Result { pos += len; if name == "meta" { seg.chunks = serde_json::from_slice(body).map_err(|e| err(e.to_string()))?; + } else if name == "meta2" { + // v3 row-addressable metadata: concatenated standalone JSON rows. + let mut de = serde_json::Deserializer::from_slice(body).into_iter::(); + for c in de.by_ref() { + seg.chunks.push(c.map_err(|e| err(e.to_string()))?); + } + } else if name == "metaidx" { + // Full decode doesn't need the index (meta2 rows stream in order); + // it exists for cold range reads. + } else if name.starts_with("cent:") { + let c = crate::search::ivf::parse_cent(body) + .ok_or_else(|| err(format!("bad {name} section")))?; + cent_dims.insert(name.clone(), c.dims); + } else if let Some(space) = name.strip_prefix("clu:") { + let cent_key = format!("cent:{space}"); + let dims = cent_dims + .get(¢_key) + .copied() + .ok_or_else(|| err(format!("clu:{space} without preceding cent section")))?; + for (id, v) in crate::search::ivf::parse_cluster_rows(body, dims) { + embs.entry(id).or_default().insert(space.to_string(), v); + } } else if let Some(space) = name.strip_prefix("emb:") { need(12, body.len())?; let dims = u32::from_le_bytes(body[0..4].try_into().unwrap()) as usize; @@ -310,7 +362,7 @@ pub fn encode_segment( relations: &[ChunkRelation], max_id: u64, ) -> Result, StorageError> { - encode_segment_v2(&Segment { + encode_segment_v3(&Segment { version: 2, chunks: chunks.to_vec(), relations: relations.to_vec(), @@ -323,8 +375,8 @@ pub fn encode_segment( fn decode_segment(bytes: &[u8]) -> Result { // v2 binary (magic-tagged) first; then v1 JSON object; then the oldest // bare-JSON-array form. - if bytes.len() >= 8 && bytes[0..8] == SEG_MAGIC_V2 { - return decode_segment_v2(bytes); + if bytes.len() >= 8 && (bytes[0..8] == SEG_MAGIC_V2 || bytes[0..8] == SEG_MAGIC_V3) { + return decode_segment_sectioned(bytes); } if let Ok(seg) = serde_json::from_slice::(bytes) { return Ok(seg); @@ -678,8 +730,8 @@ mod tests { tombstones: vec![7, 9], relation_tombstones: vec!["dead".into()], }; - let bytes = encode_segment_v2(&seg).unwrap(); - assert_eq!(&bytes[0..8], b"CSEG0002"); + let bytes = encode_segment_v3(&seg).unwrap(); + assert_eq!(&bytes[0..8], b"CSEG0003"); let back = decode_segment(&bytes).unwrap(); assert_eq!(back.max_id, 42); assert_eq!(back.tombstones, vec![7, 9]); @@ -691,6 +743,45 @@ mod tests { assert_eq!(back.relations.len(), 1); } + // Past the clustering threshold the encoder emits cent:/clu: instead of + // emb:; the full decode must reconstruct every chunk's (normalized) + // embedding from the clustered layout. + #[test] + fn segment_v3_clustered_roundtrip() { + let n = crate::search::ivf::CLUSTER_MIN_ROWS + 100; + let chunks: Vec = (0..n as u64) + .map(|i| { + let mut c = chunk(i, &format!("t{i}")); + let v: Vec = (0..8).map(|d| ((i + d) % 13) as f32 + 1.0).collect(); + c.embeddings.insert("default".into(), v); + c + }) + .collect(); + let seg = Segment { + version: 2, + chunks, + relations: vec![], + max_id: n as u64, + tombstones: vec![], + relation_tombstones: vec![], + }; + let bytes = encode_segment_v3(&seg).unwrap(); + let toc_len = u32::from_le_bytes(bytes[16..20].try_into().unwrap()) as usize; + let toc = std::str::from_utf8(&bytes[20..20 + toc_len]).unwrap(); + assert!(toc.contains("cent:default"), "toc: {toc}"); + assert!(toc.contains("clu:default"), "toc: {toc}"); + assert!(!toc.contains("emb:default"), "toc: {toc}"); + + let back = decode_segment(&bytes).unwrap(); + assert_eq!(back.chunks.len(), n); + for c in &back.chunks { + let v = &c.embeddings["default"]; + assert_eq!(v.len(), 8); + let norm: f32 = v.iter().map(|x| x * x).sum::().sqrt(); + assert!((norm - 1.0).abs() < 1e-4, "clu vectors are unit-norm"); + } + } + // A delete folded into a NEWER tail segment must erase a chunk living in // an OLDER segment at materialize time (cross-segment tombstones). #[tokio::test] @@ -721,7 +812,7 @@ mod tests { "ns", &v1, &m1, - Bytes::from(encode_segment_v2(&tail).unwrap()), + Bytes::from(encode_segment_v3(&tail).unwrap()), records, folded_through, ) diff --git a/crates/compass/src/collections/cold_serve_tests.rs b/crates/compass/src/collections/cold_serve_tests.rs new file mode 100644 index 0000000..0faedbb --- /dev/null +++ b/crates/compass/src/collections/cold_serve_tests.rs @@ -0,0 +1,404 @@ +// collections/cold_serve_tests.rs — Phase 5 serve-from-storage coverage. +// A node with cold serving on answers semantic queries on namespaces it has +// NEVER attached, straight from object-storage range reads. + +use super::*; +use crate::embed::EmbedState; +use crate::storage::object_store_backend::ObjectStoreBackend; +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; + +fn unique_data_dir() -> std::path::PathBuf { + static N: AtomicU64 = AtomicU64::new(0); + std::env::temp_dir().join(format!( + "compass-cold-{}-{}", + std::process::id(), + N.fetch_add(1, Ordering::Relaxed) + )) +} + +fn embed_state() -> EmbedState { + EmbedState { + bge: None, + distilled: None, + } +} + +fn mem_storage() -> Arc { + Arc::new(ObjectStoreBackend::from_store( + std::sync::Arc::new(object_store::memory::InMemory::new()), + "object-store:memory", + )) +} + +const DIMS: usize = 8; + +/// Deterministic embedding: direction depends on i % 4, plus a nudge that is +/// UNIQUE per id (ties would make self-recall assertions ambiguous). +fn vec_for(i: u64) -> Vec { + let mut v = vec![0.05f32; DIMS]; + v[(i % 4) as usize * 2] = 1.0; + v[7] = i as f32 * 1e-4; + v +} + +fn mk_chunk(i: u64, kind: &str) -> IngestChunk { + let mut metadata = HashMap::new(); + metadata.insert("kind".to_string(), MetadataValue::String(kind.to_string())); + let mut embeddings = HashMap::new(); + embeddings.insert("default".to_string(), vec_for(i)); + IngestChunk { + client_id: Some(format!("c{i}")), + file_id: format!("f{i}"), + chunk_index: 0, + page: None, + text: format!("cold document number {i}"), + metadata, + doc_type: "chunk".to_string(), + parent_id: None, + parent_ref: None, + group_id: None, + embeddings, + embedding: None, + } +} + +fn semantic_req(target: u64, top_k: usize) -> SearchRequest { + SearchRequest { + query: String::new(), + mode: "semantic".to_string(), + vector_space: None, + top_k, + query_vector: Some(vec_for(target)), + filters: HashMap::new(), + score_weights: None, + recency: None, + recency_preset: None, + recency_field: None, + boosts: Vec::new(), + relationship_boost: None, + explain: false, + include_relations: false, + relation_types: None, + relation_direction: RelationDirection::default(), + min_seq: None, + } +} + +async fn cold_manager(dir: &std::path::Path, storage: Arc) -> Arc { + std::fs::create_dir_all(dir).unwrap(); + let m = CollectionManager::new_with_storage_opts(dir, storage, NodeRole::Full, true, 0, 0) + .await + .unwrap(); + m.set_cold_serve(true); + m.set_warm_after(0); // promotion off unless a test opts in + m +} + +// Core promise: a fresh node answers semantic queries on a compacted (v3, +// clustered) namespace WITHOUT attaching it — and sees WAL-tail writes and +// tombstones that landed after compaction (read-your-writes from storage). +#[tokio::test] +async fn cold_search_serves_without_attach() { + let storage = mem_storage(); + let embed = embed_state(); + + // Writer side: ingest past the clustering threshold, then compact. + let n = (crate::search::ivf::CLUSTER_MIN_ROWS + 200) as u64; + { + let dir = unique_data_dir(); + std::fs::create_dir_all(&dir).unwrap(); + let a = CollectionManager::new_with_storage(&dir, storage.clone()) + .await + .unwrap(); + a.create_collection("frozen", None, Some(DIMS), None) + .await + .unwrap(); + let mut batch = Vec::new(); + for i in 0..n { + batch.push(mk_chunk(i, if i % 2 == 0 { "even" } else { "odd" })); + if batch.len() == 1000 { + a.ingest("frozen", std::mem::take(&mut batch), &embed) + .await + .unwrap(); + } + } + if !batch.is_empty() { + a.ingest("frozen", batch, &embed).await.unwrap(); + } + let live = a.compact_collection("frozen").await.unwrap(); + assert_eq!(live, n); + // Post-compaction writes + a delete stay in the WAL tail. + a.ingest("frozen", vec![mk_chunk(n, "tail")], &embed) + .await + .unwrap(); + a.delete_chunks("frozen", &[2]).await.unwrap(); + let _ = std::fs::remove_dir_all(&dir); + } + + // Cold node: never attaches, still answers. + let dir_b = unique_data_dir(); + let b = cold_manager(&dir_b, storage.clone()).await; + + // Exact-vector self-recall: the target chunk must be the top hit. + let target = 1234u64; + let (results, _, _, _) = b + .search("frozen", &semantic_req(target, 5), &embed) + .await + .unwrap(); + assert!(!results.is_empty(), "cold search returned nothing"); + assert_eq!( + results[0].0.file_id, + format!("f{target}"), + "self-recall: exact stored vector must rank first" + ); + assert_eq!(results[0].2, "semantic-cold"); + + // The namespace is still NOT attached (that's the whole point). + assert!( + !b.collections.read().await.contains_key("frozen"), + "cold search must not attach" + ); + + // Tail write is visible; tombstoned chunk is not. + let (results, _, _, _) = b + .search("frozen", &semantic_req(n, 5), &embed) + .await + .unwrap(); + assert!( + results.iter().any(|r| r.0.file_id == format!("f{n}")), + "post-compaction tail write must be cold-visible" + ); + let (results, _, _, _) = b + .search("frozen", &semantic_req(2, 20), &embed) + .await + .unwrap(); + assert!( + results.iter().all(|r| r.0.file_id != "f2"), + "tombstoned chunk leaked into cold results" + ); + + // Metadata filters apply cold. + let mut req = semantic_req(target, 10); + req.filters.insert( + "kind".to_string(), + FilterValue::Exact(MetadataValue::String("even".to_string())), + ); + let (results, _, _, _) = b.search("frozen", &req, &embed).await.unwrap(); + assert!(!results.is_empty()); + for r in &results { + assert_eq!( + r.0.metadata.get("kind"), + Some(&MetadataValue::String("even".to_string())) + ); + } + + // FTS stays honest: clear error, not silent emptiness. + let mut req = semantic_req(0, 5); + req.mode = "fts".to_string(); + req.query = "cold".to_string(); + req.query_vector = None; + let err = b.search("frozen", &req, &embed).await.unwrap_err(); + assert!(err.to_string().contains("cold"), "{err}"); + + let _ = std::fs::remove_dir_all(&dir_b); +} + +// Cold serving composes with tenant partitions: a partition namespace is +// cold-served through the same router, tenant isolation intact. +#[tokio::test] +async fn cold_search_composes_with_partitions() { + let storage = mem_storage(); + let embed = embed_state(); + { + let dir = unique_data_dir(); + std::fs::create_dir_all(&dir).unwrap(); + let a = CollectionManager::new_with_storage(&dir, storage.clone()) + .await + .unwrap(); + a.create_collection( + "mt", + None, + Some(DIMS), + Some(CollectionConfig { + embed_model: "bge-small".to_string(), + partition_by: Some("tenant".to_string()), + }), + ) + .await + .unwrap(); + let mut chunks = Vec::new(); + for i in 0..40u64 { + let mut c = mk_chunk(i, "x"); + c.metadata.insert( + "tenant".to_string(), + MetadataValue::String(if i % 2 == 0 { "acme" } else { "globex" }.to_string()), + ); + chunks.push(c); + } + a.ingest("mt", chunks, &embed).await.unwrap(); + let _ = std::fs::remove_dir_all(&dir); + } + + let dir_b = unique_data_dir(); + let b = cold_manager(&dir_b, storage.clone()).await; + let mut req = semantic_req(4, 10); // id 4 is acme (even) + req.filters.insert( + "tenant".to_string(), + FilterValue::Exact(MetadataValue::String("acme".to_string())), + ); + let (results, _, _, _) = b.search("mt", &req, &embed).await.unwrap(); + assert!(!results.is_empty()); + for r in &results { + assert_eq!( + r.0.metadata.get("tenant"), + Some(&MetadataValue::String("acme".to_string())), + "tenant isolation must hold on the cold path" + ); + } + assert!( + !b.collections.read().await.contains_key("mt--part--acme"), + "partition must be cold-served, not attached" + ); + let _ = std::fs::remove_dir_all(&dir_b); +} + +// Repeated cold hits promote a background attach; once attached, queries +// take the hot path. +#[tokio::test] +async fn cold_hits_promote_background_attach() { + let storage = mem_storage(); + let embed = embed_state(); + { + let dir = unique_data_dir(); + std::fs::create_dir_all(&dir).unwrap(); + let a = CollectionManager::new_with_storage(&dir, storage.clone()) + .await + .unwrap(); + a.create_collection("warmup", None, Some(DIMS), None) + .await + .unwrap(); + a.ingest( + "warmup", + (0..20u64).map(|i| mk_chunk(i, "x")).collect(), + &embed, + ) + .await + .unwrap(); + let _ = std::fs::remove_dir_all(&dir); + } + + let dir_b = unique_data_dir(); + let b = cold_manager(&dir_b, storage.clone()).await; + b.set_warm_after(2); + for _ in 0..2 { + let (results, _, _, _) = b + .search("warmup", &semantic_req(3, 3), &embed) + .await + .unwrap(); + assert!(!results.is_empty()); + } + // The promotion attach runs in the background; poll briefly. + let mut attached = false; + for _ in 0..100 { + if b.collections.read().await.contains_key("warmup") { + attached = true; + break; + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + assert!(attached, "warm promotion never attached the namespace"); + // Post-promotion searches run the hot path (engine != semantic-cold). + let (results, _, _, _) = b + .search("warmup", &semantic_req(3, 3), &embed) + .await + .unwrap(); + assert!(!results.is_empty()); + assert_ne!(results[0].2, "semantic-cold"); + let _ = std::fs::remove_dir_all(&dir_b); +} + +// H1 regression: a LAZY (or cold-serve) node must ingest into a partitioned +// collection for a brand-new tenant WITHOUT the parent ever being attached — +// the partition template comes from the bucket config. +#[tokio::test] +async fn lazy_node_ingests_new_tenant_without_attaching_parent() { + let storage = mem_storage(); + let embed = embed_state(); + { + let dir = unique_data_dir(); + std::fs::create_dir_all(&dir).unwrap(); + let a = CollectionManager::new_with_storage(&dir, storage.clone()) + .await + .unwrap(); + a.create_collection( + "lz", + None, + Some(DIMS), + Some(CollectionConfig { + embed_model: "bge-small".to_string(), + partition_by: Some("tenant".to_string()), + }), + ) + .await + .unwrap(); + let _ = std::fs::remove_dir_all(&dir); + } + + let dir_b = unique_data_dir(); + let b = cold_manager(&dir_b, storage.clone()).await; // lazy + cold serve + let mut c = mk_chunk(0, "x"); + c.metadata.insert( + "tenant".to_string(), + MetadataValue::String("fresh-tenant".to_string()), + ); + let (n, _, _) = b + .ingest("lz", vec![c], &embed) + .await + .expect("lazy node must route partitioned ingest from bucket config"); + assert_eq!(n, 1); + assert!( + !b.collections.read().await.contains_key("lz"), + "the parent must not have been attached to serve the ingest" + ); + // And the data is queryable through the partition filter. + let mut req = semantic_req(0, 3); + req.filters.insert( + "tenant".to_string(), + FilterValue::Exact(MetadataValue::String("fresh-tenant".to_string())), + ); + let (results, _, _, _) = b.search("lz", &req, &embed).await.unwrap(); + assert_eq!(results.len(), 1); + let _ = std::fs::remove_dir_all(&dir_b); +} + +// Live-stack repro: the cold node boots BEFORE the collection exists (empty +// registry), another node creates + ingests, cold must still answer. +#[tokio::test] +async fn cold_serves_collection_created_after_boot() { + let storage = mem_storage(); + let embed = embed_state(); + let dir_b = unique_data_dir(); + let b = cold_manager(&dir_b, storage.clone()).await; // boots on empty bucket + + let dir_a = unique_data_dir(); + std::fs::create_dir_all(&dir_a).unwrap(); + let a = CollectionManager::new_with_storage(&dir_a, storage.clone()) + .await + .unwrap(); + a.create_collection("late", None, Some(DIMS), None) + .await + .unwrap(); + a.ingest("late", vec![mk_chunk(1, "x")], &embed) + .await + .unwrap(); + + let (results, _, _, _) = b + .search("late", &semantic_req(1, 3), &embed) + .await + .expect("cold node must serve a collection created after its boot"); + assert_eq!(results.len(), 1); + assert_eq!(results[0].2, "semantic-cold"); + let _ = std::fs::remove_dir_all(&dir_a); + let _ = std::fs::remove_dir_all(&dir_b); +} diff --git a/crates/compass/src/collections/mod.rs b/crates/compass/src/collections/mod.rs index 7205902..a1feb36 100644 --- a/crates/compass/src/collections/mod.rs +++ b/crates/compass/src/collections/mod.rs @@ -11,6 +11,8 @@ pub mod cloud; #[cfg(all(test, feature = "object-storage"))] +mod cold_serve_tests; +#[cfg(all(test, feature = "object-storage"))] mod partition_cloud_tests; #[cfg(test)] mod partition_tests; @@ -181,6 +183,21 @@ pub struct CollectionManager { lazy_attach: bool, /// LRU budget for attached collections (COMPASS_MAX_ATTACHED; 0 = unbounded). max_attached: usize, + /// Serve-from-storage (COMPASS_COLD_SERVE): semantic queries on + /// UNATTACHED namespaces are answered with object-storage range reads + /// instead of triggering an attach. Cloud + lazy mode only. + cold_serve: std::sync::atomic::AtomicBool, + /// Cold queries on a namespace before a background attach is kicked off + /// (COMPASS_WARM_AFTER; 0 = never warm automatically). + warm_after: std::sync::atomic::AtomicU32, + /// Cold-hit counters per namespace (drives warm promotion). + cold_hits: std::sync::Mutex>, + /// Cached cold-read artifacts, keyed by "{ns}/{segment_id}". Segments are + /// immutable, so entries never go stale; bounded by simple clearing. + cold_segments: tokio::sync::RwLock>>, + /// Weak self-reference for background tasks spawned from &self methods + /// (warm promotion). Set once right after construction. + self_weak: std::sync::OnceLock>, } /// What this node does. Parsed from `COMPASS_ROLE` (default `full`). @@ -230,9 +247,15 @@ impl CollectionManager { storage: Arc, role: NodeRole, ) -> Result, Box> { - let lazy = std::env::var("COMPASS_LAZY_ATTACH") + let cold = std::env::var("COMPASS_COLD_SERVE") .map(|v| v == "true" || v == "1") .unwrap_or(false); + // Cold serving implies lazy attach: its whole point is answering + // queries WITHOUT attaching, so eager boot-time attach is senseless. + let lazy = cold + || std::env::var("COMPASS_LAZY_ATTACH") + .map(|v| v == "true" || v == "1") + .unwrap_or(false); let max_attached = std::env::var("COMPASS_MAX_ATTACHED") .ok() .and_then(|v| v.parse().ok()) @@ -241,7 +264,7 @@ impl CollectionManager { .ok() .and_then(|v| v.parse().ok()) .unwrap_or(5); - Self::new_with_storage_opts( + let manager = Self::new_with_storage_opts( data_dir, storage, role, @@ -249,7 +272,14 @@ impl CollectionManager { max_attached, refresh_interval_secs, ) - .await + .await?; + if cold { + manager.set_cold_serve(true); + } + if let Ok(Some(n)) = std::env::var("COMPASS_WARM_AFTER").map(|v| v.parse().ok()) { + manager.set_warm_after(n); + } + Ok(manager) } /// Fully-explicit constructor (role + lazy-attach + LRU budget), used by @@ -289,7 +319,13 @@ impl CollectionManager { attach_locks: tokio::sync::Mutex::new(HashMap::new()), lazy_attach: cloud_mode && lazy_attach, max_attached, + cold_serve: std::sync::atomic::AtomicBool::new(false), + warm_after: std::sync::atomic::AtomicU32::new(3), + cold_hits: std::sync::Mutex::new(HashMap::new()), + cold_segments: tokio::sync::RwLock::new(HashMap::new()), + self_weak: std::sync::OnceLock::new(), }); + let _ = manager.self_weak.set(Arc::downgrade(&manager)); // Writer role: no local collections, no recovery — the node serves // durable appends only, validated against bucket configs. Boot is @@ -688,6 +724,21 @@ impl CollectionManager { } } Err(crate::storage::StorageError::AlreadyExists(_)) => { + if partitions::is_partition_ns(name) { + // A racing writer bootstrapped this partition's + // manifest between our config write and here. That is + // the expected create race for partitions — the + // namespace (config + manifest) is exactly what we + // wanted; adopt it. (Destroying the config here left + // a config-less namespace that cold serving and warm + // attach then disagreed about.) + rollback_local().await; + return Err(format!( + "Collection '{}' already exists in object storage", + name + ) + .into()); + } // Data exists in the bucket without a config (pre-v0.4 // namespace): this create collides with real data. Remove // the config we just wrote and refuse. @@ -712,7 +763,14 @@ impl CollectionManager { && collection.config.partition_by.is_some() && !partitions::is_partition_ns(name) { - crate::storage::id_alloc::seed(self.storage.as_ref(), name, 0).await?; + if let Err(e) = crate::storage::id_alloc::seed(self.storage.as_ref(), name, 0).await { + // Roll back: a partitioned parent without an allocator could + // never ingest (the migrate path needs a manifest that local + // mode deliberately doesn't have). + self.collections.write().await.remove(name); + let _ = store::delete_collection_data(&self.data_dir, name); + return Err(format!("id allocator seed failed: {e}").into()); + } } tracing::info!("Created collection '{}'", name); @@ -1122,11 +1180,221 @@ impl CollectionManager { store::vectors_dir(&self.data_dir, collection_name) } + // ── Serve-from-storage (Phase 5) ───────────────────────────────────── + + pub fn set_cold_serve(&self, on: bool) { + self.cold_serve + .store(on, std::sync::atomic::Ordering::Relaxed); + } + + /// Cold hits before background warm promotion (0 disables promotion). + pub fn set_warm_after(&self, n: u32) { + self.warm_after + .store(n, std::sync::atomic::Ordering::Relaxed); + } + + fn cold_serve(&self) -> bool { + self.cold_serve.load(std::sync::atomic::Ordering::Relaxed) + } + + /// Cold semantic search: manifest read (freshness anchor) → cached + /// per-segment artifacts → cluster probes → tail brute-force → hydrate. + /// FTS needs an inverted index and is not cold-servable — callers get a + /// clear error steering them to semantic mode (or a warmed node). + #[allow(clippy::type_complexity)] + async fn search_cold( + &self, + ns: &str, + req: &SearchRequest, + embed_state: &EmbedState, + ) -> Result< + ( + Vec<( + DocumentChunk, + f32, + String, + Option>, + Option>, + )>, + usize, + u64, + Option, + ), + Box, + > { + let start = std::time::Instant::now(); + crate::metrics::inc(&crate::metrics::COLD_SEARCHES_TOTAL); + // The cold path is pure vector search. Anything that needs local + // indexes or the scoring pipeline is REJECTED, not silently ignored — + // identical requests must never return materially different rankings + // cold vs. warm without a signal. ("hybrid" with an empty text query + // degenerates to semantic legitimately and is allowed.) + if req.mode == "fts" || (req.mode == "hybrid" && !req.query.is_empty()) { + return Err(format!( + "collection '{ns}' is cold (not attached): '{}' search needs local \ + indexes. Use mode \"semantic\", or query again after the namespace warms.", + req.mode + ) + .into()); + } + if req.recency.is_some() + || req.recency_preset.is_some() + || !req.boosts.is_empty() + || req.relationship_boost.is_some() + || req.include_relations + { + return Err(format!( + "collection '{ns}' is cold (not attached): recency/boosts/relationship \ + options need the warm scoring pipeline. Drop them, or query again after \ + the namespace warms." + ) + .into()); + } + + // The bucket config names the default space. A namespace can hold + // data WITHOUT a config (older create paths could destroy the config + // after a race) — warm attach accepts manifest-exists, so cold must + // too, falling back to the requested/"default" space. + let default_space = match self.bucket_config(ns, false).await { + Ok(cfg) => cfg.default_vector_space.clone(), + Err(e) if e.downcast_ref::().is_some() => { + if !self.storage.exists(&format!("{ns}/manifest")).await? { + return Err(e); + } + None + } + Err(e) => return Err(e), + }; + let space = req + .vector_space + .clone() + .or(default_space) + .unwrap_or_else(|| "default".to_string()); + + let query_vec: Vec = match &req.query_vector { + Some(v) => v.clone(), + None => embed_state.embed_query(&req.query).map_err(|e| { + format!("cold search needs a query_vector or a loaded embed model: {e}") + })?, + }; + + let (manifest, _) = crate::storage::lsm::read_manifest(self.storage.as_ref(), ns).await?; + if let Some(min_seq) = req.min_seq { + // Cold reads see everything committed to the manifest, so + // read-your-writes holds by construction — only a seq beyond the + // write history is unsatisfiable. + if min_seq >= manifest.next_seq { + return Err(format!( + "min_seq {} is beyond the collection's write history ({})", + min_seq, manifest.next_seq + ) + .into()); + } + } + + // Cached artifacts per segment (immutable → cache by id). + let mut segments = Vec::with_capacity(manifest.segments.len()); + for sref in &manifest.segments { + let key = format!("{ns}/{}", sref.id); + if let Some(cs) = self.cold_segments.read().await.get(&key).cloned() { + segments.push(cs); + continue; + } + let cs = Arc::new( + crate::search::cold::ColdSegment::open(self.storage.as_ref(), ns, &sref.id).await?, + ); + let mut cache = self.cold_segments.write().await; + if cache.len() >= 1024 { + cache.clear(); // crude bound; entries rebuild in a few reads + } + cache.insert(key, cs.clone()); + segments.push(cs); + } + + let nprobe = std::env::var("COMPASS_COLD_NPROBE") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(crate::search::cold::DEFAULT_NPROBE); + let hits = crate::search::cold::search( + self.storage.as_ref(), + ns, + &segments, + &manifest, + &space, + &query_vec, + req.top_k, + nprobe, + &req.filters, + ) + .await?; + + self.maybe_warm(ns); + + let took_us = start.elapsed().as_micros() as u64; + let total = hits.len(); + let explain = req.explain.then(|| ExplainPlan { + filter: FilterExplain { + eligible_count: total as u64, + universe_count: 0, // unknown without attaching — cold reads don't scan + selectivity: 0.0, + }, + ann: AnnExplain { + engine: "cold-ivf".to_string(), + candidates_inspected: None, + ef_search_used: 0, + }, + }); + let results = hits + .into_iter() + .map(|(c, score)| (c, score, "semantic-cold".to_string(), None, None)) + .collect(); + Ok((results, total, took_us, explain)) + } + + /// Count a cold hit; at the warm threshold, spawn a background attach so + /// a repeatedly-queried namespace migrates to the fast path on its own. + fn maybe_warm(&self, ns: &str) { + let after = self.warm_after.load(std::sync::atomic::Ordering::Relaxed); + if after == 0 { + return; + } + let fire = { + let mut map = self.cold_hits.lock().unwrap(); + let e = map.entry(ns.to_string()).or_insert(0); + *e += 1; + // Reset at the threshold so a namespace that got promoted and + // later LRU-evicted can warm AGAIN after `after` fresh cold hits + // (an == latch would seal shut forever after the first firing). + if *e >= after { + *e = 0; + true + } else { + false + } + }; + if fire { + if let Some(m) = self.self_weak.get().and_then(|w| w.upgrade()) { + let ns = ns.to_string(); + tokio::spawn(async move { + crate::metrics::inc(&crate::metrics::WARM_PROMOTIONS_TOTAL); + tracing::info!("cold namespace '{ns}' hit warm threshold; attaching"); + if let Err(e) = m.ensure_attached(&ns).await { + tracing::warn!("warm promotion attach for '{ns}' failed: {e}"); + // Reset so a later burst can retry. + m.cold_hits.lock().unwrap().remove(&ns); + } + }); + } + } + } + // ── Tenant partitions (Phase 6) ────────────────────────────────────── /// The partition field of a collection, or None for normal collections - /// and partition namespaces themselves. Attaches the parent if needed - /// (cheap: a partitioned parent holds config only, no chunk data). + /// and partition namespaces themselves. Reads attached metadata when + /// present, else the (cached) bucket config — deliberately WITHOUT + /// attaching: cold-served and lazy namespaces must be routable from + /// config alone. async fn partition_field( &self, name: &str, @@ -1134,11 +1402,24 @@ impl CollectionManager { if partitions::is_partition_ns(name) { return Ok(None); } - self.ensure_attached(name).await?; - let collections = self.collections.read().await; - Ok(collections - .get(name) - .and_then(|l| l.metadata.config.partition_by.clone())) + if let Some(loaded) = self.collections.read().await.get(name) { + return Ok(loaded.metadata.config.partition_by.clone()); + } + if self.cloud_mode { + // Unattached: the bucket config answers without an attach. A + // MISSING config means "not partitioned" (the caller's own lookup + // produces the not-found) — but a transient storage error must + // PROPAGATE: swallowing it would reclassify a partitioned + // collection as unpartitioned and misroute tenant writes into the + // parent namespace, where no partition-routed search looks. + return Ok( + match cloud::read_bucket_config(self.storage.as_ref(), name).await? { + Some(cfg) => cfg.config.partition_by, + None => None, + }, + ); + } + Ok(None) } /// Make sure a partition namespace exists and is servable, creating it on @@ -1157,18 +1438,32 @@ impl CollectionManager { if self.cloud_mode && self.registered.read().await.contains(&ns) { return Ok(ns); // lazy attach loads it at the entry point } - let (spaces, config) = { + // Parent template: attached metadata when present, else the bucket + // config — a lazy/cold-serve node routes partitioned ingest without + // ever attaching the parent, so requiring attachment here would break + // first ingest of a new tenant on exactly those nodes. + let attached_template = { let collections = self.collections.read().await; - let parent_meta = collections - .get(parent) - .ok_or_else(|| not_found(format_args!("Collection '{}' not found", parent)))?; - ( - parent_meta.metadata.vector_spaces.clone(), - CollectionConfig { - embed_model: parent_meta.metadata.config.embed_model.clone(), - partition_by: None, - }, - ) + collections.get(parent).map(|p| { + ( + p.metadata.vector_spaces.clone(), + p.metadata.config.embed_model.clone(), + ) + }) + }; + let (spaces, embed_model) = match attached_template { + Some(t) => t, + None if self.cloud_mode => { + let cfg = self.bucket_config(parent, false).await?; + (cfg.vector_spaces.clone(), cfg.config.embed_model.clone()) + } + None => { + return Err(not_found(format_args!("Collection '{}' not found", parent))); + } + }; + let config = CollectionConfig { + embed_model, + partition_by: None, }; match self .create_collection_inner(&ns, Some(spaces), None, Some(config)) @@ -1278,6 +1573,11 @@ impl CollectionManager { if self.bucket_configs.read().await.contains_key(&ns) { return Ok(ns); } + // First sight of this partition: re-validate the PARENT with a fresh + // read before creating bucket objects. A writer's cached parent + // config outlives a cascade delete — bootstrapping from it would + // resurrect the collection as an orphan namespace. + let parent_cfg = self.bucket_config(&parent_cfg.name, true).await?; let mut part_cfg = parent_cfg.clone(); part_cfg.name = ns.clone(); part_cfg.config.partition_by = None; @@ -1321,11 +1621,15 @@ impl CollectionManager { return Ok(false); } if self.role == NodeRole::Writer { - return Ok(self - .bucket_config(name, false) - .await - .map(|c| c.config.partition_by.is_some()) - .unwrap_or(false)); + return match self.bucket_config(name, false).await { + Ok(c) => Ok(c.config.partition_by.is_some()), + // Unknown collection: not partitioned (downstream 404s). + Err(e) if e.downcast_ref::().is_some() => Ok(false), + // Transient storage errors PROPAGATE — treating them as + // "not partitioned" would let a writer append tombstones + // into the parent namespace no serving node materializes. + Err(e) => Err(e), + }; } Ok(self.partition_field(name).await?.is_some()) } @@ -2254,6 +2558,16 @@ impl CollectionManager { .search_partitioned(collection_name, &field, req, embed_state) .await; } + + // Serve-from-storage: an UNATTACHED namespace answers semantic + // queries with a handful of object-storage range reads — no attach, + // no index rebuild. Repeated cold hits promote a background attach. + if self.cold_serve() + && self.cloud_mode + && !self.collections.read().await.contains_key(collection_name) + { + return self.search_cold(collection_name, req, embed_state).await; + } self.ensure_attached(collection_name).await?; // Read-your-writes: wait (bounded) until fragments up to `min_seq` are @@ -3086,7 +3400,10 @@ impl CollectionManager { if self.max_attached == 0 { return; } - // Pick the victim under a short read lock. + // Pick the victim under a short read lock. In NON-lazy mode only + // dynamic partition namespaces may be evicted — a normal collection + // evicted there could never re-attach (ensure_attached early-returns + // for non-partitions when lazy attach is off). let victim: Option = { let collections = self.collections.read().await; if collections.len() <= self.max_attached { @@ -3095,6 +3412,7 @@ impl CollectionManager { collections .iter() .filter(|(name, _)| name.as_str() != just_attached) + .filter(|(name, _)| self.lazy_attach || partitions::is_partition_ns(name)) .min_by_key(|(_, l)| l.last_used.load(std::sync::atomic::Ordering::Relaxed)) .map(|(name, _)| name.clone()) } @@ -3351,11 +3669,14 @@ impl CollectionManager { // writer nodes too: a tombstone appended to the PARENT namespace // would never be materialized by any serving node. if self.is_partitioned_any_role(collection_name).await? { - return Err(format!( - "collection '{collection_name}' is partitioned: delete via filters \ - (POST .../delete with the partition field), not bare ids" - ) - .into()); + let hint = if self.role == NodeRole::Writer { + "route the delete through a serving node (writers cannot resolve \ + partition-scoped filters)" + } else { + "delete via filters (POST .../delete with the partition field), \ + not bare ids" + }; + return Err(format!("collection '{collection_name}' is partitioned: {hint}").into()); } // Writer role: durable tombstone only. Without local indexes we can't // filter to ids-that-exist; a tombstone for an absent id is an @@ -3969,7 +4290,7 @@ pub(crate) async fn compact_storage( crate::storage::lsm::read_uncompacted_fragments_strict(storage, ns, &manifest).await?; let segment = cloud::fold_tail(&frags)?; let records = segment.chunks.len() as u64; - let bytes = cloud::encode_segment_v2(&segment)?; + let bytes = cloud::encode_segment_v3(&segment)?; match crate::storage::lsm::append_segment( storage, ns, diff --git a/crates/compass/src/collections/partition_cloud_tests.rs b/crates/compass/src/collections/partition_cloud_tests.rs index 8fa2bf1..49f2884 100644 --- a/crates/compass/src/collections/partition_cloud_tests.rs +++ b/crates/compass/src/collections/partition_cloud_tests.rs @@ -162,7 +162,11 @@ async fn writer_partitioned_ingest_visible_on_serving_node() { // Writer refuses partitioned operations that would black-hole data. let err = writer.delete_chunks("wp", &[0]).await.unwrap_err(); - assert!(err.to_string().contains("delete via filters"), "{err}"); + assert!( + err.to_string() + .contains("route the delete through a serving node"), + "{err}" + ); let err = writer .create_relations( "wp", diff --git a/crates/compass/src/main.rs b/crates/compass/src/main.rs index 2c87211..f559891 100644 --- a/crates/compass/src/main.rs +++ b/crates/compass/src/main.rs @@ -102,7 +102,7 @@ async fn main() -> Result<(), Box> { .allow_methods(Any) .allow_headers(Any); - // Anonymous telemetry — opt out with COMPASS_TELEMETRY=off or DO_NOT_TRACK=1 + // Anonymous telemetry — OPT-IN ONLY (COMPASS_TELEMETRY=on); off by default telemetry::spawn_telemetry(data_dir.clone(), app_state.manager.clone()); // Bearer-token auth via COMPASS_API_KEY. When unset, auth is disabled. diff --git a/crates/compass/src/metrics.rs b/crates/compass/src/metrics.rs index fe89c98..96f9ae8 100644 --- a/crates/compass/src/metrics.rs +++ b/crates/compass/src/metrics.rs @@ -33,6 +33,8 @@ counters!( ATTACH_SECONDS_SUM_MILLIS, COMPACTIONS_TOTAL, QUARANTINED_CHUNKS_TOTAL, + COLD_SEARCHES_TOTAL, + WARM_PROMOTIONS_TOTAL, ); #[inline] diff --git a/crates/compass/src/search/cold.rs b/crates/compass/src/search/cold.rs new file mode 100644 index 0000000..979b31d --- /dev/null +++ b/crates/compass/src/search/cold.rs @@ -0,0 +1,762 @@ +// search/cold.rs — serve-from-storage: answer semantic queries on a +// collection that is NOT attached, with a handful of object-storage range +// reads instead of a full index rebuild. +// +// Query flow (per namespace): +// 1. GET manifest (freshness anchor: everything committed is visible, so +// cold reads satisfy read-your-writes by construction) +// 2. per segment (immutable → artifacts cached by segment id): +// header+TOC, `cent:` centroids, `tombs`, `metaidx` — all small +// 3. rank clusters by centroid dot-product, range-GET the top `nprobe` +// clusters, score their (unit-norm) vectors against the query +// 4. brute-force the uncompacted WAL tail (bounded by the auto-compact +// threshold) and apply tombstones; newest version of an id wins +// 5. hydrate the top candidates' chunk JSON by byte range via `metaidx`, +// apply metadata filters, return +// +// RAM cost per cold namespace: centroids + directories + tombstones + the +// metadata index — megabytes, independent of collection size. + +use crate::models::{DocumentChunk, FilterValue, MetadataValue}; +use crate::search::filter_pushdown::{FilterExpr, Predicate}; +use crate::search::ivf; +use crate::storage::{lsm, Storage, StorageError}; +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; + +/// Clusters probed per segment per query (env-tunable via the manager). +pub const DEFAULT_NPROBE: usize = 8; +/// Overfetch factor before deduping down to top_k. +const OVERFETCH: usize = 4; +/// Additional overfetch multiplier when metadata filters are present (cold +/// filtering is post-selection; see the recall-contract note in search()). +const FILTER_OVERFETCH: usize = 8; +/// Header bytes fetched optimistically (magic + max_id + toc_len + TOC). +const HEADER_PROBE: u64 = 16 * 1024; +/// metaidx at or below this size is fetched whole; larger ones page in +/// blocks on demand. +const METAIDX_FULL_MAX: u64 = 8 * 1024 * 1024; +const METAIDX_BLOCK_ROWS: usize = 2048; + +type BoxErr = Box; + +/// Cached, immutable cold-read artifacts for ONE segment object. +pub struct ColdSegment { + ns: String, + segment_id: String, + /// section name -> (absolute byte offset in the object, length) + sections: HashMap, + /// space -> parsed centroids + cluster directory + cents: HashMap, + /// ids tombstoned BY this segment (they apply to OLDER segments only — + /// see the generation rule in `search`) + tombstones: HashSet, + metaidx: MetaIdx, +} + +enum MetaIdx { + /// Whole index resident: sorted (id, off, len) rows. + Full(Vec<(u64, u64, u32)>), + /// Sparse anchors (first id of each block) + block byte range info; blocks + /// are fetched on demand per query (not cached — queries touch few). + Paged { + anchors: Vec, // first id of block i + n_rows: u64, + idx_offset: u64, // absolute offset of the first row (after the count) + }, +} + +fn seg_key(ns: &str, id: &str) -> String { + format!("{ns}/segments/{id}") +} + +async fn get_range( + storage: &dyn Storage, + key: &str, + start: u64, + len: u64, +) -> Result { + storage.get_range(key, start..start + len).await +} + +impl ColdSegment { + /// Build the cached artifacts with a few small reads. Total fetched: + /// TOC + centroids + tombstones + (metaidx or its anchors). + pub async fn open(storage: &dyn Storage, ns: &str, segment_id: &str) -> Result { + Self::open_with_limits(storage, ns, segment_id, METAIDX_FULL_MAX).await + } + + /// `open` with an explicit full-fetch threshold — lets tests exercise the + /// paged metadata-index path without a 400k-chunk segment. + async fn open_with_limits( + storage: &dyn Storage, + ns: &str, + segment_id: &str, + metaidx_full_max: u64, + ) -> Result { + let key = seg_key(ns, segment_id); + // Header + TOC (optimistic single read; re-read if the TOC is huge). + let head = storage.get_range(&key, 0..HEADER_PROBE).await?; + if head.len() < 20 { + return Err(format!("segment {segment_id}: truncated header").into()); + } + // Only v3 segments carry the cold-read sections (metaidx/meta2 and + // clusters). v2 would brute-force its whole flat section and then + // drop every hit at hydration — reject loudly instead. + if head[0..8] != crate::collections::cloud::SEG_MAGIC_V3 { + return Err(format!( + "segment {segment_id} predates the cold-servable format; \ + run POST /collections/:name/compact once to upgrade it" + ) + .into()); + } + let toc_len = u32::from_le_bytes(head[16..20].try_into().unwrap()) as u64; + let toc_bytes = if 20 + toc_len <= head.len() as u64 { + head.slice(20..(20 + toc_len) as usize) + } else { + get_range(storage, &key, 20, toc_len).await? + }; + let toc: Vec<(String, u64)> = serde_json::from_slice(&toc_bytes) + .map_err(|e| format!("segment {segment_id}: bad TOC: {e}"))?; + let mut sections = HashMap::new(); + let mut pos = 20 + toc_len; + for (name, len) in toc { + sections.insert(name, (pos, len)); + pos += len; + } + + // Centroids for every clustered space (small — cache them all). + let mut cents = HashMap::new(); + for (name, &(off, len)) in §ions { + if let Some(space) = name.strip_prefix("cent:") { + let body = get_range(storage, &key, off, len).await?; + let c = ivf::parse_cent(&body) + .ok_or_else(|| format!("segment {segment_id}: bad {name}"))?; + cents.insert(space.to_string(), c); + } + } + + // Tombstones (u64 LE array; bounded by deletes-per-fold). + let mut tombstones = HashSet::new(); + if let Some(&(off, len)) = sections.get("tombs") { + if len > 0 { + let body = get_range(storage, &key, off, len).await?; + for c in body.chunks_exact(8) { + tombstones.insert(u64::from_le_bytes(c.try_into().unwrap())); + } + } + } + + // Metadata index: whole if small, paged anchors otherwise. + let metaidx = match sections.get("metaidx") { + Some(&(off, len)) if len > 8 => { + if len <= metaidx_full_max { + let body = get_range(storage, &key, off, len).await?; + let n = u64::from_le_bytes(body[0..8].try_into().unwrap()) as usize; + if body.len() < 8 + n * 20 { + return Err(format!( + "segment {segment_id}: truncated metaidx ({n} rows declared)" + ) + .into()); + } + let mut rows = Vec::with_capacity(n); + for i in 0..n { + let p = 8 + i * 20; + rows.push(( + u64::from_le_bytes(body[p..p + 8].try_into().unwrap()), + u64::from_le_bytes(body[p + 8..p + 16].try_into().unwrap()), + u32::from_le_bytes(body[p + 16..p + 20].try_into().unwrap()), + )); + } + MetaIdx::Full(rows) + } else { + // Big index: fetch it whole ONCE at open (bounded by + // index size, ~20B/row) but keep only per-block anchor + // ids resident; lookups page 20B×2048 blocks on demand. + let body = get_range(storage, &key, off, len).await?; + let n = u64::from_le_bytes(body[0..8].try_into().unwrap()); + let mut anchors = Vec::new(); + let mut i = 0u64; + while i < n { + let p = (8 + i * 20) as usize; + anchors.push(u64::from_le_bytes(body[p..p + 8].try_into().unwrap())); + i += METAIDX_BLOCK_ROWS as u64; + } + MetaIdx::Paged { + anchors, + n_rows: n, + idx_offset: off + 8, + } + } + } + _ => MetaIdx::Full(Vec::new()), + }; + + Ok(Self { + ns: ns.to_string(), + segment_id: segment_id.to_string(), + sections, + cents, + tombstones, + metaidx, + }) + } + + /// Rank this segment's clusters for `q` (unit-norm) and return the top + /// `nprobe` cluster byte ranges to fetch. + fn probe_plan(&self, space: &str, q: &[f32], nprobe: usize) -> Vec<(u64, u64)> { + let Some(cent) = self.cents.get(space) else { + return Vec::new(); + }; + let Some(&(clu_off, _)) = self.sections.get(&format!("clu:{space}")) else { + return Vec::new(); + }; + let mut ranked: Vec<(usize, f32)> = cent + .centroids + .iter() + .enumerate() + .filter(|(i, _)| cent.dir[*i].count > 0) + .map(|(i, c)| (i, ivf::dot(c, q))) + .collect(); + ranked.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal)); + ranked + .into_iter() + .take(nprobe) + .map(|(i, _)| { + let d = cent.dir[i]; + (clu_off + d.offset, d.len) + }) + .collect() + } + + fn dims_for(&self, space: &str) -> Option { + self.cents.get(space).map(|c| c.dims) + } + + /// The flat `emb:` byte range (segments below the clustering + /// threshold) — brute-forced whole. + fn flat_range(&self, space: &str) -> Option<(u64, u64)> { + self.sections.get(&format!("emb:{space}")).copied() + } + + /// Look up the meta2 byte ranges for a set of ids. + async fn meta_ranges( + &self, + storage: &dyn Storage, + ids: &[u64], + ) -> Result, BoxErr> { + let Some(&(meta_off, _)) = self.sections.get("meta2") else { + return Ok(Vec::new()); + }; + let key = seg_key(&self.ns, &self.segment_id); + let mut out = Vec::new(); + match &self.metaidx { + MetaIdx::Full(rows) => { + for &id in ids { + if let Ok(i) = rows.binary_search_by_key(&id, |r| r.0) { + let (rid, off, len) = rows[i]; + out.push((rid, meta_off + off, len)); + } + } + } + MetaIdx::Paged { + anchors, + n_rows, + idx_offset, + } => { + // Group wanted ids by block, fetch each needed block once. + let mut by_block: HashMap> = HashMap::new(); + for &id in ids { + let block = match anchors.binary_search(&id) { + Ok(i) => i, + Err(0) => continue, // below the first anchor: absent + Err(i) => i - 1, + }; + by_block.entry(block).or_default().push(id); + } + for (block, wanted) in by_block { + let start_row = (block * METAIDX_BLOCK_ROWS) as u64; + let rows_here = (*n_rows - start_row).min(METAIDX_BLOCK_ROWS as u64); + let body = + get_range(storage, &key, idx_offset + start_row * 20, rows_here * 20) + .await?; + for r in body.chunks_exact(20) { + let rid = u64::from_le_bytes(r[0..8].try_into().unwrap()); + if wanted.contains(&rid) { + out.push(( + rid, + meta_off + u64::from_le_bytes(r[8..16].try_into().unwrap()), + u32::from_le_bytes(r[16..20].try_into().unwrap()), + )); + } + } + } + } + } + Ok(out) + } +} + +/// One scored candidate before hydration. `generation` orders duplicates of +/// the same id: higher wins (segments in manifest order, tail above all). +struct Candidate { + id: u64, + score: f32, + generation: usize, + /// Tail candidates already carry their chunk. + chunk: Option, + segment: Option, // index into segments, for hydration +} + +/// A cold semantic search over one namespace. `segments` are the cached +/// artifacts in manifest order; the WAL tail is read fresh per query. +#[allow(clippy::too_many_arguments)] +pub async fn search( + storage: &dyn Storage, + ns: &str, + segments: &[Arc], + manifest: &lsm::Manifest, + space: &str, + query: &[f32], + top_k: usize, + nprobe: usize, + filters: &HashMap, +) -> Result, BoxErr> { + let mut q = query.to_vec(); + ivf::normalize(&mut q); + + // Tombstone semantics must match materialize(): a segment's carried + // tombstones apply only to OLDER segments (its own chunks are written + // after them, and a NEWER segment's re-ingest of the same id must + // survive). So a candidate from generation g dies only to a tombstone + // from generation > g. A single flat union would permanently suppress + // re-ingested chunks that warm search serves. + let tomb_of = |generation: usize| -> &HashSet { &segments[generation].tombstones }; + let killed_by_newer = |id: u64, generation: usize| -> bool { + ((generation + 1)..segments.len()).any(|j| tomb_of(j).contains(&id)) + }; + // Tail tombstones (replayed in seq order below) are the newest + // generation of all: they kill any segment candidate. + let mut tail_dead: HashSet = HashSet::new(); + + // WAL tail: bounded by the auto-compaction threshold in healthy + // operation. A pathologically long tail (compaction disabled/failing) + // would make this a full-dataset materialization per query — refuse + // loudly instead of degrading into that silently. + let tail_len = manifest.uncompacted().count(); + if tail_len > 2 * crate::collections::AUTO_COMPACT_FRAGMENT_THRESHOLD { + return Err(format!( + "namespace '{ns}' has {tail_len} uncompacted WAL fragments — too many to \ + cold-serve. Run POST /collections/:name/compact (or check why \ + auto-compaction is not running), then retry." + ) + .into()); + } + let tail = lsm::read_uncompacted_fragments(storage, ns, manifest).await?; + let mut tail_chunks: HashMap = HashMap::new(); + for (fref, payload) in &tail { + match fref.kind { + lsm::FragmentKind::Data => { + let chunks: Vec = serde_json::from_slice(payload) + .map_err(|e| format!("tail fragment decode: {e}"))?; + for c in chunks { + tail_dead.remove(&c.id); // re-ingest after delete resurrects + tail_chunks.insert(c.id, c); + } + } + lsm::FragmentKind::Tombstone => { + let ids: Vec = serde_json::from_slice(payload) + .map_err(|e| format!("tail tombstone decode: {e}"))?; + for id in ids { + tail_dead.insert(id); + tail_chunks.remove(&id); + } + } + _ => {} + } + } + + // Filters are applied POST-candidate-selection on the cold path (there + // is no roaring index to push down without attaching), so a selective + // filter needs a deeper candidate pool. Recall contract: cold filtered + // queries can under-return when matches are rarer than ~1/FILTER_OVERFETCH + // of the probed neighborhoods; warm search has no such limit. + let overfetch = if filters.is_empty() { + OVERFETCH + } else { + OVERFETCH * FILTER_OVERFETCH + }; + let want = (top_k * overfetch).max(top_k).min(512); + let mut candidates: Vec = Vec::new(); + + // Segment candidates: probe clusters (or brute-force flat sections). + for (gen, seg) in segments.iter().enumerate() { + let key = seg_key(ns, &seg.segment_id); + let mut ranges = seg.probe_plan(space, &q, nprobe); + let dims = match seg.dims_for(space) { + Some(d) => d, + None => match seg.flat_range(space) { + Some((off, len)) if len >= 12 => { + // Flat section: [u32 dims][u64 n][rows] — brute force it. + let head = get_range(storage, &key, off, 12).await?; + let dims = u32::from_le_bytes(head[0..4].try_into().unwrap()) as usize; + ranges = vec![(off + 12, len - 12)]; + dims + } + _ => continue, // space absent in this segment + }, + }; + if dims != q.len() { + return Err(format!( + "query has {} dims but segment space '{space}' has {dims}", + q.len() + ) + .into()); + } + // Fetch probed ranges concurrently. + let bodies = futures::future::try_join_all( + ranges + .iter() + .map(|&(off, len)| get_range(storage, &key, off, len)), + ) + .await?; + for body in bodies { + for (id, v) in ivf::parse_cluster_rows(&body, dims) { + if tail_dead.contains(&id) + || tail_chunks.contains_key(&id) + || killed_by_newer(id, gen) + { + continue; + } + // Flat sections store raw vectors; clustered store unit-norm. + // Normalizing again is idempotent for the latter. + let mut v = v; + ivf::normalize(&mut v); + candidates.push(Candidate { + id, + score: ivf::dot(&v, &q), + generation: gen, + chunk: None, + segment: Some(gen), + }); + } + } + } + + // Tail candidates: brute-force the fresh writes. + let tail_gen = segments.len(); + for (id, c) in &tail_chunks { + if let Some(emb) = c.embeddings.get(space) { + let mut v = emb.clone(); + ivf::normalize(&mut v); + candidates.push(Candidate { + id: *id, + score: ivf::dot(&v, &q), + generation: tail_gen, + chunk: Some(c.clone()), + segment: None, + }); + } + } + + // Dedupe by id, newest generation wins; then keep the global top `want`. + candidates.sort_by(|a, b| a.id.cmp(&b.id).then(b.generation.cmp(&a.generation))); + candidates.dedup_by_key(|c| c.id); + candidates.sort_by(|a, b| { + b.score + .partial_cmp(&a.score) + .unwrap_or(std::cmp::Ordering::Equal) + }); + candidates.truncate(want); + + // Hydrate: group segment candidates per segment, batch the meta lookups. + let expr = FilterExpr::compile(filters); + let mut hydrated: Vec<(DocumentChunk, f32)> = Vec::new(); + let mut by_seg: HashMap> = HashMap::new(); + let mut scores: HashMap = HashMap::new(); + for c in &candidates { + scores.insert(c.id, c.score); + match (&c.chunk, c.segment) { + (Some(ch), _) => { + if eval_filters(&expr, ch) { + hydrated.push((ch.clone(), c.score)); + } + } + (None, Some(seg_i)) => by_seg.entry(seg_i).or_default().push(c.id), + _ => {} + } + } + for (seg_i, ids) in by_seg { + let seg = &segments[seg_i]; + let key = seg_key(ns, &seg.segment_id); + let mut ranges = seg.meta_ranges(storage, &ids).await?; + // Coalesce adjacent-ish rows into fewer GETs. + ranges.sort_by_key(|r| r.1); + let mut batches: Vec<(u64, u64, Vec<(u64, u64, u32)>)> = Vec::new(); + for r in ranges { + match batches.last_mut() { + Some((_start, end, rows)) if r.1 <= *end + 64 * 1024 => { + *end = (*end).max(r.1 + r.2 as u64); + rows.push(r); + } + _ => batches.push((r.1, r.1 + r.2 as u64, vec![r])), + } + } + for (start, end, rows) in batches { + let body = get_range(storage, &key, start, end - start).await?; + for (id, off, len) in rows { + let lo = (off - start) as usize; + let chunk: DocumentChunk = serde_json::from_slice(&body[lo..lo + len as usize]) + .map_err(|e| format!("meta2 row decode (id {id}): {e}"))?; + if eval_filters(&expr, &chunk) { + hydrated.push((chunk, scores.get(&id).copied().unwrap_or(0.0))); + } + } + } + } + + hydrated.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal)); + hydrated.truncate(top_k); + Ok(hydrated) +} + +/// Metadata filter evaluation for cold hits (the roaring FilterIndex only +/// exists for attached collections). Mirrors FilterIndex::eligible semantics, +/// including the doc_type-as-metadata rule. +fn eval_filters(expr: &FilterExpr, chunk: &DocumentChunk) -> bool { + if expr.is_empty() { + return true; + } + let get = |field: &str| -> Option { + if field == "doc_type" { + Some(MetadataValue::String(chunk.doc_type.clone())) + } else { + chunk.metadata.get(field).cloned() + } + }; + expr.predicates.iter().all(|p| match p { + Predicate::Eq { field, value } => get(field).as_ref() == Some(value), + Predicate::Range { field, gte, lte } => match get(field).and_then(|m| m.as_f64()) { + Some(n) => gte.map(|g| n >= g).unwrap_or(true) && lte.map(|l| n <= l).unwrap_or(true), + None => false, + }, + // Parity with FilterIndex: `contains` matches STRING LISTS only (the + // warm index populates string_list_contains from StringList values) — + // matching bare strings here would make cold return hits warm never + // would. + Predicate::Contains { field, value } => match get(field) { + Some(MetadataValue::StringList(xs)) => xs.iter().any(|x| x == value), + _ => false, + }, + Predicate::In { field, values } => match get(field) { + Some(MetadataValue::String(s)) => values.contains(&s), + _ => false, + }, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::storage::local::LocalDiskStorage; + + fn storage(name: &str) -> (std::path::PathBuf, Arc) { + use std::sync::atomic::{AtomicU64, Ordering}; + static N: AtomicU64 = AtomicU64::new(0); + let root = std::env::temp_dir().join(format!( + "compass_cold_unit_{}_{}_{}", + name, + std::process::id(), + N.fetch_add(1, Ordering::Relaxed) + )); + let _ = std::fs::remove_dir_all(&root); + let s: Arc = Arc::new(LocalDiskStorage::new(&root).unwrap()); + (root, s) + } + + fn seg_with_chunks(ids: &[u64], tombstones: &[u64]) -> Vec { + use crate::models::DocumentChunk; + let chunks: Vec = ids + .iter() + .map(|&i| { + let mut c = DocumentChunk { + id: i, + collection: "ns".into(), + file_id: format!("f{i}"), + chunk_index: 0, + page: None, + text: format!("t{i}"), + metadata: Default::default(), + doc_type: "chunk".into(), + parent_id: None, + group_id: None, + embeddings: Default::default(), + embedding: None, + }; + c.embeddings + .insert("default".into(), vec![i as f32, 1.0, 0.0, 0.0]); + c + }) + .collect(); + crate::collections::cloud::encode_segment_v3(&crate::collections::cloud::Segment { + version: 2, + chunks, + relations: vec![], + max_id: ids.iter().copied().max().unwrap_or(0), + tombstones: tombstones.to_vec(), + relation_tombstones: vec![], + }) + .unwrap() + } + + // C1 regression: an OLDER segment's carried tombstone must not suppress + // the same id re-ingested into a NEWER segment (materialize parity). + #[tokio::test] + async fn newer_segment_survives_older_tombstone() { + let (root, s) = storage("gen"); + // seg A (gen 0): chunk 1 live, carries tombstone for id 5. + s.put( + "ns/segments/a", + bytes::Bytes::from(seg_with_chunks(&[1], &[5])), + ) + .await + .unwrap(); + // seg B (gen 1): id 5 re-ingested. + s.put( + "ns/segments/b", + bytes::Bytes::from(seg_with_chunks(&[5], &[])), + ) + .await + .unwrap(); + let manifest = lsm::Manifest { + segments: vec![ + lsm::SegmentRef { + id: "a".into(), + records: 1, + }, + lsm::SegmentRef { + id: "b".into(), + records: 1, + }, + ], + ..Default::default() + }; + let segs = vec![ + Arc::new(ColdSegment::open(s.as_ref(), "ns", "a").await.unwrap()), + Arc::new(ColdSegment::open(s.as_ref(), "ns", "b").await.unwrap()), + ]; + let hits = search( + s.as_ref(), + "ns", + &segs, + &manifest, + "default", + &[5.0, 1.0, 0.0, 0.0], + 10, + DEFAULT_NPROBE, + &Default::default(), + ) + .await + .unwrap(); + assert!( + hits.iter().any(|(c, _)| c.id == 5), + "id 5 lives in the NEWER segment; the older tombstone must not kill it" + ); + // And the reverse still holds: a NEWER segment's tombstone kills an + // OLDER segment's chunk. + s.put( + "ns/segments/c", + bytes::Bytes::from(seg_with_chunks(&[9], &[1])), + ) + .await + .unwrap(); + let manifest2 = lsm::Manifest { + segments: vec![ + lsm::SegmentRef { + id: "a".into(), + records: 1, + }, + lsm::SegmentRef { + id: "c".into(), + records: 1, + }, + ], + ..Default::default() + }; + let segs2 = vec![ + segs[0].clone(), + Arc::new(ColdSegment::open(s.as_ref(), "ns", "c").await.unwrap()), + ]; + let hits = search( + s.as_ref(), + "ns", + &segs2, + &manifest2, + "default", + &[1.0, 1.0, 0.0, 0.0], + 10, + DEFAULT_NPROBE, + &Default::default(), + ) + .await + .unwrap(); + assert!( + hits.iter().all(|(c, _)| c.id != 1), + "newer segment's tombstone must kill the older chunk" + ); + let _ = std::fs::remove_dir_all(&root); + } + + // Paged metadata-index path: force it with a tiny full-fetch threshold + // and verify hydration still resolves every candidate. + #[tokio::test] + async fn paged_metaidx_hydrates() { + let (root, s) = storage("paged"); + let ids: Vec = (0..50).collect(); + s.put( + "ns/segments/p", + bytes::Bytes::from(seg_with_chunks(&ids, &[])), + ) + .await + .unwrap(); + let seg = ColdSegment::open_with_limits(s.as_ref(), "ns", "p", 16) + .await + .unwrap(); + assert!( + matches!(seg.metaidx, MetaIdx::Paged { .. }), + "tiny threshold must force the paged variant" + ); + let ranges = seg + .meta_ranges(s.as_ref(), &[0, 7, 49, 999_999]) + .await + .unwrap(); + let found: std::collections::HashSet = ranges.iter().map(|r| r.0).collect(); + assert_eq!( + found, + [0u64, 7, 49].into_iter().collect(), + "paged lookups must resolve present ids and skip absent ones" + ); + let _ = std::fs::remove_dir_all(&root); + } + + // Pre-v3 segments are rejected loudly — v2 has no metaidx, so cold + // serving it would read the whole flat section and then drop every hit. + #[tokio::test] + async fn v2_segment_rejected_with_upgrade_hint() { + let (root, s) = storage("v2"); + let mut bytes = Vec::new(); + bytes.extend_from_slice(&crate::collections::cloud::SEG_MAGIC_V2); + bytes.extend_from_slice(&0u64.to_le_bytes()); + bytes.extend_from_slice(&2u32.to_le_bytes()); + bytes.extend_from_slice(b"[]"); + s.put("ns/segments/old", bytes::Bytes::from(bytes)) + .await + .unwrap(); + let err = match ColdSegment::open(s.as_ref(), "ns", "old").await { + Err(e) => e, + Ok(_) => panic!("v2 segment must be rejected"), + }; + assert!(err.to_string().contains("compact"), "{err}"); + let _ = std::fs::remove_dir_all(&root); + } +} diff --git a/crates/compass/src/search/filter_pushdown.rs b/crates/compass/src/search/filter_pushdown.rs index 2bf2db6..fc0c03b 100644 --- a/crates/compass/src/search/filter_pushdown.rs +++ b/crates/compass/src/search/filter_pushdown.rs @@ -84,8 +84,9 @@ mod tests { use super::*; // Semantics (eq / range / contains / in, AND across fields) are covered - // end-to-end in filter_index.rs tests via FilterIndex::eligible — the one - // live evaluator. These only pin the compile() shape. + // in filter_index.rs tests (FilterIndex::eligible, the warm evaluator) + // and cold.rs (eval_filters, its cold-path mirror). These only pin the + // compile() shape. #[test] fn compile_shapes() { let mut f = HashMap::new(); diff --git a/crates/compass/src/search/ivf.rs b/crates/compass/src/search/ivf.rs new file mode 100644 index 0000000..8c3d84c --- /dev/null +++ b/crates/compass/src/search/ivf.rs @@ -0,0 +1,314 @@ +// search/ivf.rs — IVF (inverted-file) clustering for serve-from-storage. +// +// Compaction k-means-clusters each vector space and writes two sections into +// the segment: `cent:` (tiny: centroids + a cluster directory) and +// `clu:` (the vectors, grouped by cluster). A cold query then needs +// only: the centroids (cached, a few hundred KB), and range-GETs of the +// `nprobe` nearest clusters — instead of materializing the whole segment. +// That is what turns "attach = rebuild everything" into "query = a handful +// of small reads": the difference between warm and true serverless. +// +// Section formats (all little-endian): +// cent: = [u32 k][u32 dims] +// [k × dims × f32 centroids] +// [k × (u64 offset, u64 len, u32 count)] cluster directory, +// offsets relative to the START of clu:'s body +// clu: = concatenation of clusters, each [count × (u64 id, dims×f32)] +// +// Vectors are stored L2-NORMALIZED in `clu` so scoring is a plain dot +// product (cosine == dot on unit vectors); centroids are means of normalized +// vectors, re-normalized. + +/// Below this many rows a space is stored as the flat `emb:` section and cold +/// queries brute-force it — clustering tiny sets costs more than it saves. +pub const CLUSTER_MIN_ROWS: usize = 5_000; + +/// Cap on k-means training sample: training cost is O(sample × k × dims); +/// assignment of ALL rows is a single pass afterwards. +const TRAIN_SAMPLE_MAX: usize = 20_000; +const KMEANS_ITERS: usize = 8; + +/// Number of clusters for n rows: sqrt(n), clamped. At 5M rows and 384 dims +/// this keeps a cluster ~3.5MB — a few parallel range-GETs per query. +pub fn cluster_count(n: usize) -> usize { + ((n as f64).sqrt() as usize).clamp(16, 4096) +} + +#[inline] +pub fn normalize(v: &mut [f32]) { + let norm: f32 = v.iter().map(|x| x * x).sum::().sqrt(); + if norm > 0.0 { + for x in v.iter_mut() { + *x /= norm; + } + } +} + +#[inline] +pub fn dot(a: &[f32], b: &[f32]) -> f32 { + a.iter().zip(b).map(|(x, y)| x * y).sum() +} + +/// K-means (Lloyd) over normalized vectors, trained on a deterministic +/// sample. Returns (centroids, assignment of EVERY input row). +/// Deterministic: seeded by row order, no RNG. +pub fn kmeans(rows: &[(u64, Vec)], dims: usize, k: usize) -> (Vec>, Vec) { + let n = rows.len(); + let k = k.min(n).max(1); + + // Deterministic training sample: evenly-strided rows. + let stride = (n / TRAIN_SAMPLE_MAX).max(1); + let sample: Vec<&[f32]> = rows + .iter() + .step_by(stride) + .map(|(_, v)| v.as_slice()) + .collect(); + + // Init: evenly-strided sample points as seeds. + let seed_stride = (sample.len() / k).max(1); + let mut centroids: Vec> = sample + .iter() + .step_by(seed_stride) + .take(k) + .map(|v| v.to_vec()) + .collect(); + while centroids.len() < k { + centroids.push(centroids[centroids.len() % sample.len().max(1)].clone()); + } + + let nearest = |cents: &[Vec], v: &[f32]| -> usize { + let mut best = 0usize; + let mut best_d = f32::MIN; + for (i, c) in cents.iter().enumerate() { + let d = dot(c, v); // unit vectors: max dot == min angle + if d > best_d { + best_d = d; + best = i; + } + } + best + }; + + for _ in 0..KMEANS_ITERS { + let mut sums = vec![vec![0f32; dims]; k]; + let mut counts = vec![0usize; k]; + for v in &sample { + let c = nearest(¢roids, v); + for (s, x) in sums[c].iter_mut().zip(v.iter()) { + *s += x; + } + counts[c] += 1; + } + for (i, (sum, cnt)) in sums.iter_mut().zip(counts.iter()).enumerate() { + if *cnt > 0 { + for x in sum.iter_mut() { + *x /= *cnt as f32; + } + normalize(sum); + centroids[i] = std::mem::take(sum); + } + // Empty cluster: keep the old centroid (harmless; directory entry + // just ends up with count 0). + } + } + + let assignment: Vec = rows.iter().map(|(_, v)| nearest(¢roids, v)).collect(); + (centroids, assignment) +} + +/// Directory entry for one cluster inside `clu:`. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct ClusterRef { + pub offset: u64, + pub len: u64, + pub count: u32, +} + +/// Build the `cent:` and `clu:` section bodies for one space. +/// Input rows may be un-normalized; they are normalized in place here. +pub fn build_sections(mut rows: Vec<(u64, Vec)>, dims: usize) -> (Vec, Vec) { + for (_, v) in rows.iter_mut() { + normalize(v); + } + let k = cluster_count(rows.len()); + let (centroids, assignment) = kmeans(&rows, dims, k); + + // Group row indexes by cluster, then lay clusters out contiguously. + let mut by_cluster: Vec> = vec![Vec::new(); k]; + for (row_idx, &c) in assignment.iter().enumerate() { + by_cluster[c].push(row_idx); + } + + let row_size = 8 + dims * 4; + let mut clu = Vec::with_capacity(rows.len() * row_size); + let mut dir: Vec = Vec::with_capacity(k); + for members in &by_cluster { + let offset = clu.len() as u64; + for &ri in members { + let (id, v) = &rows[ri]; + clu.extend_from_slice(&id.to_le_bytes()); + for x in v { + clu.extend_from_slice(&x.to_le_bytes()); + } + } + dir.push(ClusterRef { + offset, + len: (members.len() * row_size) as u64, + count: members.len() as u32, + }); + } + + let mut cent = Vec::with_capacity(8 + k * dims * 4 + k * 20); + cent.extend_from_slice(&(k as u32).to_le_bytes()); + cent.extend_from_slice(&(dims as u32).to_le_bytes()); + for c in ¢roids { + for x in c { + cent.extend_from_slice(&x.to_le_bytes()); + } + } + for d in &dir { + cent.extend_from_slice(&d.offset.to_le_bytes()); + cent.extend_from_slice(&d.len.to_le_bytes()); + cent.extend_from_slice(&d.count.to_le_bytes()); + } + (cent, clu) +} + +/// Parsed `cent:` section. +pub struct Centroids { + pub dims: usize, + pub centroids: Vec>, + pub dir: Vec, +} + +pub fn parse_cent(body: &[u8]) -> Option { + if body.len() < 8 { + return None; + } + let k = u32::from_le_bytes(body[0..4].try_into().ok()?) as usize; + let dims = u32::from_le_bytes(body[4..8].try_into().ok()?) as usize; + let cent_bytes = k.checked_mul(dims)?.checked_mul(4)?; + let dir_bytes = k.checked_mul(20)?; + if body.len() < 8 + cent_bytes + dir_bytes { + return None; + } + let mut centroids = Vec::with_capacity(k); + let mut pos = 8; + for _ in 0..k { + let mut v = Vec::with_capacity(dims); + for _ in 0..dims { + v.push(f32::from_le_bytes(body[pos..pos + 4].try_into().ok()?)); + pos += 4; + } + centroids.push(v); + } + let mut dir = Vec::with_capacity(k); + for _ in 0..k { + let offset = u64::from_le_bytes(body[pos..pos + 8].try_into().ok()?); + let len = u64::from_le_bytes(body[pos + 8..pos + 16].try_into().ok()?); + let count = u32::from_le_bytes(body[pos + 16..pos + 20].try_into().ok()?); + pos += 20; + dir.push(ClusterRef { offset, len, count }); + } + Some(Centroids { + dims, + centroids, + dir, + }) +} + +/// Iterate `(id, vector)` rows out of a cluster blob. +pub fn parse_cluster_rows(body: &[u8], dims: usize) -> impl Iterator)> + '_ { + let row = 8 + dims * 4; + body.chunks_exact(row).map(move |r| { + let id = u64::from_le_bytes(r[0..8].try_into().unwrap()); + let mut v = Vec::with_capacity(dims); + for d in 0..dims { + let o = 8 + d * 4; + v.push(f32::from_le_bytes(r[o..o + 4].try_into().unwrap())); + } + (id, v) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn synthetic(n: usize, dims: usize) -> Vec<(u64, Vec)> { + // Deterministic, mildly clustered data: 8 anchor directions + noise. + (0..n) + .map(|i| { + let anchor = i % 8; + let v: Vec = (0..dims) + .map(|d| { + let base = if d % 8 == anchor { 1.0 } else { 0.1 }; + base + ((i * 31 + d * 17) % 97) as f32 / 970.0 + }) + .collect(); + (i as u64, v) + }) + .collect() + } + + #[test] + fn sections_roundtrip_and_cover_all_rows() { + let dims = 16; + let rows = synthetic(1000, dims); + let (cent, clu) = build_sections(rows.clone(), dims); + let parsed = parse_cent(¢).expect("cent parses"); + assert_eq!(parsed.dims, dims); + let total: u32 = parsed.dir.iter().map(|d| d.count).sum(); + assert_eq!(total as usize, rows.len(), "every row lands in a cluster"); + // Every directory range decodes to exactly `count` rows and all ids + // survive. + let mut seen = std::collections::HashSet::new(); + for d in &parsed.dir { + let body = &clu[d.offset as usize..(d.offset + d.len) as usize]; + let rows: Vec<_> = parse_cluster_rows(body, dims).collect(); + assert_eq!(rows.len(), d.count as usize); + for (id, v) in rows { + assert_eq!(v.len(), dims); + assert!(seen.insert(id), "id {id} duplicated across clusters"); + } + } + assert_eq!(seen.len(), 1000); + } + + #[test] + fn nearest_cluster_probe_finds_exact_vector() { + // Self-recall: probing the nearest clusters for a vector that IS in + // the index must find it with a modest nprobe. + let dims = 16; + let rows = synthetic(2000, dims); + let (cent, clu) = build_sections(rows.clone(), dims); + let parsed = parse_cent(¢).unwrap(); + + let mut hits = 0; + let probes = 4; + for probe_i in (0..2000).step_by(97) { + let mut q = rows[probe_i].1.clone(); + normalize(&mut q); + let mut ranked: Vec<(usize, f32)> = parsed + .centroids + .iter() + .enumerate() + .map(|(i, c)| (i, dot(c, &q))) + .collect(); + ranked.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap()); + let found = ranked.iter().take(probes).any(|(ci, _)| { + let d = parsed.dir[*ci]; + let body = &clu[d.offset as usize..(d.offset + d.len) as usize]; + parse_cluster_rows(body, dims).any(|(id, _)| id == rows[probe_i].0) + }); + if found { + hits += 1; + } + } + let total = (0..2000).step_by(97).count(); + assert!( + hits * 10 >= total * 9, + "self-recall with nprobe={probes}: {hits}/{total}" + ); + } +} diff --git a/crates/compass/src/search/mod.rs b/crates/compass/src/search/mod.rs index f233d6b..37b47a8 100644 --- a/crates/compass/src/search/mod.rs +++ b/crates/compass/src/search/mod.rs @@ -7,11 +7,13 @@ pub mod chunk_cache; pub mod chunk_store; +pub mod cold; #[cfg(test)] mod filter_bench; pub mod filter_index; pub mod filter_pushdown; pub mod hybrid; +pub mod ivf; pub mod mmap_vectors; pub mod tantivy_fts; pub mod vector; diff --git a/crates/compass/src/storage/local.rs b/crates/compass/src/storage/local.rs index a0412da..b645fd3 100644 --- a/crates/compass/src/storage/local.rs +++ b/crates/compass/src/storage/local.rs @@ -132,7 +132,11 @@ impl Storage for LocalDiskStorage { .metadata() .map_err(|e| StorageError::Io(e.to_string()))? .len(); - if range.start > range.end || range.end > size { + // Contract parity with the object-store backend: a range end past + // the object is CLAMPED (S3/GCS Range semantics), not an error — + // cold reads probe fixed-size headers on objects of unknown length. + let range = range.start..range.end.min(size); + if range.start > range.end { return Err(StorageError::InvalidRange { start: range.start, end: range.end, @@ -361,11 +365,9 @@ mod tests { let s = store("range"); s.put("k", Bytes::from_static(b"0123456789")).await.unwrap(); assert_eq!(&s.get_range("k", 2..5).await.unwrap()[..], b"234"); - // Out-of-bounds range errors. - assert!(matches!( - s.get_range("k", 5..100).await, - Err(StorageError::InvalidRange { .. }) - )); + // End past EOF is CLAMPED (object-store Range semantics — cold reads + // probe fixed-size headers on objects of unknown length). + assert_eq!(&s.get_range("k", 5..100).await.unwrap()[..], b"56789"); } // Boundary cases for the seek-based get_range (each a plausible off-by-one). @@ -387,9 +389,12 @@ mod tests { #[allow(clippy::reversed_empty_ranges)] let reversed = s.get_range("k", 6..3).await; assert!(matches!(reversed, Err(StorageError::InvalidRange { .. }))); - // end past EOF → error. + // end past EOF → clamped to the object (matches S3/GCS semantics). + assert_eq!(&s.get_range("k", 8..11).await.unwrap()[..], b"89"); + // start past EOF stays an error (object_store 416 parity): only the + // END is clamped. assert!(matches!( - s.get_range("k", 8..11).await, + s.get_range("k", 20..30).await, Err(StorageError::InvalidRange { .. }) )); } diff --git a/crates/compass/src/storage/mod.rs b/crates/compass/src/storage/mod.rs index 0e5608b..b3f1bb5 100644 --- a/crates/compass/src/storage/mod.rs +++ b/crates/compass/src/storage/mod.rs @@ -108,11 +108,10 @@ pub trait Storage: Send + Sync { /// Whole-object read. async fn get(&self, key: &str) -> Result; - /// Range read — fetch only `range` bytes of the object. The primitive that - /// makes large segments servable without loading the whole object. - // Range reads are the sectioned-segment read primitive (v2 TOC points at - // byte ranges); both backends implement it, callers land with Phase 5/6. - #[allow(dead_code)] + /// Range read — the serve-from-storage primitive (segment TOCs point at + /// byte ranges; cold queries fetch only the sections they need). A range + /// end past the object is CLAMPED, never an error (S3/GCS semantics; + /// LocalDiskStorage matches). async fn get_range(&self, key: &str, range: Range) -> Result; /// Read the object together with its current version, for a CAS cycle. diff --git a/crates/compass/src/telemetry.rs b/crates/compass/src/telemetry.rs index 9d87b97..cc26504 100644 --- a/crates/compass/src/telemetry.rs +++ b/crates/compass/src/telemetry.rs @@ -15,15 +15,23 @@ const POSTHOG_API_KEY: &str = "phc_BFvsmH5rpe8GqJ8zwfqhH9jGAdZMXcNZhEao8mnDEd3X" const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60); const STARTUP_DELAY: Duration = Duration::from_secs(60); -/// Returns true if telemetry is enabled (default). +/// Returns true if telemetry is enabled. DEFAULT OFF: Compass's core promise +/// is "data never leaves your machine" — an engine pitched on privacy must +/// not phone home unless the operator explicitly opts in +/// (COMPASS_TELEMETRY=on). DO_NOT_TRACK is honored even when opted in. pub fn is_enabled() -> bool { - if let Ok(v) = std::env::var("COMPASS_TELEMETRY") { - return !matches!(v.to_lowercase().as_str(), "off" | "false" | "0" | "no"); - } if let Ok(v) = std::env::var("DO_NOT_TRACK") { - return !matches!(v.as_str(), "1" | "true"); + if matches!(v.as_str(), "1" | "true") { + return false; + } } - true + matches!( + std::env::var("COMPASS_TELEMETRY") + .unwrap_or_default() + .to_lowercase() + .as_str(), + "on" | "true" | "1" | "yes" + ) } /// Persistent instance ID — generated once, stored in data_dir/instance_id. @@ -100,7 +108,7 @@ pub fn spawn_telemetry( let instance_id = get_or_create_instance_id(&data_dir); tracing::info!( - "Anonymous telemetry enabled (instance: {}). Set COMPASS_TELEMETRY=off to disable.", + "Anonymous telemetry enabled by explicit opt-in (instance: {}). Unset COMPASS_TELEMETRY to disable.", &instance_id[..8] ); diff --git a/docs/deployment.md b/docs/deployment.md new file mode 100644 index 0000000..a6fe8b8 --- /dev/null +++ b/docs/deployment.md @@ -0,0 +1,84 @@ +# Deployment topologies + +Compass runs in three shapes. All of them are the same binary; the shape is +chosen entirely by environment variables. + +## 1. Local single node (default) + +Zero config. Local disk is the source of truth; everything is embedded. + +```bash +./compass # or: docker run -p 4001:4001 -v ./data:/app/data compass +``` + +- No cloud credentials, no telemetry, no network calls. +- Tenant partitions (`partition_by`) work fully in this mode. +- Backup = copy `DATA_DIR`. + +## 2. Cloud: serving nodes + stateless writers + +Object storage is the source of truth; nodes are disposable. + +```bash +# Serving node(s): full local indexes, fast reads, background convergence +COMPASS_STORAGE=s3://bucket AWS_ACCESS_KEY_ID=… AWS_SECRET_ACCESS_KEY=… ./compass + +# Writer node(s): stateless append-only ingest, boots in milliseconds +COMPASS_ROLE=writer COMPASS_STORAGE=s3://bucket … ./compass +``` + +- Writers validate against the bucket config, mint chunk ids from a CAS + allocator (never collide with anyone), append one WAL fragment, return a + `seq`. Durable immediately; searchable on serving nodes within + `COMPASS_REFRESH_INTERVAL` (default 5s). +- Read-your-writes: pass a write's `seq` as `min_seq` on search. +- A serving node that loses its disk rebuilds every collection from the + bucket on boot. Kill -9 is a supported operation. +- Optional: `COMPASS_LAZY_ATTACH=true` + `COMPASS_MAX_ATTACHED=N` bound RAM + to the hot collection set (LRU detach; re-attach on demand). + +Upgrade caveat: do not run pre-v0.4 and v0.4 writers against one bucket; +old readers fail loudly on the v0.4 segment format rather than mis-reading. + +## 3. Cloud: cold serving (serverless reads) + +```bash +COMPASS_COLD_SERVE=true COMPASS_STORAGE=s3://bucket … ./compass +``` + +- Boots in <1s regardless of how much data the bucket holds; RAM starts at + ~tens of MB. +- Semantic queries on collections the node has NEVER attached are answered + from object-storage range reads (manifest → cached centroids → a few + cluster reads → byte-range hydration). Freshness is read-your-writes by + construction — every cold query reads the live manifest. +- `COMPASS_WARM_AFTER` (default 3) cold hits promote a background attach: + cold → warm → hot automatically. +- Honest limits: cold is semantic-only (FTS errors until the namespace + warms); recall on unstructured vector spaces needs a higher + `COMPASS_COLD_NPROBE` (see [search-quality.md](search-quality.md)); + scoring options (recency/boosts/relations) are rejected cold rather than + silently ignored. + +## Multi-tenant collections (any topology) + +```bash +curl -X POST :4001/collections -d '{"name":"app","embedding_dims":384, + "config":{"partition_by":"tenant_id"}}' +``` + +Every chunk routes to an internal per-tenant partition by +`metadata.tenant_id`. Searches/deletes must filter on the partition field +(exact, or `{"in":[…]}` for ≤16 tenants). Ids are collection-unique; +partitions auto-create on first ingest (writer nodes included), hide from +listings, cascade-delete with the parent. Serving cost tracks the HOT tenant +set — 50 or 200 tenants boot identically. + +## What the fleet does NOT give you (yet) + +- Tenant-affinity routing between nodes: put a proxy in front and hash a + tenant header to a node, or every node will warm every hot tenant. +- Per-tenant auth: `COMPASS_API_KEY` is one key for the whole node; tenant + scoping is the caller's responsibility today. +- `/metrics` and `/health` are unauthenticated by design; firewall them if + collection names/counts are sensitive. diff --git a/docs/scale-envelope.md b/docs/scale-envelope.md index 74e541e..47e5480 100644 --- a/docs/scale-envelope.md +++ b/docs/scale-envelope.md @@ -32,7 +32,7 @@ runs meaningfully faster; treat these as conservative floors. file — roughly linear in collection size). Lazy attach + LRU keep this a first-request cost per namespace, not a boot cost, but a 100M-chunk collection still takes tens of minutes to attach on first use. -- **Billion-vector serving therefore remains out of envelope** until +Billion-vector serving is reached via tenant partitions + serve-from-storage: the per-NAMESPACE envelope above bounds the largest tenant, not the collection, and cold reads serve unattached namespaces from object storage (see docs/search-quality.md for the recall contract). serve-from-storage indexes land (roadmap Phase 6: centroid routing over range-readable segments — attach becomes "fetch centroids", milliseconds). Do not deploy a single collection past ~10–50M chunks and expect diff --git a/docs/search-quality.md b/docs/search-quality.md new file mode 100644 index 0000000..9f65373 --- /dev/null +++ b/docs/search-quality.md @@ -0,0 +1,51 @@ +# Search quality — measured recall & latency + +Method: 20k docs × 64 dims, 200 held-out queries (perturbed documents), +ground truth = exact cosine top-10 (numpy). Two datasets: **structured** +(32 topic clusters + mild noise — the shape real text/image embeddings have) +and **adversarial uniform** (noise-dominated, nearly structureless — the +worst case for any ANN index). Latency measured over HTTP against Docker + +MinIO on a laptop; treat relative numbers, not absolutes. + +## Warm path (attached: HNSW, ef_search=128) + +| dataset | recall@10 | p50 | p95 | +|---|---|---|---| +| structured | 1.000 | 1.0ms | 1.1ms | +| structured, filtered (50% selectivity) | 1.000 | 1.0ms | 1.2ms | +| adversarial uniform | 0.895 | 1.2ms | 1.3ms | +| adversarial, filtered | 0.962 | 1.4ms | 1.7ms | + +Full-text (BM25): exact-token top-1 50/50; topical precision@10 = 1.000. + +## Cold path (serve-from-storage: IVF over object storage) + +`COMPASS_COLD_NPROBE` clusters probed per segment (default 8): + +| dataset | nprobe | recall@10 | p50 | +|---|---|---|---| +| structured | 4 | 0.947 | 13ms | +| structured | **8 (default)** | **1.000** | 13ms | +| structured | 16 | 1.000 | 14ms | +| adversarial uniform | 4 | 0.269 | 12ms | +| adversarial uniform | 8 | 0.409 | 13ms | +| adversarial uniform | 16 | 0.590 | 14ms | +| adversarial uniform | k (exhaustive) | 1.000 | 30ms | + +## The honest contract + +- On **clustered embedding spaces** — which is what real embedding models + produce — cold recall reaches warm parity at the default nprobe, at + ~10× warm latency (a handful of object-storage range reads). +- On **unstructured/uniform vector spaces**, IVF recall drops steeply (this + is inherent to inverted-file indexes, not a Compass bug — the exhaustive + row proves the pipeline is exact). If your vectors are random-ish + (hashes, uncalibrated projections), raise `COMPASS_COLD_NPROBE` + aggressively or rely on warm serving (`COMPASS_WARM_AFTER` promotes hot + namespaces automatically). +- Cold filtered queries apply filters post-selection with an 8× deeper + candidate pool; extremely selective filters (≪1% match rate) can + under-return on the cold path — warm search has no such limit. + +Reproduce: `scratchpad` eval scripts live in the PR discussion; the harness +is ~100 lines of numpy + HTTP and pins seeds. diff --git a/docs/serverless-roadmap.md b/docs/serverless-roadmap.md index d26d38f..e7c6fbf 100644 --- a/docs/serverless-roadmap.md +++ b/docs/serverless-roadmap.md @@ -1,6 +1,6 @@ # Serverless Roadmap -> Status: Phases 0-3 SHIPPED on feat/warm-serverless (v0.4.0 candidate); Phases 4+ planned. Target: evolve Compass from a cloud-durable single-node +> Status: warm serverless (stateless writers, refresh, lazy attach), tenant partitions, and serve-from-storage cold reads are SHIPPED on the v0.4 branches (phase numbers below predate the final split). Remaining: routing/affinity hooks, per-tenant auth binding, cold FTS, intra-tenant sharding. Target: evolve Compass from a cloud-durable single-node > engine (v0.3.0) into a fully serverless database — storage/compute separated, > stateless workers, bounded cold starts, scale-to-zero — with **every item > additive and open source** under Apache 2.0. Local-first, zero-config diff --git a/docs/v0.4-filter-aware-ann.md b/docs/v0.4-filter-aware-ann.md index 31c3efd..fe41d9b 100644 --- a/docs/v0.4-filter-aware-ann.md +++ b/docs/v0.4-filter-aware-ann.md @@ -2,7 +2,6 @@ Status: done. Decision: ship Path A (USearch native `filtered_search`). -Sister design doc: [docs/v0.4-vision.md](./v0.4-vision.md). ## Verdict diff --git a/scripts/e2e.sh b/scripts/e2e.sh index 44a725a..d1d4ee4 100755 --- a/scripts/e2e.sh +++ b/scripts/e2e.sh @@ -4,6 +4,7 @@ set -u FULL=${FULL:-localhost:4001} WRITER=${WRITER:-localhost:4009} +COLD=${COLD:-} # optional: a COMPASS_COLD_SERVE node against the same bucket pass=0; fail=0 ok(){ echo " ✅ $1"; pass=$((pass+1)); } bad(){ echo " ❌ $1 ($2)"; fail=$((fail+1)); } @@ -146,6 +147,26 @@ code=$(curl -s -o /dev/null -w '%{http_code}' -X DELETE $FULL/collections/e2e) code=$(curl -s -o /dev/null -w '%{http_code}' $FULL/collections/e2e) [ "$code" = "404" ] || [ "$(curl -s $FULL/collections/e2e)" = "null" ] && ok "collection gone" || bad gone "$code" +if [ -n "$COLD" ]; then +echo "── serve-from-storage (cold node) ──" +post $FULL/collections '{"name":"icy","embedding_dims":4}' >/dev/null +post $FULL/collections/icy/ingest '{"chunks":[ + {"file_id":"i1","chunk_index":0,"doc_type":"chunk","text":"glacier core","metadata":{"kind":"ice"},"embeddings":{"default":[0.9,0.1,0.1,0.1]}}, + {"file_id":"i2","chunk_index":0,"doc_type":"chunk","text":"magma core","metadata":{"kind":"fire"},"embeddings":{"default":[0.1,0.9,0.1,0.1]}}]}' >/dev/null +r=$(post $COLD/collections/icy/search '{"query":"","mode":"semantic","top_k":3,"query_vector":[0.9,0.1,0.1,0.1]}') +f1=$(echo "$r" | jqn "d['results'][0]['chunk']['file_id']") +[ "$f1" = "i1" ] && ok "cold node answers without attach" || bad cold-search "$f1" +n=$(post $COLD/collections/icy/search '{"query":"","mode":"semantic","top_k":3,"query_vector":[0.9,0.1,0.1,0.1],"filters":{"kind":"fire"}}' | jqn "d['results'][0]['chunk']['file_id']") +[ "$n" = "i2" ] && ok "cold filters apply" || bad cold-filter "$n" +code=$(curl -s -o /dev/null -w '%{http_code}' -X POST $COLD/collections/icy/search -H 'content-type: application/json' -d '{"query":"glacier","mode":"fts"}') +[ "$code" -ge 400 ] && ok "cold FTS rejected with guidance" || bad cold-fts "$code" +wseq2=$(post $WRITER/collections/icy/ingest '{"chunks":[{"file_id":"i3","chunk_index":0,"doc_type":"chunk","text":"fresh tail","metadata":{"kind":"new"},"embeddings":{"default":[0.1,0.1,0.9,0.1]}}]}' | jqn "d['seq']") +f3=$(post $COLD/collections/icy/search '{"query":"","mode":"semantic","top_k":1,"query_vector":[0.1,0.1,0.9,0.1]}' | jqn "d['results'][0]['chunk']['file_id']") +[ "$f3" = "i3" ] && ok "cold read-your-writes (writer tail visible instantly, seq $wseq2)" || bad cold-ryw "$f3" +curl -s $COLD/metrics | grep -q "compass_cold_searches_total [1-9]" && ok "cold metrics counting" || bad cold-metrics x +curl -s -o /dev/null -X DELETE $FULL/collections/icy +fi + echo "" echo "E2E RESULT: $pass passed, $fail failed" [ "$fail" = "0" ]