Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
98 changes: 98 additions & 0 deletions docs/user-guide/concurrency.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,98 @@
# Concurrency

`snapvec` indexes are **single-writer, multi-reader** within a single
process.

## What's safe

- Multiple threads calling `search()` on the **same** index concurrently,
**after** `freeze()` (see below). Search paths allocate their own
scratch buffers and, once the index is frozen, only read shared state.
- Multiple processes opening **different** index files and querying
them independently.
- A single writer and any number of readers, as long as you never
overlap a writer with a reader on the same instance.

## Freeze before sharing across threads

`SnapIndex.search()` lazily materialises an internal float16 centroid
cache on the first query. If two threads both issue their *first*
search concurrently, they race on that cache assignment -- concurrent
`search()` is only safe once the cache exists.

The library exposes `freeze()` precisely to pre-warm that state:

```python
idx = SnapIndex(dim=384, bits=4)
idx.add_batch(ids, vectors)
idx.freeze() # pre-warms the cache, makes concurrent search safe

# Now you can hand idx to multiple reader threads.
```

After `freeze()`, mutations (`add_batch`, `delete`) raise, so freeze
also doubles as an "I'm done writing" signal. `PQSnapIndex`,
`IVFPQSnapIndex`, and `ResidualSnapIndex` have the same contract:
call `freeze()` -- or at least issue one warm-up `search()` on the
main thread -- before fanning out.

## What's not safe

- Two threads calling `add_batch`, `delete`, or `fit` on the same
index concurrently. There is no internal lock; the library assumes
the caller serializes mutations.
- One thread mutating while another searches. Even when the mutation
looks atomic at the Python level (for example, appending to a list),
internal arrays are resized and re-sorted without coordination.

## Recommended pattern

If your application overlaps readers and writers, every public call
must acquire the same lock. A simple wrapper:

```python
import threading

class SafeIndex:
def __init__(self, idx):
self._idx = idx
self._lock = threading.Lock()

def add_batch(self, ids, vectors):
with self._lock:
self._idx.add_batch(ids, vectors)

def delete(self, id_):
with self._lock:
return self._idx.delete(id_)

def search(self, query, k=10, **kwargs):
with self._lock:
return self._idx.search(query, k=k, **kwargs)
```
Comment on lines +61 to +72

Copilot AI Apr 20, 2026

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The SafeIndex example only locks add_batch/delete, but leaves search() unlocked. That does not prevent reader/writer overlap on the same instance, which the page says is unsafe. Either (a) take the same lock in search() as well, (b) demonstrate a reader/writer lock, or (c) recommend a copy-on-write pattern (build a new index, then atomically swap the reference) so searches never overlap with mutations.

Copilot uses AI. Check for mistakes.

If you *never* mutate the index during reads (typical for a
build-then-serve workflow: one `add_batch` at startup, many `search()`
calls forever after), the reader lock can be skipped -- `search()`
only touches immutable shared state (codes, centroids) and
thread-local scratch. If you need higher read concurrency *and*
occasional writes, use a `threading.RLock` plus a read/write wrapper
(for example, the `readerwriterlock` package) at the application
layer.

## Cross-process access

The on-disk format is designed for cold reload, not shared access:

- `save()` writes to `<path>.tmp` then renames, so a concurrent reader
calling `load(path)` either sees the old file or the new one, never
a partial write.
- Nothing prevents two processes from opening the same file and writing
back. If you need multi-process writes, put a file lock (for example,
`fcntl.flock`) around the `save()` call in your application layer.

## Future work

