vrr/ruvector
2
0
Fork 0
mirror of https://github.com/ruvnet/RuVector.git synced 2026-08-26 00:52:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

fix(context,core): close three HIGH findings in the scope-sharded index

Browse source 
Security review of this PR found three HIGH issues, each demonstrated by
running code rather than inferred. All three are fixed here, with
regression tests that were verified to fail against the unfixed code
before the fix was applied.

HIGH-1 -- concurrent open regressed in ruvector-core. Moving DB_POOL to
Weak without giving VectorStorage a Drop meant Weak::upgrade went None the
instant the strong count hit zero, while redb's Database::drop (a write txn
plus fsync) and the flock release were still pending. A concurrent
VectorStorage::new then called Database::create against a still-locked
file: 194/200 failures with "Database already open", against 0/200 on main.
The Weak change is kept -- it fixes a real data-correctness bug -- and the
lifetime is made explicit instead: VectorStorage now has a Drop that clears
the slot and drops the Database while still holding the path guard, so the
lock is released before a racing open can observe an empty slot. The pool
is now per-path, so one path's fsync no longer blocks opens of unrelated
paths, and dead entries are reaped rather than accumulating forever.

HIGH-2 -- create_shard adopted a pre-existing file. open() verified the
scope-to-filename binding but the create path did not, and VectorDB::new on
an existing path silently inherits that file's stored config and vectors.
A shard planted at a victim's computed filename while the index was open
was adopted on the next insert, and because create_shard then rewrote the
manifest to the victim scope, the rows were laundered: after a restart the
binding check passed and the victim scope permanently contained them.
Creation now refuses an existing file, and refuses again after opening if
the database carries a manifest, is non-empty, or opened with unrequested
dimensions or metric -- closing the check-to-open window. Neither refusal
unlinks, so a planted file survives to be quarantined rather than deleted.

HIGH-3 -- erase_scope resurrected data. Two index handles over one root
shared the process-global pool while keeping independent in-memory indexes,
so an erase reported success while the other handle kept the rows alive,
and the erasing handle's next insert got the unlinked database back. open()
now takes an exclusive advisory lock on the root, so a second handle fails
loudly instead of silently sharing state. This also closes the divergence
where one handle's search could not see another's insert while scope_stats
reported the higher count.

Also fixed: erase_scope now unlinks before mutating the catalog, so a
failed unlink can no longer report an in-memory erase over a surviving
file; a single unreadable shard-shaped file is quarantined and surfaced
through quarantined_shards() instead of denying open() to every tenant;
max_results is capped so usize::MAX is refused by validate() rather than
reaching Vec::with_capacity and panicking out of a #![forbid(unsafe_code)]
crate; and errors report the shard filename rather than an absolute path.

The per-shard search counter is now type-enforced. Shard moved into a
private module where ann_search is the only route to the ANN index and
always increments, so an uncounted traversal no longer compiles. The
counter previously guarded one of five db touch sites, which made the
isolation claim close to tautological.

The integration suite is parameterized over both FlatIndex and HNSW. It
previously set hnsw_config: None throughout, so the claim that a tenant
search never calls another tenant's ANN index was established with no ANN
index in the process. The claim holds under HNSW.

ADR renumbered 332 -> 334: 332 was already taken by the RF-sensing ADR
merged earlier the same day, on a branch that already contained that merge.
The ADR now also records that the manifest binds scope to filename and not
to content, that there is no MAC over stored vectors, and that the global
scope ceiling and eager open are explicit non-goals.

