use anyhow::{Context, Result};
use rusqlite::{Connection, OptionalExtension, params};
use std::path::PathBuf;
pub const DEFAULT_CAP_MB: u64 = 1024;
pub const CAP_ENV: &str = "SEMLITH_VECTOR_CACHE_MB";
const ROW_BYTES: u64 = 1_700;
pub fn cap_bytes() -> u64 {
let mb = std::env::var(CAP_ENV)
.ok()
.and_then(|v| v.parse::<u64>().ok())
.or_else(|| crate::home::Settings::load().vector_cache_mb)
.unwrap_or(DEFAULT_CAP_MB);
mb.saturating_mul(1024 * 1024)
}
pub fn path() -> Result<PathBuf> {
Ok(crate::home::home_or_error()?
.join("cache")
.join("vectors.db"))
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Location {
pub path: PathBuf,
pub cap: u64,
}
impl Location {
pub fn at(path: impl Into<PathBuf>, cap_mb: u64) -> Self {
Self {
path: path.into(),
cap: cap_mb.saturating_mul(1024 * 1024),
}
}
pub fn machine() -> Option<Self> {
let cap = cap_bytes();
(cap > 0)
.then(path)
.and_then(Result::ok)
.map(|path| Self { path, cap })
}
}
pub fn clear(at: &Location) -> Result<()> {
for suffix in ["", "-wal", "-shm"] {
let mut name = at.path.clone().into_os_string();
name.push(suffix);
match std::fs::remove_file(&name) {
Err(e) if e.kind() != std::io::ErrorKind::NotFound => {
return Err(e)
.with_context(|| format!("removing {}", PathBuf::from(name).display()));
}
_ => {}
}
}
Ok(())
}
#[derive(Debug, Clone)]
pub struct Scope {
pub model: String,
pub chunker: u32,
pub truncation: usize,
}
impl Scope {
pub fn of(model: &crate::embed::Model) -> Self {
let model = match model {
crate::embed::Model::Granite => {
format!(
"{}@{}",
crate::embed::GRANITE_NAME,
crate::embed::GRANITE_REVISION
)
}
other => other.to_string(),
};
Self {
model,
chunker: crate::store::CODE_CONTEXT,
truncation: crate::chunk::MAX_CHARS / 2,
}
}
pub fn text(text: &str) -> [u8; 32] {
*blake3::hash(text.as_bytes()).as_bytes()
}
pub fn key(&self, text: &[u8; 32], variant: &str) -> [u8; 32] {
let mut hasher = blake3::Hasher::new();
for part in [
self.model.as_bytes(),
variant.as_bytes(),
&self.chunker.to_le_bytes(),
&(self.truncation as u64).to_le_bytes(),
] {
hasher.update(&(part.len() as u64).to_le_bytes());
hasher.update(part);
}
hasher.update(text);
*hasher.finalize().as_bytes()
}
}
pub struct Cache {
db: Connection,
cap: u64,
}
impl Cache {
pub fn open() -> Option<Self> {
Self::open_in(&Location::machine()?)
}
pub fn open_in(at: &Location) -> Option<Self> {
if at.cap == 0 {
return None;
}
let mut cache = Self::open_at(&at.path).ok()?;
cache.cap = at.cap;
Some(cache)
}
pub fn open_at(path: &std::path::Path) -> Result<Self> {
if let Some(dir) = path.parent() {
std::fs::create_dir_all(dir).with_context(|| format!("creating {}", dir.display()))?;
crate::home::secure_dir(dir)?;
}
let db = Connection::open(path)?;
db.busy_timeout(std::time::Duration::from_secs(5))?;
db.pragma_update(None, "page_size", 16_384)?;
db.pragma_update(None, "journal_mode", "WAL")?;
db.pragma_update(None, "synchronous", "NORMAL")?;
db.execute_batch(
"CREATE TABLE IF NOT EXISTS vectors (
key BLOB PRIMARY KEY,
variant TEXT NOT NULL,
vector BLOB NOT NULL,
used INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS vectors_used ON vectors(used);
CREATE TABLE IF NOT EXISTS counts (k TEXT PRIMARY KEY, v INTEGER NOT NULL);",
)?;
let kept: Option<i64> = db
.query_row("SELECT v FROM counts WHERE k = 'rows'", [], |r| r.get(0))
.optional()?;
if kept.is_none() {
db.execute(
"INSERT OR IGNORE INTO counts (k, v) SELECT 'rows', COUNT(*) FROM vectors",
[],
)?;
}
crate::home::tighten_file(path);
Ok(Self {
db,
cap: cap_bytes(),
})
}
fn rows(&self) -> Result<u64> {
count(&self.db, "rows")
}
pub fn forget_vectors(&mut self, vectors: &[Vec<f32>]) -> Result<usize> {
if vectors.is_empty() {
return Ok(0);
}
self.db.pragma_update(None, "secure_delete", "ON")?;
let tx = self.db.transaction()?;
let mut gone = 0usize;
{
let mut stmt = tx.prepare("DELETE FROM vectors WHERE vector = ?1")?;
for vector in vectors {
let bytes: Vec<u8> = vector.iter().flat_map(|v| v.to_le_bytes()).collect();
gone += stmt.execute(params![bytes])?;
}
}
tx.execute(
"UPDATE counts SET v = MAX(0, v - ?1) WHERE k = 'rows'",
params![gone as i64],
)?;
tx.commit()?;
Ok(gone)
}
pub fn get(&self, key: &[u8; 32]) -> Option<Vec<f32>> {
let bytes: Vec<u8> = self
.db
.query_row(
"SELECT vector FROM vectors WHERE key = ?1",
params![&key[..]],
|r| r.get(0),
)
.optional()
.ok()??;
bytes.len().is_multiple_of(4).then(|| {
bytes
.as_chunks::<4>()
.0
.iter()
.map(|b| f32::from_le_bytes(*b))
.collect()
})
}
pub fn record(
&mut self,
fresh: &[([u8; 32], &'static str, &[f32])],
hits: &[[u8; 32]],
lookups: u64,
) -> Result<()> {
let now = crate::now();
let tx = self.db.transaction()?;
{
let mut put = tx.prepare_cached(
"INSERT OR IGNORE INTO vectors (key, variant, vector, used) VALUES (?1, ?2, ?3, ?4)",
)?;
let mut touch = tx.prepare_cached("UPDATE vectors SET used = ?2 WHERE key = ?1")?;
let mut added = 0i64;
for (key, variant, vector) in fresh {
let bytes: Vec<u8> = vector.iter().flat_map(|v| v.to_le_bytes()).collect();
if put.execute(params![&key[..], variant, bytes, now])? == 0 {
touch.execute(params![&key[..], now])?;
} else {
added += 1;
}
}
for key in hits {
touch.execute(params![&key[..], now])?;
}
let mut count = tx.prepare_cached(
"INSERT INTO counts (k, v) VALUES (?1, ?2)
ON CONFLICT(k) DO UPDATE SET v = v + excluded.v",
)?;
count.execute(params!["hits", hits.len() as i64])?;
count.execute(params!["lookups", lookups as i64])?;
count.execute(params!["rows", added])?;
}
tx.commit()?;
self.evict(self.cap)
}
pub fn evict(&mut self, cap: u64) -> Result<()> {
let rows = self.rows()?;
if cap == 0 || rows * ROW_BYTES <= cap {
return Ok(());
}
let keep = cap * 9 / 10 / ROW_BYTES;
let drop = rows.saturating_sub(keep);
let tx = self.db.transaction()?;
let gone = tx.execute(
"DELETE FROM vectors WHERE key IN
(SELECT key FROM vectors ORDER BY used ASC, key ASC LIMIT ?1)",
params![drop as i64],
)?;
tx.execute(
"UPDATE counts SET v = v - ?1 WHERE k = 'rows'",
params![gone as i64],
)?;
tx.commit()?;
Ok(())
}
pub fn stats(&self) -> Result<Stats> {
stats_of(&self.db, self.cap)
}
}
fn count(db: &Connection, k: &str) -> Result<u64> {
Ok(db
.query_row("SELECT v FROM counts WHERE k = ?1", params![k], |r| {
r.get::<_, i64>(0)
})
.optional()?
.unwrap_or(0)
.max(0) as u64)
}
fn stats_of(db: &Connection, cap: u64) -> Result<Stats> {
let rows = count(db, "rows")?;
Ok(Stats {
rows,
bytes: rows * ROW_BYTES,
cap,
hits: count(db, "hits")?,
lookups: count(db, "lookups")?,
})
}
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct Stats {
pub rows: u64,
pub bytes: u64,
pub cap: u64,
pub hits: u64,
pub lookups: u64,
}
impl Stats {
pub fn line(&self) -> String {
if self.cap == 0 {
return "off (vector cache cap is 0)".to_string();
}
let rate = if self.lookups > 0 {
format!(
", {:.0} % of lookups hit",
self.hits as f64 * 100.0 / self.lookups as f64
)
} else {
String::new()
};
format!(
"{} vectors, {} of {}{rate}",
self.rows,
crate::human_bytes(self.bytes as i64),
crate::human_bytes(self.cap as i64)
)
}
}
pub fn stats() -> Stats {
static LAST: std::sync::Mutex<Option<Stats>> = std::sync::Mutex::new(None);
let cap = cap_bytes();
let off = Stats {
cap,
..Stats::default()
};
let Some(at) = Location::machine() else {
return off;
};
let read = read_stats(&at);
let mut last = LAST.lock().unwrap_or_else(|e| e.into_inner());
match read {
Ok(stats) => {
*last = Some(stats.clone());
stats
}
Err(_) => last.clone().unwrap_or(off),
}
}
pub fn stats_at(at: &Location) -> Stats {
read_stats(at).unwrap_or(Stats {
cap: at.cap,
..Stats::default()
})
}
fn read_stats(at: &Location) -> Result<Stats> {
let off = Stats {
cap: at.cap,
..Stats::default()
};
if at.cap == 0 || !at.path.exists() {
return Ok(off);
}
let db = Connection::open_with_flags(&at.path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)?;
db.busy_timeout(std::time::Duration::from_millis(200))?;
stats_of(&db, at.cap)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_key_moves_with_every_part_that_decides_the_vector() {
let scope = Scope {
model: "m@1".into(),
chunker: 4,
truncation: 400,
};
let text = Scope::text("fn main() {}");
let base = scope.key(&text, "fp16-ane");
assert_ne!(base, scope.key(&Scope::text("fn main() {} "), "fp16-ane"));
assert_ne!(base, scope.key(&text, "int8-cpu"));
let mut other = scope.clone();
other.chunker = 5;
assert_ne!(base, other.key(&text, "fp16-ane"));
other = scope.clone();
other.truncation = 512;
assert_ne!(base, other.key(&text, "fp16-ane"));
other = scope.clone();
other.model = "m@2".into();
assert_ne!(base, other.key(&text, "fp16-ane"));
assert_eq!(base, scope.key(&text, "fp16-ane"));
}
#[test]
fn vectors_round_trip_byte_for_byte_and_eviction_is_least_recently_used() {
let dir = tempfile::tempdir().unwrap();
let mut cache = Cache::open_at(&dir.path().join("v.db")).unwrap();
let scope = Scope {
model: "m".into(),
chunker: 1,
truncation: 1,
};
let v: Vec<f32> = (0..384).map(|i| (i as f32).sin()).collect();
let keys: Vec<[u8; 32]> = (0..10)
.map(|i| scope.key(&Scope::text(&i.to_string()), "x"))
.collect();
let fresh: Vec<([u8; 32], &'static str, &[f32])> =
keys.iter().map(|k| (*k, "x", v.as_slice())).collect();
cache.record(&fresh, &[], 10).unwrap();
cache.record(&fresh, &[], 0).unwrap();
assert_eq!(
cache.stats().unwrap().rows,
10,
"a key recorded twice is one row"
);
let got = cache.get(&keys[3]).unwrap();
assert_eq!(
got.iter().map(|f| f.to_bits()).collect::<Vec<_>>(),
v.iter().map(|f| f.to_bits()).collect::<Vec<_>>()
);
std::thread::sleep(std::time::Duration::from_millis(1100));
cache.record(&[], &[keys[0]], 1).unwrap();
cache.evict(5 * ROW_BYTES).unwrap();
let stats = cache.stats().unwrap();
assert_eq!(stats.rows, 4);
assert!(cache.get(&keys[0]).is_some());
assert_eq!(stats.hits, 1);
assert_eq!(stats.lookups, 11);
}
fn fill(cache: &mut Cache, n: usize, hits: &[[u8; 32]]) -> Vec<[u8; 32]> {
let scope = Scope {
model: "m".into(),
chunker: 1,
truncation: 1,
};
let v: Vec<f32> = (0..384).map(|i| i as f32).collect();
let keys: Vec<[u8; 32]> = (0..n)
.map(|i| scope.key(&Scope::text(&i.to_string()), "x"))
.collect();
let fresh: Vec<([u8; 32], &'static str, &[f32])> =
keys.iter().map(|k| (*k, "x", v.as_slice())).collect();
cache.record(&fresh, hits, (n + hits.len()) as u64).unwrap();
keys
}
#[test]
fn two_instance_caches_keep_apart_and_each_reports_its_own_figures() {
let dir = tempfile::tempdir().unwrap();
let a = Location::at(dir.path().join("org-a/vectors.db"), 64);
let b = Location::at(dir.path().join("org-b/vectors.db"), 64);
let keys = fill(&mut Cache::open_in(&a).unwrap(), 10, &[]);
let cb = Cache::open_in(&b).unwrap();
assert!(
keys.iter().all(|k| cb.get(k).is_none()),
"b saw a's vectors"
);
let sa = stats_at(&a);
assert_eq!((sa.rows, sa.cap), (10, 64 * 1024 * 1024));
assert_eq!(stats_at(&b).rows, 0);
let mut cb = cb;
fill(&mut cb, 3, &[]);
assert_eq!(stats_at(&b).rows, 3);
assert_eq!(stats_at(&a).rows, 10, "b's rows landed in a");
}
#[test]
fn a_cap_of_zero_turns_one_instance_cache_off_and_reads_as_off() {
let dir = tempfile::tempdir().unwrap();
let off = Location::at(dir.path().join("v.db"), 0);
assert!(Cache::open_in(&off).is_none());
assert!(!off.path.exists(), "an off cache wrote a file");
assert_eq!(stats_at(&off).line(), "off (vector cache cap is 0)");
}
#[test]
fn an_instance_cache_evicts_to_its_own_cap_and_clears() {
let dir = tempfile::tempdir().unwrap();
let at = Location {
path: dir.path().join("v.db"),
cap: 5 * ROW_BYTES,
};
let mut cache = Cache::open_in(&at).unwrap();
fill(&mut cache, 10, &[]);
assert_eq!(stats_at(&at).rows, 4);
drop(cache);
clear(&at).unwrap();
assert!(!at.path.exists());
assert_eq!(stats_at(&at).rows, 0);
clear(&at).unwrap();
}
#[test]
fn a_row_takes_about_what_the_cap_counts_it_at() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("v.db");
let mut cache = Cache::open_at(&path).unwrap();
let scope = Scope {
model: "m".into(),
chunker: 1,
truncation: 1,
};
let v: Vec<f32> = (0..384).map(|i| (i as f32).cos()).collect();
let keys: Vec<[u8; 32]> = (0..2_000)
.map(|i| scope.key(&Scope::text(&i.to_string()), "fp16-ane"))
.collect();
let fresh: Vec<([u8; 32], &'static str, &[f32])> = keys
.iter()
.map(|k| (*k, "fp16-ane", v.as_slice()))
.collect();
cache.record(&fresh, &[], 0).unwrap();
cache
.db
.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")
.unwrap();
let per_row = std::fs::metadata(&path).unwrap().len() / keys.len() as u64;
assert!(
per_row <= ROW_BYTES * 6 / 5,
"{per_row} bytes a row on disk against {ROW_BYTES} counted"
);
}
}