Native single-writer protection via an internal `threading.Lock`, and a
delta-buffer mode for low-latency incremental updates, are tracked on
the roadmap for a future release.
1 change: 1 addition & 0 deletions mkdocs.yml
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@ nav:
- ResidualSnapIndex: user-guide/residual.md
- Save and load: user-guide/save-load.md
- Filtered search: user-guide/filter-search.md
- Concurrency: user-guide/concurrency.md
- Architecture: architecture.md
- Benchmarks: benchmarks.md
- API reference:
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ Issues = "https://github.com/stffns/snapvec/issues"
dev = [
"pytest",
"pytest-cov",
"hypothesis>=6.100",
"mypy>=1.8",
"ruff",
"Cython>=3.0",
Expand Down
2 changes: 2 additions & 0 deletions snapvec/_index.py
Original file line number Diff line number Diff line change
Expand Up @@ -406,6 +406,8 @@ def search(
materialising the full float16 cache. This trades peak RAM for
additional compute — useful when N > ~500k vectors.
"""
if k < 1:
raise ValueError(f"k must be >= 1; got {k}")
if not self._ids:
return []

Expand Down
224 changes: 224 additions & 0 deletions tests/test_adversarial.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,224 @@
"""Adversarial edge-case tests.

Tiny dims, tiny N, degenerate distributions, empty inputs. These are
the cases where off-by-one bugs and implicit shape assumptions tend to
surface.
"""
from __future__ import annotations

import numpy as np
import pytest

from snapvec import IVFPQSnapIndex, PQSnapIndex, ResidualSnapIndex, SnapIndex


# --------------------------------------------------------------------------- #
# Empty index #
# --------------------------------------------------------------------------- #


def test_empty_snapindex_search_returns_empty() -> None:
idx = SnapIndex(dim=8, bits=4, seed=0)
q = np.zeros(8, dtype=np.float32)
q[0] = 1.0
assert idx.search(q, k=5) == []
assert len(idx) == 0


def test_empty_pqsnapindex_search_returns_empty() -> None:
vecs = np.random.default_rng(0).standard_normal((64, 8)).astype(np.float32)
idx = PQSnapIndex(dim=8, M=4, K=8, seed=0)
idx.fit(vecs)
# Never call add_batch -- index is fitted but empty.
q = vecs[0]
assert idx.search(q, k=5) == []
assert len(idx) == 0


# --------------------------------------------------------------------------- #
# Single-vector corpus #
# --------------------------------------------------------------------------- #


def test_snapindex_n1() -> None:
"""n=1 is a legal if-degenerate corpus. search(k>=1) returns 1 hit."""
v = np.random.default_rng(0).standard_normal((1, 16)).astype(np.float32)
idx = SnapIndex(dim=16, bits=4, seed=0)
idx.add_batch(["only"], v)
hits = idx.search(v[0], k=5)
assert len(hits) == 1
assert hits[0][0] == "only"


# --------------------------------------------------------------------------- #
# k larger than n #
# --------------------------------------------------------------------------- #


def test_search_k_larger_than_n_returns_n() -> None:
vecs = np.random.default_rng(0).standard_normal((3, 16)).astype(np.float32)
idx = SnapIndex(dim=16, bits=4, seed=0)
idx.add_batch([0, 1, 2], vecs)
hits = idx.search(vecs[0], k=100)
assert len(hits) == 3


# --------------------------------------------------------------------------- #
# Zero-norm inputs #
# --------------------------------------------------------------------------- #


def test_search_with_zero_query_returns_empty() -> None:
"""Zero-norm query can't be normalized; library returns [] instead of NaN hits."""
vecs = np.random.default_rng(0).standard_normal((20, 16)).astype(np.float32)
idx = SnapIndex(dim=16, bits=4, seed=0)
idx.add_batch(list(range(20)), vecs)
q_zero = np.zeros(16, dtype=np.float32)
assert idx.search(q_zero, k=5) == []


# --------------------------------------------------------------------------- #
# All-same-vector corpus (degenerate clusters) #
# --------------------------------------------------------------------------- #


def test_snapindex_all_same_vector() -> None:
"""Every vector identical -> search should still return k distinct ids."""
v = np.ones((1, 32), dtype=np.float32)
vecs = np.tile(v, (10, 1))
idx = SnapIndex(dim=32, bits=4, seed=0)
idx.add_batch(list(range(10)), vecs)
hits = idx.search(v[0], k=5)
assert len(hits) == 5
ids = [h[0] for h in hits]
assert len(set(ids)) == len(ids) # distinct


# --------------------------------------------------------------------------- #
# Filter edge cases #
# --------------------------------------------------------------------------- #


def test_filter_with_only_unknown_ids_returns_empty() -> None:
vecs = np.random.default_rng(0).standard_normal((50, 16)).astype(np.float32)
idx = SnapIndex(dim=16, bits=4, seed=0)
idx.add_batch(list(range(50)), vecs)
hits = idx.search(vecs[0], k=5, filter_ids={"never-added", "also-never"})
assert hits == []


def test_filter_with_empty_set_returns_empty() -> None:
vecs = np.random.default_rng(0).standard_normal((50, 16)).astype(np.float32)
idx = SnapIndex(dim=16, bits=4, seed=0)
idx.add_batch(list(range(50)), vecs)
hits = idx.search(vecs[0], k=5, filter_ids=set())
assert hits == []


# --------------------------------------------------------------------------- #
# Delete-all #
# --------------------------------------------------------------------------- #


def test_snapindex_delete_all_then_search() -> None:
vecs = np.random.default_rng(0).standard_normal((5, 16)).astype(np.float32)
idx = SnapIndex(dim=16, bits=4, seed=0)
idx.add_batch(list(range(5)), vecs)
for i in range(5):
assert idx.delete(i) is True
assert len(idx) == 0
assert idx.search(vecs[0], k=3) == []


# --------------------------------------------------------------------------- #
# Aggressive compression (bits=2) #
# --------------------------------------------------------------------------- #


def test_snapindex_bits2_basic_recall() -> None:
"""bits=2 still returns *something* sensible on clustered data."""
rng = np.random.default_rng(0)
centers = rng.standard_normal((5, 32)).astype(np.float32) * 3
assign = rng.integers(0, 5, size=200)
jitter = rng.standard_normal((200, 32)).astype(np.float32) * 0.2
corpus = centers[assign] + jitter

idx = SnapIndex(dim=32, bits=2, seed=0)
idx.add_batch(list(range(200)), corpus)

# Self-query should rank the exact corpus row near the top on such
# strongly clustered data.
hits = idx.search(corpus[0], k=5)
assert len(hits) == 5
returned = [h[0] for h in hits]
# Because clusters have ~40 members and bits=2 is aggressive,
# we don't assert hits[0] == 0. We only assert the top-5 are all
# from the same cluster as the query.
query_cluster = assign[0]
top_clusters = [assign[i] for i in returned]
assert top_clusters.count(query_cluster) >= 3


# --------------------------------------------------------------------------- #
# k=0 is an error #
# --------------------------------------------------------------------------- #


def test_search_k_zero_raises() -> None:
vecs = np.random.default_rng(0).standard_normal((10, 16)).astype(np.float32)
idx = SnapIndex(dim=16, bits=4, seed=0)
idx.add_batch(list(range(10)), vecs)
with pytest.raises(ValueError):
idx.search(vecs[0], k=0)


# --------------------------------------------------------------------------- #
# IVF-PQ extreme nprobe #
# --------------------------------------------------------------------------- #


def test_ivfpq_nprobe_equals_nlist_is_full_scan() -> None:
"""With nprobe=nlist, IVF-PQ must visit every cluster."""
rng = np.random.default_rng(0)
corpus = rng.standard_normal((300, 32)).astype(np.float32)
idx = IVFPQSnapIndex(dim=32, nlist=8, M=4, K=16, seed=0)
idx.fit(corpus)
idx.add_batch(list(range(300)), corpus)

hits = idx.search(corpus[0], k=10, nprobe=8)
assert len(hits) == 10


def test_ivfpq_nprobe_out_of_range_raises() -> None:
rng = np.random.default_rng(0)
corpus = rng.standard_normal((300, 32)).astype(np.float32)
idx = IVFPQSnapIndex(dim=32, nlist=8, M=4, K=16, seed=0)
idx.fit(corpus)
idx.add_batch(list(range(300)), corpus)
with pytest.raises(ValueError):
idx.search(corpus[0], k=5, nprobe=0)
with pytest.raises(ValueError):
idx.search(corpus[0], k=5, nprobe=99)


# --------------------------------------------------------------------------- #
# Residual rerank #
# --------------------------------------------------------------------------- #


def test_residual_rerank_saturates_near_full_recall() -> None:
"""ResidualSnapIndex with a generous rerank_M should match full scan on
clustered data."""
rng = np.random.default_rng(0)
centers = rng.standard_normal((8, 32)).astype(np.float32) * 3
assign = rng.integers(0, 8, size=200)
jitter = rng.standard_normal((200, 32)).astype(np.float32) * 0.2
corpus = centers[assign] + jitter

idx = ResidualSnapIndex(dim=32, b1=3, b2=3, seed=0)
idx.add_batch(list(range(200)), corpus)

full = [h[0] for h in idx.search(corpus[0], k=5, rerank_M=None)]
reranked = [h[0] for h in idx.search(corpus[0], k=5, rerank_M=50)]
# Both modes should agree on most of the top-5.
assert len(set(full) & set(reranked)) >= 3
Loading
Loading