use crate::io_timeout::STAT_TIMEOUT;
use crate::library::{bounded_op, root_cause_is_not_found, LibraryContext};
use crate::library_locks::ActivityMode;
use anyhow::{bail, Context, Result};
use rusqlite::Connection;
use std::io::Read;
use std::path::{Component, Path};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
const SCHEMA_VERSION: i64 = 1;
const SQLITE_MAGIC: [u8; 16] = *b"SQLite format 3\0";
const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
const FILE_HASHES_COLUMNS: &[(&str, &str)] = &[
("path", "TEXT PRIMARY KEY"),
("hash", "TEXT NOT NULL"),
("size_bytes", "INTEGER"),
("created_at", "TEXT"),
("modified_at", "TEXT"),
("ext", "TEXT"),
("mime", "TEXT"),
("phash", "INTEGER"),
("exif_date", "TEXT"),
("gps_lat", "REAL"),
("gps_lon", "REAL"),
("width", "INTEGER"),
("height", "INTEGER"),
("duration_secs", "REAL"),
("codec", "TEXT"),
("location_name", "TEXT"),
("location_cluster_id", "INTEGER"),
];
const FACES_COLUMNS: &[(&str, &str)] = &[
("id", "INTEGER PRIMARY KEY"),
("hash", "TEXT NOT NULL"),
("bbox", "TEXT NOT NULL"),
("landmark", "TEXT"),
("embedding", "BLOB NOT NULL"),
("cluster_id", "INTEGER"),
("person_label", "TEXT"),
("confirmed", "INTEGER DEFAULT 0"),
("is_primary", "INTEGER DEFAULT 0"),
("det_score", "REAL"),
("blur", "REAL"),
];
const REQUIRED_TABLES: &[&str] = &[
"file_hashes",
"people",
"faces",
"faces_scanned",
"marks",
"photo_tags",
"classifications",
"location_clusters",
"pipeline_runs",
];
static TMP_SEQ: AtomicU64 = AtomicU64::new(0);
fn file_hashes_ddl() -> String {
let columns: Vec<String> = FILE_HASHES_COLUMNS
.iter()
.map(|(name, decl)| format!(" {name:<20} {decl}"))
.collect();
format!(
"CREATE TABLE IF NOT EXISTS file_hashes (\n{}\n);",
columns.join(",\n")
)
}
fn column_exists(conn: &Connection, table: &str, column: &str) -> rusqlite::Result<bool> {
let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?;
let names = stmt.query_map([], |row| row.get::<_, String>(1))?;
for name in names {
if name? == column {
return Ok(true);
}
}
Ok(false)
}
fn add_missing_columns(
conn: &Connection,
table: &str,
expected: &[(&str, &str)],
) -> rusqlite::Result<()> {
for (name, decl) in expected {
if column_exists(conn, table, name)? {
continue;
}
let alter = format!("ALTER TABLE {table} ADD COLUMN {name} {decl}");
if let Err(e) = conn.execute_batch(&alter) {
if !column_exists(conn, table, name)? {
return Err(e);
}
}
}
Ok(())
}
pub fn ensure_scan_schema(conn: &Connection) -> rusqlite::Result<()> {
conn.execute_batch(&file_hashes_ddl())?;
add_missing_columns(conn, "file_hashes", FILE_HASHES_COLUMNS)
}
fn verify_schema(conn: &Connection) -> Result<()> {
for table in REQUIRED_TABLES {
if !crate::db::table_exists(conn, table)? {
bail!("required table {table} is missing after schema preparation");
}
}
let mut required: Vec<(&str, &str)> = Vec::new();
for (name, _) in FILE_HASHES_COLUMNS {
required.push(("file_hashes", name));
}
for (name, _) in FACES_COLUMNS {
required.push(("faces", name));
}
required.push(("classifications", "model_id"));
required.push(("classifications", "hash"));
for (table, column) in &required {
if !column_exists(conn, table, column)? {
bail!("required column {table}.{column} is missing after schema preparation");
}
}
Ok(())
}
fn schema_complete(conn: &Connection) -> Result<bool> {
Ok(verify_schema(conn).is_ok())
}
fn prepare_schema(conn: &Connection) -> Result<()> {
ensure_scan_schema(conn)?;
crate::face_db::create_faces_table(conn)?;
crate::marks::ensure_marks_table(conn)?;
crate::tags::ensure_photo_tags_table(conn)?;
crate::classify::ensure_classifications_table(conn)?;
crate::location_cluster::ensure_location_clusters_table(conn)?;
crate::pipeline_runs::ensure_pipeline_runs_table(conn)?;
add_missing_columns(conn, "faces", FACES_COLUMNS)?;
verify_schema(conn)?;
Ok(())
}
fn user_version(conn: &Connection) -> Result<i64> {
let version: i64 = conn
.query_row("PRAGMA user_version", [], |r| r.get(0))
.context("read the library schema version")?;
Ok(version)
}
fn is_sqlite_file(path: &Path) -> Result<bool> {
let owned = path.to_path_buf();
match bounded_op(path, "read", STAT_TIMEOUT, move || {
let mut file = std::fs::File::open(&owned)?;
let mut header = [0u8; 16];
match file.read_exact(&mut header) {
Ok(()) => Ok(Some(header)),
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => Ok(None),
Err(e) => Err(e),
}
}) {
Ok(Some(header)) => Ok(header == SQLITE_MAGIC),
Ok(None) => Ok(false),
Err(e) => Err(e),
}
}
enum DbFile {
Absent,
ZeroByte,
Present,
}
fn inspect_db_file(ctx: &LibraryContext) -> Result<DbFile> {
crate::library_locks::reject_redirect(&ctx.paths.db, "the library database")?;
let owned = ctx.paths.db.clone();
let len = match bounded_op(&ctx.paths.db, "read", STAT_TIMEOUT, move || {
std::fs::symlink_metadata(&owned).map(|m| m.len())
}) {
Ok(len) => len,
Err(e) if root_cause_is_not_found(&e) => return Ok(DbFile::Absent),
Err(e) => return Err(e),
};
if len == 0 {
return Ok(DbFile::ZeroByte);
}
if !is_sqlite_file(&ctx.paths.db)? {
bail!(
"{} is not a SQLite database; not a videre library",
ctx.paths.db.display()
);
}
Ok(DbFile::Present)
}
pub(crate) fn open_without_create(path: &Path) -> rusqlite::Result<Connection> {
use rusqlite::OpenFlags;
Connection::open_with_flags(
path,
OpenFlags::SQLITE_OPEN_READ_WRITE
| OpenFlags::SQLITE_OPEN_NO_MUTEX
| OpenFlags::SQLITE_OPEN_NOFOLLOW,
)
}
fn open_existing_conn(ctx: &LibraryContext) -> Result<Connection> {
let conn = open_without_create(&ctx.paths.db)
.with_context(|| format!("open {}", ctx.paths.db.display()))?;
conn.busy_timeout(BUSY_TIMEOUT)
.context("set the library database busy timeout")?;
Ok(conn)
}
fn set_wal(conn: &Connection) -> Result<()> {
conn.pragma_update(None, "journal_mode", "WAL")
.context("set the library database to WAL journal mode")
}
fn require_supported_library(conn: &Connection, db: &Path) -> Result<()> {
if crate::db::table_exists(conn, "file_hashes")? {
return Ok(());
}
let user_tables: i64 = conn.query_row(
"SELECT count(*) FROM sqlite_master
WHERE type = 'table' AND name NOT LIKE 'sqlite_%'",
[],
|r| r.get(0),
)?;
anyhow::ensure!(
user_tables == 0,
"{} is not a videre library: a SQLite database with tables of its own and no file_hashes",
db.display()
);
Ok(())
}
#[cfg(test)]
thread_local! {
static ROWS_VALIDATED: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
}
#[cfg(test)]
pub(crate) fn count_rows_validated() -> u64 {
ROWS_VALIDATED.with(std::cell::Cell::get)
}
#[cfg(test)]
fn note_row_validated() {
ROWS_VALIDATED.with(|c| c.set(c.get() + 1));
}
fn validate_row_containment(ctx: &LibraryContext, conn: &Connection) -> Result<()> {
if ctx.index_validated() {
return Ok(());
}
if crate::db::table_exists(conn, "file_hashes")? {
let root = ctx.paths.root.clone();
let mut stmt = conn
.prepare("SELECT path FROM file_hashes")
.context("read the indexed paths")?;
let mut rows = stmt.query([])?;
while let Some(row) = rows.next()? {
let path: String = row.get(0)?;
#[cfg(test)]
note_row_validated();
let stored = Path::new(&path);
let dot_component = stored
.components()
.any(|component| matches!(component, Component::CurDir | Component::ParentDir));
if dot_component || !stored.starts_with(&root) {
bail!(
"the database at {} indexes {}, which is outside the library root {}; it belongs to a different library, so nothing was read or changed",
ctx.paths.db.display(),
path,
root.display()
);
}
}
}
ctx.mark_index_validated();
Ok(())
}
fn open_prepared(ctx: &LibraryContext, conn: &Connection) -> Result<()> {
let version = user_version(conn)?;
if version > SCHEMA_VERSION {
bail!(
"the library at {} was written by a newer videre (schema version {version}, this build understands up to {SCHEMA_VERSION}); upgrade videre to open it",
ctx.paths.root.display()
);
}
require_supported_library(conn, &ctx.paths.db)?;
validate_row_containment(ctx, conn)?;
set_wal(conn)?;
if version < SCHEMA_VERSION || !schema_complete(conn)? {
prepare_schema(conn)?;
conn.pragma_update(None, "user_version", SCHEMA_VERSION)
.context("record the library schema version")?;
}
verify_schema(conn)?;
Ok(())
}
fn remove_file_bounded(path: &Path, what: &str) -> Result<()> {
let owned = path.to_path_buf();
bounded_op(path, "remove", STAT_TIMEOUT, move || {
std::fs::remove_file(&owned)
})
.with_context(|| format!("remove {what} {}", path.display()))
}
pub(crate) fn sync_dir(path: &Path) -> Result<()> {
let owned = path.to_path_buf();
bounded_op(path, "sync", STAT_TIMEOUT, move || {
let dir = std::fs::File::open(&owned)?;
dir.sync_all()
})
.with_context(|| format!("sync {}", path.display()))
}
fn sweep_stale_builds(ctx: &LibraryContext) {
let state = ctx.paths.state.clone();
let names = bounded_op(&ctx.paths.state, "read", STAT_TIMEOUT, move || {
let mut out = Vec::new();
for entry in std::fs::read_dir(&state)? {
out.push(entry?.file_name());
}
Ok(out)
});
if let Ok(names) = names {
for name in names {
let text = name.to_string_lossy();
let stale = text.starts_with("hashes.db.")
&& (text.ends_with(".tmp")
|| text.ends_with(".tmp-wal")
|| text.ends_with(".tmp-shm"));
if stale {
let _ =
remove_file_bounded(&ctx.paths.state.join(&name), "a stale build temporary");
}
}
}
}
fn publish_database(from: &Path, to: &Path) -> Result<()> {
let rename_from = from.to_path_buf();
let rename_to = to.to_path_buf();
bounded_op(to, "publish", STAT_TIMEOUT, move || {
rustix::fs::renameat_with(
rustix::fs::CWD,
&rename_from,
rustix::fs::CWD,
&rename_to,
rustix::fs::RenameFlags::NOREPLACE,
)
.map_err(Into::into)
})
.with_context(|| format!("publish {}", to.display()))
}
fn publish_fresh(ctx: &LibraryContext) -> Result<Connection> {
sweep_stale_builds(ctx);
let tmp = ctx.paths.state.join(format!(
"hashes.db.{}.{}.tmp",
std::process::id(),
TMP_SEQ.fetch_add(1, Ordering::Relaxed)
));
let build = || -> Result<()> {
let conn = Connection::open(&tmp).with_context(|| format!("build {}", tmp.display()))?;
conn.busy_timeout(BUSY_TIMEOUT)?;
prepare_schema(&conn)?;
conn.pragma_update(None, "user_version", SCHEMA_VERSION)
.context("record the library schema version")?;
set_wal(&conn)?;
let _ = conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);");
Ok(())
};
build()?;
let len = {
let owned = tmp.clone();
bounded_op(&tmp, "read", STAT_TIMEOUT, move || {
std::fs::metadata(&owned).map(|m| m.len())
})
.with_context(|| format!("read {}", tmp.display()))?
};
anyhow::ensure!(
len > 0,
"the built database is empty; refusing to publish it"
);
anyhow::ensure!(
is_sqlite_file(&tmp)?,
"the built database does not carry the SQLite header; refusing to publish it"
);
crate::library_locks::reject_redirect(&ctx.paths.db, "the library database")?;
let owned = ctx.paths.db.clone();
let exists = match bounded_op(&ctx.paths.db, "read", STAT_TIMEOUT, move || {
std::fs::symlink_metadata(&owned).map(|_| ())
}) {
Ok(()) => true,
Err(e) if root_cause_is_not_found(&e) => false,
Err(e) => return Err(e),
};
anyhow::ensure!(
!exists,
"a database appeared at {} while the library was being initialized",
ctx.paths.db.display()
);
publish_database(&tmp, &ctx.paths.db)?;
sync_dir(&ctx.paths.state)?;
let conn = open_existing_conn(ctx)?;
open_prepared(ctx, &conn)?;
Ok(conn)
}
pub fn initialize(ctx: &LibraryContext) -> Result<Connection> {
crate::library_locks::ensure_state_and_locks(ctx)?;
let _activity = crate::library_locks::try_activity(ctx, ActivityMode::Exclusive)
.with_context(|| format!("initialize {}", ctx.paths.root.display()))?;
let _init = crate::library_locks::try_init(ctx)?;
crate::library_locks::verify_state(ctx)?;
crate::library_locks::reject_redirect(&ctx.paths.config, "the library config")?;
crate::library_locks::reject_dir_redirect(&ctx.paths.embeddings, "the embeddings directory")?;
let conn = match inspect_db_file(ctx)? {
DbFile::Absent => publish_fresh(ctx)?,
DbFile::ZeroByte => {
remove_file_bounded(&ctx.paths.db, "the empty database")?;
publish_fresh(ctx)?
}
DbFile::Present => {
let conn = open_existing_conn(ctx)?;
open_prepared(ctx, &conn)?;
conn
}
};
crate::library_config::write_initial_if_absent(ctx)?;
Ok(conn)
}
pub fn open_existing(ctx: &LibraryContext) -> Result<Connection> {
crate::library_locks::verify_state(ctx)?;
let _activity = crate::library_locks::try_activity(ctx, ActivityMode::Shared)?;
crate::library_locks::reject_redirect(&ctx.paths.config, "the library config")?;
crate::library_locks::reject_dir_redirect(&ctx.paths.embeddings, "the embeddings directory")?;
match inspect_db_file(ctx)? {
DbFile::Present => {}
DbFile::Absent => bail!(
"library {} has not been initialized: no database at {}",
ctx.paths.root.display(),
ctx.paths.db.display()
),
DbFile::ZeroByte => bail!(
"library {} was never fully initialized: {} is empty; initialize it (for example with videre scan)",
ctx.paths.root.display(),
ctx.paths.db.display()
),
}
let conn = open_existing_conn(ctx)?;
let version = user_version(&conn)?;
if version > SCHEMA_VERSION {
bail!(
"the library at {} was written by a newer videre (schema version {version}, this build understands up to {SCHEMA_VERSION}); upgrade videre to open it",
ctx.paths.root.display()
);
}
require_supported_library(&conn, &ctx.paths.db)?;
validate_row_containment(ctx, &conn)?;
if version == SCHEMA_VERSION && schema_complete(&conn)? {
set_wal(&conn)?;
return Ok(conn);
}
drop(conn);
drop(_activity);
let _exclusive = crate::library_locks::try_activity(ctx, ActivityMode::Exclusive)?;
let _init = crate::library_locks::try_init(ctx)?;
crate::library_locks::verify_state(ctx)?;
match inspect_db_file(ctx)? {
DbFile::Present => {}
DbFile::ZeroByte => bail!(
"the database at {} became empty while the library was being opened",
ctx.paths.db.display()
),
DbFile::Absent => bail!(
"the database at {} disappeared while the library was being opened",
ctx.paths.db.display()
),
}
let conn = open_existing_conn(ctx)?;
open_prepared(ctx, &conn)?;
Ok(conn)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::library::LibraryContext;
use crate::library_locks;
use rusqlite::params;
fn library() -> (tempfile::TempDir, LibraryContext) {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("photos");
std::fs::create_dir(&root).unwrap();
let ctx = LibraryContext::new(&root, &temp.path().join("cache")).unwrap();
(temp, ctx)
}
#[test]
fn readers_never_initialize_and_repeated_initialization_preserves_config() {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("photos");
std::fs::create_dir(&root).unwrap();
let ctx = crate::library::LibraryContext::new(&root, &temp.path().join("cache")).unwrap();
assert!(open_existing(&ctx).is_err());
assert!(!ctx.paths.state.exists());
drop(initialize(&ctx).unwrap());
let before = std::fs::read(&ctx.paths.config).unwrap();
drop(initialize(&ctx).unwrap());
assert_eq!(std::fs::read(&ctx.paths.config).unwrap(), before);
let conn = open_existing(&ctx).unwrap();
let count: i64 = conn
.query_row("SELECT count(*) FROM file_hashes", [], |r| r.get(0))
.unwrap();
assert_eq!(count, 0);
}
#[test]
fn fresh_database_publication_never_replaces_an_existing_destination() {
let temp = tempfile::tempdir().unwrap();
let built = temp.path().join("built.db");
let destination = temp.path().join("hashes.db");
std::fs::write(&built, b"completed build").unwrap();
std::fs::write(&destination, b"newer initializer").unwrap();
let err = publish_database(&built, &destination).unwrap_err();
assert!(format!("{err:#}").contains("publish"), "{err:#}");
assert_eq!(std::fs::read(&destination).unwrap(), b"newer initializer");
assert!(built.exists(), "the refused build remains recoverable");
}
#[test]
fn corrupt_database_bytes_are_refused_unchanged() {
let (_t, ctx) = library();
std::fs::create_dir_all(&ctx.paths.locks).unwrap();
let garbage = b"definitely not a sqlite database at all";
std::fs::write(&ctx.paths.db, garbage).unwrap();
assert!(open_existing(&ctx).is_err());
let err = initialize(&ctx).unwrap_err();
assert!(
format!("{err:#}").contains("not a SQLite database"),
"{err:#}"
);
assert_eq!(std::fs::read(&ctx.paths.db).unwrap(), garbage);
}
#[test]
fn a_zero_byte_database_is_never_exposed_as_a_library() {
let (_t, ctx) = library();
std::fs::create_dir_all(&ctx.paths.locks).unwrap();
std::fs::write(&ctx.paths.db, b"").unwrap();
let err = open_existing(&ctx).unwrap_err();
assert!(format!("{err:#}").contains("empty"), "{err:#}");
let conn = initialize(&ctx).unwrap();
let count: i64 = conn
.query_row("SELECT count(*) FROM file_hashes", [], |r| r.get(0))
.unwrap();
assert_eq!(count, 0);
let version: i64 = conn
.query_row("PRAGMA user_version", [], |r| r.get(0))
.unwrap();
assert_eq!(version, 1);
}
#[test]
fn a_foreign_sqlite_database_is_refused_unchanged() {
let (_t, ctx) = library();
std::fs::create_dir_all(&ctx.paths.locks).unwrap();
{
let conn = rusqlite::Connection::open(&ctx.paths.db).unwrap();
conn.execute_batch(
"CREATE TABLE somebody_elses (id INTEGER PRIMARY KEY);
INSERT INTO somebody_elses VALUES (1);",
)
.unwrap();
}
let before = std::fs::read(&ctx.paths.db).unwrap();
let err = open_existing(&ctx).unwrap_err();
assert!(
format!("{err:#}").contains("not a videre library"),
"{err:#}"
);
let err = initialize(&ctx).unwrap_err();
assert!(
format!("{err:#}").contains("not a videre library"),
"{err:#}"
);
assert_eq!(std::fs::read(&ctx.paths.db).unwrap(), before);
}
#[test]
fn an_older_supported_schema_is_upgraded_in_place_with_its_rows() {
let (_t, ctx) = library();
std::fs::create_dir_all(&ctx.paths.locks).unwrap();
{
let conn = rusqlite::Connection::open(&ctx.paths.db).unwrap();
conn.execute_batch(
"CREATE TABLE file_hashes (
path TEXT PRIMARY KEY, hash TEXT NOT NULL, size_bytes INTEGER,
created_at TEXT, modified_at TEXT, ext TEXT, phash INTEGER,
exif_date TEXT, gps_lat REAL, gps_lon REAL, width INTEGER,
height INTEGER
);",
)
.unwrap();
for name in ["a.jpg", "b.jpg"] {
conn.execute(
"INSERT INTO file_hashes (path, hash, ext) VALUES (?1, 'h', 'jpg')",
params![ctx.paths.root.join(name).to_str().unwrap()],
)
.unwrap();
}
}
let conn = open_existing(&ctx).unwrap();
for column in [
"mime",
"duration_secs",
"codec",
"location_name",
"location_cluster_id",
] {
let present: i64 = conn
.query_row(
&format!(
"SELECT count(*) FROM pragma_table_info('file_hashes') WHERE name = '{column}'"
),
[],
|r| r.get(0),
)
.unwrap();
assert_eq!(present, 1, "file_hashes.{column}");
}
for table in [
"people",
"faces",
"faces_scanned",
"marks",
"photo_tags",
"classifications",
"location_clusters",
"pipeline_runs",
] {
assert!(crate::db::table_exists(&conn, table).unwrap(), "{table}");
}
let count: i64 = conn
.query_row("SELECT count(*) FROM file_hashes", [], |r| r.get(0))
.unwrap();
assert_eq!(count, 2, "the upgrade must keep the existing rows");
let version: i64 = conn
.query_row("PRAGMA user_version", [], |r| r.get(0))
.unwrap();
assert_eq!(version, 1);
drop(conn);
let conn = open_existing(&ctx).unwrap();
let version: i64 = conn
.query_row("PRAGMA user_version", [], |r| r.get(0))
.unwrap();
assert_eq!(version, 1);
}
#[test]
fn a_config_without_a_database_is_not_a_library_and_initialize_keeps_the_config() {
let (_t, ctx) = library();
std::fs::create_dir_all(&ctx.paths.state).unwrap();
std::fs::write(&ctx.paths.config, "custom = \"keep\"\n").unwrap();
assert!(open_existing(&ctx).is_err());
drop(initialize(&ctx).unwrap());
assert_eq!(
std::fs::read_to_string(&ctx.paths.config).unwrap(),
"custom = \"keep\"\n"
);
}
#[test]
fn a_database_without_a_config_reads_and_initialize_writes_five_declarations_once() {
let (_t, ctx) = library();
drop(initialize(&ctx).unwrap());
std::fs::remove_file(&ctx.paths.config).unwrap();
drop(open_existing(&ctx).unwrap());
assert!(!ctx.paths.config.exists());
drop(initialize(&ctx).unwrap());
let text = std::fs::read_to_string(&ctx.paths.config).unwrap();
let table: toml::Table = toml::from_str(&text).unwrap();
assert_eq!(table.len(), 5, "{text}");
assert_eq!(table["db"].as_str(), Some("hashes.db"));
assert_eq!(table["jsonl"].as_str(), Some("hashes.jsonl"));
assert_eq!(
table["default_model"].as_str(),
Some(crate::embeddings::DEFAULT_MODEL_ID)
);
assert_eq!(table["xmp_precedence"].as_str(), Some("db"));
assert_eq!(table["export_xmp_on_watch"].as_bool(), Some(false));
let before = std::fs::read(&ctx.paths.config).unwrap();
drop(initialize(&ctx).unwrap());
assert_eq!(std::fs::read(&ctx.paths.config).unwrap(), before);
}
#[test]
fn two_simultaneous_initializers_produce_one_valid_library() {
let temp = tempfile::tempdir().unwrap();
let root = temp.path().join("photos");
std::fs::create_dir(&root).unwrap();
let ctx = LibraryContext::new(&root, &temp.path().join("cache")).unwrap();
let barrier = std::sync::Arc::new(std::sync::Barrier::new(2));
let handles: Vec<_> = (0..2)
.map(|_| {
let ctx = ctx.clone();
let barrier = barrier.clone();
std::thread::spawn(move || {
barrier.wait();
initialize(&ctx).map(|conn| {
conn.query_row::<i64, _, _>("SELECT count(*) FROM file_hashes", [], |r| {
r.get(0)
})
.unwrap()
})
})
})
.collect();
let results: Vec<_> = handles.into_iter().map(|h| h.join().unwrap()).collect();
let winners = results.iter().filter(|r| r.is_ok()).count();
assert_eq!(winners, 1, "exactly one initializer must win: {results:?}");
for result in &results {
match result {
Ok(count) => assert_eq!(*count, 0),
Err(e) => {
let msg = format!("{e:#}");
assert!(
msg.contains("in use"),
"the loser must be refused by the lock, not by anything else: {msg}"
);
}
}
}
let table: toml::Table =
toml::from_str(&std::fs::read_to_string(&ctx.paths.config).unwrap()).unwrap();
assert_eq!(table.len(), 5);
drop(open_existing(&ctx).unwrap());
}
#[test]
fn a_held_activity_lock_blocks_initialization_cleanly() {
let (_t, ctx) = library();
std::fs::create_dir_all(&ctx.paths.locks).unwrap();
let _held =
library_locks::try_activity(&ctx, library_locks::ActivityMode::Exclusive).unwrap();
let err = initialize(&ctx).unwrap_err();
assert!(format!("{err:#}").contains("in use"), "{err:#}");
drop(_held);
drop(initialize(&ctx).unwrap());
}
#[test]
fn a_held_exclusive_activity_lock_blocks_readers_too() {
let (_t, ctx) = library();
drop(initialize(&ctx).unwrap());
let _held =
library_locks::try_activity(&ctx, library_locks::ActivityMode::Exclusive).unwrap();
assert!(open_existing(&ctx).is_err());
}
#[test]
fn redirected_state_database_and_config_are_refused() {
let (temp, ctx) = library();
let elsewhere = temp.path().join("elsewhere");
std::fs::create_dir(&elsewhere).unwrap();
std::os::unix::fs::symlink(&elsewhere, &ctx.paths.state).unwrap();
let err = initialize(&ctx).unwrap_err();
assert!(format!("{err:#}").contains("symlink"), "{err:#}");
assert!(!elsewhere.join("locks").exists());
std::fs::remove_file(&ctx.paths.state).unwrap();
std::fs::create_dir_all(&ctx.paths.locks).unwrap();
let outside_db = temp.path().join("outside.db");
std::fs::write(&outside_db, b"").unwrap();
std::os::unix::fs::symlink(&outside_db, &ctx.paths.db).unwrap();
let err = initialize(&ctx).unwrap_err();
assert!(format!("{err:#}").contains("symlink"), "{err:#}");
assert_eq!(std::fs::read(&outside_db).unwrap(), b"");
let err = open_existing(&ctx).unwrap_err();
assert!(format!("{err:#}").contains("symlink"), "{err:#}");
std::fs::remove_file(&ctx.paths.db).unwrap();
let outside_cfg = temp.path().join("outside.toml");
std::fs::write(&outside_cfg, b"custom = \"x\"\n").unwrap();
std::os::unix::fs::symlink(&outside_cfg, &ctx.paths.config).unwrap();
let err = LibraryContext::new(&ctx.paths.root, &temp.path().join("cache")).unwrap_err();
assert!(format!("{err:#}").contains("symlink"), "{err:#}");
assert_eq!(std::fs::read(&outside_cfg).unwrap(), b"custom = \"x\"\n");
}
#[test]
fn multiply_linked_state_files_are_refused() {
let (temp, ctx) = library();
std::fs::create_dir_all(&ctx.paths.locks).unwrap();
let twin_db = temp.path().join("twin.db");
{
let conn = rusqlite::Connection::open(&twin_db).unwrap();
conn.execute_batch(
"CREATE TABLE file_hashes (path TEXT PRIMARY KEY, hash TEXT NOT NULL);",
)
.unwrap();
}
std::fs::hard_link(&twin_db, &ctx.paths.db).unwrap();
let err = initialize(&ctx).unwrap_err();
assert!(format!("{err:#}").contains("hard-linked"), "{err:#}");
let err = open_existing(&ctx).unwrap_err();
assert!(format!("{err:#}").contains("hard-linked"), "{err:#}");
assert!(twin_db.exists());
std::fs::remove_file(&ctx.paths.db).unwrap();
let twin_cfg = temp.path().join("twin.toml");
std::fs::write(&twin_cfg, "custom = \"x\"\n").unwrap();
std::fs::hard_link(&twin_cfg, &ctx.paths.config).unwrap();
let err = LibraryContext::new(&ctx.paths.root, &temp.path().join("cache")).unwrap_err();
assert!(format!("{err:#}").contains("hard-linked"), "{err:#}");
}
#[test]
fn a_symlinked_embeddings_directory_is_refused() {
let (temp, ctx) = library();
std::fs::create_dir_all(&ctx.paths.locks).unwrap();
let outside = temp.path().join("outside-embeddings");
std::fs::create_dir(&outside).unwrap();
std::os::unix::fs::symlink(&outside, &ctx.paths.embeddings).unwrap();
let err = initialize(&ctx).unwrap_err();
assert!(format!("{err:#}").contains("symlink"), "{err:#}");
assert!(!ctx.paths.db.exists());
assert!(std::fs::read_dir(&outside).unwrap().next().is_none());
let err = open_existing(&ctx).unwrap_err();
assert!(format!("{err:#}").contains("symlink"), "{err:#}");
std::fs::remove_file(&ctx.paths.embeddings).unwrap();
std::fs::create_dir(&ctx.paths.embeddings).unwrap();
drop(initialize(&ctx).unwrap());
drop(open_existing(&ctx).unwrap());
}
#[test]
fn a_foreign_row_refuses_the_whole_library_without_touching_it() {
let (_t, ctx) = library();
{
let conn = initialize(&ctx).unwrap();
let adjacent = format!("{}-sibling/x.jpg", ctx.paths.root.to_str().unwrap());
conn.execute(
"INSERT INTO file_hashes (path, hash) VALUES (?1, 'h')",
params![adjacent],
)
.unwrap();
}
let before = std::fs::read(&ctx.paths.db).unwrap();
let before_cfg = std::fs::read(&ctx.paths.config).unwrap();
let fresh = LibraryContext::new(&ctx.paths.root, &ctx.cache.base).unwrap();
let err = open_existing(&fresh).unwrap_err();
let msg = format!("{err:#}");
assert!(msg.contains("outside the library root"), "{msg}");
assert!(msg.contains("-sibling"), "{msg}");
assert!(initialize(&fresh).is_err());
assert_eq!(std::fs::read(&ctx.paths.db).unwrap(), before);
assert_eq!(std::fs::read(&ctx.paths.config).unwrap(), before_cfg);
}
#[test]
fn a_row_with_a_dot_component_is_refused_as_foreign() {
let (_t, ctx) = library();
{
let conn = initialize(&ctx).unwrap();
conn.execute(
"INSERT INTO file_hashes (path, hash) VALUES (?1, 'h')",
params![ctx.paths.root.join("plain.jpg").to_str().unwrap()],
)
.unwrap();
let escaped = format!("{}/../outside/x.jpg", ctx.paths.root.to_str().unwrap());
conn.execute(
"INSERT INTO file_hashes (path, hash) VALUES (?1, 'h')",
params![escaped],
)
.unwrap();
}
let before = std::fs::read(&ctx.paths.db).unwrap();
let before_cfg = std::fs::read(&ctx.paths.config).unwrap();
let fresh = LibraryContext::new(&ctx.paths.root, &ctx.cache.base).unwrap();
let err = open_existing(&fresh).unwrap_err();
let msg = format!("{err:#}");
assert!(msg.contains("outside the library root"), "{msg}");
assert!(msg.contains(".."), "{msg}");
assert!(initialize(&fresh).is_err());
assert_eq!(std::fs::read(&ctx.paths.db).unwrap(), before);
assert_eq!(std::fs::read(&ctx.paths.config).unwrap(), before_cfg);
}
#[test]
fn known_missing_media_under_the_root_remains_a_valid_library() {
let (_t, ctx) = library();
{
let conn = initialize(&ctx).unwrap();
conn.execute(
"INSERT INTO file_hashes (path, hash) VALUES (?1, 'h')",
params![ctx.paths.root.join("long-gone.jpg").to_str().unwrap()],
)
.unwrap();
}
let fresh = LibraryContext::new(&ctx.paths.root, &ctx.cache.base).unwrap();
let conn = open_existing(&fresh).unwrap();
let count: i64 = conn
.query_row("SELECT count(*) FROM file_hashes", [], |r| r.get(0))
.unwrap();
assert_eq!(count, 1);
}
#[test]
fn root_aliases_open_the_same_initialized_library() {
let (temp, ctx) = library();
drop(initialize(&ctx).unwrap());
let alias = temp.path().join("alias");
std::os::unix::fs::symlink(&ctx.paths.root, &alias).unwrap();
let via_alias = LibraryContext::new(&alias, &temp.path().join("cache")).unwrap();
assert_eq!(via_alias.paths.db, ctx.paths.db);
let conn = open_existing(&via_alias).unwrap();
let count: i64 = conn
.query_row("SELECT count(*) FROM file_hashes", [], |r| r.get(0))
.unwrap();
assert_eq!(count, 0);
}
#[test]
fn a_future_schema_version_is_refused_with_an_actionable_error() {
let (_t, ctx) = library();
drop(initialize(&ctx).unwrap());
{
let conn = rusqlite::Connection::open(&ctx.paths.db).unwrap();
conn.pragma_update(None, "user_version", 2).unwrap();
}
let err = open_existing(&ctx).unwrap_err();
let msg = format!("{err:#}");
assert!(msg.contains("newer"), "{msg}");
assert!(msg.contains("version 2"), "{msg}");
assert!(initialize(&ctx).is_err());
}
#[test]
fn the_published_database_is_self_contained_and_leaves_no_temporaries() {
let (_t, ctx) = library();
drop(initialize(&ctx).unwrap());
let conn = open_existing(&ctx).unwrap();
let mode: String = conn
.query_row("PRAGMA journal_mode", [], |r| r.get(0))
.unwrap();
assert_eq!(mode.to_lowercase(), "wal");
drop(conn);
let entries: Vec<String> = std::fs::read_dir(&ctx.paths.state)
.unwrap()
.map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
.collect();
for entry in &entries {
assert!(
!entry.contains(".tmp") && !entry.ends_with("-wal") && !entry.ends_with("-shm"),
"leftover {entry}: {entries:?}"
);
}
}
#[test]
fn stale_build_wal_sidecars_are_swept_with_their_temporary() {
let (_t, ctx) = library();
std::fs::create_dir_all(&ctx.paths.state).unwrap();
let stem = "hashes.db.4242.0.tmp";
for suffix in ["", "-wal", "-shm"] {
std::fs::write(ctx.paths.state.join(format!("{stem}{suffix}")), b"debris").unwrap();
}
std::fs::write(ctx.paths.state.join("notes.txt"), b"keep").unwrap();
drop(initialize(&ctx).unwrap());
for suffix in ["", "-wal", "-shm"] {
assert!(
!ctx.paths.state.join(format!("{stem}{suffix}")).exists(),
"sweeping must remove {stem}{suffix}"
);
}
assert!(ctx.paths.state.join("notes.txt").exists());
assert!(ctx.paths.db.exists());
}
#[test]
fn a_refused_foreign_row_validation_is_never_memoized_as_success() {
let (_t, ctx) = library();
let reader = LibraryContext::new(&ctx.paths.root, &ctx.cache.base).unwrap();
{
let conn = initialize(&ctx).unwrap();
conn.execute(
"INSERT INTO file_hashes (path, hash) VALUES (?1, 'h')",
params!["/elsewhere-not-this-library/x.jpg"],
)
.unwrap();
}
let err = open_existing(&reader).unwrap_err();
assert!(
format!("{err:#}").contains("outside the library root"),
"{err:#}"
);
assert!(
!reader.index_validated(),
"a refused validation must not be memoized"
);
{
let conn = rusqlite::Connection::open(&ctx.paths.db).unwrap();
let removed = conn
.execute(
"DELETE FROM file_hashes WHERE path = '/elsewhere-not-this-library/x.jpg'",
[],
)
.unwrap();
assert_eq!(removed, 1);
}
drop(open_existing(&reader).unwrap());
assert!(reader.index_validated());
let fresh = LibraryContext::new(&ctx.paths.root, &ctx.cache.base).unwrap();
drop(open_existing(&fresh).unwrap());
}
#[test]
fn opening_a_seventy_thousand_row_library_validates_with_zero_filesystem_probes() {
let (_t, ctx) = library();
const ROWS: u32 = 70_000;
{
let mut conn = initialize(&ctx).unwrap();
let tx = conn.transaction().unwrap();
{
let mut stmt = tx
.prepare("INSERT INTO file_hashes (path, hash) VALUES (?1, 'h')")
.unwrap();
for i in 0..ROWS {
let path = ctx.paths.root.join(format!(
"Trips/album-{:02}/img-{:05}.jpg",
i / 1000,
i
));
stmt.execute(params![path.to_str().unwrap()]).unwrap();
}
}
tx.commit().unwrap();
}
let fresh = LibraryContext::new(&ctx.paths.root, &ctx.cache.base).unwrap();
let probes_before = crate::library_guard::count_resolutions();
let rows_before = count_rows_validated();
let start = std::time::Instant::now();
let conn = open_existing(&fresh).unwrap();
let elapsed = start.elapsed();
assert!(
fresh.index_validated(),
"a successful open records the memo"
);
assert_eq!(
crate::library_guard::count_resolutions() - probes_before,
0,
"containment validation must not touch the filesystem per row"
);
assert_eq!(
count_rows_validated() - rows_before,
u64::from(ROWS),
"every row must have been judged"
);
assert!(
elapsed < Duration::from_secs(5),
"validating {ROWS} rows took {elapsed:?}"
);
let count: i64 = conn
.query_row("SELECT count(*) FROM file_hashes", [], |r| r.get(0))
.unwrap();
assert_eq!(count, i64::from(ROWS));
}
}