Co-Authored-By: claude-flow <ruv@ruv.net>
Claude-Session: https://claude.ai/code/session_016QSCkKnxDjqU49NVVpWMK5
This commit is contained in:
ruv 2026-08-22 17:04:43 -04:00
parent 41d690f88a
commit 13e7dbb50b
10 changed files with 1052 additions and 283 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -9180,6 +9180,7 @@ dependencies = [
name = "ruvector-context"
version = "2.3.0"
dependencies = [
"fs2",
"ruvector-core 2.3.0",
"serde",
"serde_json",

1
crates/ruvector-context/Cargo.toml
View file

@ -11,6 +11,7 @@ keywords = ["vector", "context", "multitenancy", "security"]
categories = ["database", "data-structures"]
[dependencies]
fs2 = "0.4"
ruvector-core = { path = "../ruvector-core", default-features = false, features = ["storage", "hnsw"] }
serde = { workspace = true }
serde_json = { workspace = true }

12
crates/ruvector-context/src/error.rs
View file

@ -44,8 +44,20 @@ pub enum ContextIndexError {
#[error("context point ID already exists with different bytes")]
ImmutableConflict,
/// A discovered shard lacks a valid, hash-bound scope manifest.
///
/// Reports the shard filename only; the absolute path is never disclosed.
#[error("corrupt context shard: {0}")]
CorruptShard(String),
/// A file already occupies the deterministic path of a scope being created.
///
/// The vector engine would adopt that file's stored vectors and relabel
/// them as the new scope's data, so creation fails and the file is left
/// untouched for quarantine at the next open.
#[error("refusing to adopt pre-existing context shard: {0}")]
ShardAdoption(String),
/// Another live handle already holds the exclusive lock on this index root.
#[error("context index root is already locked by another handle")]
RootLocked,
/// A shared index lock was poisoned by a panicking caller.
#[error("context index lock poisoned")]
LockPoisoned,

223
crates/ruvector-context/src/index.rs
View file

@ -1,16 +1,24 @@
//! Persistent scope-sharded vector index.
use crate::shard::{acquire_root_lock, create_shard, load_shard, shard_name, Shard};
use crate::{ContextIndexError, ContextNamespace, ContextScope, Result};
use ruvector_core::types::DbOptions;
use ruvector_core::{SearchQuery, VectorDB, VectorEntry};
use ruvector_core::{SearchQuery, VectorEntry};
use std::collections::BTreeMap;
use std::fs::File;
use std::io::ErrorKind;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
const MANIFEST_KEY: &str = "ruvector_context_scope_v1";
const MAX_POINT_ID_BYTES: usize = 512;
/// Hard ceiling on [`ContextIndexOptions::max_results`].
///
/// Top K bounds a caller-driven allocation, so an unbounded "unlimited"
/// spelling such as `usize::MAX` must be rejected by configuration validation
/// rather than reaching the allocator.
pub const MAX_RESULT_LIMIT: usize = 4_096;
/// Resource and vector-engine configuration for a scoped context index.
#[derive(Debug, Clone)]
pub struct ContextIndexOptions {
@ -20,7 +28,7 @@ pub struct ContextIndexOptions {
pub max_scopes: usize,
/// Maximum descendant shards one search may touch.
pub max_search_shards: usize,
/// Maximum top K accepted from a caller.
/// Maximum top K accepted from a caller, capped at [`MAX_RESULT_LIMIT`].
pub max_results: usize,
}
@ -37,6 +45,11 @@ impl ContextIndexOptions {
"resource limits must be nonzero",
));
}
if self.max_results > MAX_RESULT_LIMIT {
return Err(ContextIndexError::InvalidConfiguration(
"max_results exceeds the supported ceiling",
));
}
Ok(())
}
}
@ -70,50 +83,70 @@ pub struct ScopeStats {
pub searches: u64,
}
struct Shard {
db: VectorDB,
searches: AtomicU64,
}
type PathShards = BTreeMap<Vec<String>, Arc<Shard>>;
type NamespaceShards = BTreeMap<ContextNamespace, PathShards>;
/// Persistent vector index physically partitioned by exact context scope.
///
/// # Caller authorization
///
/// This type is a physical isolation mechanism, not an authorization service.
/// It never decides who may touch a scope: every method assumes the caller has
/// already authorized the supplied [`ContextScope`] for the requesting
/// principal. Passing an unauthorized scope yields that scope's data.
///
/// One live handle owns an index root exclusively. [`ScopedContextIndex::open`]
/// takes an advisory lock on the root directory, so a second handle over the
/// same root fails with [`ContextIndexError::RootLocked`] instead of silently
/// diverging from the first.
pub struct ScopedContextIndex {
root: PathBuf,
options: ContextIndexOptions,
shards: RwLock<NamespaceShards>,
quarantined: Vec<String>,
_lock: File,
}
impl ScopedContextIndex {
/// Open an index directory and recover every hash-bound scope shard.
///
/// Takes an exclusive advisory lock on `root`; the lock is released when
/// the returned handle is dropped, and by the operating system if the
/// process exits without dropping it.
///
/// Files that match the shard naming convention but fail to load, or whose
/// stored manifest does not match their filename, are quarantined and
/// skipped rather than failing the whole open. See
/// [`ScopedContextIndex::quarantined_shards`].
pub fn open(root: impl AsRef<Path>, options: ContextIndexOptions) -> Result<Self> {
options.validate()?;
let root = root.as_ref().to_path_buf();
std::fs::create_dir_all(&root)?;
let mut shards = BTreeMap::new();
let lock = acquire_root_lock(&root)?;
let mut shards: NamespaceShards = BTreeMap::new();
let mut quarantined = Vec::new();
for entry in std::fs::read_dir(&root)? {
let path = entry?.path();
if !is_shard_path(&path) {
let Some(filename) = shard_name(&path) else {
continue;
};
match load_shard(&path, &filename, &options) {
Ok((scope, shard)) if filename == scope.shard_filename() => {
shards
.entry(scope.namespace().clone())
.or_default()
.insert(scope.path().to_vec(), Arc::new(shard));
}
_ => quarantined.push(filename),
}
let (scope, shard) = load_shard(&path, &options)?;
let filename = path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("");
if filename != scope.shard_filename() {
return Err(ContextIndexError::CorruptShard(filename.to_string()));
}
shards
.entry(scope.namespace().clone())
.or_insert_with(BTreeMap::new)
.insert(scope.path().to_vec(), Arc::new(shard));
}
quarantined.sort_unstable();
let index = Self {
root,
options,
shards: RwLock::new(shards),
quarantined,
_lock: lock,
};
if index.scope_count()? > index.options.max_scopes {
return Err(ContextIndexError::ScopeCapacity {
@ -123,7 +156,14 @@ impl ScopedContextIndex {
Ok(index)
}
/// Insert an immutable point, returning success for an identical replay.
/// Insert a point, returning success for a byte-identical replay.
///
/// A point is immutable only while it is present: once
/// [`ScopedContextIndex::delete_point`] removes an identifier, the same
/// identifier may be inserted again with different bytes.
///
/// The caller must already have authorized `scope` for the requesting
/// principal.
pub fn insert(&self, scope: &ContextScope, point: ContextPoint) -> Result<()> {
validate_point(&point, self.options.vector.dimensions)?;
let mut all = self
@ -144,26 +184,28 @@ impl ScopedContextIndex {
}
let shard = Arc::new(create_shard(&self.root, scope, &self.options)?);
all.entry(scope.namespace().clone())
.or_insert_with(BTreeMap::new)
.or_default()
.insert(scope.path().to_vec(), Arc::clone(&shard));
shard
};
if let Some(existing) = shard.db.get(&point.id)? {
if let Some(existing) = shard.get(&point.id)? {
return if existing.vector == point.vector {
Ok(())
} else {
Err(ContextIndexError::ImmutableConflict)
};
}
shard.db.insert(VectorEntry {
shard.insert(VectorEntry {
id: Some(point.id),
vector: point.vector,
metadata: None,
})?;
Ok(())
})
}
/// Search only exact shards at or below an already authorized root scope.
///
/// The caller must already have authorized `root` for the requesting
/// principal; every shard in its subtree is considered readable.
pub fn search(
&self,
root: &ContextScope,
@ -203,10 +245,9 @@ impl ScopedContextIndex {
maximum: self.options.max_search_shards,
});
}
let mut matches = Vec::with_capacity(k);
let mut matches = Vec::with_capacity(k.min(MAX_RESULT_LIMIT));
for (path, shard) in selected {
shard.searches.fetch_add(1, Ordering::Relaxed);
let results = shard.db.search(SearchQuery {
let results = shard.ann_search(SearchQuery {
vector: vector.to_vec(),
k,
filter: None,
@ -230,6 +271,9 @@ impl ScopedContextIndex {
}
/// Delete one point from one exact scope.
///
/// The caller must already have authorized `scope` for the requesting
/// principal.
pub fn delete_point(&self, scope: &ContextScope, id: &str) -> Result<bool> {
validate_point_id(id)?;
let all = self
@ -242,10 +286,17 @@ impl ScopedContextIndex {
else {
return Ok(false);
};
shard.db.delete(id).map_err(Into::into)
shard.delete(id)
}
/// Erase one exact scope and its persistent vector shard.
///
/// The shard file is unlinked before any in-memory state changes, so a
/// failure to unlink is reported as an error with the scope still present
/// rather than as a phantom erase.
///
/// The caller must already have authorized `scope` for the requesting
/// principal.
pub fn erase_scope(&self, scope: &ContextScope) -> Result<bool> {
let mut all = self
.shards
@ -254,20 +305,27 @@ impl ScopedContextIndex {
let Some(paths) = all.get_mut(scope.namespace()) else {
return Ok(false);
};
let Some(shard) = paths.remove(scope.path()) else {
if !paths.contains_key(scope.path()) {
return Ok(false);
};
if !paths.is_empty() {
drop(shard);
} else {
all.remove(scope.namespace());
drop(shard);
}
std::fs::remove_file(self.root.join(scope.shard_filename()))?;
match std::fs::remove_file(self.root.join(scope.shard_filename())) {
Ok(()) => {}
Err(error) if error.kind() == ErrorKind::NotFound => {}
Err(error) => return Err(error.into()),
}
drop(paths.remove(scope.path()));
let namespace_is_empty = paths.is_empty();
if namespace_is_empty {
all.remove(scope.namespace());
}
Ok(true)
}
/// Return counters for one exact scope without exposing other scopes.
///
/// This is an existence oracle over exact scopes and reports exact vector
/// counts, so the caller must already have authorized `scope` for the
/// requesting principal exactly as it would for a read.
pub fn scope_stats(&self, scope: &ContextScope) -> Result<Option<ScopeStats>> {
let all = self
.shards
@ -280,8 +338,8 @@ impl ScopedContextIndex {
return Ok(None);
};
Ok(Some(ScopeStats {
vectors: shard.db.len()?,
searches: shard.searches.load(Ordering::Relaxed),
vectors: shard.len()?,
searches: shard.searches(),
}))
}
@ -293,46 +351,16 @@ impl ScopedContextIndex {
.map_err(|_| ContextIndexError::LockPoisoned)?;
Ok(count_scopes(&all))
}
}
fn create_shard(root: &Path, scope: &ContextScope, options: &ContextIndexOptions) -> Result<Shard> {
let path = root.join(scope.shard_filename());
let mut vector = options.vector.clone();
vector.storage_path = path.to_string_lossy().into_owned();
let db = VectorDB::new(vector)?;
let manifest = serde_json::to_string(scope)?;
if let Err(error) = db.save_config_value(MANIFEST_KEY, &manifest) {
drop(db);
let _ = std::fs::remove_file(&path);
return Err(error.into());
/// Shard filenames skipped at open time because they failed to load or
/// were not bound to their own filename.
///
/// Operations on the scopes those files claim fail rather than silently
/// serving or overwriting unverified data.
#[must_use]
pub fn quarantined_shards(&self) -> &[String] {
&self.quarantined
}
Ok(Shard {
db,
searches: AtomicU64::new(0),
})
}
fn load_shard(path: &Path, options: &ContextIndexOptions) -> Result<(ContextScope, Shard)> {
let mut vector = options.vector.clone();
vector.storage_path = path.to_string_lossy().into_owned();
let db = VectorDB::new(vector)?;
if db.options().dimensions != options.vector.dimensions
|| db.options().distance_metric != options.vector.distance_metric
{
return Err(ContextIndexError::CorruptShard(path.display().to_string()));
}
let manifest = db
.load_config_value(MANIFEST_KEY)?
.ok_or_else(|| ContextIndexError::CorruptShard(path.display().to_string()))?;
let scope: ContextScope = serde_json::from_str(&manifest)
.map_err(|_| ContextIndexError::CorruptShard(path.display().to_string()))?;
Ok((
scope,
Shard {
db,
searches: AtomicU64::new(0),
},
))
}
fn validate_point(point: &ContextPoint, dimensions: usize) -> Result<()> {
@ -366,17 +394,6 @@ fn count_scopes(shards: &NamespaceShards) -> usize {
shards.values().map(BTreeMap::len).sum()
}
fn is_shard_path(path: &Path) -> bool {
let Some(name) = path.file_name().and_then(|name| name.to_str()) else {
return false;
};
name.len() == 69
&& name.ends_with(".redb")
&& name[..64]
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}
fn compare_matches(left: &ContextMatch, right: &ContextMatch) -> core::cmp::Ordering {
left.score
.partial_cmp(&right.score)
@ -399,3 +416,25 @@ fn push_top_match(matches: &mut Vec<ContextMatch>, limit: usize, candidate: Cont
matches[worst] = candidate;
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn unbounded_max_results_is_rejected_by_validation() {
let options = ContextIndexOptions {
vector: DbOptions {
dimensions: 3,
..DbOptions::default()
},
max_scopes: 1,
max_search_shards: 1,
max_results: usize::MAX,
};
assert!(matches!(
options.validate(),
Err(ContextIndexError::InvalidConfiguration(_))
));
}
}

22
crates/ruvector-context/src/lib.rs
View file

@ -3,6 +3,22 @@
//! Metadata filtering after approximate-nearest-neighbor traversal is not a
//! tenant boundary. This crate places each exact context scope in a distinct
//! vector index and selects authorized descendant shards before any ANN call.
//!
//! # This crate is not the tenant boundary
//!
//! [`ScopedContextIndex`] enforces *physical* isolation between scopes; it does
//! not authenticate or authorize anyone. Every method — [`ScopedContextIndex::insert`],
//! [`ScopedContextIndex::search`], [`ScopedContextIndex::delete_point`],
//! [`ScopedContextIndex::erase_scope`], and [`ScopedContextIndex::scope_stats`]
//! — treats the [`ContextScope`] it is handed as already authorized for the
//! requesting principal. Deciding which principal may name which scope is the
//! caller's responsibility, and it must happen before the call. In particular
//! [`ScopedContextIndex::scope_stats`] is an existence oracle that reports
//! exact vector counts, so it needs the same authorization as a read.
//!
//! The scope manifest stored in each shard binds a scope to a *filename*, not
//! to shard contents; there is no message authentication code over the stored
//! vectors.
#![forbid(unsafe_code)]
#![deny(missing_docs)]
@ -10,7 +26,11 @@
mod error;
mod index;
mod scope;
mod shard;
pub use error::{ContextIndexError, Result};
pub use index::{ContextIndexOptions, ContextMatch, ContextPoint, ScopeStats, ScopedContextIndex};
pub use index::{
ContextIndexOptions, ContextMatch, ContextPoint, ScopeStats, ScopedContextIndex,
MAX_RESULT_LIMIT,
};
pub use scope::{ContextNamespace, ContextScope};

198
crates/ruvector-context/src/shard.rs Normal file
View file

@ -0,0 +1,198 @@
//! Shard files: naming, creation, recovery, and the root lock.
//!
//! The vector engine backing one exact scope is wrapped here so the per-scope
//! isolation counter cannot be bypassed, and so the rules that decide whether a
//! file on disk may be treated as this scope's shard live in one place.
use crate::{ContextIndexError, ContextIndexOptions, ContextScope, Result};
use fs2::FileExt;
use ruvector_core::types::DbOptions;
use ruvector_core::{SearchQuery, SearchResult, VectorDB, VectorEntry};
use std::fs::File;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
const MANIFEST_KEY: &str = "ruvector_context_scope_v1";
/// Name of the exclusive root lock file held for the lifetime of one index.
///
/// It is deliberately not a valid shard name, so `is_shard_name` skips it.
const ROOT_LOCK_FILENAME: &str = ".lock";
/// Handle to the vector engine backing one exact context scope.
///
/// The wrapped [`VectorDB`] is private to this module, so the per-scope
/// isolation counter cannot be bypassed by accident: [`Shard::ann_search`] is
/// the only route to the approximate-nearest-neighbor index and it always
/// counts the traversal. Touching a shard without counting does not compile.
pub(crate) struct Shard {
db: VectorDB,
searches: AtomicU64,
}
impl Shard {
/// Wrap an opened vector database as a countable shard.
pub(crate) fn new(db: VectorDB) -> Self {
Self {
db,
searches: AtomicU64::new(0),
}
}
/// Configuration the engine actually opened with, stored config included.
pub(crate) fn options(&self) -> &DbOptions {
self.db.options()
}
/// Fetch one stored point by identifier.
pub(crate) fn get(&self, id: &str) -> Result<Option<VectorEntry>> {
Ok(self.db.get(id)?)
}
/// Store one point.
pub(crate) fn insert(&self, entry: VectorEntry) -> Result<()> {
self.db.insert(entry)?;
Ok(())
}
/// Remove one point, reporting whether it was present.
pub(crate) fn delete(&self, id: &str) -> Result<bool> {
Ok(self.db.delete(id)?)
}
/// Number of points currently retained.
pub(crate) fn len(&self) -> Result<usize> {
Ok(self.db.len()?)
}
/// Whether the shard holds no points at all.
pub(crate) fn is_empty(&self) -> Result<bool> {
Ok(self.db.is_empty()?)
}
/// Persist one engine-level configuration value.
pub(crate) fn save_config_value(&self, key: &str, value: &str) -> Result<()> {
Ok(self.db.save_config_value(key, value)?)
}
/// Read back one engine-level configuration value.
pub(crate) fn load_config_value(&self, key: &str) -> Result<Option<String>> {
Ok(self.db.load_config_value(key)?)
}
/// Number of ANN traversals issued to this shard since it was opened.
pub(crate) fn searches(&self) -> u64 {
self.searches.load(Ordering::Relaxed)
}
/// The only accessor that reaches the ANN index; every call is counted.
pub(crate) fn ann_search(&self, query: SearchQuery) -> Result<Vec<SearchResult>> {
self.searches.fetch_add(1, Ordering::Relaxed);
Ok(self.db.search(query)?)
}
}
pub(crate) fn acquire_root_lock(root: &Path) -> Result<File> {
let file = std::fs::OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(root.join(ROOT_LOCK_FILENAME))?;
match file.try_lock_exclusive() {
Ok(()) => Ok(file),
Err(error) if error.raw_os_error() == fs2::lock_contended_error().raw_os_error() => {
Err(ContextIndexError::RootLocked)
}
Err(error) => Err(error.into()),
}
}
pub(crate) fn create_shard(
root: &Path,
scope: &ContextScope,
options: &ContextIndexOptions,
) -> Result<Shard> {
let filename = scope.shard_filename();
let path = root.join(&filename);
// `VectorDB::new` opens an existing file, adopts its stored configuration
// and keeps its stored vectors. Writing this scope's manifest over such a
// file would relabel another scope's data as this one's, so a pre-existing
// file is refused and never removed.
if path.try_exists()? {
return Err(ContextIndexError::ShardAdoption(filename));
}
let mut vector = options.vector.clone();
vector.storage_path = path.to_string_lossy().into_owned();
let shard = Shard::new(VectorDB::new(vector)?);
// Defence in depth for the window between the check and the open: a shard
// this call created is empty, carries no manifest, and opened with exactly
// the requested vector configuration.
if shard.load_config_value(MANIFEST_KEY)?.is_some()
|| !shard.is_empty()?
|| shard.options().dimensions != options.vector.dimensions
|| shard.options().distance_metric != options.vector.distance_metric
{
drop(shard);
return Err(ContextIndexError::ShardAdoption(filename));
}
let manifest = serde_json::to_string(scope)?;
if let Err(error) = shard.save_config_value(MANIFEST_KEY, &manifest) {
// Only reachable when this call created the file, because adoption is
// refused above, so this cannot delete another scope's shard.
drop(shard);
let _ = std::fs::remove_file(&path);
return Err(error);
}
Ok(shard)
}
pub(crate) fn load_shard(
path: &Path,
filename: &str,
options: &ContextIndexOptions,
) -> Result<(ContextScope, Shard)> {
let mut vector = options.vector.clone();
vector.storage_path = path.to_string_lossy().into_owned();
let shard = Shard::new(VectorDB::new(vector)?);
if shard.options().dimensions != options.vector.dimensions
|| shard.options().distance_metric != options.vector.distance_metric
{
return Err(ContextIndexError::CorruptShard(filename.to_string()));
}
let manifest = shard
.load_config_value(MANIFEST_KEY)?
.ok_or_else(|| ContextIndexError::CorruptShard(filename.to_string()))?;
let scope: ContextScope = serde_json::from_str(&manifest)
.map_err(|_| ContextIndexError::CorruptShard(filename.to_string()))?;
Ok((scope, shard))
}
/// Shard filename for a directory entry, or `None` if it is not a shard.
pub(crate) fn shard_name(path: &Path) -> Option<String> {
let name = path.file_name().and_then(|name| name.to_str())?;
is_shard_name(name).then(|| name.to_string())
}
fn is_shard_name(name: &str) -> bool {
// The root lock is not a shard; its name cannot collide with one, but the
// exclusion is stated rather than inferred from the length check.
name != ROOT_LOCK_FILENAME
&& name.len() == 69
&& name.ends_with(".redb")
&& name[..64]
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn root_lock_is_never_mistaken_for_a_shard() {
assert!(!is_shard_name(ROOT_LOCK_FILENAME));
assert!(is_shard_name(&format!("{}.redb", "a".repeat(64))));
assert!(!is_shard_name(&format!("{}.redb", "g".repeat(64))));
}
}

510
crates/ruvector-context/tests/scoped_context.rs
View file

@ -1,11 +1,58 @@
//! Integration tests for persistent, scope-isolated context indexes.
//!
//! Every behavioural test runs against both vector engines. Isolation claims
//! about approximate-nearest-neighbor traversal are meaningless when the
//! process contains no ANN index at all, so `hnsw_config: None` alone cannot
//! establish them.
use ruvector_context::{
ContextIndexError, ContextIndexOptions, ContextNamespace, ContextPoint, ContextScope,
ScopedContextIndex,
};
use ruvector_core::types::DbOptions;
use ruvector_core::types::{DbOptions, HnswConfig};
use ruvector_core::DistanceMetric;
use std::path::{Path, PathBuf};
#[derive(Clone, Copy)]
enum Engine {
Flat,
Hnsw,
}
impl Engine {
fn hnsw_config(self) -> Option<HnswConfig> {
match self {
Self::Flat => None,
Self::Hnsw => Some(HnswConfig {
max_elements: 1_024,
..HnswConfig::default()
}),
}
}
}
/// Generate one `flat` and one `hnsw` test per declared body.
macro_rules! engine_tests {
($(fn $name:ident($engine:ident: Engine) $body:block)*) => {
$(
mod $name {
use super::*;
fn run($engine: Engine) $body
#[test]
fn flat() {
run(Engine::Flat);
}
#[test]
fn hnsw() {
run(Engine::Hnsw);
}
}
)*
};
}
fn namespace(tenant: &str) -> ContextNamespace {
ContextNamespace::new("context.example", tenant, "agent", "reader", "memory").unwrap()
@ -19,13 +66,13 @@ fn scope(tenant: &str, path: &[&str]) -> ContextScope {
.unwrap()
}
fn options(max_search_shards: usize) -> ContextIndexOptions {
fn options(engine: Engine, max_search_shards: usize) -> ContextIndexOptions {
ContextIndexOptions {
vector: DbOptions {
dimensions: 3,
distance_metric: DistanceMetric::Euclidean,
storage_path: String::new(),
hnsw_config: None,
hnsw_config: engine.hnsw_config(),
quantization: None,
},
max_scopes: 16,
@ -41,159 +88,332 @@ fn point(id: &str, vector: [f32; 3]) -> ContextPoint {
}
}
#[test]
fn cross_tenant_search_never_touches_other_tenant_index() {
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(8)).unwrap();
let acme = scope("acme", &["project"]);
let other = scope("other", &["project"]);
index
.insert(&acme, point("acme:one", [1.0, 0.0, 0.0]))
.unwrap();
index
.insert(&other, point("other:one", [1.0, 0.0, 0.0]))
.unwrap();
let matches = index.search(&acme, &[1.0, 0.0, 0.0], 4).unwrap();
assert_eq!(matches.len(), 1);
assert_eq!(matches[0].id, "acme:one");
assert_eq!(index.scope_stats(&acme).unwrap().unwrap().searches, 1);
assert_eq!(index.scope_stats(&other).unwrap().unwrap().searches, 0);
}
#[test]
fn path_prefix_selects_descendants_without_touching_siblings() {
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(8)).unwrap();
let root = scope("acme", &["project"]);
let child = scope("acme", &["project", "doc"]);
let sibling = scope("acme", &["project-archive"]);
index
.insert(&child, point("child", [0.0, 1.0, 0.0]))
.unwrap();
index
.insert(&sibling, point("sibling", [0.0, 1.0, 0.0]))
.unwrap();
let matches = index.search(&root, &[0.0, 1.0, 0.0], 4).unwrap();
assert_eq!(
matches
.iter()
.map(|item| item.id.as_str())
.collect::<Vec<_>>(),
vec!["child"]
);
assert_eq!(index.scope_stats(&child).unwrap().unwrap().searches, 1);
assert_eq!(index.scope_stats(&sibling).unwrap().unwrap().searches, 0);
}
#[test]
fn restart_recovers_hash_bound_scope_and_vectors() {
let temp = tempfile::tempdir().unwrap();
let target = scope("acme", &["durable"]);
{
let index = ScopedContextIndex::open(temp.path(), options(8)).unwrap();
index
.insert(&target, point("revision:view", [0.0, 0.0, 1.0]))
.unwrap();
}
let recovered = ScopedContextIndex::open(temp.path(), options(8)).unwrap();
let matches = recovered.search(&target, &[0.0, 0.0, 1.0], 1).unwrap();
assert_eq!(matches[0].id, "revision:view");
}
#[test]
fn immutable_replay_is_idempotent_but_conflict_is_refused() {
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(8)).unwrap();
let target = scope("acme", &["immutable"]);
index
.insert(&target, point("same", [1.0, 0.0, 0.0]))
.unwrap();
index
.insert(&target, point("same", [1.0, 0.0, 0.0]))
.unwrap();
assert!(matches!(
index.insert(&target, point("same", [0.0, 1.0, 0.0])),
Err(ContextIndexError::ImmutableConflict)
));
}
#[test]
fn fanout_failure_occurs_before_any_shard_search() {
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(1)).unwrap();
let root = ContextScope::root(namespace("acme"));
let one = scope("acme", &["one"]);
let two = scope("acme", &["two"]);
index.insert(&one, point("one", [1.0, 0.0, 0.0])).unwrap();
index.insert(&two, point("two", [0.0, 1.0, 0.0])).unwrap();
assert!(matches!(
index.search(&root, &[1.0, 0.0, 0.0], 1),
Err(ContextIndexError::ScopeFanout {
actual: 2,
maximum: 1
/// Shard filenames are opaque hashes, so tests recover them from the directory
/// rather than recomputing the private hash construction.
fn shard_files(root: &Path) -> Vec<PathBuf> {
let mut found = std::fs::read_dir(root)
.unwrap()
.map(|entry| entry.unwrap().path())
.filter(|path| {
path.extension()
.is_some_and(|extension| extension == "redb")
})
));
assert_eq!(index.scope_stats(&one).unwrap().unwrap().searches, 0);
assert_eq!(index.scope_stats(&two).unwrap().unwrap().searches, 0);
.collect::<Vec<_>>();
found.sort();
found
}
#[test]
fn exact_scope_erasure_removes_persistent_shard() {
let temp = tempfile::tempdir().unwrap();
let target = scope("acme", &["erase"]);
let index = ScopedContextIndex::open(temp.path(), options(8)).unwrap();
index
.insert(&target, point("erase-me", [1.0, 0.0, 0.0]))
.unwrap();
assert!(index.erase_scope(&target).unwrap());
assert_eq!(index.scope_count().unwrap(), 0);
/// Build a shard for `target` in a throwaway root and return its file path.
fn isolated_shard(engine: Engine, root: &Path, target: &ContextScope, id: &str) -> PathBuf {
let index = ScopedContextIndex::open(root, options(engine, 8)).unwrap();
index.insert(target, point(id, [1.0, 0.0, 0.0])).unwrap();
drop(index);
assert_eq!(
ScopedContextIndex::open(temp.path(), options(8))
let files = shard_files(root);
assert_eq!(files.len(), 1);
files.into_iter().next().unwrap()
}
engine_tests! {
fn cross_tenant_search_never_touches_other_tenant_index(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
let acme = scope("acme", &["project"]);
let other = scope("other", &["project"]);
index
.insert(&acme, point("acme:one", [1.0, 0.0, 0.0]))
.unwrap();
index
.insert(&other, point("other:one", [1.0, 0.0, 0.0]))
.unwrap();
let matches = index.search(&acme, &[1.0, 0.0, 0.0], 4).unwrap();
assert_eq!(matches.len(), 1);
assert_eq!(matches[0].id, "acme:one");
assert_eq!(index.scope_stats(&acme).unwrap().unwrap().searches, 1);
assert_eq!(index.scope_stats(&other).unwrap().unwrap().searches, 0);
}
fn path_prefix_selects_descendants_without_touching_siblings(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
let root = scope("acme", &["project"]);
let child = scope("acme", &["project", "doc"]);
let sibling = scope("acme", &["project-archive"]);
index
.insert(&child, point("child", [0.0, 1.0, 0.0]))
.unwrap();
index
.insert(&sibling, point("sibling", [0.0, 1.0, 0.0]))
.unwrap();
let matches = index.search(&root, &[0.0, 1.0, 0.0], 4).unwrap();
assert_eq!(
matches
.iter()
.map(|item| item.id.as_str())
.collect::<Vec<_>>(),
vec!["child"]
);
assert_eq!(index.scope_stats(&child).unwrap().unwrap().searches, 1);
assert_eq!(index.scope_stats(&sibling).unwrap().unwrap().searches, 0);
}
fn restart_recovers_hash_bound_scope_and_vectors(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
let target = scope("acme", &["durable"]);
{
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
index
.insert(&target, point("revision:view", [0.0, 0.0, 1.0]))
.unwrap();
}
let recovered = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
let matches = recovered.search(&target, &[0.0, 0.0, 1.0], 1).unwrap();
assert_eq!(matches[0].id, "revision:view");
}
fn immutable_replay_is_idempotent_but_conflict_is_refused(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
let target = scope("acme", &["immutable"]);
index
.insert(&target, point("same", [1.0, 0.0, 0.0]))
.unwrap();
index
.insert(&target, point("same", [1.0, 0.0, 0.0]))
.unwrap();
assert!(matches!(
index.insert(&target, point("same", [0.0, 1.0, 0.0])),
Err(ContextIndexError::ImmutableConflict)
));
}
fn deleted_point_ids_are_reusable_with_different_bytes(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
let target = scope("acme", &["reuse"]);
index
.insert(&target, point("id", [1.0, 0.0, 0.0]))
.unwrap();
assert!(index.delete_point(&target, "id").unwrap());
index
.insert(&target, point("id", [0.0, 1.0, 0.0]))
.unwrap();
assert_eq!(index.scope_stats(&target).unwrap().unwrap().vectors, 1);
}
fn fanout_failure_occurs_before_any_shard_search(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(engine, 1)).unwrap();
let root = ContextScope::root(namespace("acme"));
let one = scope("acme", &["one"]);
let two = scope("acme", &["two"]);
index.insert(&one, point("one", [1.0, 0.0, 0.0])).unwrap();
index.insert(&two, point("two", [0.0, 1.0, 0.0])).unwrap();
assert!(matches!(
index.search(&root, &[1.0, 0.0, 0.0], 1),
Err(ContextIndexError::ScopeFanout {
actual: 2,
maximum: 1
})
));
assert_eq!(index.scope_stats(&one).unwrap().unwrap().searches, 0);
assert_eq!(index.scope_stats(&two).unwrap().unwrap().searches, 0);
}
fn exact_scope_erasure_removes_persistent_shard(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
let target = scope("acme", &["erase"]);
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
index
.insert(&target, point("erase-me", [1.0, 0.0, 0.0]))
.unwrap();
assert!(index.erase_scope(&target).unwrap());
assert_eq!(index.scope_count().unwrap(), 0);
assert!(shard_files(temp.path()).is_empty());
drop(index);
assert_eq!(
ScopedContextIndex::open(temp.path(), options(engine, 8))
.unwrap()
.scope_count()
.unwrap(),
0
);
}
fn erased_scope_recreation_never_reuses_the_unlinked_database(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
let target = scope("acme", &["erase-recreate"]);
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
index
.insert(&target, point("old", [1.0, 0.0, 0.0]))
.unwrap();
assert!(index.erase_scope(&target).unwrap());
index
.insert(&target, point("new", [0.0, 1.0, 0.0]))
.unwrap();
let matches = index.search(&target, &[1.0, 0.0, 0.0], 8).unwrap();
assert_eq!(
matches
.iter()
.map(|item| item.id.as_str())
.collect::<Vec<_>>(),
vec!["new"]
);
}
fn non_finite_vectors_fail_before_index_access(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
let target = scope("acme", &["finite"]);
assert!(matches!(
index.insert(&target, point("nan", [f32::NAN, 0.0, 0.0])),
Err(ContextIndexError::NonFiniteVector)
));
assert!(matches!(
index.search(&target, &[f32::INFINITY, 0.0, 0.0], 1),
Err(ContextIndexError::NonFiniteVector)
));
assert_eq!(index.scope_count().unwrap(), 0);
}
fn planted_foreign_shard_is_refused_and_never_laundered(engine: Engine) {
let attacker = scope("attacker", &["docs"]);
let victim = scope("victim", &["docs"]);
// A real, well-formed shard belonging to the attacker's scope.
let attacker_root = tempfile::tempdir().unwrap();
let planted =
isolated_shard(engine, attacker_root.path(), &attacker, "attacker-planted-row");
// The victim scope's opaque filename, learned from a throwaway root.
let probe_root = tempfile::tempdir().unwrap();
let victim_file = isolated_shard(engine, probe_root.path(), &victim, "probe");
let victim_filename = victim_file.file_name().unwrap().to_owned();
// Plant it while the victim's index is already open, so the binding
// check performed at open time never sees the file.
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
std::fs::copy(&planted, temp.path().join(&victim_filename)).unwrap();
let refused = index.insert(&victim, point("victim-own-row", [1.0, 0.0, 0.0]));
let Err(ContextIndexError::ShardAdoption(reported)) = refused else {
panic!("create adopted a pre-existing shard: {refused:?}");
};
assert_eq!(Some(reported.as_str()), victim_filename.to_str());
assert!(!reported.contains(std::path::MAIN_SEPARATOR));
assert!(index.search(&victim, &[1.0, 0.0, 0.0], 8).unwrap().is_empty());
assert_eq!(index.scope_count().unwrap(), 0);
// Nothing was laundered: after a restart the planted file is still
// bound to the attacker's scope and is quarantined, not adopted.
drop(index);
let restarted = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
assert_eq!(
restarted.quarantined_shards(),
&[victim_filename.to_str().unwrap().to_string()]
);
assert_eq!(restarted.scope_count().unwrap(), 0);
assert!(restarted
.search(&victim, &[1.0, 0.0, 0.0], 8)
.unwrap()
.scope_count()
.unwrap(),
0
);
.is_empty());
}
fn second_handle_on_one_root_fails_loudly(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
assert!(matches!(
ScopedContextIndex::open(temp.path(), options(engine, 8)),
Err(ContextIndexError::RootLocked)
));
// Dropping the handle releases the advisory lock.
drop(index);
ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
}
fn lock_file_left_by_a_crashed_process_does_not_brick_the_root(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
// A process that exits without dropping its handle leaves the lock file
// behind, but the kernel released the advisory lock on exit. The
// leftover file must be reused, and never mistaken for a shard.
std::fs::write(temp.path().join(".lock"), b"").unwrap();
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
assert!(index.quarantined_shards().is_empty());
let target = scope("acme", &["after-crash"]);
index
.insert(&target, point("row", [1.0, 0.0, 0.0]))
.unwrap();
assert_eq!(index.scope_count().unwrap(), 1);
}
fn unloadable_shard_is_quarantined_without_denying_open(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
let good = scope("acme", &["healthy"]);
{
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
index
.insert(&good, point("healthy-row", [1.0, 0.0, 0.0]))
.unwrap();
}
let junk_name = format!("{}.redb", "ab".repeat(32));
std::fs::write(temp.path().join(&junk_name), b"not a vector database").unwrap();
let index = ScopedContextIndex::open(temp.path(), options(engine, 8)).unwrap();
assert_eq!(index.quarantined_shards(), &[junk_name]);
assert_eq!(index.scope_count().unwrap(), 1);
let matches = index.search(&good, &[1.0, 0.0, 0.0], 1).unwrap();
assert_eq!(matches[0].id, "healthy-row");
}
fn unbounded_max_results_is_refused_instead_of_panicking(engine: Engine) {
let temp = tempfile::tempdir().unwrap();
let mut unbounded = options(engine, 8);
unbounded.max_results = usize::MAX;
assert!(matches!(
ScopedContextIndex::open(temp.path(), unbounded),
Err(ContextIndexError::InvalidConfiguration(_))
));
}
}
/// A failed unlink must not be reported as an erase.
///
/// Unix only: the failure is produced by revoking write permission on the
/// index root, which a process running as root would bypass.
#[cfg(unix)]
#[test]
fn erased_scope_recreation_never_reuses_the_unlinked_database() {
fn failed_shard_unlink_does_not_report_a_phantom_erase() {
use std::os::unix::fs::PermissionsExt;
let temp = tempfile::tempdir().unwrap();
let target = scope("acme", &["erase-recreate"]);
let index = ScopedContextIndex::open(temp.path(), options(8)).unwrap();
let target = scope("acme", &["erase-fails"]);
let index = ScopedContextIndex::open(temp.path(), options(Engine::Flat, 8)).unwrap();
index
.insert(&target, point("old", [1.0, 0.0, 0.0]))
.unwrap();
assert!(index.erase_scope(&target).unwrap());
index
.insert(&target, point("new", [0.0, 1.0, 0.0]))
.insert(&target, point("gdpr-erase-me", [1.0, 0.0, 0.0]))
.unwrap();
let matches = index.search(&target, &[1.0, 0.0, 0.0], 8).unwrap();
assert_eq!(
matches
.iter()
.map(|item| item.id.as_str())
.collect::<Vec<_>>(),
vec!["new"]
);
}
let original = std::fs::metadata(temp.path()).unwrap().permissions();
let mut readonly = original.clone();
readonly.set_mode(0o555);
std::fs::set_permissions(temp.path(), readonly).unwrap();
let probe = temp.path().join("probe");
let enforced = std::fs::File::create(&probe).is_err();
let erased = enforced.then(|| index.erase_scope(&target));
std::fs::set_permissions(temp.path(), original).unwrap();
let _ = std::fs::remove_file(&probe);
#[test]
fn non_finite_vectors_fail_before_index_access() {
let temp = tempfile::tempdir().unwrap();
let index = ScopedContextIndex::open(temp.path(), options(8)).unwrap();
let target = scope("acme", &["finite"]);
assert!(matches!(
index.insert(&target, point("nan", [f32::NAN, 0.0, 0.0])),
Err(ContextIndexError::NonFiniteVector)
));
assert!(matches!(
index.search(&target, &[f32::INFINITY, 0.0, 0.0], 1),
Err(ContextIndexError::NonFiniteVector)
));
assert_eq!(index.scope_count().unwrap(), 0);
let Some(erased) = erased else {
return; // running as root; the permission bits are not enforced
};
assert!(erased.is_err(), "unlink failure reported as success");
assert_eq!(index.scope_count().unwrap(), 1);
assert_eq!(index.scope_stats(&target).unwrap().unwrap().vectors, 1);
let matches = index.search(&target, &[1.0, 0.0, 0.0], 1).unwrap();
assert_eq!(matches[0].id, "gdpr-erase-me");
assert_eq!(shard_files(temp.path()).len(), 1);
}

295
crates/ruvector-core/src/storage.rs
View file

@ -32,14 +32,34 @@ const CONFIG_TABLE: TableDefinition<&str, &str> = TableDefinition::new("config")
/// Key used to store database configuration in CONFIG_TABLE
const DB_CONFIG_KEY: &str = "__ruvector_db_config__";
/// Per-path guard around the pooled database handle.
///
/// The `Weak` (rather than a strong `Arc`) is what stops an erased-and-recreated
/// path from handing out a `Database` that still points at the unlinked inode:
/// the pool never keeps a database alive on its own. Opening and closing a
/// database both happen while this guard is held, so a close (redb's final write
/// transaction, fsync, and file-lock release, all of which run in
/// `Database::drop`) is never observed half-finished by a concurrent open.
type PathSlot = Mutex<Option<Weak<Database>>>;
// Global database connection pool to allow multiple VectorDB instances
// to share the same underlying database file
static DB_POOL: Lazy<Mutex<HashMap<PathBuf, Weak<Database>>>> =
// to share the same underlying database file. The global lock only guards the
// map itself; the slow work happens under the per-path slot lock so an fsync on
// one path never blocks opens of unrelated paths.
static DB_POOL: Lazy<Mutex<HashMap<PathBuf, Arc<PathSlot>>>> =
Lazy::new(|| Mutex::new(HashMap::new()));
/// The pooled database and the guard that owns its lifecycle.
struct Pooled {
db: Arc<Database>,
slot: Arc<PathSlot>,
}
/// Storage backend for vector database
pub struct VectorStorage {
db: Arc<Database>,
/// Always `Some` for a live handle; only `Drop` takes it.
pooled: Option<Pooled>,
path: PathBuf,
dimensions: usize,
}
@ -99,32 +119,86 @@ impl VectorStorage {
}
}
// Check if we already have a Database instance for this path
let db = {
// Claim this path's slot. Slot handles are only ever cloned while the
// pool lock is held, which is what makes the reference-count check in
// `release_slot_if_unused` sound.
let slot = {
let mut pool = DB_POOL.lock();
Arc::clone(pool.entry(path_buf.clone()).or_default())
};
if let Some(existing_db) = pool.get(&path_buf).and_then(Weak::upgrade) {
// Reuse existing database connection
existing_db
} else {
// Create new database and add to pool
let new_db = Arc::new(Database::create(&path_buf)?);
// Initialize tables
let write_txn = new_db.begin_write()?;
{
let _ = write_txn.open_table(VECTORS_TABLE)?;
let _ = write_txn.open_table(METADATA_TABLE)?;
let _ = write_txn.open_table(CONFIG_TABLE)?;
}
write_txn.commit()?;
pool.insert(path_buf, Arc::downgrade(&new_db));
new_db
let db = match Self::open_pooled(&slot, &path_buf) {
Ok(db) => db,
Err(e) => {
// Nothing was pooled, so this slot may now be garbage.
Self::release_slot_if_unused(&path_buf, slot);
return Err(e);
}
};
Ok(Self { db, dimensions })
Ok(Self {
pooled: Some(Pooled { db, slot }),
path: path_buf,
dimensions,
})
}
/// Reuse this path's pooled database, or open a fresh one under its guard.
fn open_pooled(slot: &Arc<PathSlot>, path: &Path) -> Result<Arc<Database>> {
let mut guard = slot.lock();
if let Some(existing_db) = guard.as_ref().and_then(Weak::upgrade) {
// Reuse existing database connection
return Ok(existing_db);
}
// Create new database and publish it to the pool. On any failure below,
// `new_db` is dropped before `guard`, so the file lock is released while
// the slot is still held.
let new_db = Arc::new(Database::create(path)?);
// Initialize tables
let write_txn = new_db.begin_write()?;
{
let _ = write_txn.open_table(VECTORS_TABLE)?;
let _ = write_txn.open_table(METADATA_TABLE)?;
let _ = write_txn.open_table(CONFIG_TABLE)?;
}
write_txn.commit()?;
*guard = Some(Arc::downgrade(&new_db));
Ok(new_db)
}
/// Drop this path's pool entry once nothing can reach it any more.
///
/// Leaving the entry in place would leak one map entry per opened path.
/// `slot` is consumed and released *under* the pool lock, so a remaining
/// strong count of one proves the pool's own entry is the last handle: new
/// handles are only ever cloned under this same lock, so no other thread can
/// be mid-`new` for this path.
fn release_slot_if_unused(path: &Path, slot: Arc<PathSlot>) {
let mut pool = DB_POOL.lock();
let is_ours = matches!(pool.get(path), Some(pooled) if Arc::ptr_eq(pooled, &slot));
drop(slot);
if is_ours && matches!(pool.get(path), Some(pooled) if Arc::strong_count(pooled) == 1) {
pool.remove(path);
}
}
/// Borrow the pooled database.
///
/// Handing out `&Database` rather than the `Arc` keeps the strong count
/// under this handle's control, which `Drop` relies on to decide whether it
/// is closing the last reference.
#[inline]
fn db(&self) -> &Database {
&self
.pooled
.as_ref()
.expect("database handle used after drop")
.db
}
/// Insert a vector entry
@ -141,7 +215,7 @@ impl VectorStorage {
.clone()
.unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
let write_txn = self.db.begin_write()?;
let write_txn = self.db().begin_write()?;
{
let mut table = write_txn.open_table(VECTORS_TABLE)?;
@ -166,7 +240,7 @@ impl VectorStorage {
/// Insert multiple vectors in a batch
pub fn insert_batch(&self, entries: &[VectorEntry]) -> Result<Vec<VectorId>> {
let write_txn = self.db.begin_write()?;
let write_txn = self.db().begin_write()?;
let mut ids = Vec::with_capacity(entries.len());
{
@ -208,7 +282,7 @@ impl VectorStorage {
/// Get a vector by ID
pub fn get(&self, id: &str) -> Result<Option<VectorEntry>> {
let read_txn = self.db.begin_read()?;
let read_txn = self.db().begin_read()?;
let table = read_txn.open_table(VECTORS_TABLE)?;
let Some(vector_data) = table.get(id)? else {
@ -240,7 +314,7 @@ impl VectorStorage {
/// Delete a vector by ID
pub fn delete(&self, id: &str) -> Result<bool> {
let write_txn = self.db.begin_write()?;
let write_txn = self.db().begin_write()?;
let deleted;
{
@ -257,7 +331,7 @@ impl VectorStorage {
/// Get the number of vectors stored
pub fn len(&self) -> Result<usize> {
let read_txn = self.db.begin_read()?;
let read_txn = self.db().begin_read()?;
let table = read_txn.open_table(VECTORS_TABLE)?;
Ok(table.len()? as usize)
}
@ -269,7 +343,7 @@ impl VectorStorage {
/// Get all vector IDs
pub fn all_ids(&self) -> Result<Vec<VectorId>> {
let read_txn = self.db.begin_read()?;
let read_txn = self.db().begin_read()?;
let table = read_txn.open_table(VECTORS_TABLE)?;
let mut ids = Vec::new();
@ -287,7 +361,7 @@ impl VectorStorage {
let config_json = serde_json::to_string(options)
.map_err(|e| RuvectorError::SerializationError(e.to_string()))?;
let write_txn = self.db.begin_write()?;
let write_txn = self.db().begin_write()?;
{
let mut table = write_txn.open_table(CONFIG_TABLE)?;
table.insert(DB_CONFIG_KEY, config_json.as_str())?;
@ -299,7 +373,7 @@ impl VectorStorage {
/// Load database configuration from persistent storage
pub fn load_config(&self) -> Result<Option<DbOptions>> {
let read_txn = self.db.begin_read()?;
let read_txn = self.db().begin_read()?;
// Try to open config table - may not exist in older databases
let table = match read_txn.open_table(CONFIG_TABLE) {
@ -323,7 +397,7 @@ impl VectorStorage {
/// caller can delete without deleting the data it describes is not a
/// safety property. See [`AgenticDB::with_embedding_provider`](crate::AgenticDB::with_embedding_provider).
pub fn save_config_value(&self, key: &str, value: &str) -> Result<()> {
let write_txn = self.db.begin_write()?;
let write_txn = self.db().begin_write()?;
{
let mut table = write_txn.open_table(CONFIG_TABLE)?;
table.insert(key, value)?;
@ -335,7 +409,7 @@ impl VectorStorage {
/// Read a value written by [`save_config_value`](Self::save_config_value).
/// A database created before the key existed returns `None`.
pub fn load_config_value(&self, key: &str) -> Result<Option<String>> {
let read_txn = self.db.begin_read()?;
let read_txn = self.db().begin_read()?;
let table = match read_txn.open_table(CONFIG_TABLE) {
Ok(t) => t,
Err(_) => return Ok(None),
@ -349,6 +423,30 @@ impl VectorStorage {
}
}
impl Drop for VectorStorage {
fn drop(&mut self) {
let Some(Pooled { db, slot }) = self.pooled.take() else {
return;
};
{
let mut guard = slot.lock();
// Evicting under the guard is what keeps a concurrent `new` for this
// path correct: `Weak::upgrade` starts failing the moment the strong
// count hits zero, but redb only commits, fsyncs, and unlocks the
// file later, inside `Database::drop`. Anyone racing us blocks on the
// guard instead of calling `Database::create` against a held lock.
if Arc::strong_count(&db) == 1 {
*guard = None;
}
drop(db);
}
Self::release_slot_if_unused(&self.path, slot);
}
}
// Add uuid dependency
use uuid;
@ -469,4 +567,133 @@ mod tests {
Ok(())
}
#[test]
fn reopening_after_the_last_handle_drops_preserves_data() -> Result<()> {
let dir = tempdir().unwrap();
let db_path = dir.path().join("reopen.db");
{
let storage = VectorStorage::new(&db_path, 3)?;
storage.insert(&VectorEntry {
id: Some("kept".to_string()),
vector: vec![1.0, 2.0, 3.0],
metadata: None,
})?;
}
// The pool entry must be reaped once the last handle is gone, otherwise
// long-running processes accumulate one dead entry per opened path.
assert!(
!DB_POOL.lock().contains_key(&db_path),
"pool kept a dead entry for {}",
db_path.display()
);
let reopened = VectorStorage::new(&db_path, 3)?;
assert_eq!(reopened.len()?, 1);
assert_eq!(
reopened.get("kept")?.expect("entry survived reopen").vector,
vec![1.0, 2.0, 3.0]
);
Ok(())
}
#[test]
fn concurrently_dropping_two_handles_still_reaps_the_pool_entry() {
use std::sync::Barrier;
let dir = tempdir().unwrap();
let db_path = dir.path().join("double-drop.db");
for _ in 0..50 {
let first = VectorStorage::new(&db_path, 3).expect("first open");
let second = VectorStorage::new(&db_path, 3).expect("second open");
let barrier = Arc::new(Barrier::new(2));
let threads: Vec<_> = [first, second]
.into_iter()
.map(|handle| {
let barrier = Arc::clone(&barrier);
std::thread::spawn(move || {
barrier.wait();
drop(handle);
})
})
.collect();
for thread in threads {
thread.join().unwrap();
}
assert!(
!DB_POOL.lock().contains_key(&db_path),
"pool kept a dead entry after both handles dropped"
);
}
}
#[test]
fn concurrent_drop_and_open_never_race_the_file_lock() {
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Barrier;
// The unfixed pool failed this race on ~97.5% of attempts (195/200
// when measured), so a handful of iterations already makes a
// regression a near-certainty to catch: missing it 40 times running
// is about 0.025^40. Each iteration is a real `Database::create` plus
// redb's fsync-bearing drop, so this is kept small deliberately —
// 200 iterations cost ~26s and would dominate the crate's suite.
const ITERATIONS: usize = 40;
let dir = tempdir().unwrap();
let db_path = dir.path().join("race.db");
let failures = Arc::new(AtomicUsize::new(0));
let last_error = Arc::new(Mutex::new(None::<String>));
for _ in 0..ITERATIONS {
// The only live handle for this path; dropping it releases the
// underlying redb file lock.
let holder = VectorStorage::new(&db_path, 3).expect("initial open");
let barrier = Arc::new(Barrier::new(2));
let dropper = {
let barrier = Arc::clone(&barrier);
std::thread::spawn(move || {
barrier.wait();
drop(holder);
})
};
let opener = {
let barrier = Arc::clone(&barrier);
let path = db_path.clone();
let failures = Arc::clone(&failures);
let last_error = Arc::clone(&last_error);
std::thread::spawn(move || {
barrier.wait();
match VectorStorage::new(&path, 3) {
Ok(storage) => drop(storage),
Err(e) => {
failures.fetch_add(1, Ordering::Relaxed);
*last_error.lock() = Some(e.to_string());
}
}
})
};
dropper.join().unwrap();
opener.join().unwrap();
}
let failed = failures.load(Ordering::Relaxed);
assert_eq!(
failed,
0,
"{failed}/{ITERATIONS} concurrent opens raced the drop of the last \
handle; last error: {:?}",
last_error.lock().as_deref()
);
}
}

54
docs/adr/ADR-332-SCOPE-SHARDED-CONTEXT-INDEX.md → docs/adr/ADR-334-scope-sharded-context-index.md
View file

@ -1,4 +1,4 @@
# ADR-332: Scope-Sharded Context Vector Index
# ADR-334: Scope-Sharded Context Vector Index
**Status:** Proposed
@ -52,7 +52,46 @@ does not parse capability handles or infer authority from a URI.
5. Scope strings never become filesystem path components.
6. Recovery authenticates the scope-to-filename binding before serving data.
7. Exact-scope erasure closes and removes the persistent shard while holding
the catalog write lock.
the catalog write lock. The unlink is ordered before the catalog mutation,
so a failed erasure never reports in-memory success over a surviving file.
8. Shard creation never adopts a pre-existing file. A file already present at
the computed filename is refused rather than opened, because
`VectorDB::new` on an existing path silently inherits that file's stored
config and stored vectors.
9. One process holds at most one index handle per root directory, enforced by
an exclusive advisory lock taken in `open`. Two handles over one root would
otherwise share the process-global database pool while maintaining
independent in-memory indexes.
### What the manifest binding does and does not prove
Invariant 6 authenticates the scope-to-**filename** binding. It does not
authenticate shard **contents**: there is no MAC over the stored vectors, and
`shard_id` is an unkeyed digest over non-secret identifiers, so any party that
can write the root directory can compute a victim's filename offline and
author a well-formed shard for it. Invariant 8 is what closes the resulting
laundering path — without it, a planted file adopted through the create path
would have its manifest rewritten to the victim scope and would then pass
invariant 6 forever after.
Treat the root directory as trusted deployment state. Deployments that cannot
guarantee that should key `shard_id` with a per-deployment secret (HMAC rather
than bare SHA-256) so filenames are unguessable, and add content
authentication.
### Explicit non-goals
- **Availability is not isolated.** `max_scopes` is a process-global ceiling
with no per-namespace quota, so one tenant exhausting it makes another
tenant's first insert fail. Per-tenant quotas are deferred.
- **Startup cost is proportional to the whole corpus.** `open` materializes
every shard before the first query rather than opening shards lazily, so
memory and startup scale with total stored vectors across all tenants,
bounded only by `max_scopes`.
- **No authorization is performed here.** Every public method, not only
`search`, assumes the caller has already authorized the scope. `scope_stats`
in particular answers existence and exact vector count for any nameable
scope.
## Pseudocode
@ -119,3 +158,14 @@ The upstream crate must demonstrate:
deterministically.
6. Formatting, unit tests, Clippy, dependency audit, and changed-file secret
scan pass before merge.
7. The isolation suite runs against **both** index kinds. A suite configured
with `hnsw_config: None` routes every shard to `FlatIndex` and therefore
cannot establish claim 1, whose subject is the ANN index.
8. The per-shard search counter is incremented on a path that cannot be
bypassed by construction, rather than by a call site that a future edit
could omit while leaving the suite green.
9. A shard file planted at another scope's computed filename while the index
is open is refused, and no restart launders it into the victim scope.
10. A second index handle on an already-open root fails loudly.
11. A single unreadable shard-shaped file does not deny `open` to unrelated
tenants.

19
docs/adr/INDEX.md
View file

@ -1,6 +1,6 @@
# ADR Index
**Next available ADR number: 334**
**Next available ADR number: 335**
> Generated by `node scripts/adr-index.mjs` — do not edit by hand.
> This file is the canonical allocation counter for new ADR numbers
@ -8,8 +8,8 @@
> historical artifacts and are cited as `ADR-NNN (slug)`.
> CI gate: `node scripts/adr-index.mjs --check`.
- ADR files indexed: **364** (317 on the canonical counter, 47 in namespaced families)
- Highest allocated number: **ADR-333**
- ADR files indexed: **365** (318 on the canonical counter, 47 in namespaced families)
- Highest allocated number: **ADR-334**
- Frozen duplicate numbers: **27** (spanning 61 files)
| Number | Title | File | Last commit | Status | Duplicate |
@ -325,12 +325,13 @@
| ADR-325 | ADR-325: D²ACCI-Pattern Stage-Level Memory Diagnostic Gate | [`ADR-325-d2acci-pattern-stage-level-memory-diagnostic-gate.md`](./ADR-325-d2acci-pattern-stage-level-memory-diagnostic-gate.md) | 2026-08-21 | Proposed | |
| ADR-326 | ADR-326: DeAR-Pattern Decentralized Capability-Grounded Reasoning | [`ADR-326-dear-pattern-decentralized-capability-grounded-reasoning.md`](./ADR-326-dear-pattern-decentralized-capability-grounded-reasoning.md) | 2026-08-21 | Proposed | |
| ADR-327 | ADR-327: Zetta-Pattern Three-Timescale Physical-Evolution Harness (New Bounded Context, Explicit Stretch) | [`ADR-327-zetta-pattern-timescale-separated-evolution-harness.md`](./ADR-327-zetta-pattern-timescale-separated-evolution-harness.md) | 2026-08-21 | Proposed | |
| ADR-328 | ADR-328: AI4AI-Bench Recursive-Improvement Benchmark Adapter | [`ADR-328-ai4ai-bench-recursive-improvement-adapter.md`](./ADR-328-ai4ai-bench-recursive-improvement-adapter.md) | | Proposed | |
| ADR-329 | ADR-329: Content-Addressed Schema-Resource Cache (ReCache-Pattern) | [`ADR-329-recache-pattern-schema-resource-cache.md`](./ADR-329-recache-pattern-schema-resource-cache.md) | | Proposed | |
| ADR-330 | ADR-330: CAMA-Pattern Correlation-Aware Memory Arbitration | [`ADR-330-cama-pattern-correlation-aware-memory-arbitration.md`](./ADR-330-cama-pattern-correlation-aware-memory-arbitration.md) | | Proposed | |
| ADR-331 | ADR-331: Value-of-Information Cost-Aware Routing (Pandora-Pattern) | [`ADR-331-pandora-pattern-voi-cost-aware-routing.md`](./ADR-331-pandora-pattern-voi-cost-aware-routing.md) | | Proposed | |
| ADR-332 | ADR-332: RF Sensing Modality Router (Stretch) | [`ADR-332-rf-sensing-modality-router.md`](./ADR-332-rf-sensing-modality-router.md) | | Proposed (stretch — ADR-only this wave, implementation deferred pending RuView c | |
| ADR-333 | ADR-333: RVM Semantic-Authority Layer Above OpenShell-Class Secure Runtimes | [`ADR-333-rvm-semantic-authority-above-openshell.md`](./ADR-333-rvm-semantic-authority-above-openshell.md) | | Proposed (cross-repo posture — ADR-only in this repo; RVM-side work lands in `ru | |
| ADR-328 | ADR-328: AI4AI-Bench Recursive-Improvement Benchmark Adapter | [`ADR-328-ai4ai-bench-recursive-improvement-adapter.md`](./ADR-328-ai4ai-bench-recursive-improvement-adapter.md) | 2026-08-22 | Proposed | |
| ADR-329 | ADR-329: Content-Addressed Schema-Resource Cache (ReCache-Pattern) | [`ADR-329-recache-pattern-schema-resource-cache.md`](./ADR-329-recache-pattern-schema-resource-cache.md) | 2026-08-22 | Proposed | |
| ADR-330 | ADR-330: CAMA-Pattern Correlation-Aware Memory Arbitration | [`ADR-330-cama-pattern-correlation-aware-memory-arbitration.md`](./ADR-330-cama-pattern-correlation-aware-memory-arbitration.md) | 2026-08-22 | Proposed | |
| ADR-331 | ADR-331: Value-of-Information Cost-Aware Routing (Pandora-Pattern) | [`ADR-331-pandora-pattern-voi-cost-aware-routing.md`](./ADR-331-pandora-pattern-voi-cost-aware-routing.md) | 2026-08-22 | Proposed | |
| ADR-332 | ADR-332: RF Sensing Modality Router (Stretch) | [`ADR-332-rf-sensing-modality-router.md`](./ADR-332-rf-sensing-modality-router.md) | 2026-08-22 | Proposed (stretch — ADR-only this wave, implementation deferred pending RuView c | |
| ADR-333 | ADR-333: RVM Semantic-Authority Layer Above OpenShell-Class Secure Runtimes | [`ADR-333-rvm-semantic-authority-above-openshell.md`](./ADR-333-rvm-semantic-authority-above-openshell.md) | 2026-08-22 | Proposed (cross-repo posture — ADR-only in this repo; RVM-side work lands in `ru | |
| ADR-334 | ADR-334: Scope-Sharded Context Vector Index | [`ADR-334-scope-sharded-context-index.md`](./ADR-334-scope-sharded-context-index.md) | | Proposed | |
| ADR-CE-001 | ADR-CE-001: Sheaf Laplacian Defines Coherence Witness | [`coherence-engine/ADR-CE-001-sheaf-laplacian-coherence.md`](./coherence-engine/ADR-CE-001-sheaf-laplacian-coherence.md) | 2026-08-13 | Accepted | |
| ADR-CE-002 | ADR-CE-002: Incremental Coherence Computation | [`coherence-engine/ADR-CE-002-incremental-computation.md`](./coherence-engine/ADR-CE-002-incremental-computation.md) | 2026-08-13 | Accepted | |
| ADR-CE-003 | ADR-CE-003: PostgreSQL + Ruvector Unified Substrate | [`coherence-engine/ADR-CE-003-hybrid-storage.md`](./coherence-engine/ADR-CE-003-hybrid-storage.md) | 2026-08-13 | Accepted | |