1336 lines
51 KiB
Rust
1336 lines
51 KiB
Rust
//! Row-level write helpers that keep the FTS5 contentless table in sync with
|
||
//! `files`/`documents`/`properties`.
|
||
//!
|
||
//! FTS5 contentless tables store postings only. Updating them requires the
|
||
//! old row values to compute which terms to remove. These helpers centralize
|
||
//! that bookkeeping so callers never have to remember the order of operations.
|
||
//!
|
||
//! States (mirrors `basic_state` / `content_state` columns):
|
||
//!
|
||
//! | value | meaning |
|
||
//! |------:|---------|
|
||
//! | 0 | pending |
|
||
//! | 1 | done |
|
||
//! | 2 | failed |
|
||
//! | 3 | not applicable (content only) |
|
||
|
||
use rusqlite::{params, Connection, OptionalExtension, Transaction};
|
||
|
||
use crate::mime::FileType;
|
||
|
||
pub const STATE_PENDING: i64 = 0;
|
||
pub const STATE_DONE: i64 = 1;
|
||
pub const STATE_FAILED: i64 = 2;
|
||
pub const STATE_NA: i64 = 3;
|
||
|
||
/// Everything needed to insert a fresh file row.
|
||
#[derive(Debug, Clone)]
|
||
pub struct NewFile<'a> {
|
||
pub name: &'a str,
|
||
pub path: &'a str,
|
||
pub parent: &'a str,
|
||
pub size: u64,
|
||
pub mtime: u64,
|
||
pub inode: Option<u64>,
|
||
pub device_id: Option<u64>,
|
||
pub mime: Option<&'a str>,
|
||
pub ftype: FileType,
|
||
pub hash: Option<&'a [u8]>,
|
||
/// Whether the content pass has work to do for this file. `false` means
|
||
/// the row is born `STATE_NA` — nothing claims its MIME, the
|
||
/// `content_extensions` filter excludes it, or it is over
|
||
/// `maximum_text_file_size`. Decided at walk time by
|
||
/// [`crate::file_handling::content_extractable`].
|
||
pub needs_content: bool,
|
||
}
|
||
|
||
/// Insert a new file row, returning its id. `basic_state` is set to DONE
|
||
/// (row existing *is* the basic-index state); `content_state` comes from
|
||
/// `needs_content` — PENDING for a file an extractor will claim, NA for one it
|
||
/// won't. Deciding here rather than on a content worker is what keeps PENDING
|
||
/// meaning "real work outstanding", which the extraction progress denominator
|
||
/// and [`pending_content_page`] both depend on.
|
||
///
|
||
/// Uses `INSERT OR IGNORE` so a UNIQUE(path) collision (which indicates the
|
||
/// caller fed the same path twice in one run) becomes a silent no-op
|
||
/// returning `None` rather than aborting the whole batch. The walker is
|
||
/// expected to dedupe visits upstream; this is a defense-in-depth backstop.
|
||
pub fn insert_file(tx: &Transaction<'_>, f: &NewFile<'_>) -> Result<Option<i64>, String> {
|
||
let rows = tx
|
||
.prepare_cached(
|
||
"INSERT OR IGNORE INTO files (
|
||
name, path, parent, size, mtime, inode, device_id,
|
||
mime, type, basic_state, content_state, hash
|
||
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
|
||
)
|
||
.and_then(|mut stmt| {
|
||
stmt.execute(params![
|
||
f.name,
|
||
f.path,
|
||
f.parent,
|
||
f.size as i64,
|
||
f.mtime as i64,
|
||
f.inode.map(|x| x as i64),
|
||
f.device_id.map(|x| x as i64),
|
||
f.mime,
|
||
f.ftype.bits() as i64,
|
||
STATE_DONE,
|
||
initial_content_state(f),
|
||
f.hash,
|
||
])
|
||
})
|
||
.map_err(|e| format!("insert file {}: {}", f.path, e))?;
|
||
if rows == 0 {
|
||
// Existing row with the same path (e.g. duplicate visit within the run).
|
||
return Ok(None);
|
||
}
|
||
Ok(Some(tx.last_insert_rowid()))
|
||
}
|
||
|
||
/// The `content_state` a freshly written row starts in. Shared by
|
||
/// [`insert_file`] and [`update_file_basic`] so a file that reappears as an
|
||
/// update lands in the same state it would have as an insert.
|
||
fn initial_content_state(f: &NewFile<'_>) -> i64 {
|
||
if f.needs_content {
|
||
STATE_PENDING
|
||
} else {
|
||
STATE_NA
|
||
}
|
||
}
|
||
|
||
/// Update a file's metadata in place (same path, changed size/mtime/hash) and
|
||
/// reset its content state from `f.needs_content`, clearing any extracted
|
||
/// content so the text-indexing pass re-processes it.
|
||
///
|
||
/// Writes `size`, `mtime`, `hash`, `mime`, `type`, `content_state` and
|
||
/// `failure_msg` — and only those. `name`, `parent`, `inode` and `device_id`
|
||
/// are not refreshed here despite being present on the `NewFile`: the row is
|
||
/// found by path, and the first two are functions of it.
|
||
pub fn update_file_basic(tx: &Transaction<'_>, f: &NewFile<'_>) -> Result<Option<i64>, String> {
|
||
// One statement, not a lookup then an update: `RETURNING` hands back the
|
||
// id of the row it just wrote, and a miss is simply no row returned.
|
||
let id: Option<i64> = tx
|
||
.prepare_cached(
|
||
"UPDATE files
|
||
SET size = ?1, mtime = ?2, hash = ?3, mime = ?4, type = ?5,
|
||
content_state = ?6, failure_msg = NULL
|
||
WHERE path = ?7
|
||
RETURNING id",
|
||
)
|
||
.and_then(|mut stmt| {
|
||
stmt.query_row(
|
||
params![
|
||
f.size as i64,
|
||
f.mtime as i64,
|
||
f.hash,
|
||
f.mime,
|
||
f.ftype.bits() as i64,
|
||
initial_content_state(f),
|
||
f.path,
|
||
],
|
||
|r| r.get(0),
|
||
)
|
||
.optional()
|
||
})
|
||
.map_err(|e| format!("update file {}: {}", f.path, e))?;
|
||
let Some(id) = id else {
|
||
return Ok(None);
|
||
};
|
||
// Any prior extracted content is stale — remove it along with its FTS row.
|
||
remove_content_for_id(tx, id)?;
|
||
Ok(Some(id))
|
||
}
|
||
|
||
/// Mark a file's content indexing as complete and write the extracted text +
|
||
/// properties atomically. The plaintext is fed to the contentless FTS5
|
||
/// tokenizer (which keeps only the inverted index) and — when `store_text`
|
||
/// is `true` — separately stored zstd-compressed in `documents_text` for
|
||
/// on-demand snippet rendering. When `store_text=false` the sidecar INSERT
|
||
/// is skipped: queries still match the right files but result rows can't
|
||
/// render snippets. `properties` are stored both as a structured side-
|
||
/// table (for exact retrieval) and concatenated into the FTS `properties`
|
||
/// column (for MATCH).
|
||
pub fn set_content_done(
|
||
tx: &Transaction<'_>,
|
||
file_id: i64,
|
||
name: &str,
|
||
text: &str,
|
||
properties: &[(String, String)],
|
||
store_text: bool,
|
||
) -> Result<(), String> {
|
||
// Clear any previous extraction (in case of re-run).
|
||
remove_content_for_id(tx, file_id)?;
|
||
|
||
for (k, v) in properties {
|
||
tx.prepare_cached("INSERT INTO properties(file_id, key, value) VALUES (?1, ?2, ?3)")
|
||
.and_then(|mut stmt| stmt.execute(params![file_id, k, v]))
|
||
.map_err(|e| format!("insert property {}={}: {}", k, v, e))?;
|
||
}
|
||
let props_blob = encode_properties_for_fts(properties);
|
||
// Contentless FTS5 still accepts values on INSERT — the tokenizer needs
|
||
// them — it simply doesn't persist the raw column values.
|
||
tx.prepare_cached(
|
||
"INSERT INTO searchabletext(rowid, name, text, properties) VALUES (?1, ?2, ?3, ?4)",
|
||
)
|
||
.and_then(|mut stmt| stmt.execute(params![file_id, name, text, props_blob]))
|
||
.map_err(|e| format!("insert FTS row {}: {}", file_id, e))?;
|
||
|
||
// Skip the compressed sidecar when: the config disables snippet storage
|
||
// outright, or there's no body text (e.g. an image whose extractor
|
||
// returned only EXIF properties). The second case saves a zstd frame
|
||
// on what would otherwise be an empty blob.
|
||
if store_text && !text.is_empty() {
|
||
let compressed = zstd::encode_all(text.as_bytes(), ZSTD_LEVEL)
|
||
.map_err(|e| format!("zstd encode for file {}: {}", file_id, e))?;
|
||
tx.prepare_cached(
|
||
"INSERT INTO documents_text(file_id, text_zstd, text_len) VALUES (?1, ?2, ?3)",
|
||
)
|
||
.and_then(|mut stmt| stmt.execute(params![file_id, compressed, text.len() as i64]))
|
||
.map_err(|e| format!("insert documents_text {}: {}", file_id, e))?;
|
||
}
|
||
|
||
tx.prepare_cached("UPDATE files SET content_state = ?1, failure_msg = NULL WHERE id = ?2")
|
||
.and_then(|mut stmt| stmt.execute(params![STATE_DONE, file_id]))
|
||
.map_err(|e| format!("update content_state DONE {}: {}", file_id, e))?;
|
||
// Clear any prior failed-file record.
|
||
tx.prepare_cached("DELETE FROM failed_files WHERE file_id = ?1")
|
||
.and_then(|mut stmt| stmt.execute(params![file_id]))
|
||
.map_err(|e| format!("clear failed_files {}: {}", file_id, e))?;
|
||
Ok(())
|
||
}
|
||
|
||
/// zstd level tuned for extracted-text prose. Level 3 hits ~3-5× on English
|
||
/// prose at high throughput (hundreds of MB/s) — level 9+ would shave a few
|
||
/// percent more but at 10× the CPU cost during indexing. Wrong knob to
|
||
/// tune: readers decompress far faster than writers compress, so keep
|
||
/// write-side cost low.
|
||
const ZSTD_LEVEL: i32 = 3;
|
||
|
||
/// Mark a file's content extraction as failed. Keeps the basic row in place.
|
||
pub fn set_content_failed(tx: &Transaction<'_>, file_id: i64, reason: &str) -> Result<(), String> {
|
||
let now = std::time::SystemTime::now()
|
||
.duration_since(std::time::UNIX_EPOCH)
|
||
.map(|d| d.as_secs())
|
||
.unwrap_or(0) as i64;
|
||
tx.prepare_cached("UPDATE files SET content_state = ?1, failure_msg = ?2 WHERE id = ?3")
|
||
.and_then(|mut stmt| stmt.execute(params![STATE_FAILED, reason, file_id]))
|
||
.map_err(|e| format!("update content_state FAILED {}: {}", file_id, e))?;
|
||
tx.prepare_cached(
|
||
"INSERT OR REPLACE INTO failed_files(file_id, reason, ts) VALUES (?1, ?2, ?3)",
|
||
)
|
||
.and_then(|mut stmt| stmt.execute(params![file_id, reason, now]))
|
||
.map_err(|e| format!("insert failed_files {}: {}", file_id, e))?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Mark content extraction as not applicable (e.g. binary format we don't
|
||
/// support). The file row still contributes to filename search.
|
||
pub fn set_content_na(tx: &Transaction<'_>, file_id: i64) -> Result<(), String> {
|
||
tx.prepare_cached("UPDATE files SET content_state = ?1, failure_msg = NULL WHERE id = ?2")
|
||
.and_then(|mut stmt| stmt.execute(params![STATE_NA, file_id]))
|
||
.map_err(|e| format!("update content_state NA {}: {}", file_id, e))?;
|
||
tx.prepare_cached("DELETE FROM failed_files WHERE file_id = ?1")
|
||
.and_then(|mut stmt| stmt.execute(params![file_id]))
|
||
.map_err(|e| format!("clear failed_files {}: {}", file_id, e))?;
|
||
Ok(())
|
||
}
|
||
|
||
/// Delete a file row by path, keeping FTS in sync. Returns whether a row was
|
||
/// removed.
|
||
pub fn delete_file_by_path(tx: &Transaction<'_>, path: &str) -> Result<bool, String> {
|
||
// `RETURNING` gives the id of the row that went, so the dependent tables
|
||
// can be cleared without a separate lookup first.
|
||
let id: Option<i64> = tx
|
||
.prepare_cached("DELETE FROM files WHERE path = ?1 RETURNING id")
|
||
.and_then(|mut stmt| stmt.query_row(params![path], |r| r.get(0)).optional())
|
||
.map_err(|e| format!("delete file {}: {}", path, e))?;
|
||
let Some(id) = id else { return Ok(false) };
|
||
remove_content_for_id(tx, id)?;
|
||
Ok(true)
|
||
}
|
||
|
||
/// Delete every row whose path falls in the half-open range `[lo, hi)`,
|
||
/// keeping FTS, `documents_text`, `properties` and `failed_files` in step.
|
||
/// Returns how many `files` rows went.
|
||
///
|
||
/// The bulk counterpart to [`delete_file_by_path`], for removing a whole
|
||
/// directory at once. Five statements regardless of how many files the range
|
||
/// holds — where deleting them one at a time costs ~5 *per file* — and the
|
||
/// range is a plain index seek on `UNIQUE(files.path)`, so this is
|
||
/// `SEARCH … (path>? AND path<?)` rather than a scan. Build the bounds with
|
||
/// [`crate::file_handling::ExtractCursor::for_root`], which is what makes them
|
||
/// separator-correct.
|
||
///
|
||
/// All four dependent tables are deleted explicitly rather than left to their
|
||
/// `ON DELETE CASCADE` foreign keys. `searchabletext` is an FTS5 virtual table
|
||
/// and has no foreign key at all, so relying on cascade would leave the rule
|
||
/// half-declarative and half-manual, free to drift the moment someone opens a
|
||
/// connection without `PRAGMA foreign_keys`. (`contentless_delete=1` is what
|
||
/// lets the FTS rows go by rowid alone.)
|
||
pub fn delete_subtree(tx: &Transaction<'_>, lo: &str, hi: &str) -> Result<usize, String> {
|
||
// Every dependent table is keyed by the file id, so they share one
|
||
// sub-select; `files` itself goes last, once nothing references it.
|
||
for (table, key) in [
|
||
("searchabletext", "rowid"),
|
||
("documents_text", "file_id"),
|
||
("properties", "file_id"),
|
||
("failed_files", "file_id"),
|
||
] {
|
||
let sql = format!(
|
||
"DELETE FROM {} WHERE {} IN \
|
||
(SELECT id FROM files WHERE path >= ?1 AND path < ?2)",
|
||
table, key
|
||
);
|
||
tx.prepare_cached(&sql)
|
||
.and_then(|mut stmt| stmt.execute(params![lo, hi]))
|
||
.map_err(|e| format!("delete {} under {}: {}", table, lo, e))?;
|
||
}
|
||
let removed = tx
|
||
.prepare_cached("DELETE FROM files WHERE path >= ?1 AND path < ?2")
|
||
.and_then(|mut stmt| stmt.execute(params![lo, hi]))
|
||
.map_err(|e| format!("delete files under {}: {}", lo, e))?;
|
||
Ok(removed)
|
||
}
|
||
|
||
/// Every indexed file directly inside `parent`, as `name -> mtime`.
|
||
///
|
||
/// The walk's unit of classification. Keyed by name rather than full path
|
||
/// because the parent is implied — at millions of files, not storing the
|
||
/// directory prefix once per entry is the difference this whole path exists
|
||
/// to make. Served by `idx_files_parent`, so this is one index range lookup.
|
||
pub fn dir_rows(
|
||
conn: &Connection,
|
||
parent: &str,
|
||
) -> Result<std::collections::HashMap<String, u64>, String> {
|
||
let mut stmt = conn
|
||
.prepare_cached("SELECT name, mtime FROM files WHERE parent = ?1")
|
||
.map_err(|e| format!("prepare dir rows for {}: {}", parent, e))?;
|
||
let rows = stmt
|
||
.query_map(params![parent], |r| {
|
||
Ok((r.get::<_, String>(0)?, r.get::<_, i64>(1)?.max(0) as u64))
|
||
})
|
||
.map_err(|e| format!("query dir rows for {}: {}", parent, e))?;
|
||
let mut out = std::collections::HashMap::new();
|
||
for row in rows {
|
||
let (name, mtime) = row.map_err(|e| format!("read dir row under {}: {}", parent, e))?;
|
||
out.insert(name, mtime);
|
||
}
|
||
Ok(out)
|
||
}
|
||
|
||
/// One page of rows still awaiting content extraction under `cursor`'s range,
|
||
/// as `(id, name, path, mime)` ordered by id.
|
||
///
|
||
/// Keyset, not `OFFSET`: `id > cursor.last_id` means each page is an index
|
||
/// seek rather than a re-scan of everything already handed out, and — because
|
||
/// the cursor only moves forward — a row is served exactly once even though
|
||
/// the writer is concurrently flipping `content_state` behind the reader. The
|
||
/// `content_state = 0` predicate is the belt to that braces, not the
|
||
/// mechanism.
|
||
pub fn pending_content_page(
|
||
conn: &Connection,
|
||
cursor: &crate::file_handling::ExtractCursor,
|
||
max_size: i64,
|
||
limit: i64,
|
||
) -> Result<Vec<(i64, String, String, Option<String>)>, String> {
|
||
let mut stmt = conn
|
||
.prepare_cached(
|
||
"SELECT id, name, path, mime FROM files
|
||
WHERE content_state = 0 AND size <= ?1 AND id > ?2
|
||
AND path >= ?3 AND path < ?4
|
||
ORDER BY id
|
||
LIMIT ?5",
|
||
)
|
||
.map_err(|e| format!("prepare pending content query: {}", e))?;
|
||
let rows = stmt
|
||
.query_map(
|
||
params![max_size, cursor.last_id, cursor.lo, cursor.hi, limit],
|
||
|row| {
|
||
Ok((
|
||
row.get::<_, i64>(0)?,
|
||
row.get::<_, String>(1)?,
|
||
row.get::<_, String>(2)?,
|
||
row.get::<_, Option<String>>(3)?,
|
||
))
|
||
},
|
||
)
|
||
.map_err(|e| format!("query pending content: {}", e))?;
|
||
rows.collect::<Result<Vec<_>, _>>()
|
||
.map_err(|e| format!("read pending content row: {}", e))
|
||
}
|
||
|
||
/// The stored mtime for one exact path, or `None` if it isn't indexed.
|
||
///
|
||
/// For files the walk reaches by a spelling whose parent isn't the directory
|
||
/// being read — a resolved symlink target — where [`dir_rows`] would not
|
||
/// have them.
|
||
pub fn mtime_for_path(conn: &Connection, path: &str) -> Result<Option<u64>, String> {
|
||
let mut stmt = conn
|
||
.prepare_cached("SELECT mtime FROM files WHERE path = ?1")
|
||
.map_err(|e| format!("prepare mtime lookup for {}: {}", path, e))?;
|
||
stmt.query_row(params![path], |r| r.get::<_, i64>(0))
|
||
.optional()
|
||
.map(|o| o.map(|m| m.max(0) as u64))
|
||
.map_err(|e| format!("mtime lookup for {}: {}", path, e))
|
||
}
|
||
|
||
/// Distinct `parent` values within the half-open path range `[lo, hi)`,
|
||
/// streamed to `f` rather than collected.
|
||
///
|
||
/// Callers use this to find directories the walk never visited, so it must
|
||
/// not itself materialize a list proportional to the tree — the whole point
|
||
/// of the change that introduced it. `idx_files_parent` makes this an
|
||
/// index-only scan.
|
||
pub fn for_each_parent_in_range<F: FnMut(String)>(
|
||
conn: &Connection,
|
||
lo: &str,
|
||
hi: &str,
|
||
mut f: F,
|
||
) -> Result<(), String> {
|
||
let mut stmt = conn
|
||
.prepare("SELECT DISTINCT parent FROM files WHERE parent >= ?1 AND parent < ?2")
|
||
.map_err(|e| format!("prepare parent scan: {}", e))?;
|
||
let rows = stmt
|
||
.query_map(params![lo, hi], |r| r.get::<_, String>(0))
|
||
.map_err(|e| format!("parent scan: {}", e))?;
|
||
for row in rows {
|
||
f(row.map_err(|e| format!("read parent row: {}", e))?);
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// Paths of every file directly inside `parent`.
|
||
///
|
||
/// The companion to [`for_each_parent_in_range`]: once a parent is known to
|
||
/// be unvisited, this is what its rows are.
|
||
pub fn paths_in_dir(conn: &Connection, parent: &str) -> Result<Vec<String>, String> {
|
||
let mut stmt = conn
|
||
.prepare_cached("SELECT path FROM files WHERE parent = ?1")
|
||
.map_err(|e| format!("prepare paths in {}: {}", parent, e))?;
|
||
let rows = stmt
|
||
.query_map(params![parent], |r| r.get::<_, String>(0))
|
||
.map_err(|e| format!("query paths in {}: {}", parent, e))?;
|
||
rows.collect::<Result<Vec<_>, _>>()
|
||
.map_err(|e| format!("read path under {}: {}", parent, e))
|
||
}
|
||
|
||
/// Remove the FTS row, compressed text blob, and any `properties` rows for
|
||
/// a given file id. Does not touch the `files` row itself. Idempotent — a
|
||
/// missing row is fine.
|
||
pub fn remove_content_for_id(tx: &Transaction<'_>, file_id: i64) -> Result<(), String> {
|
||
// `contentless_delete=1` on the FTS5 table makes this work without
|
||
// re-supplying the old column values (it tombstones the rowid).
|
||
for (table, key) in [
|
||
("searchabletext", "rowid"),
|
||
("documents_text", "file_id"),
|
||
("properties", "file_id"),
|
||
] {
|
||
let sql = format!("DELETE FROM {} WHERE {} = ?1", table, key);
|
||
tx.prepare_cached(&sql)
|
||
.and_then(|mut stmt| stmt.execute(params![file_id]))
|
||
.map_err(|e| format!("delete {} for {}: {}", table, file_id, e))?;
|
||
}
|
||
Ok(())
|
||
}
|
||
|
||
/// Serialize properties for the FTS `properties` column. `key:value` pairs
|
||
/// separated by spaces so `MATCH 'properties:artist:beatles'` works.
|
||
fn encode_properties_for_fts(props: &[(String, String)]) -> String {
|
||
let mut buf = String::new();
|
||
for (i, (k, v)) in props.iter().enumerate() {
|
||
if i > 0 {
|
||
buf.push(' ');
|
||
}
|
||
buf.push_str(k);
|
||
buf.push(':');
|
||
buf.push_str(v);
|
||
}
|
||
buf
|
||
}
|
||
|
||
/// Free pages, as a percentage of the file, that make a [`maintain`] VACUUM
|
||
/// worth its cost. Below this the file is rewritten for nothing: a run that
|
||
/// changed little leaves little slack, and rewriting a multi-gigabyte index
|
||
/// to reclaim a few megabytes is minutes of I/O for no gain.
|
||
const VACUUM_MIN_SLACK_PERCENT: i64 = 10;
|
||
|
||
/// Flush the whole WAL into the main database and truncate the log to zero
|
||
/// bytes. `Err` means the log was *not* emptied.
|
||
///
|
||
/// `execute`/`execute_batch` discard the pragma's result row, and that row is
|
||
/// the only place SQLite reports that a checkpoint gave up — an incomplete
|
||
/// checkpoint is not an error, it is a number in a row nobody read. That is
|
||
/// how a log can grow past the index it journals without a word in the logs.
|
||
///
|
||
/// The signal is the *log* column (frames left in the WAL), not `busy`: a
|
||
/// TRUNCATE that cannot take the writer lock silently downgrades itself to
|
||
/// PASSIVE and still reports `busy = 0` with the log untouched. A real restart
|
||
/// sets `mxFrame` to 0. A database not in WAL mode reports -1, hence `<= 0`.
|
||
pub fn checkpoint_truncate(conn: &Connection) -> Result<(), String> {
|
||
let (busy, log): (i64, i64) = conn
|
||
.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |r| {
|
||
Ok((r.get(0)?, r.get(1)?))
|
||
})
|
||
.map_err(|e| format!("wal checkpoint: {}", e))?;
|
||
if log <= 0 {
|
||
Ok(())
|
||
} else {
|
||
Err(format!(
|
||
"wal checkpoint incomplete: busy={}, {} frames left in the log",
|
||
busy, log
|
||
))
|
||
}
|
||
}
|
||
|
||
/// Flush the WAL into the main DB file and close. Call on clean shutdown so
|
||
/// the next open starts with an empty log. WAL mode itself is persistent in
|
||
/// the file — deliberately left on.
|
||
pub fn checkpoint_and_close(conn: Connection) {
|
||
if let Err(e) = checkpoint_truncate(&conn) {
|
||
crate::log_warn!("{}", e);
|
||
}
|
||
drop(conn);
|
||
}
|
||
|
||
/// Land the log, reclaim the file's slack, and refresh the query planner's
|
||
/// statistics. Returns whether it vacuumed.
|
||
///
|
||
/// The sequence is checkpoint → VACUUM → `PRAGMA optimize` → checkpoint. The
|
||
/// trailing checkpoint is not a repeat of the first: VACUUM's copy-back pushes
|
||
/// every page of the rebuilt file through the log, and `optimize` writes
|
||
/// `sqlite_stat1`, so leaving without one would trade the slack just reclaimed
|
||
/// for a log the size of the index.
|
||
///
|
||
/// Run on a connection from [`crate::db::open::open_maintenance`], never on
|
||
/// the indexer's: VACUUM builds the replacement database in the temp store,
|
||
/// and under the indexer's `temp_store = MEMORY` that means building the whole
|
||
/// index in RAM.
|
||
///
|
||
/// `db_dir` is where the temporary database goes, and it must be the index's
|
||
/// own directory. `temp_store = FILE` alone resolves through `SQLITE_TMPDIR`,
|
||
/// `TMPDIR`, `/var/tmp`, then `/tmp` — and `/tmp` is a RAM-backed tmpfs on many
|
||
/// Linux systems, which would put us straight back where we started. Keeping
|
||
/// the temporary database beside the index also puts it on a volume already
|
||
/// known to hold one.
|
||
///
|
||
/// Peak transient space on that volume is roughly three times the index: the
|
||
/// original, the replacement being built beside it, and the log that VACUUM's
|
||
/// copy-back runs through. Running out is a failed VACUUM, not a damaged
|
||
/// index — the transaction rolls back.
|
||
pub fn maintain(conn: &Connection, db_dir: &str) -> Result<bool, String> {
|
||
// Best-effort: a reader that briefly pins the log is no reason to skip
|
||
// the compaction below, which does not need the log empty to start.
|
||
if let Err(e) = checkpoint_truncate(conn) {
|
||
crate::log_warn!("{}", e);
|
||
}
|
||
|
||
let page_count: i64 = conn
|
||
.query_row("PRAGMA page_count", [], |r| r.get(0))
|
||
.map_err(|e| format!("read page_count: {}", e))?;
|
||
let freelist: i64 = conn
|
||
.query_row("PRAGMA freelist_count", [], |r| r.get(0))
|
||
.map_err(|e| format!("read freelist_count: {}", e))?;
|
||
|
||
let vacuumed = freelist * 100 >= page_count * VACUUM_MIN_SLACK_PERCENT;
|
||
if vacuumed {
|
||
// `temp_store_directory` is a deprecated pragma that writes a global,
|
||
// so it is set for the VACUUM and cleared straight after rather than
|
||
// left standing for every other connection in the process.
|
||
let escaped = db_dir.replace('\'', "''");
|
||
conn.execute_batch(&format!("PRAGMA temp_store_directory = '{}';", escaped))
|
||
.map_err(|e| format!("set temp dir for vacuum: {}", e))?;
|
||
let outcome = conn
|
||
.execute_batch("VACUUM;")
|
||
.map_err(|e| format!("vacuum: {}", e));
|
||
let _ = conn.execute_batch("PRAGMA temp_store_directory = '';");
|
||
outcome?;
|
||
}
|
||
|
||
// Re-analyses only the tables whose shape has drifted far enough to matter,
|
||
// so it is close to free on a run that changed little. A run that added
|
||
// millions of rows is exactly when the planner's old statistics start
|
||
// choosing the wrong index for a search.
|
||
conn.execute_batch("PRAGMA optimize;")
|
||
.map_err(|e| format!("optimize: {}", e))?;
|
||
|
||
checkpoint_truncate(conn)?;
|
||
Ok(vacuumed)
|
||
}
|
||
|
||
/// Read the `last_full_index` marker (unix seconds of the last *successful*
|
||
/// full indexing run) from `schema_info`. Absent key — fresh DB, or a DB
|
||
/// from before this marker existed — means "never".
|
||
pub fn get_last_full_index(conn: &Connection) -> Option<u64> {
|
||
conn.query_row(
|
||
"SELECT value FROM schema_info WHERE key = 'last_full_index'",
|
||
[],
|
||
|r| r.get::<_, String>(0),
|
||
)
|
||
.optional()
|
||
.ok()
|
||
.flatten()
|
||
.and_then(|v| v.parse().ok())
|
||
}
|
||
|
||
/// Stamp `last_full_index` with `ts` (unix seconds). Called at the end of
|
||
/// every successful full indexing run; the coordinator reads it to schedule
|
||
/// periodic reindexing.
|
||
pub fn set_last_full_index(conn: &Connection, ts: u64) -> Result<(), String> {
|
||
conn.execute(
|
||
"INSERT OR REPLACE INTO schema_info(key, value) VALUES ('last_full_index', ?1)",
|
||
params![ts.to_string()],
|
||
)
|
||
.map_err(|e| format!("write last_full_index: {}", e))?;
|
||
Ok(())
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use std::sync::atomic::{AtomicBool, Ordering};
|
||
use std::sync::Arc;
|
||
|
||
use super::*;
|
||
use crate::db::open_or_recreate;
|
||
|
||
fn tmp_path() -> std::path::PathBuf {
|
||
let mut p = std::env::temp_dir();
|
||
p.push(format!(
|
||
"quicksearch-repo-{}-{}.sqlite",
|
||
std::process::id(),
|
||
std::time::SystemTime::now()
|
||
.duration_since(std::time::UNIX_EPOCH)
|
||
.unwrap()
|
||
.as_nanos()
|
||
));
|
||
p
|
||
}
|
||
|
||
#[test]
|
||
fn insert_update_delete_round_trip() {
|
||
let p = tmp_path();
|
||
let mut conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
{
|
||
let tx = conn.transaction().unwrap();
|
||
let id = insert_file(
|
||
&tx,
|
||
&NewFile {
|
||
name: "a.txt",
|
||
path: "/tmp/a.txt",
|
||
parent: "/tmp",
|
||
size: 42,
|
||
mtime: 1_700_000_000,
|
||
inode: Some(7),
|
||
device_id: Some(64768),
|
||
mime: Some("text/plain"),
|
||
ftype: FileType::TEXT,
|
||
hash: Some(&[1, 2, 3]),
|
||
needs_content: true,
|
||
},
|
||
)
|
||
.unwrap()
|
||
.expect("unique path");
|
||
set_content_done(
|
||
&tx,
|
||
id,
|
||
"a.txt",
|
||
"hello world",
|
||
&[("title".to_string(), "hi".to_string())],
|
||
true,
|
||
)
|
||
.unwrap();
|
||
tx.commit().unwrap();
|
||
}
|
||
|
||
// Text is findable via FTS.
|
||
let hit: i64 = conn
|
||
.query_row(
|
||
"SELECT rowid FROM searchabletext WHERE searchabletext MATCH 'hello'",
|
||
[],
|
||
|r| r.get(0),
|
||
)
|
||
.unwrap();
|
||
assert!(hit > 0);
|
||
|
||
// Delete cleans up.
|
||
{
|
||
let tx = conn.transaction().unwrap();
|
||
assert!(delete_file_by_path(&tx, "/tmp/a.txt").unwrap());
|
||
tx.commit().unwrap();
|
||
}
|
||
let count: i64 = conn
|
||
.query_row("SELECT COUNT(*) FROM files", [], |r| r.get(0))
|
||
.unwrap();
|
||
assert_eq!(count, 0);
|
||
let fts_count: i64 = conn
|
||
.query_row("SELECT COUNT(*) FROM searchabletext", [], |r| r.get(0))
|
||
.unwrap();
|
||
assert_eq!(fts_count, 0);
|
||
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
#[test]
|
||
fn insert_writes_content_state_from_needs_content() {
|
||
// The whole point of the field: a row nothing will extract is born NA,
|
||
// so "pending" downstream — the extraction denominator, the feeder's
|
||
// page query, the Baloo pending count — means real outstanding work.
|
||
let p = tmp_path();
|
||
let mut conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
let tx = conn.transaction().unwrap();
|
||
let mut row = NewFile {
|
||
name: "claimed.txt",
|
||
path: "/tmp/claimed.txt",
|
||
parent: "/tmp",
|
||
size: 1,
|
||
mtime: 1,
|
||
inode: None,
|
||
device_id: None,
|
||
mime: Some("text/plain"),
|
||
ftype: FileType::TEXT,
|
||
hash: None,
|
||
needs_content: true,
|
||
};
|
||
let claimed = insert_file(&tx, &row).unwrap().expect("unique path");
|
||
row.name = "unclaimed.mp4";
|
||
row.path = "/tmp/unclaimed.mp4";
|
||
row.mime = Some("video/mp4");
|
||
row.needs_content = false;
|
||
let unclaimed = insert_file(&tx, &row).unwrap().expect("unique path");
|
||
|
||
let state = |id: i64| -> i64 {
|
||
tx.query_row(
|
||
"SELECT content_state FROM files WHERE id = ?1",
|
||
params![id],
|
||
|r| r.get(0),
|
||
)
|
||
.unwrap()
|
||
};
|
||
assert_eq!(state(claimed), STATE_PENDING);
|
||
assert_eq!(state(unclaimed), STATE_NA);
|
||
|
||
drop(tx);
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
#[test]
|
||
fn update_writes_content_state_from_needs_content() {
|
||
let p = tmp_path();
|
||
let mut conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
let mut row = NewFile {
|
||
name: "a.txt",
|
||
path: "/tmp/a.txt",
|
||
parent: "/tmp",
|
||
size: 10,
|
||
mtime: 1,
|
||
inode: None,
|
||
device_id: None,
|
||
mime: None,
|
||
ftype: FileType::EMPTY,
|
||
hash: None,
|
||
needs_content: false,
|
||
};
|
||
let id = {
|
||
let tx = conn.transaction().unwrap();
|
||
let id = insert_file(&tx, &row).unwrap().expect("unique path");
|
||
set_content_done(&tx, id, "a.txt", "old text", &[], true).unwrap();
|
||
tx.commit().unwrap();
|
||
id
|
||
};
|
||
|
||
let content_state = |conn: &Connection| -> i64 {
|
||
conn.query_row(
|
||
"SELECT content_state FROM files WHERE id = ?1",
|
||
params![id],
|
||
|r| r.get(0),
|
||
)
|
||
.unwrap()
|
||
};
|
||
|
||
// Rewritten as something an extractor claims: back to pending, and the
|
||
// stale FTS row goes with it.
|
||
{
|
||
let tx = conn.transaction().unwrap();
|
||
row.size = 20;
|
||
row.mtime = 2;
|
||
row.mime = Some("text/plain");
|
||
row.ftype = FileType::TEXT;
|
||
row.needs_content = true;
|
||
let got = update_file_basic(&tx, &row).unwrap();
|
||
assert_eq!(got, Some(id));
|
||
tx.commit().unwrap();
|
||
}
|
||
let basic: i64 = conn
|
||
.query_row(
|
||
"SELECT basic_state FROM files WHERE id = ?1",
|
||
params![id],
|
||
|r| r.get(0),
|
||
)
|
||
.unwrap();
|
||
assert_eq!(basic, STATE_DONE);
|
||
assert_eq!(content_state(&conn), STATE_PENDING);
|
||
|
||
let fts_hits: i64 = conn
|
||
.query_row(
|
||
"SELECT COUNT(*) FROM searchabletext WHERE searchabletext MATCH 'old'",
|
||
[],
|
||
|r| r.get(0),
|
||
)
|
||
.unwrap();
|
||
assert_eq!(fts_hits, 0);
|
||
|
||
// And rewritten as something nothing claims: NA, not pending. Without
|
||
// this the row would re-enter the content pass on every run.
|
||
{
|
||
let tx = conn.transaction().unwrap();
|
||
row.mtime = 3;
|
||
row.mime = Some("video/mp4");
|
||
row.needs_content = false;
|
||
update_file_basic(&tx, &row)
|
||
.unwrap()
|
||
.expect("row still there");
|
||
tx.commit().unwrap();
|
||
}
|
||
assert_eq!(content_state(&conn), STATE_NA);
|
||
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
#[test]
|
||
fn insert_file_twice_on_same_path_is_idempotent() {
|
||
// Defense-in-depth: when the walker visits a canonical path twice
|
||
// (overlapping roots, symlink resolution quirks), the second INSERT
|
||
// must be a silent no-op, not a run-ending error.
|
||
let p = tmp_path();
|
||
let mut conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
let tx = conn.transaction().unwrap();
|
||
let row = NewFile {
|
||
name: "dup.txt",
|
||
path: "/tmp/dup.txt",
|
||
parent: "/tmp",
|
||
size: 1,
|
||
mtime: 1,
|
||
inode: None,
|
||
device_id: None,
|
||
mime: Some("text/plain"),
|
||
ftype: FileType::TEXT,
|
||
hash: None,
|
||
needs_content: true,
|
||
};
|
||
let id1 = insert_file(&tx, &row).unwrap().expect("first insert");
|
||
let id2 = insert_file(&tx, &row).unwrap();
|
||
assert!(id2.is_none(), "second insert of same path must return None");
|
||
// Only one row exists.
|
||
let count: i64 = tx
|
||
.query_row("SELECT COUNT(*) FROM files", [], |r| r.get(0))
|
||
.unwrap();
|
||
assert_eq!(count, 1);
|
||
let (id_read,): (i64,) = tx
|
||
.query_row(
|
||
"SELECT id FROM files WHERE path = ?1",
|
||
params!["/tmp/dup.txt"],
|
||
|r| Ok((r.get(0)?,)),
|
||
)
|
||
.unwrap();
|
||
assert_eq!(id_read, id1);
|
||
tx.commit().unwrap();
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
#[test]
|
||
fn delete_subtree_clears_every_dependent_table() {
|
||
let p = tmp_path();
|
||
let mut conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
let add = |tx: &Transaction<'_>, path: &str| -> i64 {
|
||
let name = path.rsplit('/').next().unwrap();
|
||
let parent = &path[..path.rfind('/').unwrap()];
|
||
let id = insert_file(
|
||
tx,
|
||
&NewFile {
|
||
name,
|
||
path,
|
||
parent,
|
||
size: 1,
|
||
mtime: 1,
|
||
inode: None,
|
||
device_id: None,
|
||
mime: Some("text/plain"),
|
||
ftype: FileType::TEXT,
|
||
hash: None,
|
||
needs_content: true,
|
||
},
|
||
)
|
||
.unwrap()
|
||
.expect("unique path");
|
||
set_content_done(tx, id, name, "body text", &[("k".into(), "v".into())], true).unwrap();
|
||
id
|
||
};
|
||
|
||
{
|
||
let tx = conn.transaction().unwrap();
|
||
add(&tx, "/tree/a.txt");
|
||
add(&tx, "/tree/deep/b.txt");
|
||
let failed = add(&tx, "/tree/deep/c.txt");
|
||
set_content_failed(&tx, failed, "bad parse").unwrap();
|
||
// Outside the range: a prefix sibling, and a LIKE-metacharacter
|
||
// neighbour that a `LIKE 'tree_%'` sweep would have swallowed.
|
||
add(&tx, "/tree2/keep.txt");
|
||
add(&tx, "/treeX/keep.txt");
|
||
tx.commit().unwrap();
|
||
}
|
||
|
||
let range = crate::file_handling::ExtractCursor::for_root("/tree");
|
||
let removed = {
|
||
let tx = conn.transaction().unwrap();
|
||
// The directory row itself does not exist (only files are indexed),
|
||
// so everything here comes from the range.
|
||
let n = delete_subtree(&tx, &range.lo, &range.hi).unwrap();
|
||
tx.commit().unwrap();
|
||
n
|
||
};
|
||
assert_eq!(removed, 3, "three files under /tree");
|
||
|
||
let count = |sql: &str| -> i64 { conn.query_row(sql, [], |r| r.get(0)).unwrap() };
|
||
assert_eq!(count("SELECT COUNT(*) FROM files"), 2, "siblings survive");
|
||
assert_eq!(count("SELECT COUNT(*) FROM searchabletext"), 2);
|
||
assert_eq!(count("SELECT COUNT(*) FROM documents_text"), 2);
|
||
assert_eq!(count("SELECT COUNT(*) FROM properties"), 2);
|
||
assert_eq!(
|
||
count("SELECT COUNT(*) FROM failed_files"),
|
||
0,
|
||
"the failed row went with its file"
|
||
);
|
||
|
||
let survivors: Vec<String> = {
|
||
let mut stmt = conn
|
||
.prepare("SELECT path FROM files ORDER BY path")
|
||
.unwrap();
|
||
let v = stmt
|
||
.query_map([], |r| r.get::<_, String>(0))
|
||
.unwrap()
|
||
.map(|r| r.unwrap())
|
||
.collect();
|
||
v
|
||
};
|
||
assert_eq!(survivors, vec!["/tree2/keep.txt", "/treeX/keep.txt"]);
|
||
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
/// The walk's row prefetcher runs `dir_rows` once per directory, against a
|
||
/// deliberately tiny page cache, so it must not have to touch the table
|
||
/// heap at all. `idx_files_parent` carries `name` and `mtime` for exactly
|
||
/// this; trimming it back to `(parent)` would silently reintroduce a row
|
||
/// fetch per entry.
|
||
#[test]
|
||
fn dir_rows_is_served_entirely_from_the_index() {
|
||
let p = tmp_path();
|
||
let conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
let plan: String = conn
|
||
.query_row(
|
||
"EXPLAIN QUERY PLAN SELECT name, mtime FROM files WHERE parent = ?1",
|
||
params!["/some/dir"],
|
||
|r| r.get(3),
|
||
)
|
||
.unwrap();
|
||
assert!(
|
||
plan.contains("COVERING INDEX idx_files_parent"),
|
||
"dir_rows must be index-only, got: {}",
|
||
plan
|
||
);
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
/// The whole point of the range form: it is an index seek. `LIKE … ESCAPE`
|
||
/// cannot use the index (SQLite disables the LIKE optimisation whenever an
|
||
/// ESCAPE clause is present), so the old sweep read every path in the table
|
||
/// on every deletion event.
|
||
#[test]
|
||
fn the_subtree_range_is_an_index_seek_not_a_scan() {
|
||
let p = tmp_path();
|
||
let conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
let plan: String = conn
|
||
.query_row(
|
||
"EXPLAIN QUERY PLAN DELETE FROM files WHERE path >= ?1 AND path < ?2",
|
||
params!["/tree/", "/tree0"],
|
||
|r| r.get(3),
|
||
)
|
||
.unwrap();
|
||
assert!(
|
||
plan.contains("SEARCH") && !plan.contains("SCAN"),
|
||
"range delete must seek, got: {}",
|
||
plan
|
||
);
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
#[test]
|
||
fn last_full_index_round_trip() {
|
||
let p = tmp_path();
|
||
let conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
assert_eq!(get_last_full_index(&conn), None, "fresh DB has no marker");
|
||
set_last_full_index(&conn, 1_700_000_123).unwrap();
|
||
assert_eq!(get_last_full_index(&conn), Some(1_700_000_123));
|
||
// Overwrite, not accumulate.
|
||
set_last_full_index(&conn, 1_700_000_999).unwrap();
|
||
assert_eq!(get_last_full_index(&conn), Some(1_700_000_999));
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
#[test]
|
||
fn checkpoint_and_close_truncates_wal() {
|
||
let p = tmp_path();
|
||
let mut conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
{
|
||
let tx = conn.transaction().unwrap();
|
||
insert_file(
|
||
&tx,
|
||
&NewFile {
|
||
name: "w.txt",
|
||
path: "/tmp/w.txt",
|
||
parent: "/tmp",
|
||
size: 1,
|
||
mtime: 1,
|
||
inode: None,
|
||
device_id: None,
|
||
mime: None,
|
||
ftype: FileType::EMPTY,
|
||
hash: None,
|
||
needs_content: false,
|
||
},
|
||
)
|
||
.unwrap();
|
||
tx.commit().unwrap();
|
||
}
|
||
checkpoint_and_close(conn);
|
||
// After a TRUNCATE checkpoint + close of the last connection the WAL
|
||
// sidecar is gone or empty; the row lives in the main file.
|
||
let wal = std::path::PathBuf::from(format!("{}-wal", p.display()));
|
||
let wal_len = std::fs::metadata(&wal).map(|m| m.len()).unwrap_or(0);
|
||
assert_eq!(wal_len, 0, "WAL should be truncated on clean close");
|
||
let conn = crate::db::open_existing(p.to_str().unwrap(), false).unwrap();
|
||
let n: i64 = conn
|
||
.query_row("SELECT COUNT(*) FROM files", [], |r| r.get(0))
|
||
.unwrap();
|
||
assert_eq!(n, 1);
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
/// Bulk rows, cheap to write, enough of them to make the file grow.
|
||
fn seed_rows(conn: &mut Connection, range: std::ops::Range<usize>) {
|
||
let tx = conn.transaction().unwrap();
|
||
for i in range {
|
||
let path = format!("/tmp/bulk/{}.txt", i);
|
||
let name = format!("{}.txt", i);
|
||
let id = insert_file(
|
||
&tx,
|
||
&NewFile {
|
||
name: &name,
|
||
path: &path,
|
||
parent: "/tmp/bulk",
|
||
size: 1,
|
||
mtime: 1,
|
||
inode: None,
|
||
device_id: None,
|
||
mime: Some("text/plain"),
|
||
ftype: FileType::TEXT,
|
||
hash: None,
|
||
needs_content: true,
|
||
},
|
||
)
|
||
.unwrap()
|
||
.unwrap();
|
||
set_content_done(
|
||
&tx,
|
||
id,
|
||
&name,
|
||
&"lorem ipsum dolor sit amet ".repeat(64),
|
||
&[],
|
||
true,
|
||
)
|
||
.unwrap();
|
||
}
|
||
tx.commit().unwrap();
|
||
}
|
||
|
||
fn wal_bytes(p: &std::path::Path) -> u64 {
|
||
std::fs::metadata(format!("{}-wal", p.display()))
|
||
.map(|m| m.len())
|
||
.unwrap_or(0)
|
||
}
|
||
|
||
#[test]
|
||
fn checkpoint_truncate_empties_the_log() {
|
||
let p = tmp_path();
|
||
let mut conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
seed_rows(&mut conn, 0..200);
|
||
assert!(wal_bytes(&p) > 0, "the writes should be sitting in the log");
|
||
|
||
checkpoint_truncate(&conn).expect("nothing is holding the log back");
|
||
assert_eq!(wal_bytes(&p), 0, "a completed TRUNCATE leaves no log");
|
||
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
/// The case the old `execute_batch("PRAGMA wal_checkpoint(TRUNCATE)")`
|
||
/// reported as success: a reader pins the log, SQLite declines to reset
|
||
/// it, and the only word of it is in the result row.
|
||
#[test]
|
||
fn checkpoint_truncate_reports_an_incomplete_checkpoint() {
|
||
let p = tmp_path();
|
||
let mut writer = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
seed_rows(&mut writer, 0..200);
|
||
// Long enough to prove the point, short enough not to stall the suite.
|
||
writer
|
||
.busy_timeout(std::time::Duration::from_millis(100))
|
||
.unwrap();
|
||
|
||
let reader = crate::db::open_existing(p.to_str().unwrap(), false).unwrap();
|
||
let mut stmt = reader.prepare("SELECT id FROM files").unwrap();
|
||
let mut rows = stmt.query([]).unwrap();
|
||
rows.next().unwrap().expect("a row to hold the snapshot on");
|
||
|
||
let err = checkpoint_truncate(&writer).expect_err("a reader holds the log open");
|
||
assert!(err.contains("incomplete"), "unexpected message: {}", err);
|
||
assert!(wal_bytes(&p) > 0, "and the log is still there");
|
||
|
||
drop(rows);
|
||
drop(stmt);
|
||
drop(reader);
|
||
drop(writer);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
/// The mechanism the whole in-run checkpoint exists for.
|
||
///
|
||
/// SQLite's autocheckpoint copies committed frames into the database
|
||
/// continuously, but the log is only *reset* when the writer opens a
|
||
/// transaction at an instant no reader holds a read mark — and it tries
|
||
/// that lock exactly once, with no retry. A reader querying back to back,
|
||
/// which is what an indexing run's per-root prefetcher is, keeps that
|
||
/// instant from arriving, so the log appends for as long as the run lasts.
|
||
/// An explicit checkpoint retries the same lock under `busy_timeout` and
|
||
/// gets it.
|
||
#[test]
|
||
fn a_busy_reader_defeats_the_autocheckpoint_but_not_a_forced_one() {
|
||
// Bare `files` rows: no text, so no zstd and no tokenising. The log
|
||
// grows from committed frames, and frames are what starves the reset —
|
||
// paying for extraction here would only make the test slow.
|
||
fn seed_bare(conn: &mut Connection, range: std::ops::Range<usize>) {
|
||
let tx = conn.transaction().unwrap();
|
||
for i in range {
|
||
let path = format!("/tmp/bare/{}.txt", i);
|
||
let name = format!("{}.txt", i);
|
||
insert_file(
|
||
&tx,
|
||
&NewFile {
|
||
name: &name,
|
||
path: &path,
|
||
parent: "/tmp/bare",
|
||
size: i as u64,
|
||
mtime: 1,
|
||
inode: None,
|
||
device_id: None,
|
||
mime: None,
|
||
ftype: FileType::TEXT,
|
||
hash: Some(&[0u8; 32]),
|
||
needs_content: false,
|
||
},
|
||
)
|
||
.unwrap();
|
||
}
|
||
tx.commit().unwrap();
|
||
}
|
||
|
||
fn run(p: &std::path::Path, force_every: usize) -> u64 {
|
||
let mut conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
seed_bare(&mut conn, 0..500);
|
||
|
||
// Stands in for a walk prefetcher: short reads, no gaps.
|
||
let stop: Arc<AtomicBool> = Arc::new(AtomicBool::new(false));
|
||
let reader = {
|
||
let (path, stop) = (p.to_path_buf(), stop.clone());
|
||
std::thread::spawn(move || {
|
||
let conn = crate::db::open_existing(path.to_str().unwrap(), false).unwrap();
|
||
while !stop.load(Ordering::Relaxed) {
|
||
let _: i64 = conn
|
||
.query_row(
|
||
"SELECT COUNT(*) FROM files WHERE parent = '/tmp/bare'",
|
||
[],
|
||
|r| r.get(0),
|
||
)
|
||
.unwrap();
|
||
}
|
||
})
|
||
};
|
||
|
||
let mut peak = 0u64;
|
||
for batch in 0..60 {
|
||
let lo = 1000 + batch * 300;
|
||
seed_bare(&mut conn, lo..lo + 300);
|
||
if force_every > 0 && batch % force_every == force_every - 1 {
|
||
let _ = checkpoint_truncate(&conn);
|
||
}
|
||
peak = peak.max(wal_bytes(p));
|
||
}
|
||
|
||
stop.store(true, Ordering::Relaxed);
|
||
reader.join().unwrap();
|
||
drop(conn);
|
||
peak
|
||
}
|
||
|
||
let unbounded = tmp_path();
|
||
let left_alone = run(&unbounded, 0);
|
||
std::fs::remove_file(&unbounded).ok();
|
||
|
||
let bounded = tmp_path();
|
||
let forced = run(&bounded, 4);
|
||
std::fs::remove_file(&bounded).ok();
|
||
|
||
eprintln!(
|
||
"peak WAL: autocheckpoint only {}, forced {}",
|
||
left_alone, forced
|
||
);
|
||
assert!(
|
||
forced * 2 < left_alone,
|
||
"forcing checkpoints did not bound the log: {} vs {}",
|
||
forced,
|
||
left_alone
|
||
);
|
||
}
|
||
|
||
#[test]
|
||
fn maintain_vacuums_when_slack_is_significant() {
|
||
let p = tmp_path();
|
||
let mut conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
seed_rows(&mut conn, 0..2000);
|
||
checkpoint_truncate(&conn).unwrap();
|
||
let before = std::fs::metadata(&p).unwrap().len();
|
||
|
||
{
|
||
let tx = conn.transaction().unwrap();
|
||
for i in 0..1900 {
|
||
delete_file_by_path(&tx, &format!("/tmp/bulk/{}.txt", i)).unwrap();
|
||
}
|
||
tx.commit().unwrap();
|
||
}
|
||
drop(conn);
|
||
|
||
let conn = crate::db::open::open_maintenance(p.to_str().unwrap()).unwrap();
|
||
let freelist: i64 = conn
|
||
.query_row("PRAGMA freelist_count", [], |r| r.get(0))
|
||
.unwrap();
|
||
assert!(freelist > 0, "the deletions should have freed pages");
|
||
|
||
let dir = p.parent().unwrap().to_string_lossy().into_owned();
|
||
assert!(
|
||
maintain(&conn, &dir).unwrap(),
|
||
"that much slack is worth a vacuum"
|
||
);
|
||
assert_eq!(
|
||
wal_bytes(&p),
|
||
0,
|
||
"the vacuum's own writes are checkpointed too"
|
||
);
|
||
assert!(
|
||
std::fs::metadata(&p).unwrap().len() < before,
|
||
"the file should have shrunk"
|
||
);
|
||
// The surviving rows are still there and still searchable.
|
||
let n: i64 = conn
|
||
.query_row("SELECT COUNT(*) FROM files", [], |r| r.get(0))
|
||
.unwrap();
|
||
assert_eq!(n, 100);
|
||
let hits: i64 = conn
|
||
.query_row(
|
||
"SELECT COUNT(*) FROM searchabletext WHERE searchabletext MATCH 'lorem'",
|
||
[],
|
||
|r| r.get(0),
|
||
)
|
||
.unwrap();
|
||
assert_eq!(hits, 100, "the FTS index survived the rewrite");
|
||
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
#[test]
|
||
fn maintain_skips_vacuum_on_a_tight_file() {
|
||
let p = tmp_path();
|
||
let mut conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
seed_rows(&mut conn, 0..200);
|
||
drop(conn);
|
||
|
||
let conn = crate::db::open::open_maintenance(p.to_str().unwrap()).unwrap();
|
||
let dir = p.parent().unwrap().to_string_lossy().into_owned();
|
||
assert!(
|
||
!maintain(&conn, &dir).unwrap(),
|
||
"a file with no slack is not worth rewriting"
|
||
);
|
||
// The checkpoint is not conditional on the vacuum, though.
|
||
assert_eq!(wal_bytes(&p), 0);
|
||
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
|
||
#[test]
|
||
fn set_content_failed_writes_failed_table() {
|
||
let p = tmp_path();
|
||
let mut conn = open_or_recreate(p.to_str().unwrap(), "trigram").unwrap();
|
||
let id = {
|
||
let tx = conn.transaction().unwrap();
|
||
let id = insert_file(
|
||
&tx,
|
||
&NewFile {
|
||
name: "oops.bin",
|
||
path: "/tmp/oops.bin",
|
||
parent: "/tmp",
|
||
size: 0,
|
||
mtime: 1,
|
||
inode: None,
|
||
device_id: None,
|
||
mime: None,
|
||
ftype: FileType::EMPTY,
|
||
hash: None,
|
||
needs_content: false,
|
||
},
|
||
)
|
||
.unwrap()
|
||
.expect("unique path");
|
||
set_content_failed(&tx, id, "bad parse").unwrap();
|
||
tx.commit().unwrap();
|
||
id
|
||
};
|
||
|
||
let reason: String = conn
|
||
.query_row(
|
||
"SELECT reason FROM failed_files WHERE file_id = ?1",
|
||
params![id],
|
||
|r| r.get(0),
|
||
)
|
||
.unwrap();
|
||
assert_eq!(reason, "bad parse");
|
||
let content_state: i64 = conn
|
||
.query_row(
|
||
"SELECT content_state FROM files WHERE id = ?1",
|
||
params![id],
|
||
|r| r.get(0),
|
||
)
|
||
.unwrap();
|
||
assert_eq!(content_state, STATE_FAILED);
|
||
|
||
drop(conn);
|
||
std::fs::remove_file(&p).ok();
|
||
}
|
||
}
|