use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::sync::{mpsc, oneshot};
use crate::error::{classify, BulkInterrupted, BulkResult, DbError, Result, WriteOp};
use crate::graph::edge::EdgeAssertion;
use crate::graph::lineage::{Ancestor, LineageShape, Lineages};
use crate::integrity::{rebuild_current, RebuildReport};
use crate::plan::ReadPlan;
use crate::schema::migrations;
use crate::temporal::archive::{archive, rehydrate, ArchiveReport, RehydrateReport};
use crate::temporal::interval::Interval;
use crate::temporal::snapshot::{self, SnapshotCadence};
use crate::util::clock::FutureStampPolicy;
use crate::util::clock::{Clock, SystemClock};
use crate::util::timestamp;
use crate::vector::ModelName;
/// Rows per chunk on the background write paths (§5.1.5, D-011, D-014, D-058).
///
/// The Write Actor holds the sole write connection, so a single large statement
/// blocks every other writer for its duration. Chunking bounds that stall; the
/// cost is that a bulk import is *not* atomic across chunks, which is why
/// all-or-nothing is [`Database::write_bulk_atomic`] — a separate entry point,
/// with its own command on the actor's protocol — rather than a tuning
/// parameter here.
///
/// # Why these are four constants and not one
///
/// Through 0.5.5 this was a single `CHUNK_ROWS = 1000` for all four bulk paths.
/// The golden rule it was meant to serve is a bound on *duration* — a background
/// chunk must commit fast enough that an interactive write queued behind it is
/// not made to wait — and one row count cannot express one duration across paths
/// whose measured per-row costs differ by 60× (D-058). At 1,000 rows the four
/// paths took 3.5 ms, 24 ms, 89 ms and 143 ms: the same constant, four answers,
/// three of them far outside the bound.
///
/// Each size below is derived from `benches/budgets.rs`'s `chunk_scaling`
/// sweep against [`CHUNK_BUDGET`], then verified by measuring that size directly.
/// They are *measurements of this machine*, not universal constants — D-055's
/// reasoning about reference hardware applies here too, and re-deriving them on
/// materially different storage is a `cargo bench` away.
///
/// # Sized for the tail, not the median
///
/// The first derivation solved `f + c·n = 3 ms` exactly and produced sizes whose
/// *median* commit was 2.93 ms and whose upper estimate was 2.96 — inside the
/// bound as reported and outside it for any chunk slower than typical. A latency
/// bound is a statement about the chunk an unlucky interactive write actually
/// queues behind, so these solve for ≈2.5 ms instead, leaving the remainder as
/// headroom for the tail. That costs a few percent of throughput on the two
/// linear paths and nothing on the two superlinear ones.
///
/// As measured by `chunk_budget`, each at its own size: edges **2.39 ms**,
/// concepts **2.35 ms**, annotations **2.36 ms**, embeddings **2.06 ms**, no
/// upper estimate above 2.42.
///
/// # Known limitation: these are empty-database figures
///
/// `chunk_budget` seeds concepts and starts with **no links and no vectors**,
/// and D-059 established that per-row cost on the edge and embedding paths grows
/// with the size of the structure being written, not with the chunk. The same
/// 90-edge chunk takes **9.06 ms** into an 8,000-edge table. So the bound is met
/// as measured here and *not* met on a populated database.
///
/// That gap was published as 47.7 ms until 0.10.0 and attributed to the schema
/// defect D-059 documents. The defect was fixed by the `v5 → v6` rung and the
/// figure was never updated. 9.08 ms is a 0.10.0 measurement, not D-059's 8.0 ms
/// carried forward: `chunk_budget` gained a seeded arm, because until it did,
/// nothing in the bench suite wrote a chunk into a populated table and this
/// number was unfalsifiable. It agrees with D-059 once the session is accounted
/// for — the empty arm read 2.69 and 2.65 ms beside it against the 2.39 ms
/// published above, so the *ratio* is 3.4× here and 3.35× there.
///
/// **The residual is attributed as of 0.11.0 (D-142).** It is not the missing
/// index, which shipped in 0.5.6; it is the `links_current` write. Dropping the
/// three `links` insert triggers one at a time puts effectively all of the
/// growth in `trg_links_current_sync` — the single-open guard contributes none,
/// the log trigger and the base insert ~0.35 ms of a 4.15 ms rise — and within
/// that trigger, 89% of the growth is maintenance of `idx_lc_traversal_cover`
/// and `idx_lc_open_interval` rather than the upsert itself, which costs 0.49 ms
/// run directly against the same table. Page-cache size, foreign keys and the
/// fixture's key distribution were each tested and are each not the cause.
///
/// Knowing the cause does not by itself change the constant: the expensive index
/// is D-042's covering index for the traversal, so narrowing it moves cost onto
/// the read path it exists to protect. Re-deriving these constants against the
/// D-088 fixture matrix is the named successor.
///
/// # These are ceilings as of 0.12.0, not sizes
///
/// D-143 re-derived all four against the D-088 matrix and the edge path came
/// back **20** against a shipped 90 — and 20 would have been wrong at 80,000
/// edges for the same reason 90 is wrong at 8,000, because per-row cost there
/// grows with `links_current`. The finding was that no row count can bound a
/// duration on such a path.
///
/// So the chunk loop stopped trying to pick one ahead of time. Each chunk is
/// timed by the actor and its measured hold chooses the next size; these
/// constants are the **largest** size that will ever be asked for, and every
/// derivation below still applies to them as such. A path may run well under its
/// constant on a populated database and at exactly it on an empty one, and both
/// are the bound being met rather than a size being missed.
pub mod chunk_rows {
/// Edge assertions (`bulk_import`).
///
/// Per-row cost on this path rises with the size of `links_current`, not
/// with the chunk (D-059) — so cutting the chunk buys latency and costs
/// throughput, ~11% for 1,000 edges. An earlier version of this comment
/// claimed it was 3.3× *faster*; that came from multiplying eleven copies of
/// a chunk measured into an empty database.
///
/// **This size does not meet the 3 ms bound on a populated database.** 90
/// edges into an 8,000-edge table take **9.06 ms** — measured, two sessions
/// at 9.08 and 9.05, against an empty-table arm of 2.69 and 2.65 beside
/// them (D-136).
///
/// The reason given here until 0.10.0 — that `trg_links_single_open`'s
/// `EXISTS` scans the whole out-degree, "a schema defect with a proven fix,
/// recorded in D-059 and not applied here" — described 0.5.5. The fix *was*
/// applied, as the `v5 → v6` rung, and took this from 47.7 ms to ~8 ms.
/// What survives is the miss: the bound is still exceeded ~3×. Its cause is
/// no longer unknown — D-142 attributes it to `trg_links_current_sync`, and
/// within that to secondary-index maintenance on `links_current` — and the
/// guard this comment used to blame contributes **no** growth at all.
///
/// **The constant is unchanged, and that is now a measured decision**
/// (D-143). Re-derived against all four D-088 shapes at 8,000 edges, they
/// agree that the largest size meeting the bound is **20**. It stays at 90
/// because 20 is the same miss at a larger population — per-row cost grows
/// with `links_current`, so a constant fitted at 8,000 edges is wrong at
/// 80,000 — while the throughput cost of turning eleven chunks into fifty
/// is certain and immediate (D-058). The fix is not a row count: it is for
/// the chunk loop to stop on elapsed time, **delivered in 0.12.0**. This
/// number is now the ceiling that loop starts from and never exceeds; on a
/// populated table it converges below it within a chunk or two.
///
/// D-134 retired the growth claim on the neighbouring *single-assertion*
/// path and did not measure this one; D-136 is why this line now carries a
/// measurement rather than a figure quoted from 0.5.6.
pub const EDGES: usize = 90;
/// Concept upserts (`write_concepts`).
///
/// Linear at ~23 µs per row, so unlike [`EDGES`] this size *is* a genuine
/// throughput sacrifice: 1,000-row chunks ran at 23.6 µs per row against
/// ~35 µs here. Paid deliberately — a 1,000-row chunk takes 24 ms, eight
/// times the bound.
pub const CONCEPTS: usize = 70;
/// Analytics annotations (`write_analytics_annotations`).
///
/// The one path where the old constant was nearly right, and the only bulk
/// table with no triggers at all: ~2.5 µs per row, linear, so the bound buys
/// a large chunk. 1,000 rows would be 3.5 ms — over, but only just.
pub const ANNOTATIONS: usize = 600;
/// Embedding vectors (`upsert_embeddings`).
///
/// The smallest by a wide margin, because DiskANN index maintenance makes an
/// embedding the most expensive row in the system. That cost grows with the
/// **corpus**, not the chunk (D-059): a fixed 30-vector chunk costs 49 µs per
/// vector into an empty corpus and 224 µs into an 8,000-vector one. Graph
/// insertion getting dearer as the graph grows is what DiskANN is, so unlike
/// [`EDGES`] there is nothing here to fix — but it does mean this size buys
/// latency at some throughput, not for free.
pub const EMBEDDINGS: usize = 30;
}
/// What one chunk transaction cost, reported by the actor to the caller-side
/// chunk loop (0.12.0, W1).
///
/// `held` is measured **inside** the actor, around its own transaction, and
/// therefore excludes the time the command spent queued. That exclusion is the
/// point: queue time is what strict preemption *does*, and a controller fed
/// `send + await` would shrink chunks as punishment for the actor correctly
/// serving an interactive write first.
///
/// Crate-internal, along with the command enums that carry it. It was `pub`
/// through 0.13.32 only because they were (D-206).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ChunkOutcome {
/// Rows the transaction actually wrote.
pub rows: usize,
/// How long the actor held the write lock for them.
pub held: std::time::Duration,
}
/// A flag a caller can raise to stop a chunked bulk write (0.13.8, W7.6, D-181).
///
/// Cheap to clone and safe to set from any thread, which is the whole point: the
/// task running the import is the one thing that cannot cancel it. Hand a clone
/// to whatever *can* — a signal handler, a UI thread, a timeout task — and it
/// takes effect at the next chunk boundary.
///
/// **A boundary, not an abort.** Nothing rolls back and no in-flight
/// transaction is interrupted: the loop notices between chunks and stops
/// sending. The chunks that committed stay committed, and
/// [`BulkInterrupted::written`](crate::BulkInterrupted::written) says how many
/// rows those were. That is the same per-chunk boundary
/// [`Database::bulk_import`] already documents, so cancellation adds a reason to
/// stop and no new failure mode.
///
/// Setting it after the last chunk has committed does nothing — a finished
/// write reports success, because it succeeded.
#[derive(Clone, Debug, Default)]
pub struct CancelToken(Arc<std::sync::atomic::AtomicBool>);
impl CancelToken {
/// A token that has not been cancelled.
pub fn new() -> Self {
Self::default()
}
/// Ask the bulk write holding a clone of this token to stop at its next
/// chunk boundary. Idempotent; a token never un-cancels.
pub fn cancel(&self) {
// `Relaxed` on both sides is sufficient and deliberate: nothing is
// published *through* this flag. The rows are ordered by the database
// and the chunk results by the response channel, so the only thing the
// reader needs is to observe the store eventually, which every ordering
// guarantees.
self.0.store(true, std::sync::atomic::Ordering::Relaxed);
}
/// Whether [`Self::cancel`] has been called on this token or any clone.
pub fn is_cancelled(&self) -> bool {
self.0.load(std::sync::atomic::Ordering::Relaxed)
}
}
/// One chunk's worth of progress, handed to the callback on
/// [`BulkControl::on_progress`] (0.13.8, W7.6).
///
/// Reported *after* the chunk has committed, so `written` is a count of rows
/// that are in the database and will stay there even if the next chunk fails.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct BulkProgress {
/// Rows committed so far, across every chunk including this one.
pub written: usize,
/// Rows in the batch the caller passed. `written` reaching this means the
/// last chunk has committed.
pub total: usize,
/// Rows this chunk wrote. Not a constant: the loop resizes chunks against
/// [`CHUNK_BUDGET`] as it measures them (D-058).
pub rows: usize,
/// How long the actor held the write lock for this chunk — the same figure
/// the controller steers on. Measured inside the actor, around its own
/// transaction, so it excludes the time the command spent queued.
pub held: std::time::Duration,
}
/// Cancellation and progress for the four chunked bulk paths (0.13.8, W7.6,
/// D-181).
///
/// Default is "neither", which is what [`Database::bulk_import`] and its three
/// siblings pass. The `_with` variants take one of these:
///
/// ```no_run
/// # use macrame::{BulkControl, CancelToken, Database};
/// # async fn f(db: &Database, edges: Vec<macrame::prelude::EdgeAssertion>) {
/// let token = CancelToken::new();
/// let stopper = token.clone();
/// tokio::spawn(async move {
/// tokio::time::sleep(std::time::Duration::from_secs(30)).await;
/// stopper.cancel();
/// });
///
/// let control = BulkControl::new()
/// .cancel_with(token)
/// .on_progress(|p| println!("{}/{} rows", p.written, p.total));
///
/// match db.bulk_import_with(edges, control).await {
/// Ok(n) => println!("imported {n}"),
/// Err(e) => println!("stopped after {}: {}", e.written, e.cause),
/// }
/// # }
/// ```
///
/// **The callback runs on the importing task, between chunks.** It is therefore
/// on the critical path: whatever it does is time the next chunk is not being
/// sent in. Printing or updating a counter is what it is for; a blocking write
/// is not, and neither is anything that calls back into the same `Database`,
/// which would deadlock the loop against a channel it is itself draining.
#[derive(Default, Clone)]
pub struct BulkControl {
cancel: Option<CancelToken>,
on_progress: Option<Arc<dyn Fn(BulkProgress) + Send + Sync>>,
}
impl std::fmt::Debug for BulkControl {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("BulkControl")
.field("cancel", &self.cancel)
.field("on_progress", &self.on_progress.is_some())
.finish()
}
}
impl BulkControl {
/// Neither cancellation nor progress — what the plain bulk methods pass.
pub fn new() -> Self {
Self::default()
}
/// Stop at the next chunk boundary when `token` is cancelled.
pub fn cancel_with(mut self, token: CancelToken) -> Self {
self.cancel = Some(token);
self
}
/// Call `f` after every chunk commits. See the note on [`BulkControl`]
/// about what this closure is allowed to do.
pub fn on_progress(mut self, f: impl Fn(BulkProgress) + Send + Sync + 'static) -> Self {
self.on_progress = Some(Arc::new(f));
self
}
fn is_cancelled(&self) -> bool {
self.cancel.as_ref().is_some_and(CancelToken::is_cancelled)
}
fn report(&self, progress: BulkProgress) {
if let Some(f) = &self.on_progress {
f(progress);
}
}
}
/// Smallest chunk the adaptive loop will fall to (0.12.0, W2).
///
/// # A floor is a deliberate, measured violation of [`CHUNK_BUDGET`]
///
/// Feedback alone converges to whatever size meets the budget, and on a
/// populated `links` table that size keeps falling — per-row cost there grows
/// with the table (D-059, D-142), so there is no size at which the *fixed* cost
/// of a transaction stops dominating. Left unbounded the loop reaches chunks of
/// one or two rows, where nearly all the work is `BEGIN`/`COMMIT` and the import
/// no longer finishes.
///
/// 35 is measured, and **re-measured against the loop that uses it** — the
/// difference matters, because the figure this constant shipped with was an
/// extrapolation. `examples/chunk_matrix.rs -- converge` runs a 900-edge
/// `bulk_import` into each of the four D-088 shapes at 8,000 edges and reports
/// the actor's own per-transaction readings. A 35-row chunk costs **3.11–3.43 ms**
/// across the four shapes, two sessions, excluding the run-up. The floor misses
/// the 3 ms bound by 0.1–0.4 ms, not by the ~1.1 ms predicted from the sweep.
///
/// The miss is **steady state** — not a one-chunk transient on the way down —
/// and the defense is the argument [`CHUNK_BUDGET`] is answerable to rather than
/// the number itself: an interactive assertion arriving at the worst moment
/// waits ~3.2 ms for the chunk in flight and then runs its own ≤ 5 ms write, so
/// ~8.2 ms against a 16.7 ms frame.
///
/// What the same measurement says about the *size*: on this path at this
/// population the loop goes `[90, 35, 35, …]` on all four shapes and never picks
/// anything between. The proportional shrink from a 90-row chunk proposes ~31
/// rows, which clamps here — so on the edge path the floor is not a safety net
/// under the controller, it **is** the operating point, and this number is
/// carrying more weight than a backstop normally would. Re-measure it, not the
/// controller, when the edge path's per-row cost changes.
const CHUNK_FLOOR: usize = 35;
/// Size of the next chunk, from what the last one cost (0.12.0, W2).
///
/// Pure on purpose — no clock, no database, no actor — so the control law can be
/// tested for the properties that matter without a fixture. Three regimes:
///
/// | last hold | response | why |
/// |---|---|---|
/// | over `budget` | shrink to `current · budget / held`, × 0.9 | back off *fast* from a bound already being exceeded; the 0.9 undershoots so the correction does not have to be repeated |
/// | under `budget / 2` | grow by a quarter of `current`, at least one row | approach the bound *slowly*; the dead band above it stops a size that is merely comfortable from oscillating |
/// | otherwise | hold | in band, and moving costs more than it buys |
///
/// The asymmetry is the whole design. Proportional shrinking converges from
/// above in one or two steps, which matters because every step over budget is a
/// latency miss a caller can feel; additive growth cannot overshoot by more than
/// 25%, which matters because the ceiling is a throughput preference and not a
/// bound.
///
/// `ceiling` is the path's [`chunk_rows`] constant, which is why those constants
/// keep their values and their derivations: they are no longer the size, they
/// are the largest size this will ever ask for. `floor` is [`CHUNK_FLOOR`] —
/// see there for the budget it knowingly misses.
///
/// Never returns 0, at any input, including `held == 0` or `current == 0`.
fn next_chunk_size(
current: usize,
held: std::time::Duration,
budget: std::time::Duration,
floor: usize,
ceiling: usize,
) -> usize {
let held = held.as_nanos().max(1);
let budget_ns = budget.as_nanos().max(1);
let current = current.max(1);
let next = if held > budget_ns {
// Integer math, and the `max(1)` matters: a chunk 200× over budget
// would otherwise propose 0 and the loop would stop making progress.
let scaled = (current as u128) * budget_ns * 9 / (held * 10);
(scaled as usize).max(1)
} else if held * 2 < budget_ns {
// Saturating because `current` is a `usize` and this is the one branch
// that adds to it. Nothing sane reaches the boundary; the clamp below
// makes the answer correct anyway rather than a debug panic.
current.saturating_add((current / 4).max(1))
} else {
current
};
// Applied last and unconditionally, so a caller that passes a reversed pair
// gets the floor rather than a panic — and `max(1)` last of all, because a
// chunk of zero rows is the single answer no loop can make progress from.
next.clamp(floor.min(ceiling), ceiling).max(1)
}
/// The latency bound [`chunk_rows`] is derived from (§5.1.5, D-058).
///
/// This is the golden rule's actual content. §9 has carried it as a row count
/// with a duration attached — "chunk commit, 500 rows ≤ 3 ms" — which reads as
/// two requirements and is one: the duration is the requirement, and the row
/// count is whatever satisfies it on a given path and machine.
///
/// 3 ms is §9's number, kept rather than renegotiated. What it buys, end to end:
/// an interactive assertion arriving at the worst possible moment waits for the
/// chunk in flight (≤ 3 ms — the SQLite write lock is not preemptible, so
/// priority buys the *next* turn and not this one) and then runs its own write
/// (≤ 5 ms, §9), so ≤ 8 ms
/// worst case. That fits inside a 60 Hz frame with room, which is the standard
/// this bound is ultimately answerable to.
///
/// # Some operations are exempt, and the exemption is a contract, not an oversight
///
/// This was recorded in three separate rustdoc notes and nowhere near the bound
/// itself, which is where a reader looks for its scope (§8.6). Stated here, with
/// Wave 3's measurements:
///
/// | Path | Bound | Why it cannot be chunked |
/// |---|---|---|
/// | [`Database::write_bulk_atomic`] | none — caller-sized `Vec` | D-014: the batch is *one act* under one stamp. Splitting it is the thing the method exists not to do |
/// | [`Database::archive`] | measured **26.8 ms** for 2,000 archivable edges; see [`Database::archive_windowed`] | D-012: copy-then-delete must be atomic, or a crash between the phases duplicates or loses rows |
/// | `rebuild_current` | measured **24.6 / 104 / 318 ms** at 4K / 16K / 40K rows in `links` (was "~50 s per 10M edges", which nothing had measured) | D-023: the window between `DELETE` and `INSERT` is the whole of current belief; a reader landing in it sees a graph with no edges and no error |
/// | [`Database::rehydrate`] | unmeasured; a function of how many rows the caller named | D-012 backwards: the same copy-then-delete atomicity, in the other direction. **A row here since 0.12.9 only because it was previously invisible** — rehydration reported as `archive` and inherited its exemption without anyone deciding on it (W4.3, D-152) |
/// | [`Database::archive_branch`] | unmeasured; a function of how much one lineage wrote | D-012 again, and D-230's chain: the links, the log entries and the `branches` row leave together or the ledger disagrees with itself about what is currently believed. There is no smaller unit — half a forgotten lineage is a lineage whose reads are answered by its parent |
/// | the swap turn of [`Database::rebuild_current_chunked`], counted as `shadow_swap` | measured **46.8 ms** at the largest fixture (D-082), and it grows with the table | Index names are global and SQLite has no `ALTER INDEX … RENAME`, so the shadow cannot carry `idx_lc_traversal_cover` while the live table still holds it — all three indexes are built here, under the lock. This is the residual T1.2 could not remove, and there is no smaller unit: half a swapped projection is not a projection. **Exempt since 0.14.16** (W12.16, D-233). The *fill* half keeps its own kind and is deliberately absent from this table, which is what makes a violation there a regression rather than a constant |
/// | [`Database::checkpoint`] | a function of the WAL's size, which is a function of how long since the last checkpoint — not of anything the caller passes | It is not a transaction at all. `PRAGMA wal_checkpoint` copies frames back into the main file and there is no unit smaller than the frame it is already working in; the caller asked for exactly this, and the alternative to a long checkpoint is a WAL that keeps growing (0.12.13, W5.2, D-156) |
/// | the drop turn of [`Database::bulk_embeddings`], counted as `drop_embedding_index` (0.16.2, D-276) | µs-scale; one `DROP INDEX IF EXISTS` | One statement, no smaller unit — the same shape as [`Database::checkpoint`] by nature and [`Database::write_bulk_atomic`] by atomicity. Its kind exists for attribution beside `rebuild_embedding_index`, not for cost |
/// | the rebuild turn of [`Database::bulk_embeddings`], counted as `rebuild_embedding_index` (0.16.2, D-276) | measured **2.61 / 19.7 / 39.0 s** for 2,000 vectors at dim 64 / 256 / 512, ~10 ms/vector at dim 256, growing with the corpus | One `CREATE INDEX` over the whole table — the one-pass DiskANN build is indivisible, exactly the criterion `shadow_swap` and `rebuild_current` meet. The difference is schedule: this hold is caller-scheduled and opt-in, so the docstring states the number instead of arguing it. Counted would add a permanent `N(bulk loads)` to every database that ever bulk-embedded — `shadow_swap`'s own argument, unchanged |
/// | the toggle turn of [`Database::bulk_import_deferred`], counted as `links_current_mirror` (0.16.3, D-277) | µs-scale; one DDL statement either direction | Same shape as `drop_embedding_index`: one statement, no smaller unit, and its kind exists for attribution beside the load it wraps. The window's *cost* is the chunked rebuild that follows, which keeps its own counted kinds rather than hiding behind the toggle's exemption — the toggle is not where the time goes |
///
/// The `archive` figure is end-to-end through this method, so it **includes**
/// the re-derivation `archive()` runs inside its transaction — but it does not
/// attribute it, and until D-077 more than half of that re-derivation was an
/// audit comparing `links_current` against the query that had just filled it.
/// Note also which variable that cost scales with: `rebuild_within` reprojects
/// **all of `links`**, so the archive's repair term grows with the *surviving*
/// table and not with the batch being archived. A budget stated per "100K closed
/// intervals" ([§9](../docs/architecture/s6-s10-flows-to-dependencies.md)) is
/// therefore parameterised on the wrong quantity.
///
/// The first four are atomic **by contract**, which is why "cap the batch" and
/// "add a third tier" were both considered and neither was taken: capping breaks the
/// guarantee the operation exists to provide, and a third tier changes which
/// caller waits without changing how long the lock is held. What was wrong was
/// never the exemption — it was that the bound was stated as though it had none.
///
/// A caller who needs the latency bound and not the atomicity has
/// [`Database::bulk_import`], which is the same write chunked at
/// [`chunk_rows::EDGES`] and explicitly *not* atomic overall (D-011).
///
/// # One of them is no longer unbounded (T1.1, D-080)
///
/// `archive` was the worst of them, because its hold is a function of *how long
/// since the last archive* rather than of anything the caller chose.
/// [`Database::archive_windowed`] runs the same work as N sessions, each
/// atomic, each its own actor turn. Measured on an 8,000-key fixture with four
/// generations of superseded history: the longest single hold falls from
/// **3.3 s to 0.77 s** at one-hour windows, for total wall time that is flat
/// within this cycle's noise.
///
/// The same measurement at 2,000 keys goes the other way — the hold falls
/// 260 ms → 117 ms while total time rises 260 ms → 671 ms — so windowing is a
/// trade and not a free improvement. It pays when the backlog is large, which
/// is when the unwindowed hold is a problem in the first place. `archive` is
/// kept, not deprecated, for exactly that reason.
pub const CHUNK_BUDGET: std::time::Duration = std::time::Duration::from_millis(3);
/// Predicted hold above which [`Database::write_bulk_atomic`] warns (T1.3).
///
/// 250 ms is fifteen frames at 60 Hz: not a hitch, a visible freeze. It is well
/// above [`CHUNK_BUDGET`] on purpose — this path is exempt from that bound by
/// contract, so warning at 3 ms would fire on batches that are working exactly
/// as designed and train the reader to filter the message out.
pub const BULK_ATOMIC_WARN_HOLD: std::time::Duration = std::time::Duration::from_millis(250);
/// Roughly how long [`Database::write_bulk_atomic`] will hold the actor for
/// this batch (T1.3, D-081; re-fitted 0.13.6, W7.5, D-179).
///
/// # Two terms, and the batch's shape is no longer one of them
///
/// T1.3 asks for "rows × measured per-row cost". Through 0.13.5 that was wrong
/// in a way worth a paragraph: `write_edges_atomic` opened with a
/// `reject_overlaps_within` that compared **every pair**, and the quadratic
/// term's constant depended on the batch's *shape* rather than its size, so two
/// 20,000-edge batches held the actor for **2.6 s** and **18.1 s** — a size-only
/// model was off by 7× between them, in the under-predicting direction.
///
/// W7.5 sorts and sweeps instead, and the 18.1 s batch now holds for **2.2 s**.
/// The shape term is gone from the code and therefore from here: measured on
/// the same machine, the two shapes are within 15% of each other at every size
/// from 100 to 20,000 rows, which is inside the noise this model claims.
///
/// What is left is not flat either, and the second term is why. Per-row cost
/// rises from ~36 µs at 100 rows to ~111 µs at 20,000, because each insert
/// maintains indexes and two triggers against a table the batch is itself
/// growing:
///
/// ```text
/// hold ≈ rows · (7.4 µs + 7.24 µs · ⌊log₂ rows⌋)
/// ```
///
/// # What this is calibrated against, and where it will be wrong
///
/// libSQL 0.9.30, one machine, best of three, 100–20,000 rows in both shapes;
/// within 15% from 500 rows up. Below that it under-predicts by up to 3×, which
/// is harmless in the same way the old 3× over-prediction was — nothing that
/// small approaches [`BULK_ATOMIC_WARN_HOLD`].
///
/// **The log term reads the batch because the batch is all it has.** It stands
/// for the depth of a structure the batch is loading, and this signature never
/// sees the table. That is exact for the bulk import this warns about, and
/// optimistic for a small batch appended to an already-large table — the same
/// blind spot the flat per-row model had, now visible instead of averaged away.
///
/// It is machine-specific and says nothing about disk. It exists to turn
/// "uncapped" into an order of magnitude a caller can act on, and should not be
/// read more precisely than that. `examples/bulk_atomic_diag.rs` prints
/// predicted against measured, so the model's drift is visible rather than
/// assumed.
pub fn estimated_bulk_hold(edges: &[EdgeAssertion]) -> std::time::Duration {
let rows = edges.len() as u64;
if rows == 0 {
return std::time::Duration::ZERO;
}
// Nanoseconds throughout, saturating: a caller who passes a batch large
// enough to overflow this has a problem the arithmetic cannot express, and
// saturating to ~584 years still crosses every threshold above.
let per_row = 7_400u64.saturating_add((rows.ilog2() as u64).saturating_mul(7_240));
std::time::Duration::from_nanos(rows.saturating_mul(per_row))
}
/// Most sessions [`Database::archive_windowed`] will run for one call (T1.1).
///
/// A limit exists because the session count is a function of *transaction-time
/// span divided by window*, and both come from the caller — a one-second window
/// over a decade of history is ten million actor turns, each opening a
/// transaction and writing a horizon row. That is not a slow archive, it is a
/// caller who meant something else.
///
/// 4,096 is chosen against the operation it bounds rather than against a clock:
/// at the measured 26.8 ms for a session with work in it, a full run of this
/// many is about two minutes of background writing, and the whole point of
/// windowing is that those two minutes are interruptible. It is a refusal
/// rather than a clamp — see [`DbError::ArchiveWindow`] for why.
pub const MAX_ARCHIVE_SESSIONS: usize = 4_096;
/// A concept assertion: the payload of an upsert.
///
/// `#[non_exhaustive]` since 0.14.8 for
/// [`EdgeAssertion`]'s reason: `branch` is the
/// first field added since it was written, and one break is better than a
/// recurring one.
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub struct ConceptUpsert {
pub id: String,
pub title: String,
pub content: String,
pub embedding_model: Option<String>,
pub valid_from: String,
pub valid_to: String,
pub retired: bool,
/// The lineage this concept is minted on, or `None` for the trunk (§15.2,
/// D-225).
///
/// **The rule here is narrower than the edge's, and it is the schema's
/// rather than this crate's.** `concepts` is a current-state projection
/// keyed by identity — `id` is `NOT NULL UNIQUE` — so two lineages holding
/// different beliefs about one concept is two rows with one `id`, which the
/// unique index refuses on its own. `trg_concepts_cross_lineage` turns that
/// refusal into [`DbError::CrossLineage`] so it says which rule was broken.
///
/// So a branch **inherits** its parent's concepts and cannot restate them;
/// what this field is for is a concept the branch *mints*, which is the
/// case the trunk has no row for. A branch that needs to disagree with its
/// parent about a concept's content is asking for the overlay design, which
/// is deferred with its reopen trigger named (D-214).
pub branch: Option<crate::branch::BranchId>,
}
impl ConceptUpsert {
pub fn new(id: impl Into<String>, title: impl Into<String>) -> Self {
Self {
id: id.into(),
title: title.into(),
content: String::new(),
embedding_model: None,
valid_from: String::new(),
valid_to: timestamp::OPEN_SENTINEL.to_string(),
retired: false,
branch: None,
}
}
pub fn content(mut self, content: impl Into<String>) -> Self {
self.content = content.into();
self
}
pub fn embedding_model(mut self, model: impl Into<String>) -> Self {
self.embedding_model = Some(model.into());
self
}
pub fn valid_from(mut self, ts: impl Into<String>) -> Self {
self.valid_from = ts.into();
self
}
pub fn valid_to(mut self, ts: impl Into<String>) -> Self {
self.valid_to = ts.into();
self
}
/// Mint this concept on `branch` rather than on the trunk (0.14.8).
///
/// See [`branch`](Self::branch) for why a branch may mint a concept and may
/// not restate one it inherited.
pub fn on_branch(mut self, branch: crate::branch::BranchId) -> Self {
self.branch = Some(branch);
self
}
/// The lineage this upsert names, spelled out. See
/// [`EdgeAssertion::branch_name`](crate::graph::EdgeAssertion).
pub(crate) fn branch_name(&self) -> &str {
self.branch
.as_ref()
.map_or(crate::schema::ddl::MAIN_BRANCH, |b| b.as_str())
}
pub fn retired(mut self, retired: bool) -> Self {
self.retired = retired;
self
}
/// Put the timestamps in canonical form (D-029) before they cross the channel.
pub fn normalized(mut self) -> Result<Self> {
crate::util::ids::validate_id(&self.id)?;
self.valid_from = timestamp::normalize(&self.valid_from)?;
self.valid_to = timestamp::normalize(&self.valid_to)?;
Ok(self)
}
}
/// One derived analytics result for one concept (§5.4, D-041).
///
/// Not a `ConceptUpsert`. The distinction is the whole of D-041: a concept
/// upsert is a statement about the world and belongs in the ledger, while an
/// annotation is a function of an algorithm applied to a graph and belongs in
/// `analytics_annotations`, which carries no log trigger. Writing one as the
/// other overwrote the concept's `content` with the label and recorded every
/// analytics rerun as a fresh version of the world.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct Annotation {
pub concept_id: String,
/// Namespaced by convention, e.g. `louvain.community`, `kcore.shell`.
pub label: String,
/// JSON-encoded payload. Opaque to this crate.
pub value: String,
}
impl Annotation {
pub fn new(
concept_id: impl Into<String>,
label: impl Into<String>,
value: impl Into<String>,
) -> Self {
Self {
concept_id: concept_id.into(),
label: label.into(),
value: value.into(),
}
}
}
/// Commands sent to the Write Actor on the high-priority channel (UI-driven work).
pub(crate) enum HighPriCommand {
AssertEdge {
edge: EdgeAssertion,
responder: oneshot::Sender<Result<()>>,
},
RetireEdge {
source: String,
target: String,
edge_type: String,
valid_from: String,
valid_to: String,
/// The lineage doing the retiring, or `None` for the trunk (0.14.8).
branch: Option<crate::branch::BranchId>,
responder: oneshot::Sender<Result<()>>,
},
UpsertConcept {
concept: ConceptUpsert,
responder: oneshot::Sender<Result<()>>,
},
WriteBulkAtomic {
edges: Vec<EdgeAssertion>,
responder: oneshot::Sender<Result<usize>>,
},
RebuildCurrent {
responder: oneshot::Sender<Result<RebuildReport>>,
},
/// Create a model's embedding table and its DiskANN index (D-037, D-048).
///
/// High priority despite being setup work: it is one small transaction, and
/// every embedding write for the model blocks on it, so queueing it behind a
/// bulk job would stall the thing it gates.
RegisterModel {
model: ModelName,
dim: usize,
responder: oneshot::Sender<Result<()>>,
},
/// Move WAL frames back into the main database file (§4.5, F-30, D-156).
///
/// High priority, and for once the reason is not latency: a caller asking
/// for a checkpoint is asking for it *now*, usually at the end of a bulk
/// load or before taking a copy of the file, and queueing it behind the
/// background work it was meant to follow inverts the intent. It is also
/// the only command here that is not a transaction.
Checkpoint {
responder: oneshot::Sender<Result<CheckpointReport>>,
},
/// Register a lineage (0.14.7, §15.4).
///
/// High priority, and not because it is urgent: it is one insert into a
/// table with no secondary indices, so it is the cheapest turn the actor
/// takes. What makes it high priority is that everything the caller does
/// next is a write *on* this branch, and queueing a fork behind a bulk
/// import would stall the work it exists to enable — `RegisterModel`'s
/// argument, for the same reason.
///
/// It goes through the actor rather than the read connection for the
/// ordinary reason every write does, plus one specific to it: the duplicate
/// and parent checks are only sound if nothing can register a colliding
/// name between the check and the insert, and the actor is what makes the
/// pair one turn.
Fork {
name: crate::branch::BranchId,
parent: crate::branch::BranchId,
responder: oneshot::Sender<Result<crate::branch::Branch>>,
},
Shutdown {
responder: oneshot::Sender<Result<()>>,
},
}
/// What `PRAGMA wal_checkpoint` returned (0.12.13, W5.2, D-156).
///
/// The three columns SQLite gives back, named, rather than `()` — a checkpoint
/// that did nothing and a checkpoint that reclaimed a 400 MB WAL are the same
/// `Ok(())`, and the difference is the entire reason a caller asked.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct CheckpointReport {
/// `true` when SQLite could not complete the requested mode because a
/// reader or writer was in the way.
///
/// **This is not an error, and it is not ignorable.** `TRUNCATE` waits for
/// readers only as long as `busy_timeout` allows; past that it gives up and
/// says so, having possibly still copied frames. A caller checkpointing
/// before copying the file away must read this, because a busy checkpoint
/// means the main file is not self-contained yet.
pub busy: bool,
/// Frames left in the WAL at the end. `0` when the checkpoint completed,
/// since the mode run is `TRUNCATE`.
pub log_frames: u64,
/// Frames moved back into the database file.
///
/// Read from a `FULL` pass rather than from the `TRUNCATE` — see
/// `run_checkpoint` for why a truncating checkpoint cannot report this
/// number itself.
pub checkpointed_frames: u64,
}
impl CheckpointReport {
/// The WAL was fully reclaimed: nothing blocked, and nothing is left.
pub fn is_complete(&self) -> bool {
!self.busy && self.log_frames == 0
}
}
/// Commands sent to the Write Actor on the low-priority channel (background work).
pub(crate) enum LowPriCommand {
/// One chunk of **concepts** — a ledger write, logged and versioned.
WriteConceptsChunk {
chunk: Vec<ConceptUpsert>,
responder: oneshot::Sender<Result<ChunkOutcome>>,
},
/// One chunk of **derived annotations** — off-ledger, no log trigger (D-041).
///
/// The pair is named apart deliberately: this variant was `WriteAnalyticsChunk`
/// beside a `WriteAnnotationsChunk` that carried concepts, which is the
/// crossing D-075 undid.
WriteAnalyticsChunk {
chunk: Vec<Annotation>,
responder: oneshot::Sender<Result<ChunkOutcome>>,
},
/// One chunk of vectors for one model (§5.9, D-048).
///
/// Low priority: embedding is bulk derived work and must never preempt an
/// interactive assertion.
UpsertEmbeddingChunk {
model: ModelName,
chunk: Vec<(String, Vec<f32>)>,
responder: oneshot::Sender<Result<ChunkOutcome>>,
},
/// Drop a model's DiskANN index — bulk-embedding setup (D-276).
///
/// Low priority like the chunk it serves: it is a caller's bulk load that
/// wants the index gone, and nothing interactive should queue behind it.
/// One `DROP INDEX IF EXISTS` statement, atomic by nature.
DropEmbeddingIndex {
model: ModelName,
responder: oneshot::Sender<Result<()>>,
},
/// One-pass rebuild of a model's DiskANN index — bulk-embedding finish
/// (D-276).
///
/// One `CREATE INDEX` statement, and the indivisible half of the recipe:
/// measured at 19.7 s for 2,000 vectors at dim 256 on the reference box.
/// Budget-exempt by the same criterion as `ShadowSwap` — atomic by
/// necessity — with the difference that this one is *caller-scheduled*, so
/// a slow build is a choice the caller made knowingly.
RebuildEmbeddingIndex {
model: ModelName,
responder: oneshot::Sender<Result<()>>,
},
BulkImportChunk {
chunk: Vec<EdgeAssertion>,
responder: oneshot::Sender<Result<ChunkOutcome>>,
},
/// Drop or restore `trg_links_current_sync` — the links_current mirror's
/// window, opened and closed by [`Database::bulk_import_deferred`]
/// (D-277).
///
/// Low priority beside the chunks it serves. One DDL statement either way
/// (`DROP TRIGGER IF EXISTS` down, `CREATE TRIGGER IF NOT EXISTS` up — the
/// same idempotent pair the v19 rung used on `trg_links_single_open`), so
/// the toggle is safe to send twice and the restore cannot fail on an
/// already-present trigger.
LinksCurrentMirror {
present: bool,
responder: oneshot::Sender<Result<()>>,
},
Archive {
cutoff: String,
archive_path: PathBuf,
responder: oneshot::Sender<Result<ArchiveReport>>,
},
/// Forget one lineage, moving its whole ledger to the cold file (0.14.13,
/// §15.4, D-230).
///
/// Low priority for `Archive`'s reason and one of its own: it is bulk
/// physical movement holding the write lock for its whole transaction, and
/// it is the least urgent write in the crate — the rows it moves belong to
/// a lineage nobody is reading.
ArchiveBranch {
branch: String,
archive_path: PathBuf,
responder: oneshot::Sender<Result<ArchiveReport>>,
},
/// Move named concepts back out of the cold file (0.9.0, C3).
///
/// Low priority for the same reason `Archive` is: it is bulk physical
/// movement with no latency bound, and it holds the write lock for its whole
/// transaction.
Rehydrate {
ids: Vec<String>,
archive_path: PathBuf,
responder: oneshot::Sender<Result<RehydrateReport>>,
},
/// Reconstruct the FTS index from `concepts` (§5.9, D-036, D-051).
///
/// Low priority: it is maintenance on a derivative table, and a search index
/// that is a few seconds stale is a smaller cost than an interactive write
/// that waits behind a full reindex.
RebuildFts {
responder: oneshot::Sender<Result<()>>,
},
/// Refresh or top up the query planner's statistics (0.12.4, D-149).
///
/// Low priority, and not a close call: statistics being a few seconds stale
/// costs a plan that was already the plan a moment ago, where preempting an
/// interactive assertion costs a caller their latency bound. It is a write —
/// it writes `sqlite_stat1` — so it takes the write lock like anything else,
/// and `PRAGMA analysis_limit` in `configure` is what keeps the hold a
/// function of the index count instead of the table size.
Analyze {
/// `true` runs `PRAGMA optimize`, which re-analyses only what SQLite
/// believes has gone stale; `false` runs `ANALYZE` unconditionally.
incremental: bool,
responder: oneshot::Sender<Result<()>>,
},
/// One step of a chunked shadow rebuild (§5.8, T1.2, D-082).
///
/// Low priority, and one command per step rather than one per rebuild: the
/// whole value of building beside the live table is that the actor returns
/// here between chunks. See [`Database::rebuild_current_chunked`].
ShadowRebuild {
step: crate::integrity::ShadowStep,
responder: oneshot::Sender<Result<crate::integrity::ShadowOutcome>>,
},
}
/// What the write actor knows between turns (0.15.6, W14.3, [D-248]).
///
/// [`run_writer_actor`] owned a connection and nothing else. Every command was
/// handed `&conn`, and every fact a command established died with it — so a
/// single-edge assertion asked `branches` how many lineages exist, compiled the
/// overlap guard, and compiled `INSERT_LINK`, on every call, having done all
/// three on the previous one. Measured on the trunk that is 76 µs of a 160 µs
/// write; once the database has forked it is 155 µs of a 343 µs write, because
/// the statement being compiled each time is the guard's resolved form.
///
/// **Everything here is cached for one reason and invalidated by name for the
/// same one: the actor is the only writer** (D-014). `branches` is written by
/// `Fork` and by `ArchiveBranch` and by nothing else in the crate; the
/// statements are bound to a connection this task owns for the process
/// lifetime. A cache whose only writer is holding it cannot go stale behind its
/// own back, which is why this is a plain `&mut` and not an epoch or a lock.
///
/// Built lazily rather than at open. Eager construction pays on a database that
/// never asserts an edge, and it puts fallible work in the spawn path, where
/// there is no caller to hand the error to.
///
/// # What is deliberately not here
///
/// The **hot-log intactness verdict** (review C-5). It is read on `read_conn`
/// by every recorded-time read, not by the actor, so caching it here would put
/// it on the wrong side of the process. It needs a shared cell and an
/// invalidation argument about a *reader* seeing a stale answer, which is a
/// different argument from this one and gets its own release.
///
/// The **batch paths' statements**. `write_edges_atomic` prepares inside its
/// own transaction and drops before it commits, because a live statement is
/// what makes SQLite refuse to end one. It already prepares once per chunk
/// rather than once per row (D-056, §8.8), which is where that path's cost was.
///
/// [D-248]: ../../docs/architecture/s13-decision-register.md#d-248
struct ActorState {
/// `branches`, as this actor last left it.
lineages: Option<Lineages>,
/// `INSERT_LINK`, compiled against the actor's connection.
insert_link: Option<libsql::Statement>,
/// The overlap guard, with the shape it was compiled for.
guard: Option<OverlapGuard>,
}
// `Lineages` moved to `graph::lineage` in 0.15.17 ([D-259]). It held
// `(branch_id, is_root)` here, because the shape was all the actor needed;
// resolving ancestry in Rust needs `parent_id` and `forked_at` as well, on
// both sides of the crate, and two structs answering one question from one
// table is what D-030 is about.
//
// [D-259]: ../docs/architecture/s13-decision-register.md#d-259
impl ActorState {
fn new() -> Self {
Self {
lineages: None,
insert_link: None,
guard: None,
}
}
/// Forget `branches`. Called by the two commands that write it.
fn forget_lineages(&mut self) {
self.lineages = None;
}
/// Drop the compiled statements.
///
/// Called by the commands that `ATTACH`, `DETACH`, or otherwise move the
/// schema under the connection. SQLite recompiles a statement across a
/// schema change on its own and this does not rely on that: a statement
/// dropped here costs one prepare on the next write, against a class of bug
/// whose symptom would be a stale plan on the archive path in production.
fn forget_statements(&mut self) {
self.insert_link = None;
self.guard = None;
}
/// Both of the above, for a command that does both.
fn forget_everything(&mut self) {
self.forget_lineages();
self.forget_statements();
}
async fn lineages(&mut self, conn: &libsql::Connection) -> Result<&Lineages> {
if self.lineages.is_none() {
self.lineages = Some(Lineages::load(conn).await?);
}
Ok(self
.lineages
.as_ref()
.expect("loaded on the line above or already present"))
}
/// [`Lineages::shape_of`], against the cache.
///
/// This is what replaced `check_lineages`, and the round trip it replaced
/// is the one the review counted (C-6): one `SELECT` over `branches` per
/// write, for an answer that changes when a lineage is forked or forgotten.
async fn shape_of(
&mut self,
conn: &libsql::Connection,
names: &[&str],
) -> Result<LineageShape> {
self.lineages(conn).await?.shape_of(names)
}
/// The overlap guard, compiled at most once per shape.
///
/// Keyed on the shape rather than on the statement text because that is the
/// thing [`check_prepared`] reads: it binds four parameters for `Trunk` and
/// five otherwise, so a guard held under one shape and used under another
/// would bind the wrong row even where the SQL happened to match.
async fn guard(
&mut self,
conn: &libsql::Connection,
shape: LineageShape,
branch: &str,
) -> Result<&OverlapGuard> {
// Keyed on the lineage as well as the shape since 0.15.17: the bound
// ancestry is one reader's answer, so a guard held for `main` cannot
// serve a branch that happens to share its shape. A caller writing to
// one lineage — which is every caller this crate has — still prepares
// once and keeps it across turns.
if !self
.guard
.as_ref()
.is_some_and(|g| g.answers_for(shape, branch))
{
// Resolved against the actor's own cache, then dropped, so the
// borrow ends before the assignment. `lineages` refreshes it when
// `Fork` or `ArchiveBranch` forgot it.
let lineages = self.lineages(conn).await?.clone();
self.guard = Some(OverlapGuard::prepare(conn, shape, &lineages, branch).await?);
}
Ok(self
.guard
.as_ref()
.expect("prepared on the line above or already present"))
}
/// `INSERT_LINK`, compiled once.
///
/// The single-edge path ran `conn.execute(INSERT_LINK, …)`, which compiles
/// the statement on every call — 61 µs of it, because `links` carries the
/// projection triggers and they are compiled with the insert. This is D-056
/// and D-057's lesson, which was learned on the batch path and never
/// carried across to the path a caller actually waits on.
async fn insert_link(&mut self, conn: &libsql::Connection) -> Result<&libsql::Statement> {
if self.insert_link.is_none() {
self.insert_link = Some(conn.prepare(INSERT_LINK).await?);
}
Ok(self
.insert_link
.as_ref()
.expect("prepared on the line above or already present"))
}
}
enum LoopCtl {
Continue,
Break,
}
/// Primary database handle for Macrame bitemporal ledger.
///
/// # Why this is not `Clone`, and what a multi-consumer caller uses instead
///
/// **Share it as `Arc<Database>`.** Every method here but one takes `&self`, so
/// an `Arc` is a complete handle and not a workaround: reads run concurrently
/// off `read_conn`, writes queue behind the actor's channel exactly as they do
/// through a `&Database`, and nothing becomes serialised that was not
/// serialised already. The exception is [`Database::close`], which takes `self`,
/// so the last owner closes with
/// `Arc::into_inner(db).expect("last handle").close().await`.
///
/// That exception is the whole reason `Clone` is absent. Cloning would have to
/// duplicate **the right to shut down**, and each field carrying that right
/// breaks differently when duplicated:
///
/// - `writer` is a [`tokio::task::JoinHandle`], which is not `Clone` at all —
/// so a hand-written impl would have to give the copy a `None`, and
/// `close()` on that copy returns `Ok(())` without ever checking the actor's
/// exit status. That status is one of the two reasons [`Drop`] tells callers
/// to prefer `close()`.
/// - `cadence_stop` is a [`tokio::sync::watch::Sender`], which **is** `Clone`,
/// and that is the worse case. Its contract is that *dropping* it stops the
/// snapshot task; a watch channel closes when the last sender goes, so one
/// surviving copy keeps that task running against a database that is going
/// away. Nothing returns an error, which is why this is the argument rather
/// than the `JoinHandle`.
/// - `closed` is per-handle, so two copies disagree about whether the ledger
/// was closed: `Drop` warns about a database that *was* closed, or stays
/// silent about one that was not.
///
/// And the ordering `close()` documents — cadence stopped, actor joined, *then*
/// the final snapshot, so that no write can land between the fold and the file
/// — is only enforceable while one handle can perform it. A second `close()`
/// writes a "final" snapshot with the actor still alive.
///
/// So the missing impl is the type saying shutdown has exactly one owner. The
/// Python binding reached the same shape from the other side and for the same
/// reason: `PyDatabase` holds a `RwLock<Option<Database>>` rather than a copy
/// per caller (0.13.30, W11.1, D-203).
pub struct Database {
db: libsql::Database,
/// The file this handle opened, kept so [`Database::diagnostic_conn`] can
/// open it again under different flags (T5.1, D-091). `archive_path` and
/// `snapshots_dir` are derived from it and were previously the only trace
/// of it on the struct.
path: PathBuf,
read_conn: libsql::Connection,
highpri_tx: mpsc::Sender<HighPriCommand>,
lowpri_tx: mpsc::Sender<LowPriCommand>,
clock: Arc<dyn Clock>,
archive_path: PathBuf,
snapshots_dir: PathBuf,
schema_version: u32,
/// Kept so [`Database::diagnostic_conn`] can configure the connections it
/// mints the same way `open()` configured the internal readers (0.12.16,
/// W5.5, D-159). Before that split, each one ran with SQLite's defaults.
reader_cache_size: Option<i32>,
/// The `SQLITE_OPEN_READ_ONLY` connection behind
/// [`Database::diagnostic_conn`], opened on first use and dropped with this
/// handle (0.15.14, W15.4, review C-9, [D-256]).
///
/// **The connection, not the `libsql::Database` handle**, and the
/// difference is the measurement rather than a preference. The first shape
/// written here cached the handle and minted a connection per call, on the
/// argument that `diagnostic_conn` promises a connection the *caller* owns.
/// `examples/diagnostic_conn_probe.rs` says `Builder::…build()` costs
/// **0.10 µs and opens nothing** — it succeeds against a path that does not
/// exist — while `connect()` costs **51.5 µs** and is where
/// `SQLITE_CANTOPEN` arrives for a missing file. The handle cache removes a
/// call that does no work.
///
/// **`Mutex<Option<_>>` rather than the `OnceCell` 0.15.14 shipped**
/// (0.15.15, W15.5, [D-257]). A `OnceCell` can be filled and never
/// emptied, and this connection is handed to arbitrary SQL, so it acquires
/// state that the *next* caller must not inherit — a leaked `BEGIN` above
/// all, which pins a WAL read snapshot and makes both this surface's reads
/// stale and [`Database::checkpoint`] a no-op. The slot has to be
/// clearable for the dirty ones to be replaced, and clearable is what a
/// `OnceCell` is not. `tokio`'s mutex rather than `std`'s because it is
/// held across the open.
///
/// A failed first attempt leaves the slot empty, so a database whose file
/// appears later is not poisoned by the call that came too early — the
/// property the `OnceCell` had, kept.
///
/// [D-256]: ../../docs/architecture/s13-decision-register.md#d-256
/// [D-257]: ../../docs/architecture/s13-decision-register.md#d-257
diagnostic_conn: tokio::sync::Mutex<Option<libsql::Connection>>,
writer: Option<tokio::task::JoinHandle<()>>,
/// Stops the snapshot cadence. Dropping it stops the task too, which is what
/// keeps a `Database` that is dropped rather than closed from leaving a task
/// running against a connection whose database is going away.
cadence_stop: Option<tokio::sync::watch::Sender<bool>>,
cadence: Option<tokio::task::JoinHandle<()>>,
/// Set by [`Database::close`]. Read only by [`Drop`], which warns when it is
/// still false — see that impl for why the omission is worth a warning.
closed: bool,
/// Shared with the actor (T1.4, T1.2). Held here rather than behind
/// `#[cfg(feature = "metrics")]` so `open_inner` has one shape; with the
/// feature off the metrics half is a zero-sized type and only
/// [`Database::metrics`] is gated — which is also why the field is unread in
/// the default build: the actor holds the other `Arc` and does the writing.
#[cfg_attr(not(feature = "metrics"), allow(dead_code))]
shared: Arc<ActorShared>,
}
/// What the snapshot cadence should do, for [`Tuning::cadence`].
///
/// # Why this is not `Option<SnapshotCadence>`
///
/// [`Database::open_with_cadence`] takes `Option<SnapshotCadence>`, where `None`
/// means *no cadence at all*. Carrying that field into [`Tuning`] unchanged
/// would have made it the one field in the struct whose `None` is a request to
/// change the behaviour rather than a request to leave it alone — and since
/// `Tuning` derives `Default`, `open_tuned(path, Tuning::default())` would then
/// have silently disabled snapshots, while `open(path)` runs them. Two calls
/// that read as synonyms, one of which stops writing anchors.
///
/// So the tri-state is written out. `Default` is the default cadence, matching
/// [`Database::open`]; `Disabled` is `open_with_cadence(path, None)`, and has to
/// be asked for by name.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum CadencePolicy {
/// [`SnapshotCadence::default`], as [`Database::open`] uses.
#[default]
Default,
/// No cadence task. `close()` is then the only thing that writes an anchor
/// (§5.5, D-053).
Disabled,
/// An explicit cadence.
Every(SnapshotCadence),
}
impl CadencePolicy {
/// Collapse to the `Option` the open path has always taken.
fn resolve(self) -> Option<SnapshotCadence> {
match self {
Self::Default => Some(SnapshotCadence::default()),
Self::Disabled => None,
Self::Every(cadence) => Some(cadence),
}
}
}
/// When SQLite should checkpoint the WAL on its own, for
/// [`Tuning::wal_autocheckpoint`] (0.12.14, W5.3, D-157).
///
/// # Why this is not `Option<u32>`
///
/// The same reason [`CadencePolicy`] is not `Option<SnapshotCadence>`, and the
/// plan for this wave specified `Option<u32>` here too. In a struct that derives
/// `Default`, a field whose `None` means *turn the mechanism off* is a field
/// that turns the mechanism off for everyone who did not mention it. Absence
/// means "leave it alone" everywhere in [`Tuning`], and disabling the automatic
/// checkpointer — which is not safe without an explicit
/// [`Database::checkpoint`] to replace it — has to be asked for by name.
///
/// **The default does not change.** 1,000 pages is SQLite's default and stays
/// SQLite's default; F-30 is a control-loop perturbation, not a correctness bug,
/// and changing a default is a behaviour change for every existing caller.
///
/// # What disabling it actually buys, measured (0.12.14, W5.3, D-157)
///
/// F-30 says the automatic checkpointer is an unbudgeted hold *inside* 0.12.0's
/// adaptive chunk controller: a checkpoint firing during a chunk transaction is
/// charged to that chunk, and since D-146 made the measured hold the input to
/// `next_chunk_size`, the controller shrinks in response to work the chunk did
/// not do. Three rounds, 6,000 concepts of 1 KB each through `write_concepts`,
/// release build:
///
/// | | longest chunk hold | mean | chunks | over budget | wall |
/// |---|---|---|---|---|---|
/// | autocheckpoint on (default) | **9.3–10.3 ms** | 2.40–2.44 ms | 125–130 | 24–28 | 304–321 ms |
/// | autocheckpoint off | **4.50 ms** | 2.08–2.20 ms | 142–153 | 18–27 | 298–339 ms |
///
/// **The tail is the finding, and it is real and reproducible.** The longest
/// hold roughly halves, and the >10 ms histogram bucket is populated only with
/// the checkpointer on — that bucket is the checkpoint, landing inside somebody
/// else's transaction and being charged to it. Every round agrees.
///
/// **What it does not buy is a calmer controller.** `over_budget` overlaps
/// between the arms, and total wall time is the same within noise. The
/// controller works near the budget boundary either way, because
/// [D-090](../docs/architecture/s13-decision-register.md)'s ~0.8 ms
/// per-transaction floor and the convergence cost do not go anywhere. So the
/// honest statement is that disabling autocheckpoint removes an outlier, not an
/// oscillation.
///
/// **And the cost is deferred, not removed.** The explicit
/// [`Database::checkpoint`] at the end of the same fixture moved **8,400–9,100
/// frames in 41–45 ms** with the checkpointer off, against **~860 frames in
/// 5.5–6.2 ms** with it on. That is the whole trade in one line: the same work,
/// moved out of the latency-bounded path and into one hold the caller chose the
/// moment for. It is a good trade for a bulk importer and a bad one for an
/// interactive process, which is why this is a knob and not a new default.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
#[non_exhaustive]
pub enum WalCheckpointPolicy {
/// SQLite's own default: checkpoint once the WAL passes 1,000 pages.
#[default]
Default,
/// No automatic checkpointing.
///
/// **Only correct if you call [`Database::checkpoint`] yourself.** Without
/// one, the WAL grows for the life of the process and the database file is
/// never brought up to date.
Disabled,
/// Checkpoint once the WAL passes this many pages.
///
/// `0` is not special-cased to [`Self::Disabled`] even though SQLite treats
/// it that way, because a caller who computed a threshold and got zero has
/// a bug, and inheriting SQLite's overload would turn it into a silently
/// unbounded WAL.
EveryPages(u32),
}
impl WalCheckpointPolicy {
/// The pragma to run, or `None` to leave the connection at SQLite's
/// default.
fn pragma(self) -> Option<String> {
match self {
Self::Default => None,
Self::Disabled => Some("PRAGMA wal_autocheckpoint = 0".to_string()),
Self::EveryPages(pages) => Some(format!("PRAGMA wal_autocheckpoint = {pages}")),
}
}
}
/// Everything [`Database::open_tuned`] can be told, in one growable struct
/// (0.12.12, W5.1, D-155).
///
/// # Why a struct rather than a fourth constructor
///
/// There were three — [`Database::open`], [`Database::open_with_cadence`],
/// [`Database::open_with_clock`] — and each new knob added one more, with the
/// combinatorics of the ones before it. 0.13.0 alone wanted three knobs
/// (`wal_autocheckpoint`, and a page cache each for the writer and the
/// readers), which is the point at which the naming stops being possible.
///
/// **Setters plus `#[non_exhaustive]` are the whole design** (0.15.13, W15.3,
/// [C-11], [D-255]). They make a new knob an additive change: callers write
/// `Tuning::default().cadence(..)`, and the fields that arrive after them are
/// the ones they did not ask about. That is not a hypothetical — W5.1 shipped
/// this struct with two fields, and W5.3/W5.4 added three more without
/// touching a caller.
///
/// # Why the attribute needed the setters, and why 0.5.1's answer was the
/// other one
///
/// D-155 specified `#[non_exhaustive]` for this struct and then could not
/// ship it, on a fact that is still true: a `#[non_exhaustive]` **struct**
/// cannot be built with literal syntax outside its own crate *at all*, and
/// the functional-update form is literal syntax, so
/// `Tuning { cadence, ..Default::default() }` is `E0639` for every external
/// caller — the exact expression the attribute was there to protect. (The rule
/// differs from `#[non_exhaustive]` on an enum, which only forces a wildcard
/// arm; [`CadencePolicy`] has kept it for that reason since W4.2.) D-155 named
/// the two ways to have both — *a builder with setters, or plain `Default`* —
/// and chose `Default`, because the field-literal form is the legible one.
///
/// **What changed is not the argument but the deadline.** `Default` alone
/// leaves the growth additive only for callers who wrote
/// `..Default::default()`; a caller who wrote the exhaustive literal breaks on
/// the next field. Before 1.0 that is a compile error with an obvious fix.
/// After it, it is a major version — and this struct is the one in the crate
/// whose whole documented purpose is to keep acquiring fields. So the release
/// that is still allowed to break callers pays D-155's other price and writes
/// the setters, which is the half of its own analysis it declined at the time.
///
/// **The fields stay `pub` and stay readable**, and on a value you own they
/// stay assignable: `let mut t = Tuning::default(); t.cadence = ..;` compiles
/// outside this crate exactly as it did. What the attribute forbids is the
/// *literal*, which is the one form that enumerates every field and therefore
/// the one form a new field can break.
///
/// [C-11]: ../../docs/Macrame%20Update%20Plan%20v0.16.0.md
/// [D-255]: ../../docs/architecture/s13-decision-register.md#d-255
///
/// # The three constructors stay
///
/// They delegate here and are not deprecated. `open(path)` is the right call for
/// most callers and should not acquire a warning for being the common case; the
/// consolidation is about where the *next* knob goes, not about moving anyone.
///
/// ```no_run
/// # use macrame::prelude::*;
/// # async fn f() -> macrame::Result<()> {
/// let db = Database::open_tuned(
/// "graph.db",
/// Tuning::default().cadence(CadencePolicy::Disabled),
/// )
/// .await?;
/// # Ok(()) }
/// ```
#[derive(Clone, Default)]
#[non_exhaustive]
pub struct Tuning {
/// What the snapshot cadence should do. Defaults to
/// [`SnapshotCadence::default`], as [`Database::open`] does.
pub cadence: CadencePolicy,
/// A clock to stamp `recorded_at` with, for tests (§5.1.2, D-062). `None`
/// is [`SystemClock`]. Floored against the database exactly as
/// [`Database::open_with_clock`] describes — read that before injecting
/// one against a non-empty file.
pub clock: Option<Arc<dyn Clock>>,
/// When SQLite checkpoints the WAL on its own (0.12.14, W5.3, F-30).
///
/// Applied to the **write connection**, which is the only connection in
/// this crate that commits, and therefore the only one whose autocheckpoint
/// setting can ever fire. Pair [`WalCheckpointPolicy::Disabled`] with an
/// explicit [`Database::checkpoint`] or the WAL grows without bound.
pub wal_autocheckpoint: WalCheckpointPolicy,
/// Page cache for the **write** connection, as SQLite's `cache_size`
/// (0.12.15, W5.4).
///
/// `None` leaves SQLite's default of −2000, which is −2000 *kibibytes*, or
/// 2 MB. **Negative values are KiB and positive values are pages** — that
/// is SQLite's convention and it is preserved rather than smoothed over,
/// because a caller who knows the pragma should not have to discover that
/// this crate redefined it. `Some(-64_000)` is 64 MB; `Some(64_000)` is
/// 64,000 pages, which at the 4 KiB page size this crate gets is 256 MB.
///
/// The writer wants a large cache: it is one connection, it holds the write
/// lock while it works, and every page it has to re-read from disk is time
/// no other writer can use.
///
/// # Unlike the two above, `None` here is not a policy enum
///
/// Because SQLite's default is a *value* rather than a mechanism. Absence
/// still means "leave it alone" — it just happens that leaving this alone
/// is expressible as not running a pragma, where leaving the automatic
/// checkpointer alone required saying which of two things "alone" meant.
pub writer_cache_size: Option<i32>,
/// Page cache for every **read-only** connection: the shared
/// [`Database::read_conn`], the snapshot cadence's own connection, and
/// (since W5.5) each [`Database::diagnostic_conn`] (0.12.15, W5.4).
///
/// Same units as [`Self::writer_cache_size`], and the same `None`.
///
/// Split from the writer's because the profiles are opposite and one number
/// cannot serve both. There is exactly one writer and it is long-lived, so
/// its cache is a fixed cost paid once. Read-only connections are plural —
/// the shared reader and the cadence's — so a large value here is
/// multiplied by however many exist, which is the wrong size for the one
/// connection that holds the write lock.
///
/// **The multiplier used to be unbounded** and is not since 0.15.14
/// (W15.4, [D-256]): `diagnostic_conn` minted a connection per call, so a
/// caller in a loop multiplied this number by their own call count. There
/// is one such connection per `Database` now, so the count is three.
///
/// [D-256]: ../../docs/architecture/s13-decision-register.md#d-256
pub reader_cache_size: Option<i32>,
/// What to do about a stored `recorded_at` in the future (0.13.5, W7.4,
/// §3.4).
///
/// The clock floors itself at `MAX(recorded_at)` so stamps stay strictly
/// increasing across restarts, which means one row from the future becomes
/// this process's floor and every stamp it issues inherits it — into rows
/// the next open reads back. Defaults to refusing beyond
/// [`crate::DEFAULT_FUTURE_STAMP_TOLERANCE`], a day.
///
/// Like [`Self::wal_autocheckpoint`] and unlike the two cache sizes, this
/// is a policy enum rather than an `Option`, for
/// [D-155](../../docs/architecture/s13-decision-register.md)'s reason: it
/// guards an invariant, and a `None` that switches it off would switch it
/// off for every caller who never heard of it.
pub future_stamps: FutureStampPolicy,
}
impl Tuning {
// The setters below are what make `#[non_exhaustive]` payable (0.15.13,
// W15.3, D-255). One per field, named after it, taking `self` — so
// `Tuning::default().cadence(x).writer_cache_size(y)` is an expression, and
// a field added later is a method added later rather than a break. They are
// deliberately not clever: no `Into`, no grouping of two knobs under one
// name, nothing that would have to be redesigned the first time a field
// does not fit the pattern.
/// What the snapshot cadence should do — the [`cadence`](Self::cadence)
/// field.
pub fn cadence(mut self, cadence: CadencePolicy) -> Self {
self.cadence = cadence;
self
}
/// Inject a clock — the [`clock`](Self::clock) field.
///
/// Takes the clock rather than an `Option`, because `None` is what
/// [`Tuning::default`] already holds and a setter whose argument can undo
/// itself invites `clock(None)` as a way of saying nothing.
/// [`Database::open_with_clock`] documents the flooring this is subject to;
/// read it before injecting one against a non-empty file.
pub fn clock(mut self, clock: Arc<dyn Clock>) -> Self {
self.clock = Some(clock);
self
}
/// When SQLite checkpoints the WAL on its own — the
/// [`wal_autocheckpoint`](Self::wal_autocheckpoint) field.
pub fn wal_autocheckpoint(mut self, policy: WalCheckpointPolicy) -> Self {
self.wal_autocheckpoint = policy;
self
}
/// Page cache for the write connection, in SQLite's units — the
/// [`writer_cache_size`](Self::writer_cache_size) field, which documents
/// why negative means KiB and positive means pages.
pub fn writer_cache_size(mut self, size: i32) -> Self {
self.writer_cache_size = Some(size);
self
}
/// Page cache for every read-only connection — the
/// [`reader_cache_size`](Self::reader_cache_size) field, which documents
/// why this is not the same number as the writer's.
pub fn reader_cache_size(mut self, size: i32) -> Self {
self.reader_cache_size = Some(size);
self
}
/// What to do about a stored `recorded_at` in the future — the
/// [`future_stamps`](Self::future_stamps) field.
pub fn future_stamps(mut self, policy: FutureStampPolicy) -> Self {
self.future_stamps = policy;
self
}
/// The `Option<SnapshotCadence>` the three older constructors take, mapped
/// onto the tri-state. `None` there means *disabled*, which is why
/// [`CadencePolicy`] exists — see its docs.
fn from_legacy(cadence: Option<SnapshotCadence>, clock: Option<Arc<dyn Clock>>) -> Self {
Self {
cadence: match cadence {
Some(cadence) => CadencePolicy::Every(cadence),
None => CadencePolicy::Disabled,
},
clock,
wal_autocheckpoint: WalCheckpointPolicy::default(),
writer_cache_size: None,
reader_cache_size: None,
future_stamps: FutureStampPolicy::default(),
}
}
}
// `Clock` is not `Debug` — it is a behavioural trait with two methods and
// requiring `Debug` of every implementor to print a handle here would be the
// tail wagging the dog. So the field is reported as present-or-absent, which is
// the only part of it a reader of a `Tuning` dump can act on.
impl std::fmt::Debug for Tuning {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Tuning")
.field("cadence", &self.cadence)
.field("clock", &self.clock.as_ref().map(|_| "<injected>"))
.field("wal_autocheckpoint", &self.wal_autocheckpoint)
.field("writer_cache_size", &self.writer_cache_size)
.field("reader_cache_size", &self.reader_cache_size)
.finish()
}
}
impl Database {
/// Open a database file at `path`, configuring pragmas, running migrations, and spawning the Write Actor.
///
/// The snapshot cadence runs with [`SnapshotCadence::default`]. Use
/// [`Database::open_with_cadence`] to tune or disable it.
pub async fn open(path: impl AsRef<Path>) -> Result<Self> {
Self::open_with_cadence(path, Some(SnapshotCadence::default())).await
}
/// Open with an explicit snapshot cadence, or `None` to run without one
/// (§5.5, D-053).
///
/// `None` restores the pre-0.5.5 behaviour, where `close()` is the only
/// thing that ever writes an anchor. That is the right setting for a
/// short-lived process that will not accumulate a delta worth bounding, and
/// for tests that assert on the contents of the snapshot directory.
pub async fn open_with_cadence(
path: impl AsRef<Path>,
cadence: Option<SnapshotCadence>,
) -> Result<Self> {
Self::open_inner(path.as_ref(), Tuning::from_legacy(cadence, None)).await
}
/// Open with an injected clock (§5.1.2, **defect K**, D-062).
///
/// The reason this exists is testing: `recorded_at` is the transaction-time
/// axis, and until now every test that wanted to assert on one had to either
/// avoid it or drive a raw connection, because `open()` hardcoded
/// [`SystemClock`]. `FakeClock` has been public and constructed in the test
/// harness since 0.5.2 with nothing to inject it into — the compiler warned
/// about the dead field on every build for three releases.
///
/// **The clock is floored against the database before the actor starts.**
/// [`Clock::raise_floor`] is called with the newest `recorded_at` in the
/// ledger, so an injected clock cannot issue a stamp below what is already
/// stored — which would abort the next concept write on
/// `trg_concepts_monotonic_ra` rather than merely being odd. This is the
/// step whose absence kept the defect open: the obvious implementation
/// (take an `Arc<dyn Clock>`, use it) produces a `Database` that fails on
/// its first write against any non-empty file.
///
/// On a fresh database there is no floor, so an injected `FakeClock` issues
/// exactly the stamps it was given.
pub async fn open_with_clock(
path: impl AsRef<Path>,
cadence: Option<SnapshotCadence>,
clock: Arc<dyn Clock>,
) -> Result<Self> {
Self::open_inner(path.as_ref(), Tuning::from_legacy(cadence, Some(clock))).await
}
/// Open with an explicit [`Tuning`] (0.12.12, W5.1, D-155).
///
/// The consolidated form of the three constructors above, and the one that
/// grows: every knob 0.13.0 adds arrives as a field here rather than as a
/// fourth `open_*`. See [`Tuning`] for why the struct is
/// `#[non_exhaustive]` and why that makes the growth additive.
///
/// That sentence was written at 0.12.12 and was false until 0.15.13: the
/// struct was *not* `#[non_exhaustive]`, and [`Tuning`]'s own docs carried
/// a section arguing at length that it should not be. Two documents in one
/// file contradicting each other for eleven releases, which is the shape
/// C-16 is about; W15.3 resolved it by making this one true.
pub async fn open_tuned(path: impl AsRef<Path>, tuning: Tuning) -> Result<Self> {
Self::open_inner(path.as_ref(), tuning).await
}
async fn open_inner(path: &Path, tuning: Tuning) -> Result<Self> {
let Tuning {
cadence,
clock: injected,
wal_autocheckpoint,
writer_cache_size,
reader_cache_size,
future_stamps,
} = tuning;
let cadence = cadence.resolve();
let db = libsql::Builder::new_local(path).build().await?;
let write_conn = configure(db.connect()?, writer_cache_size).await?;
// The writer is the only connection that commits, so it is the only one
// whose `wal_autocheckpoint` can ever fire. Setting it on the readers
// would be a pragma with no path to running (0.12.14, W5.3, D-157).
if let Some(pragma) = wal_autocheckpoint.pragma() {
let _ = write_conn.query(&pragma, ()).await?;
}
let read_conn = configure(db.connect()?, reader_cache_size).await?;
// PRAGMA query_only = ON on reader connection (§5.1.2)
read_conn.execute("PRAGMA query_only = ON", ()).await?;
let migration = migrations::run(&write_conn).await?;
let (highpri_tx, highpri_rx) = mpsc::channel(256);
let (lowpri_tx, lowpri_rx) = mpsc::channel(64);
// Floored after `migrations::run`, so the tables the floor is read from
// are guaranteed to exist.
let clock: Arc<dyn Clock> = match injected {
Some(clock) => {
if let Some(floor) =
crate::util::clock::recorded_at_floor(&read_conn, future_stamps).await?
{
clock.raise_floor(floor);
}
clock
}
None => Arc::new(SystemClock::new(&read_conn, future_stamps).await?),
};
let shared = Arc::new(ActorShared::default());
let writer = tokio::spawn(run_writer_actor(
write_conn,
Arc::clone(&clock),
highpri_rx,
lowpri_rx,
Arc::clone(&shared),
));
let archive_path = derive_archive_path(path);
let snapshots_dir = derive_snapshots_dir(path);
// **The cadence gets its own connection (Wave 4.1).** It used to share
// `read_conn`, on the reasoning that `libsql::Connection` is an
// Arc-backed handle and R15 makes every extra local connection a cost worth
// not paying for nothing. The cost it was not paying for turned out to be
// real: `reconstruct` brackets a fold with `ATTACH cold … DETACH cold`,
// that region is per-connection state, and it is not synchronised. Two
// folds on one connection can therefore interleave so that one DETACHes
// the handle the other is mid-fold on.
//
// Recorded in §8.5 as a hazard rather than a defect because it **did not
// reproduce**: 200 concurrent reconstructions against a 1 ms cadence with
// an archive present produced zero errors, since the cadence anchors at
// `MAX(recorded_at)` and so almost always takes the hot path. Narrow, and
// real — a write landing between `log_head` and the fold opens it.
//
// Separate connections remove the interleaving rather than ordering it,
// which is why this is preferred to a mutex around the region: there is
// no shared state left to race on, and nothing to remember to hold. The
// R15 objection does not apply — that fault is about *concurrent* opens,
// and this is one more sequential open during `open()`.
let (cadence_stop, cadence) = match cadence {
Some(cadence) => {
let cadence_conn = configure(db.connect()?, reader_cache_size).await?;
cadence_conn.execute("PRAGMA query_only = ON", ()).await?;
let (tx, rx) = tokio::sync::watch::channel(false);
let handle = tokio::spawn(snapshot::run_cadence(
cadence_conn,
snapshots_dir.clone(),
archive_path.clone(),
cadence,
rx,
Arc::clone(&shared) as Arc<dyn snapshot::CommittedTurns>,
));
(Some(tx), Some(handle))
}
None => (None, None),
};
let handle = Self {
db,
path: path.to_path_buf(),
read_conn,
highpri_tx,
lowpri_tx,
clock,
archive_path,
snapshots_dir,
schema_version: migrations::current_version(),
reader_cache_size,
diagnostic_conn: tokio::sync::Mutex::new(None),
writer: Some(writer),
cadence_stop,
cadence,
closed: false,
shared,
};
// **Re-anchor after a migration (Wave 4.4).**
//
// D-043 makes a `SCHEMA_VERSION` bump invalidate every snapshot on disk,
// which is correct — a snapshot is a serialised `MaterializedState` and a
// schema change can change what that means. What was missing is the other
// half: nothing wrote a replacement, so the first `reconstruct` after an
// upgrade skipped every file as incompatible and folded from genesis. On
// a database with a large log that is the difference between reading one
// snapshot and folding the whole history, and the only trace was a
// `warn!` per skipped file.
//
// Written here rather than left to the cadence because the cadence fires
// on log *growth* (D-053): an upgraded database that is then read but not
// written would never re-anchor at all.
//
// Failure is logged, not returned. A missing anchor costs time and no
// information — snapshots are derivative under Doctrine VI — so refusing
// to open a database because its optimisation could not be rebuilt would
// trade a real capability for a performance one.
//
// Gated on the cadence being enabled, as well as on an actual upgrade:
// `open_with_cadence(None)` means *this handle writes no snapshots except
// at close()*, and a one-off write at open would contradict that for a
// caller who asked for the quiet mode precisely to control when files
// appear. They still get an anchor from `close()`.
if migration.upgraded() && handle.cadence.is_some() {
let ts = handle.clock.now();
let archive = crate::temporal::archive::archive_present(&handle.archive_path)
.then_some(handle.archive_path.as_path());
match snapshot::write_final(&handle.read_conn, &handle.snapshots_dir, &ts, archive)
.await
{
Ok(path) => tracing::info!(
"schema moved v{} -> v{}; re-anchored snapshots at {:?}",
migration.from,
migration.to,
path
),
Err(e) => tracing::warn!(
"schema moved v{} -> v{} but the re-anchor failed: {e}. \
Reconstruction stays correct and folds from genesis until the \
cadence writes one.",
migration.from,
migration.to
),
}
}
Ok(handle)
}
/// Read connection handle for queries, traversals, and folds.
pub fn read_conn(&self) -> &libsql::Connection {
&self.read_conn
}
/// The file this handle opened.
pub fn path(&self) -> &Path {
&self.path
}
/// The **OS-level read-only** connection to this database, for diagnostics
/// (§4.7, T5.1, D-091).
///
/// # Why this exists when `read_conn()` already does
///
/// Two different things, and the difference is the point:
///
/// * `read_conn()` returns a shared `&Connection` carrying
/// `PRAGMA query_only = ON`. That pragma is **per-connection and
/// reversible by its holder in one statement**, so it is a guardrail
/// against accident, not a capability boundary. And because it is the
/// connection the crate's own traversals and folds run on, a caller who
/// runs a long reporting query there is competing with all of them.
/// * This returns a connection opened with `SQLITE_OPEN_READ_ONLY`, which is
/// enforced by the engine below the pragma layer, and which nothing inside
/// the crate runs on.
///
/// # One connection per `Database`, shared between callers (0.15.14, W15.4)
///
/// Through 0.15.13 this minted a connection per call and the sentence above
/// read *"a **new, independently owned** … connection"*. Review item C-9
/// asked for the file to be opened once per handle instead, and the
/// measurement behind it (`examples/diagnostic_conn_probe.rs`) was sharper
/// than the ask: `connect()` is **51.5 µs of an 82.7 µs call**, and
/// `Builder::…build()` — which every document in this crate called *the
/// open* — is **0.10 µs and opens nothing**, succeeding against a path that
/// does not exist. Caching the handle would have removed 0.10 µs. What
/// ships caches the connection: **82.7 µs → 19.9 µs**, and what remains is
/// the `stat` below, not the connection.
///
/// **So per-connection state is shared between diagnostic callers.** An
/// `ATTACH`, a `PRAGMA`, a temp table one caller creates is visible to the
/// next — which matters here and nowhere else in this API, because
/// `diagnostic_query` is the one arbitrary-SQL surface the crate exposes.
/// Asserted in `diagnostic_callers_share_one_connection_and_its_state`
/// rather than left to this paragraph.
///
/// What that costs is isolation between *diagnostic* callers. What it does
/// not cost is the thing D-091 was for: this is still not `read_conn()`, a
/// reporting query here still does not compete with the crate's traversals
/// and folds, and the read-only boundary below is untouched.
///
/// # What is scrubbed before you get it, and what is not (0.15.15, W15.5)
///
/// 0.15.14 shipped the sharing and documented it. [D-257] measured what it
/// actually admits, and one of the four is not an isolation nuisance but a
/// correctness hole that leaves this surface entirely: **a leaked `BEGIN`
/// pins a WAL read snapshot.** Measured — 200 writes through the typed
/// surface while a diagnostic caller held an unclosed read transaction —
/// later diagnostic reads answered **1 row instead of 201**, silently, and
/// [`Database::checkpoint`] became a no-op with the WAL stuck at 8.5 MB
/// until the transaction was rolled back. Stale answers on the surface a
/// caller reaches for when they already distrust the typed one, and an
/// unbounded WAL on a method that has nothing to do with diagnostics.
///
/// So this method scrubs on entry rather than trusting the caller, and the
/// prices are from `examples/diagnostic_hygiene_probe.rs`:
///
/// | left behind | how it is found | what happens |
/// |---|---|---|
/// | an open transaction | `is_autocommit()`, **0.04 µs** | `ROLLBACK`, 2.4 µs |
/// | a temp table or view | `PRAGMA temp.schema_version` | connection dropped, re-minted at 56.6 µs |
/// | an `ATTACH` | `PRAGMA database_list` | connection dropped, re-minted |
/// | `busy_timeout`, `cache_size` | not detected — restated, 1.0 µs | reset to the crate's values |
/// | any other pragma | **not detected** | **inherited by the next caller** |
///
/// The two dirt questions are asked as pragmas because the same two asked
/// over `temp.sqlite_master` and `pragma_database_list` cost **7.8 µs**
/// against **2.4 µs**, on a call whose entire warm cost is the `stat`.
///
/// The last row is the honest residue. SQLite has no cheap enumeration of
/// connection-scoped pragma state, so the crate restates the two pragmas
/// *it* set (D-159's `busy_timeout` above all — a caller who sets it to 0
/// would otherwise remove the 5 s margin from every later diagnostic call)
/// and leaves the rest. A caller who sets `case_sensitive_like` or
/// `recursive_triggers` changes what **later diagnostic queries on this
/// handle** see, and nothing else: the crate's own readers and writer are
/// different connections, so no typed answer can move. That is a smaller
/// blast radius than 0.15.14 had and a larger one than zero, and it is
/// written down rather than rounded off.
///
/// That paragraph is about pragmas that belong to a *connection*, which is
/// what "residue" means and what the scrub is for. **Not every pragma
/// reachable here is one**, and the section below is the exception —
/// measured after D-257 claimed this one "cannot change any typed answer"
/// without checking (0.15.16, [D-258]).
///
/// **Scrubbed on entry, not on exit**, because there is no exit: this
/// returns a `Connection` clone the caller keeps for as long as it likes,
/// and the crate is never told they are done. Entry is the one place that
/// covers both this method and `diagnostic_query`. The consequence is that
/// a caller who leaks a transaction and never calls again holds the pin
/// until the handle drops — so the Python binding, which *does* know when
/// a query is over, scrubs on exit as well.
///
/// **Measured on libSQL 0.9.30 rather than assumed**
/// (`examples/readonly_open_probe.rs`), against a live WAL database with the
/// write actor running:
///
/// | | `read_conn()` | `diagnostic_conn()` |
/// |---|---|---|
/// | `SELECT`, `EXPLAIN QUERY PLAN` | allowed | allowed |
/// | `INSERT` | refused | refused |
/// | `PRAGMA query_only = OFF` | **allowed** | allowed |
/// | `INSERT` after that | **allowed** | **refused** |
/// | `ATTACH` an existing file | allowed | allowed |
/// | `INSERT` into the attachment | refused¹ | **refused** |
/// | `ATTACH` a path that does not exist | — | refused (`SQLITE_CANTOPEN`) |
///
/// The third and fourth rows are the whole difference: turning the pragma
/// off restores writes on `read_conn()` and does not here. That is what
/// "boundary rather than guardrail" means, and it is now a number rather
/// than a claim.
///
/// ¹ On `read_conn()` that refusal is `query_only` — the same reversible
/// thing as row 2. On `diagnostic_conn()` it is the open flags, and the
/// probe runs it *after* `query_only = OFF` so that the pragma cannot be
/// what is doing the work.
///
/// # `ATTACH` is permitted, and does not widen the write boundary
///
/// Checked because `diagnostic_query` (Python) is the only arbitrary-SQL
/// surface this crate exposes, and an attachment is a second `open` whose
/// flags it does not obviously inherit. It does inherit them: the
/// attachment is read-only, and a nonexistent path is `SQLITE_CANTOPEN`
/// rather than a new file, because `SQLITE_OPEN_CREATE` is dropped for the
/// attachment as it is for `main`. So `SQLITE_OPEN_READ_ONLY` bounds the
/// **connection**, not just the one file it names (0.10.0, W4.3).
///
/// What it does widen is *reading*: an `ATTACH` can name any file the
/// process can open, so this connection is a read surface over the
/// filesystem, not over this database. That is a property of arbitrary SQL
/// rather than of the flags, and it is unchanged by them.
///
/// # One pragma here can end the process, and sharing is not why
///
/// `PRAGMA hard_heap_limit = 1` through this connection leaves the whole
/// **process** unable to use SQLite. Not this connection, not this handle:
/// measured (`tests_py/probes/diagnostic_global_pragmas.py`), the next
/// ordinary write, the next read, `checkpoint()`, `close()`, and opening a
/// *different* database file all fail with `out of memory`, permanently.
///
/// `SQLITE_OPEN_READ_ONLY` does not stand in the way because setting it is
/// not a write to the database file, and the scrub above does not help
/// because there is nothing left on the connection to scrub: the limit
/// lives in the SQLite library, one per process. **Re-measured with a
/// connection minted per call — the 0.15.13 shape, before any of the
/// sharing this method now does — the outcome is identical.** So this is
/// not a cost of [D-256]'s shared connection and no amount of hygiene
/// addresses it.
///
/// Six other candidates were measured and are harmless: `soft_heap_limit`
/// (a hint, not a wall), `locking_mode = EXCLUSIVE` (accepted; the writer
/// kept working), `temp_store_directory`, `max_page_count` (clamped, and
/// per-connection), `case_sensitive_like` (the control), and
/// `wal_checkpoint`, which is refused outright because this connection is
/// read-only.
///
/// It belongs with the `ATTACH` note above rather than with the scrub: both
/// are properties of handing a caller **arbitrary SQL**, not of the flags
/// the connection was opened with. The practical form is one sentence —
/// *`diagnostic_query` is not a safe place to put a string that came from
/// somewhere else* — which `ATTACH` already made true and this makes
/// sharper. Not blocked by refusing statements that look like this one,
/// because matching SQL text is guesswork wearing the costume of a
/// guarantee, and it would do nothing for a Rust caller holding the
/// connection directly ([D-258]).
///
/// [D-258]: ../../docs/architecture/s13-decision-register.md#d-258
///
/// # One way this is *more* permissive, which is worth knowing
///
/// `CREATE TEMP TABLE` **succeeds** here and is refused by `read_conn()`.
/// Temp tables live in a separate temporary database that is writable
/// regardless of how the main one was opened, whereas `query_only` refuses
/// them outright — which is the mechanism [D-050] measured when it removed
/// `TwoPhaseTempTable` for returning `SQLITE_READONLY (8)` on the read
/// connection. So the stronger boundary is not uniformly stronger, and a
/// strategy that needs a temp table has a connection it could run on. That
/// is recorded, not acted on: D-050 removed the strategy for two reasons and
/// this addresses one of them.
///
/// # Calling this concurrently was R15's shape, and 0.15.14 is why it is not
///
/// Through 0.15.13 this section said *"this is the one method on `Database`
/// that opens the file … each call is a fresh `libsql::Builder::…build()`,
/// so *N* threads calling it at once are *N* concurrent opens"*. The first
/// half was true and the second was wrong about which call does it:
/// `build()` opens nothing, and `connect()` is the open. The conclusion
/// happened to be right for the wrong reason, which is why it took a
/// measurement to move.
///
/// **Measured through the unlocked Python binding**, 48 threads on a
/// barrier, 30 runs per arm (`tests_py/probes/r15_diagnostic_path.py`):
///
/// | arm | bad runs |
/// |---|---|
/// | a connection per call, as before 0.15.14 | **3 / 30** |
/// | the `libsql::Database` handle cached, a connection per call | 2 / 30 |
/// | the `connect()` serialised behind a mutex | 1 / 18 |
/// | **one connection, as shipped** | **0 / 30** |
///
/// Rows two and three are why the shape that preserved the old contract was
/// not taken: caching the handle leaves the crash because it leaves the
/// `connect()`, and serialising the `connect()` alone does not reach zero —
/// the race is between minting a connection and the *use* of the others,
/// not between two mintings.
///
/// **There is nothing left on this path to bound**, because after the first
/// call it no longer opens anything. `Database::open` from many threads is
/// still R15's shape and `examples/r15_soak.rs` still reproduces it; this
/// method is no longer a way to reach it. The Python binding keeps its
/// mutex as margin rather than as a measured necessity — see
/// `PyDatabase::diagnostic_rows`.
///
/// # Errors
///
/// The file must already exist. `SQLITE_OPEN_READ_ONLY` drops
/// `SQLITE_OPEN_CREATE` with it, so a missing file is `SQLITE_CANTOPEN`
/// rather than a fresh empty database — which is the right failure, and is
/// surfaced as a typed error rather than as libSQL's error 14.
///
/// The check is a `stat` on **every** call, not only the first, and it is
/// still most of what a warm call costs (18.6 µs of the 22 µs a clean one
/// takes since 0.15.15). It is kept at that price because the alternative
/// is the worst failure a *diagnostic* surface can have: a cached
/// connection whose file has been deleted and replaced answers from the old
/// inode, silently, on the one method a caller reaches for when they
/// already doubt the typed answer.
pub async fn diagnostic_conn(&self) -> Result<libsql::Connection> {
let fail = |reason: String| DbError::DiagnosticConn {
path: self.path.display().to_string(),
reason,
};
// Checked per call rather than once, because it is the documented
// error of *this* method and costs a `stat`. The handle below is
// opened once; the question "is the file there" is asked every time,
// so a caller who deletes the file still gets the typed refusal on the
// next call rather than a connection to an inode nothing can name.
if !self.path.exists() {
return Err(fail(
"the file does not exist, and a read-only open cannot create it".to_string(),
));
}
let mut slot = self.diagnostic_conn.lock().await;
// Scrub what the last caller left, before this one can inherit it
// (0.15.15, W15.5, D-257). Free on a clean connection: the transaction
// check is a C call at 0.04 us and the two dirt pragmas are 2.4 us,
// against a `stat` of 18.6 that has already happened above.
scrub(&mut slot).await;
match slot.as_ref() {
// Restated per call since 0.15.15 rather than once at mint. The
// pragmas are per-connection, and there is one connection now, so
// 0.15.14 set them once — which was right until the shared
// connection was also something a caller could move them on. No
// dirt check can see a pragma, so the crate restores its own
// instead of detecting that they went: 1.0 us, against a
// `busy_timeout` of 0 outliving the call that set it.
Some(conn) => configure_common(conn, self.reader_cache_size).await?,
None => {
*slot = Some(self.open_diagnostic_conn(&fail).await?);
}
}
Ok(slot
.as_ref()
.expect("the slot was filled above or the open returned Err")
.clone())
}
/// Roll back and discard whatever the last diagnostic caller left behind,
/// without handing a connection out (0.15.15, W15.5, [D-257]).
///
/// [`Database::diagnostic_conn`] does this on the way *in*, which is the
/// only place the crate can do it: the method returns a `Connection` clone
/// and is never told the caller is finished with it. That covers every
/// caller and leaves one gap — somebody who leaks a transaction and then
/// never calls again holds the WAL read snapshot until the handle drops,
/// which makes [`Database::checkpoint`] a no-op for that whole time.
///
/// This is for the callers that *do* know when they are done. It costs the
/// scrub and not the `stat` — around 3.5 µs on a clean connection — because
/// it hands nothing back and so has nothing to promise about the file still
/// being there.
///
/// **The Python binding does not call it**, though the first draft did. A
/// mutation deleting that call left the whole suite green, and the reason is
/// that the gap is not reachable from there: `diagnostic_query` runs one
/// statement, a bare `BEGIN` pins nothing — the snapshot is taken by the
/// first *read* inside the transaction — and any statement that would take
/// it arrives through the same method, whose entry scrub has already rolled
/// the transaction back. The gap is real for a Rust caller holding a clone
/// across both, which is what this method and
/// `scrubbing_releases_the_pin_without_handing_out_a_connection` are for.
///
/// Infallible by construction: everything it might have reported is
/// something it responds to by discarding the connection, and the next
/// [`Database::diagnostic_conn`] opens a fresh one.
///
/// [D-257]: ../../docs/architecture/s13-decision-register.md#d-257
pub async fn scrub_diagnostic_conn(&self) {
let mut slot = self.diagnostic_conn.lock().await;
scrub(&mut slot).await;
}
/// The cold half of [`Database::diagnostic_conn`]: the actual open.
///
/// Reached on the first call and on any call whose predecessor left the
/// connection dirty enough to discard. It costs 56.5 µs against the warm
/// path's scrub, and it is split out for reading rather than for speed:
/// `Box::pin`ning it here — on the theory that carrying `Builder::build()`'s
/// state machine inside the warm path's was what made a clean call 29.8 µs
/// rather than the 21.8 its parts measure — changed nothing at all
/// (0.15.15, W15.5, [D-257]).
///
/// [D-257]: ../../docs/architecture/s13-decision-register.md#d-257
async fn open_diagnostic_conn(
&self,
fail: &dyn Fn(String) -> DbError,
) -> Result<libsql::Connection> {
let db = libsql::Builder::new_local(&self.path)
.flags(libsql::OpenFlags::SQLITE_OPEN_READ_ONLY)
.build()
.await
.map_err(|e| fail(e.to_string()))?;
let conn = db.connect().map_err(|e| fail(e.to_string()))?;
// Configured since 0.12.16 (W5.5, D-159). Until then this connection
// ran with SQLite's defaults while every other connection in the
// process ran with the crate's — most consequentially a `busy_timeout`
// of 0 against everyone else's 5 s, on the one surface whose job is to
// answer questions when the typed path is already suspect. Only the
// common half: `SQLITE_OPEN_READ_ONLY` cannot set `journal_mode`, and
// the rest govern writes this connection cannot make.
configure_common(&conn, self.reader_cache_size).await?;
// The `libsql::Database` is dropped here and the connection outlives
// it, which is the ownership libSQL's own API implies: `connect()`
// returns a `Connection` that does not borrow the builder's handle.
Ok(conn)
}
/// Cross-check the snapshot chain against a fold from genesis (§5.5, T5.3,
/// D-092).
///
/// `write_final` composes onto the previous snapshot, so snapshot *n* is
/// derived from snapshot *n−1* and nothing in the chain ever folds the whole
/// log. An error at any link propagates forward forever and every read
/// agrees with it, because every read descends from it. This is the check
/// that would notice.
///
/// # When to run it
///
/// **Not on a schedule this crate chooses.** A genesis fold is precisely the
/// cost snapshots exist to avoid, so running it periodically by default
/// would give every application the bill snapshots were bought to remove —
/// on a database whose log is large enough for snapshots to matter, which is
/// the only kind where this is worth doing. The plan calls it a scheduling
/// problem and it is the caller's schedule: an idle period, a nightly job,
/// or once per *N* anchors, chosen against a log size this crate cannot see.
///
/// The cadence is deliberately left alone for the same reason — it runs on a
/// connection shared with nothing and a fold there would compete with
/// interactive reads at a moment nobody chose.
///
/// # It reports; it does not repair
///
/// A divergence means the snapshots are a wrong **cache**, not that the
/// ledger is corrupt: [Doctrine VI] makes them disposable, so deleting
/// [`Self::snapshots_dir`] restores correctness and costs only speed.
/// Rewriting the file here would destroy the evidence that composition has a
/// defect, which is the only thing this can tell you that you did not
/// already know.
///
/// Pair it with the actor counters ([`Self::metrics`], D-079) so a
/// divergence found by a scheduled run is visible beside the write latency
/// of the period that produced it.
///
/// [Doctrine VI]: ../../docs/architecture/s0-s3-foundations.md#doctrine-vi
pub async fn verify_snapshot_chain(&self, ts: &str) -> Result<crate::temporal::ChainCheck> {
let archive = crate::temporal::archive::archive_present(&self.archive_path)
.then_some(self.archive_path.as_path());
crate::temporal::verify_snapshot_chain(&self.read_conn, ts, archive, &self.snapshots_dir)
.await
}
/// Check the newest link of the snapshot chain (0.15.19, review C-18).
///
/// The affordable half of [`Self::verify_snapshot_chain`]: re-derive the
/// newest snapshot from the one before it and compare, which is one
/// anchored delta rather than a fold from genesis. `Ok(None)` when there
/// are not two snapshots yet.
///
/// The snapshot cadence already runs this after every anchor it writes and
/// logs a divergence at `warn`, so a caller reaching for it directly is
/// usually one that wants the [`crate::temporal::ChainCheck`] itself — the
/// disagreeing ids — rather than a yes or no.
///
/// **It reports; it does not repair.** A snapshot is derivative
/// (Doctrine VI), so the repair is to delete the snapshot directory, which
/// is the caller's call and one line. What this cannot tell you is whether
/// the chain went wrong further back than one link; that is what
/// [`Self::verify_snapshot_chain`] is for, and why it stays.
pub async fn verify_last_link(&self) -> Result<Option<crate::temporal::ChainCheck>> {
let archive = crate::temporal::archive::archive_present(&self.archive_path)
.then_some(self.archive_path.as_path());
crate::temporal::verify_last_link(&self.read_conn, archive, &self.snapshots_dir).await
}
/// The clock every write is stamped with (§5.1.1).
pub fn clock(&self) -> &Arc<dyn Clock> {
&self.clock
}
/// Schema version this handle opened against.
pub fn schema_version(&self) -> u32 {
self.schema_version
}
/// Cold database path, derived by convention from the main file.
pub fn archive_path(&self) -> &Path {
&self.archive_path
}
/// Snapshot directory, derived by convention from the main file.
pub fn snapshots_dir(&self) -> &Path {
&self.snapshots_dir
}
/// What the write actor has done since this handle was opened (T1.4, D-079).
///
/// Requires the `metrics` feature. The counters are per-handle and start at
/// zero on `open()` — they are not read from the database, because the thing
/// being measured is *this process's* actor and merging two processes'
/// histograms would produce a number about neither.
///
/// The intended first question is [`crate::metrics::MetricsSnapshot::budget_violations`]:
///
/// ```no_run
/// # async fn f(db: ¯ame::Database) {
/// # #[cfg(feature = "metrics")] {
/// for k in db.metrics().budget_violations() {
/// eprintln!("{} broke the 3 ms bound {} times", k.kind, k.over_budget);
/// }
/// # }
/// # }
/// ```
///
/// Reading this does not stop the actor — see
/// [`crate::metrics::ActorMetrics::snapshot`] for what that costs in
/// consistency, and why the trade goes that way.
#[cfg(feature = "metrics")]
pub fn metrics(&self) -> crate::metrics::MetricsSnapshot {
self.shared.metrics.snapshot()
}
/// The underlying libSQL database, for callers that need their own connection.
///
/// # Actor containment is a convention above this line, not a guarantee
///
/// **Kept public, and the honest statement of what that costs (Wave 4.3).**
/// §5.1 says the write actor is the sole writer, and two mechanisms make that
/// true of the handle: every write method goes through a channel, and
/// [`Self::read_conn`] carries `PRAGMA query_only = ON`. **Nothing protects a
/// connection obtained from here.** A caller can open one, write to `links`
/// directly, and the actor will not know — the triggers still fire and the
/// ledger stays internally consistent, but the single-writer property that
/// [`crate::CHUNK_BUDGET`]'s latency argument rests on is gone, and so is the
/// serialisation the overlap guard (D-060) relies on.
///
/// This is the same shape as the limit stated in §4.2 for that guard, and it
/// is one fact rather than two: **the storage layer permits what this API
/// refuses.** Making it private would not change that — the database file is
/// reachable by any SQLite client on the machine — it would only remove the
/// supported way to do the thing, which is how escape hatches become
/// `unsafe`-adjacent folklore.
///
/// The free functions [`crate::register_model`] and
/// [`crate::upsert_embedding`] take a bare connection for the same reason and
/// carry the same caveat; prefer [`Self::register_model`] and
/// [`Self::upsert_embeddings`], which go through the actor.
///
/// # The legitimate-use list is now one item long (T5.1, D-091)
///
/// It used to read: `EXPLAIN QUERY PLAN` and other diagnostics, read-only
/// reporting queries wanting their own connection rather than sharing the
/// reader, and provoking a guard in a test. The first two are exactly what
/// [`Self::diagnostic_conn`] now does, and it does them behind an OS-level
/// read-only open rather than on a handle that can write. **Use that.**
///
/// What is left is the one use that genuinely requires write access through
/// a connection the actor does not own: *provoking a guard* — writing the
/// state §4.7 says the storage layer permits and this API refuses, so a test
/// can assert the gap is still where the document says it is. That is the
/// only thing this crate's own suite uses it for.
///
/// # Why `#[doc(hidden)]` and not a `raw-access` feature
///
/// T5.1 offers either. The feature is the stronger declaration — it shows up
/// in the consumer's `Cargo.toml`, where a reviewer sees it — and it was
/// **not** taken, for a reason specific to what uses this:
///
/// Cargo features are additive and cannot be *required* by a test target
/// except through `required-features`, which makes a plain `cargo test`
/// **skip** that binary silently. The binaries that call this are
/// `storage_boundary_tests` and `wave1_regression_tests` — the §4.7
/// tripwires, whose entire job is to fail when a documented gap moves. Gating
/// them behind a feature would mean the ordinary `cargo test` stopped running
/// the tests that enforce the section this item is about, to make a
/// declaration about a hatch. That trade is the wrong way round, and it is
/// the same failure the project already names: a suite that quietly does less
/// than it appears to.
///
/// So the hatch stays reachable and stops being *discoverable*: it is absent
/// from the docs, and the documented path for every non-write use is
/// [`Self::diagnostic_conn`]. [D-068] is unchanged — removing it would buy
/// the appearance of a guarantee, since the file is reachable by any SQLite
/// client on the machine.
///
/// [D-068]: ../../docs/architecture/s13-decision-register.md#d-068
// convention (D-068/D-091): `raw()` is #[doc(hidden)] and is NOT exposed by
// any binding. Everything above this line is invisible on docs.rs and
// invisible to a contributor reading the Python surface list, which is where
// the decision to expose it would actually be taken — hence this sentinel and
// its twin in `bindings/python/src/lib.rs` (0.10.0, W4.10). The documented
// path for every non-write use is `diagnostic_conn`.
#[doc(hidden)]
pub fn raw(&self) -> &libsql::Database {
&self.db
}
// -- write surface (§5.1, Appendix A) --
//
// Every method here validates and canonicalises before the value crosses the
// channel, so a bad edge type or a second-precision timestamp is a typed
// error at the call site rather than an engine `CHECK` failure surfacing
// from the far side of an actor with no context attached.
//
// NOTE (§5.1.8, D-028): awaiting one of these waits on a Rust channel, not
// in SQLite, so `busy_timeout` does not bound it. During an in-flight
// `rebuild_current` or `archive` the caller stalls for that transaction's
// duration. Wrap in `tokio::time::timeout` if you need a bound — but a
// timeout is not a cancellation: the command stays queued and commits when
// the actor reaches it.
/// Assert an edge (Doctrine III: a new row, never an update).
///
/// # One row costs a transaction, so N rows cost N transactions
///
/// This is the correct method for a caller who genuinely has one edge, and
/// it is the wrong one in a loop. Each call is its own transaction and pays
/// the ~0.8 ms per-transaction floor (D-090) whole, so a thousand edges
/// asserted one at a time spend roughly **0.8 s in transaction overhead
/// alone** — before any of the work — and mint a thousand distinct
/// `recorded_at` stamps for what the caller probably means as one act.
///
/// There are two bulk forms and the difference between them is the one to
/// get right:
///
/// - [`Self::bulk_import`] is **chunked** against [`CHUNK_BUDGET`] and
/// atomic per chunk. It amortises the transaction floor across the batch
/// while still yielding to interactive work at every chunk boundary. This
/// is the one a loop should almost always become.
/// - [`Self::write_bulk_atomic`] is one transaction under one stamp and is
/// **the one write with no latency bound** — the hold is a function of
/// `edges.len()`, tabulated in its own docs, and is time every other
/// writer spends waiting. Reach for it when the batch is genuinely one
/// act that must not be observable half-applied, not for speed.
///
/// The choice is the caller's and neither form is deprecated. Doctrine III
/// makes "one act, one stamp" a semantic claim rather than a performance
/// one, and only the caller knows whether their thousand edges are one act.
pub async fn assert_edge(&self, edge: EdgeAssertion) -> Result<()> {
let edge = edge.normalized()?;
self.high(|responder| HighPriCommand::AssertEdge { edge, responder })
.await
}
/// Close an open interval by asserting its replacement (Doctrine III).
pub async fn retire_edge(
&self,
source: impl Into<String>,
target: impl Into<String>,
edge_type: impl Into<String>,
valid_from: &str,
valid_to: &str,
) -> Result<()> {
let edge_type = edge_type.into();
crate::graph::edge::validate_edge_type(&edge_type)?;
let valid_from = timestamp::normalize(valid_from)?;
let valid_to = timestamp::normalize(valid_to)?;
let (source, target) = (source.into(), target.into());
self.high(|responder| HighPriCommand::RetireEdge {
source,
target,
edge_type,
valid_from,
valid_to,
branch: None,
responder,
})
.await
}
/// Retire an edge **on a lineage**, which is a different write (0.14.8).
///
/// The `_on` suffix is the crate's established spelling for the
/// branch-taking variant of a call whose trunk form predates branching —
/// [`query_as_of_edges_on`](crate::temporal::query_as_of_edges_on) is the
/// other one. A sixth positional `Option<BranchId>` on
/// [`Self::retire_edge`] would have made every existing call site read as
/// though it had made a lineage decision it never made.
///
/// # This closes a row; it does not close *the* row
///
/// Retiring an edge the branch **inherited** writes the branch's own row at
/// the ancestor's key, carrying the closed interval and this lineage's id.
/// The ancestor's row is untouched, and the read prefers the nearer one, so
/// the edge is gone from this lineage's view and unchanged in its parent's.
/// That is **shadow retirement**, and it is the only retirement across
/// lineages that does not commit the parent corruption
/// [Doctrine III](../../docs/architecture/s0-s3-foundations.md#doctrine-iii)
/// forbids — which is not a rule this method obeys but a shape the ledger
/// cannot express: `links` is append-only and no statement in this crate
/// closes a row in place.
///
/// `weight` and `properties` are carried over from the visible row rather
/// than restated, which is what makes this a retirement rather than a new
/// assertion that happens to be closed.
///
/// # Errors
///
/// - [`DbError::UnknownBranch`] when `branch` is not registered.
/// - [`DbError::NotFound`] when this lineage can see no open row at that
/// `valid_from`. On a branch that includes *never inherited it* and
/// *inherited it and already shadowed it*, which are one answer here
/// because they are one answer to the question asked: there is nothing
/// at that key to retire.
pub async fn retire_edge_on(
&self,
source: impl Into<String>,
target: impl Into<String>,
edge_type: impl Into<String>,
valid_from: &str,
valid_to: &str,
branch: crate::branch::BranchId,
) -> Result<()> {
let edge_type = edge_type.into();
crate::graph::edge::validate_edge_type(&edge_type)?;
let valid_from = timestamp::normalize(valid_from)?;
let valid_to = timestamp::normalize(valid_to)?;
let (source, target) = (source.into(), target.into());
self.high(|responder| HighPriCommand::RetireEdge {
source,
target,
edge_type,
valid_from,
valid_to,
branch: Some(branch),
responder,
})
.await
}
/// Insert or update a concept.
///
/// # One row costs a transaction
///
/// The same trade [`Self::assert_edge`] describes, for the same reason and
/// with the same ~0.8 ms floor (D-090): correct for one concept, wrong in a
/// loop. [`Self::write_concepts`] takes a `Vec` and commits it as one
/// transaction under one stamp.
///
/// There is no atomic-across-chunks concept path and none is needed to make
/// the choice: `write_concepts` is chunked against [`CHUNK_BUDGET`] and
/// atomic per chunk, so a large `Vec` is cooperative rather than a stall.
/// The responsiveness argument for writing one row at a time therefore does
/// not apply — the bulk form already yields at every chunk boundary.
pub async fn upsert_concept(&self, concept: ConceptUpsert) -> Result<()> {
let concept = concept.normalized()?;
self.high(|responder| HighPriCommand::UpsertConcept { concept, responder })
.await
}
/// A handle on one lineage (§15.4, 0.14.9, [D-226]).
///
/// Takes `&Arc<Self>` rather than `&self` because the view holds the handle
/// and must not be able to end it: `close` takes `self` by value and an
/// `Arc` cannot surrender that while a clone survives, so the restriction
/// is structural rather than documented. Sharing the handle is already
/// `Arc<Database>` (§5.1.11), so this asks for nothing a caller did not
/// have.
///
/// Does no I/O and cannot fail. Whether the lineage is *registered* is
/// asked by every operation on the view, which is where
/// [`DbError::UnknownBranch`] names it.
///
/// [D-226]: ../../docs/architecture/s13-decision-register.md#d-226
pub fn view(
self: &std::sync::Arc<Self>,
branch: crate::branch::BranchId,
) -> crate::branch::BranchView {
crate::branch::BranchView::new(std::sync::Arc::clone(self), branch)
}
/// Cut a new lineage from an existing one (§15.2, §15.4).
///
/// # A fork is O(1) in rows written
///
/// One row in `branches`, and nothing else. No ledger table is read, copied
/// or touched: a branch inherits its parent's history by *resolution at
/// read* rather than by owning a copy of it, which is what
/// [`TraversalBuilder::on_branch`](crate::graph::TraversalBuilder::on_branch)
/// resolves and 0.14.6 bounds by the fork point. The cost of that choice is
/// on the read side and is measured — [D-220] for the resolution, [D-223]
/// for the cutoff — and the cost of the alternative would be here, as an
/// O(rows) fork and storage multiplied by branch count (§15.3, option 3).
///
/// # The fork point is *now*, and that is a bound on this release rather
/// than on the design
///
/// `forked_at` is stamped from the same clock as every other write, so the
/// new lineage sees its parent's history up to this instant. Forking from a
/// *past* instant is a coherent thing to want and the schema has always
/// allowed it — `branches` carries `forked_at` and `created_at` as separate
/// columns under `CHECK (forked_at <= created_at)` — but it is not in this
/// release and is additive when it is.
///
/// # What this lineage can do
///
/// It can be **read**: every traversal entry point takes a branch, and on a
/// forked ledger the read resolves along the ancestry and stops at the fork
/// point. Since 0.14.8 it can also be **written** — [`EdgeAssertion`] and
/// [`ConceptUpsert`] carry a lineage, and [`Self::retire_edge_on`] shadows
/// an inherited edge (D-225). Through 0.14.7 they did not, and a caller who
/// forked and then called `assert_edge` got a successful write **on the
/// trunk**; that is fixed rather than documented now.
///
/// What a branch still may not do is **restate an inherited concept**.
/// `concepts` is keyed by identity, so that is refused as
/// [`DbError::CrossLineage`] — see [`ConceptUpsert::branch`]. Edges are the
/// thing a lineage may hold its own belief about, and superseding one is a
/// row written *beside* the ancestor's rather than over it.
///
/// # Errors
///
/// - [`DbError::UnknownBranch`] when `from` is not registered. Named rather
/// than left to the foreign key, because the caller asked about a branch.
/// - [`DbError::BranchExists`] when `name` is taken — including `"main"`,
/// which every database has from its first migration.
/// - [`DbError::ForkPrecedesParent`] when the clock would place this fork
/// point before the parent's *own* — not before the parent's
/// `created_at`, which is what the schema comment promised until 0.14.7
/// and is not checkable: the trunk's `created_at` is stamped during
/// migration from the wall clock, before an injected clock exists, so
/// that rule refuses every fork on every `FakeClock` database (D-224).
/// Reachable with [`FakeClock`](crate::util::FakeClock), and the one
/// refusal here that no `CHECK` could have made — it is cross-row, and a
/// `CHECK` sees one row.
///
/// # Example
///
/// ```no_run
/// # use macrame::prelude::*;
/// # async fn f(db: &Database) -> Result<()> {
/// let alt = db.fork(BranchId::new("turn/17/alt/1")?, BranchId::main()).await?;
/// let seen = TraversalBuilder::new("socrates")
/// .on_branch(alt.id.clone())
/// .execute_ids(db.read_conn(), "2026-08-29T00:00:00.000000Z")
/// .await?;
/// # let _ = seen;
/// # Ok(())
/// # }
/// ```
///
/// [D-220]: ../../docs/architecture/s13-decision-register.md#d-220
/// [D-223]: ../../docs/architecture/s13-decision-register.md#d-223
pub async fn fork(
&self,
name: crate::branch::BranchId,
from: crate::branch::BranchId,
) -> Result<crate::branch::Branch> {
self.high(|responder| HighPriCommand::Fork {
name,
parent: from,
responder,
})
.await
}
/// Every lineage the ledger knows about, trunk first (§15.4).
///
/// Read through [`Self::read_conn`] rather than the write actor, which is
/// the difference between this and [`Self::fork`] and is deliberate:
/// `branches` is append-only, so the only way this listing can be stale is
/// by missing a branch created after it was taken, and a caller who wanted
/// to know about that branch would have had to create it. Queueing a read
/// behind the write actor would make listing branches wait on a bulk import
/// for no answer it could change.
///
/// A database that has never forked returns exactly one row: the trunk,
/// with no parent and no fork point.
pub async fn branches(&self) -> Result<Vec<crate::branch::Branch>> {
crate::branch::list(self.read_conn()).await
}
/// The beliefs `a` holds that `b` does not (§15.4, 0.14.11, D-228).
///
/// One [`Divergence`](crate::branch::Divergence) per edge key the two
/// lineages disagree about, in key order: `b` holds no belief about it, or
/// holds one with a different interval or weight. Not symmetric —
/// `diff(b, a)` is the other half, and composing the two is *two* snapshots
/// even though each is one.
///
/// Read through the read connection rather than the actor, like
/// [`Self::branches`], and taken at one snapshot rather than two: see
/// `graph::lineage::diff_sql` for why that decides the shape of the query.
///
/// There is no instant parameter. A diff filtered to a valid-time instant
/// cannot report the one divergence that is *about* an instant having
/// passed — a branch that retired an edge its parent still holds open — so
/// this compares the whole of both views.
///
/// # Errors
///
/// [`DbError::UnknownBranch`], naming whichever of the two is not
/// registered, and `a` first when neither is.
pub async fn diff(
&self,
a: &crate::branch::BranchId,
b: &crate::branch::BranchId,
) -> Result<Vec<crate::branch::Divergence>> {
crate::branch::diff(self.read_conn(), a, b).await
}
/// Assert many edges in one transaction under one stamp (D-014).
///
/// # This is the one write with no latency bound, and here is what it costs
///
/// The batch is one act under one `recorded_at`, so it cannot be chunked —
/// splitting it is the thing this method exists not to do. That makes the
/// actor's hold a function of `edges.len()`, and until now the only
/// statement of that anywhere was the prose "uncapped" in
/// [`CHUNK_BUDGET`]'s table. A caller who stalls every other writer for
/// eight seconds should have been able to predict it from the signature.
///
/// Measured on libSQL 0.9.30 (T1.3, D-081), holding the actor for:
///
/// | rows | hold |
/// |---|---|
/// | 500 | ~34 ms |
/// | 2,000 | ~155 ms |
/// | 10,000 | ~1.0 s |
/// | 20,000 | ~2.6 s |
///
/// [`estimated_bulk_hold`] is that curve as a function, and this method
/// emits a `tracing::warn!` when it predicts more than
/// [`BULK_ATOMIC_WARN_HOLD`]. **The estimate is a shape, not a promise** —
/// see [`estimated_bulk_hold`] for what it is calibrated against and where
/// it will be wrong.
///
/// A caller who needs the latency bound and not the atomicity wants
/// [`Self::bulk_import`], which is the same write chunked and explicitly not
/// atomic overall (D-011).
pub async fn write_bulk_atomic(&self, edges: Vec<EdgeAssertion>) -> Result<usize> {
let estimate = estimated_bulk_hold(&edges);
if estimate > BULK_ATOMIC_WARN_HOLD {
// Warned here rather than in the actor, and before the send: this is
// the caller's own task, so the log line lands with their span
// attached and names the call site that chose the batch size. By the
// time the actor has it, the only context left is "a large batch".
tracing::warn!(
rows = edges.len(),
estimated_hold_ms = estimate.as_millis() as u64,
"write_bulk_atomic will hold the write actor for roughly \
{estimate:?} — it is atomic by contract (D-014) and cannot be \
chunked. Every other writer waits that long. Use bulk_import \
if the batch does not need to be all-or-nothing."
);
}
let edges = normalize_all(edges)?;
self.high(|responder| HighPriCommand::WriteBulkAtomic { edges, responder })
.await
}
/// Move the WAL back into the main database file (§4.5, F-30, 0.12.13,
/// W5.2, D-156).
///
/// Runs `PRAGMA wal_checkpoint(FULL)` and then `(TRUNCATE)` on the write
/// connection, as one actor turn, and returns what SQLite reported. **Read
/// [`CheckpointReport::busy`]** — a checkpoint that could not run is an
/// `Ok` whose WAL is still there.
///
/// Two passes rather than one because **a truncating checkpoint cannot
/// report its own work**: the counts describe the WAL *after* the
/// operation, and after a truncation there is nothing left to describe, so
/// `TRUNCATE` alone answers `busy=0, log=0, checkpointed=0` on success —
/// indistinguishable from having done nothing. `FULL` supplies the frame
/// count and `TRUNCATE` resets the file; `busy` is the union of the two.
///
/// # When a caller needs this
///
/// Three cases, and only three:
///
/// - **Before copying the database file elsewhere.** In WAL mode the `.db`
/// file alone is not the database; recent commits live in the `-wal`. A
/// complete checkpoint is what makes the main file self-contained.
/// - **At the end of a bulk load that turned the automatic checkpointer
/// off.** That is the pairing this method exists for — see
/// [`Tuning::wal_autocheckpoint`]. Disabling autocheckpoint without
/// calling this leaves a WAL that grows for the life of the process.
/// - **Before a long idle period**, to give back the disk.
///
/// Nobody else should call it on a timer. SQLite checkpoints automatically
/// every 1,000 pages and that default is not changed by this method
/// existing; a periodic explicit checkpoint on top of it buys nothing and
/// takes the write lock to do so.
///
/// # It takes the write lock, and it is budget-exempt
///
/// The hold is a function of how many frames have accumulated, which is a
/// function of how long since the last checkpoint — not of anything passed
/// in. It is on [`CHUNK_BUDGET`]'s exemption table for that reason, and it
/// is the one entry there that is not a transaction: there is no smaller
/// unit to chunk into, because the operation *is* the copy.
pub async fn checkpoint(&self) -> Result<CheckpointReport> {
self.high(|responder| HighPriCommand::Checkpoint { responder })
.await
}
/// Rebuild `links_current` from `links` and verify zero drift (§5.8).
///
/// One transaction holding the write lock for its whole duration, because
/// [D-023] will not let the `DELETE` and the `INSERT` be split: a reader
/// landing between them would see a graph with no edges and no error.
/// [`Self::rebuild_current_chunked`] is the same result with a different
/// latency profile, and is what a populated database wants.
///
/// The report's `drift_after` is the audit run inside the same transaction,
/// so a repair that did not converge is reported by the call that made it
/// rather than by the next one to look.
///
/// [D-023]: ../docs/architecture/s13-decision-register.md#d-023
pub async fn rebuild_current(&self) -> Result<RebuildReport> {
self.high(|responder| HighPriCommand::RebuildCurrent { responder })
.await
}
/// Rebuild `links_current` beside itself, in chunks (§5.8, T1.2, D-082).
///
/// Same result as [`Self::rebuild_current`], different latency profile.
/// `rebuild_current` is one transaction holding the write lock for its whole
/// duration, because D-023 will not let the `DELETE` and the `INSERT` be
/// split: a reader landing between them sees a graph with no edges and no
/// error. This builds the replacement in a shadow table instead — the live
/// table stays live and trigger-maintained throughout — and swaps it in at
/// the end.
///
/// Each step is its own actor turn, so an interactive assertion can jump the
/// queue between chunks. That is the whole of the improvement, and it is why
/// the loop is here rather than inside the actor's arm (the same reasoning
/// as [`Self::archive_windowed`] and [`Self::bulk_import`]).
///
/// # What the swap still costs
///
/// Not microseconds. Index names are global and SQLite has no `ALTER INDEX
/// … RENAME`, so the shadow cannot be built carrying `links_current`'s index
/// names while `links_current` still holds them — and building it under
/// other names would leave the table permanently indexed under names absent
/// from [`CREATE_INDICES`](crate::schema::ddl::CREATE_INDICES), so the next
/// migration would create a second copy of each.
/// `DROP TABLE` frees the names, so the swap transaction is where
/// the three indexes get built. What the chunking moves off the lock is the
/// **projection** — the window function over all of `links` — which is the
/// O(E log E) term.
///
/// # When this returns an error rather than a repair
///
/// [`DbError::RebuildInterrupted`] means an archive committed while the
/// shadow was being built. Its deletions are invisible to a catch-up pass
/// keyed on `recorded_at` — a deleted row has no `recorded_at` left to find
/// it by — so the work is discarded rather than swapped in. `links_current`
/// is untouched and the call can simply be retried.
///
/// Use [`Self::rebuild_current`] when the repair must be one atomic act, or
/// when nothing else is contending for the actor and the extra turns are
/// pure overhead.
pub async fn rebuild_current_chunked(&self) -> Result<RebuildReport> {
use crate::integrity::{ShadowOutcome, ShadowStep};
// Each `else` arm is unreachable: the actor maps each step to its own
// outcome variant. Written as a refutable pattern rather than an
// `unwrap` so that adding a step cannot turn a mismatch into a panic on
// the write path — and `WriterDroppedResponder` is the honest name for
// "the actor answered with something this cannot use".
let ShadowOutcome::Started { build_start, epoch } =
self.shadow_step(ShadowStep::Begin).await?
else {
return Err(DbError::WriterDroppedResponder);
};
let mut after: Option<String> = None;
loop {
let ShadowOutcome::Filled { last } = self
.shadow_step(ShadowStep::Fill {
after: after.take(),
})
.await?
else {
return Err(DbError::WriterDroppedResponder);
};
match last {
Some(last) => after = Some(last),
None => break,
}
}
let ShadowOutcome::Swapped { rows } = self
.shadow_step(ShadowStep::Swap { build_start, epoch })
.await?
else {
return Err(DbError::WriterDroppedResponder);
};
Ok(RebuildReport {
rows_rebuilt: rows,
// Not audited. The chunked path's whole argument is that the
// expensive work happens off the lock, and `audit_current` is two
// `EXCEPT` passes over the projection — the cost D-077 removed from
// the archive for the same reason. A caller who wants the check has
// `audit_current` on the read connection, where it costs nobody the
// write lock.
drift_after: 0,
})
}
/// Run one step of a chunked rebuild, for a caller doing its own scheduling.
///
/// [`Self::rebuild_current_chunked`] is this in a loop and is what almost
/// everyone wants. This exists because that loop offers no seam: it drives
/// `Begin`, then `Fill` to exhaustion, then `Swap`, and a caller who needs to
/// do something *between* steps — pace them against a frame budget, abandon
/// a rebuild that has run long enough, or provoke the archive interlock in a
/// test — cannot get in.
///
/// The obligation that comes with it: `epoch` from
/// [`ShadowOutcome::Started`](crate::integrity::ShadowOutcome) must be handed
/// back to [`ShadowStep::Swap`](crate::integrity::ShadowStep), or the
/// archive interlock is defeated and a stale projection can be swapped in.
/// The looping version cannot get that wrong; this one can.
pub async fn shadow_step(
&self,
step: crate::integrity::ShadowStep,
) -> Result<crate::integrity::ShadowOutcome> {
self.low(|responder| LowPriCommand::ShadowRebuild { step, responder })
.await
}
/// Import edges on the background channel, chunked (D-011).
///
/// Atomic *per chunk*, not overall: a failure partway leaves earlier chunks
/// committed. That is the tradeoff [`chunk_rows`] documents — use
/// [`Database::write_bulk_atomic`] when the batch must be all-or-nothing.
///
/// Chunked adaptively, at most [`chunk_rows::EDGES`] rows at a time: that
/// constant is where the loop starts and the largest chunk it will send, and
/// each chunk's measured hold sizes the next against [`CHUNK_BUDGET`]. It is
/// also faster in total than the larger chunks this used through 0.5.5
/// (D-058).
///
/// A consequence worth planning for: the chunk boundaries — and so the
/// `recorded_at` stamps this import writes — depend on how fast the machine
/// was, not only on how many edges were passed (§5.1.6).
///
/// # The WAL during a bulk (measured, 0.16.1 — see D-274's diagnostics)
///
/// Every chunk's commit runs WAL pages back into the database file when
/// SQLite's own autocheckpoint threshold (1,000 pages) is reached, and on a
/// bulk that fires every few chunks. Measured on the 16,000-edge random-pair
/// ladder (medians of 3, `benchmarks/diagnostics/`): 5.0 s with the default,
/// 3.7 s with the threshold at 10,000 pages (`wal_autocheckpoint = 10_000`
/// at open), 3.6 s with it disabled outright — and the checkpoint the bulk
/// was paying is 240 ms once at the end instead. The WAL is the price of the
/// third: ~554 MB for that fixture against 43 MB at the 10,000-page
/// threshold and ~6 MB at the default, so the 10,000-page setting is the
/// recipe unless the disk is known to be large. On an ordered (chain-shaped)
/// import the sweep is a wash — the recipe is for the random-order shape a
/// real importer produces. The knob is per-handle and set at
/// [`Database::open`](crate::Database::open); call
/// [`Self::checkpoint`] once after a disabled-checkpoint bulk.
///
/// Returns [`BulkInterrupted`] rather than [`DbError`] on failure, because
/// a path that is not all-or-nothing owes its caller the count of what
/// landed (0.13.8, W7.6). `?` into a `Result<_, DbError>` still compiles
/// and drops the count, which is the caller's decision to take.
///
/// [`Self::bulk_import_with`] adds cancellation and per-chunk progress.
pub async fn bulk_import(&self, edges: Vec<EdgeAssertion>) -> BulkResult<usize> {
self.bulk_import_with(edges, BulkControl::new()).await
}
/// [`Self::bulk_import`] with cancellation and progress (0.13.8, W7.6,
/// D-181).
///
/// The chunk boundaries this path already has are what make both possible:
/// the loop is between transactions several times a second, which is where
/// a token can be read and a callback run without holding anything.
pub async fn bulk_import_with(
&self,
edges: Vec<EdgeAssertion>,
control: BulkControl,
) -> BulkResult<usize> {
let edges = normalize_all(edges).map_err(before_any_chunk)?;
self.low_chunked(edges, chunk_rows::EDGES, control, |chunk, responder| {
LowPriCommand::BulkImportChunk { chunk, responder }
})
.await
}
/// Load edges **without the maintained projection in the way**, then
/// re-derive it in one chunked rebuild (D-277, plan §8.1 / F1).
///
/// Three phases through the actor: drop `trg_links_current_sync` (the
/// per-row upsert into `links_current`), load every edge through the same
/// chunked path [`Self::bulk_import`] uses, restore the trigger, then
/// [`Self::rebuild_current_chunked`] — the re-derivation D-082 made
/// chunked, whose fill costs ~104 ms per 16k rows and whose swap is the
/// exempt 46.8 ms turn. Measured on the reference box, 16,000 random-pair
/// edges (medians of three):
///
/// | arm | bulk | rebuild + restore | total |
/// |---|---|---|---|
/// | shipped (`bulk_import`) | 5.08 s | — | 5.08 s |
/// | this | 2.29 s | 0.19 s | **2.48 s (2.05×)** |
///
/// The mirror is what the ladder showed growing with the graph on random
/// pairs (plan §9.2): the near-chain shape pays it too, just flatter, so
/// this is the lever for both — and the rebuild's cost is a function of
/// the table, not the shape, which is why the skip arm's total is flat
/// where the shipped arm's is not.
///
/// # What the window actually touches, stated precisely
///
/// **The ledger is complete at every instant of the window.** The log
/// mirror and the single-open guard stay up: `links` and
/// `transaction_log` gain every row this call loads, versioned and
/// guarded exactly as the shipped path does. What lags is
/// `links_current` — Doctrine VI's derivative state, which the crate
/// has always maintained as *the projection you could fold from the
/// ledger*. Current-time reads during the window (`traverse`,
/// `query_as_of_edges` without an instant) see a partial projection and
/// are not wrong about the past — they are stale about the present, the
/// one state Doctrine VI calls disposable.
///
/// The window **ends when this method returns**, success, failure, or
/// cancellation: the restore runs before the rebuild, so any write that
/// lands after it mirrors again, and the rebuild then sweeps the backlog.
/// If the future is dropped or the process unwinds mid-load, the window
/// stays open — the projection stays stale but the ledger stays complete,
/// `audit_current` reports the drift as [`DbError::CurrentDrift`], and
/// [`Self::rebuild_current_chunked`] closes it. That is the honest doc
/// entry plan §8.1 asked for, and why this is a signature rather than a
/// default: the shipped path never has a window at all.
///
/// # What a failure does
///
/// The load's failure — [`BulkInterrupted`] with its `written` count —
/// propagates only **after** the trigger is restored and the projection
/// rebuilt from what did commit, so a failed deferred bulk is in exactly
/// the state a caller expects: the prefix is committed, the projection is
/// true, the mirror is on. An empty `edges` is a no-op: no DDL, no
/// rebuild, nothing attributed.
pub async fn bulk_import_deferred(
&self,
edges: Vec<EdgeAssertion>,
) -> BulkResult<usize> {
self.bulk_import_deferred_with(edges, BulkControl::new())
.await
}
/// [`Self::bulk_import_deferred`] with cancellation and progress, on the
/// load half — the same chunked loop, the same token, the same callbacks
/// [`Self::bulk_import_with`] takes. The toggle and the rebuild are
/// single turns around it.
pub async fn bulk_import_deferred_with(
&self,
edges: Vec<EdgeAssertion>,
control: BulkControl,
) -> BulkResult<usize> {
let edges = normalize_all(edges).map_err(before_any_chunk)?;
if edges.is_empty() {
return Ok(0);
}
if let Err(cause) = self
.low(|responder| LowPriCommand::LinksCurrentMirror {
present: false,
responder,
})
.await
{
// A toggle that failed has left the mirror exactly where it was —
// on. Stopping here is safe: nothing was loaded, nothing is stale.
return Err(BulkInterrupted { written: 0, cause });
}
let loaded = self
.low_chunked(edges, chunk_rows::EDGES, control, |chunk, responder| {
LowPriCommand::BulkImportChunk { chunk, responder }
})
.await;
// Restore before reporting, and before rebuilding: any write that
// lands after the restore maintains the projection itself, and the
// rebuild then sweeps whatever the window left behind.
let restored = self
.low(|responder| LowPriCommand::LinksCurrentMirror {
present: true,
responder,
})
.await;
let rebuilt = self.rebuild_current_chunked().await;
// The mirror's absence out-ranks the rebuild's failure out-ranks the
// load's: a caller must not read success beside a mirror that is off,
// and a load error beside a restored-and-rebuilt projection is just
// the load's answer.
let written = match &loaded {
Ok(n) => *n,
Err(e) => e.written,
};
match (loaded, restored, rebuilt) {
(Ok(n), Ok(()), Ok(_)) => Ok(n),
// A failed load with the mirror restored and the projection
// rebuilt is the load's own answer — the prefix committed, the
// projection is true, the mirror is on.
(Err(e), Ok(()), Ok(_)) => Err(e),
(_, Err(cause), _) => Err(BulkInterrupted { written, cause }),
(Err(e), Ok(()), Err(report)) => Err(BulkInterrupted {
written: e.written,
cause: report,
}),
(Ok(n), Ok(()), Err(report)) => Err(BulkInterrupted {
written: n,
cause: report,
}),
}
}
/// Upsert many **concepts** on the background channel, chunked (D-011).
///
/// This is the bulk concept path, and every row it writes is a ledger write:
/// it versions the concept and lands in `transaction_log`. Derived analytics
/// output does not belong here — see
/// [`Database::write_analytics_annotations`] and D-041.
///
/// Called `write_annotations` through 0.5.6, from when the two writes were
/// one call. D-041 split them and the name stayed on the wrong one for three
/// releases, so the crate had a `write_annotations` that wrote concepts
/// sitting beside a `write_analytics_annotations` that wrote annotations
/// (D-075).
///
/// Chunked, so it returns [`BulkInterrupted`] and its `written` count on
/// failure (0.13.8, W7.6); [`Self::write_concepts_with`] adds cancellation
/// and progress.
pub async fn write_concepts(&self, concepts: Vec<ConceptUpsert>) -> BulkResult<usize> {
self.write_concepts_with(concepts, BulkControl::new()).await
}
/// [`Self::write_concepts`] with cancellation and progress (0.13.8, W7.6).
pub async fn write_concepts_with(
&self,
concepts: Vec<ConceptUpsert>,
control: BulkControl,
) -> BulkResult<usize> {
let concepts: Vec<ConceptUpsert> = concepts
.into_iter()
.map(ConceptUpsert::normalized)
.collect::<Result<_>>()
.map_err(before_any_chunk)?;
self.low_chunked(
concepts,
chunk_rows::CONCEPTS,
control,
|chunk, responder| LowPriCommand::WriteConceptsChunk { chunk, responder },
)
.await
}
/// State as believed at `ts` (§5.5, D-026, D-049).
///
/// A read: it runs on `read_conn` and never touches the Write Actor, so a
/// reconstruction and a full-speed write-back do not slow each other.
///
/// Prefer this to calling [`crate::temporal::reconstruct`] directly. The
/// free function takes the archive path and the snapshot directory as
/// arguments, and a caller who passes `None` for the second gets a correct
/// answer that folds the whole log every time — the composition is opt-in
/// at that layer and easy to leave off by accident. Here both come from the
/// handle, so the fast path is the default one.
pub async fn reconstruct(&self, ts: &str) -> Result<crate::temporal::MaterializedState> {
let ts = timestamp::normalize(ts)?;
crate::temporal::reconstruct(
&self.read_conn,
&ts,
Some(&self.archive_path),
Some(&self.snapshots_dir),
)
.await
}
/// State at `ts` as `branch` saw it (0.15.17, [D-259], review C-10).
///
/// [`Self::reconstruct`] with the ancestry resolved: each ancestor bounded
/// at its fork point, one belief per edge key from the nearest lineage
/// holding it. See [`crate::temporal::reconstruct_on`] for how it is
/// assembled, what it costs, and the two things it does **not** do —
/// concepts are not resolved by lineage, and the result must not be saved
/// as a snapshot.
///
/// A read, on `read_conn`, like [`Self::reconstruct`]. The archive path and
/// the snapshot directory come from the handle, so snapshot composition is
/// on by default for each of the folds this runs.
///
/// [D-259]: ../../docs/architecture/s13-decision-register.md#d-259
pub async fn reconstruct_on(
&self,
ts: &str,
branch: &str,
) -> Result<crate::temporal::MaterializedState> {
let ts = timestamp::normalize(ts)?;
crate::temporal::reconstruct_on(
&self.read_conn,
&ts,
branch,
Some(&self.archive_path),
Some(&self.snapshots_dir),
)
.await
}
/// `branch`'s ancestry, nearest first, each with its fork-point cutoff.
///
/// The input [`crate::temporal::resolve_beliefs`] takes. Resolved from
/// `branches` in Rust since 0.15.17 ([D-259]) — the walk is a few
/// microseconds and the table is tiny and append-only, so this is a read
/// like any other rather than something to cache.
///
/// The trunk of an unforked database answers with one row and no cutoff,
/// which is its true ancestry. A lineage that is not registered is refused
/// by name with [`DbError::UnknownBranch`].
///
/// [D-259]: ../../docs/architecture/s13-decision-register.md#d-259
pub async fn ancestry(&self, branch: &str) -> Result<Vec<crate::branch::Ancestor>> {
let lineages = crate::graph::lineage::Lineages::load(&self.read_conn).await?;
// Checked before it is walked: `resolve` answers for a name it has never
// seen with a one-row ancestry, which is the right answer for a root and
// the D-069 wrong-looking-right answer for a typo.
lineages.shape(branch)?;
Ok(lineages.ancestry(branch))
}
/// Every edge one [`ReadPlan`] names (0.15.9, W13.4, [D-251]).
///
/// The whole projection filtered to the plan's instants and lineage —
/// topology only, no start node, and no budget on the answer. On a large
/// ledger that is a large `Vec`; [`Self::load_subgraph`] is the bounded
/// neighbourhood read and [`crate::graph::TraversalBuilder`] is the
/// anchored one.
///
/// # What this can express that nothing else could
///
/// [`crate::temporal::query_as_of_edges_on`] is the same read at a
/// valid-time instant, and it takes no transaction-time one: before this
/// release, *"which edges did we believe existed, as of March, as they
/// stood in January"* had exactly two answers available — walk it from a
/// start node, or fold the entire log with [`Self::reconstruct`] and filter
/// the result. The first needs an anchor the question does not have and the
/// second is a different order of work. The fold this uses is the
/// traversal's own, so the bitemporal cell is now readable whole at the
/// cost of reading it.
///
/// The two functions share one statement, which is why neither can drift
/// from the other; `query_as_of_edges_on` is this with `recorded` unset and
/// the lineage dropped from each row.
///
/// # Errors
///
/// [`DbError::UnknownBranch`] naming a
/// lineage that was never registered — refused rather than answered for the
/// trunk, for `graph::lineage::Lineages::shape`'s reason.
/// [`DbError::RecordedInstantUnreachable`]
/// when [`ReadPlan::recorded`] is below what the hot log still covers
/// ([D-247](../../docs/architecture/s13-decision-register.md#d-247)).
/// [`DbError::InvalidTimestamp`] for a
/// stamp that is not canonical, from the same normaliser every other read
/// uses — a plan is inert and validates nothing, so this is where a
/// malformed instant is noticed.
///
/// [D-251]: ../../docs/architecture/s13-decision-register.md#d-251
pub async fn edges(&self, plan: ReadPlan) -> Result<Vec<crate::temporal::EdgeBelief>> {
// `None` is now, and now is this handle's clock rather than the
// system's: a database opened on a `FakeClock` reads at the instant it
// is writing at, which is the whole reason the clock is a handle
// property (§5.1.1).
let valid = match plan.valid.as_deref() {
Some(ts) => timestamp::normalize(ts)?,
None => self.clock.now(),
};
let recorded = plan
.recorded
.as_deref()
.map(timestamp::normalize)
.transpose()?;
crate::plan::edges_at(
&self.read_conn,
&valid,
recorded.as_deref(),
plan.branch_name(),
plan.limit,
)
.await
}
/// Create a model's embedding table and DiskANN index (§5.9, D-048).
///
/// Idempotent: registering a model that already exists at the same
/// dimension succeeds, and at a different dimension fails with
/// [`DbError::DimMismatch`] naming both, rather than no-opping through
/// `IF NOT EXISTS` and leaving the caller believing the dimension they
/// asked for is the one in force.
///
/// This issues DDL, which everywhere else in the crate is the migration
/// runner's exclusive business (D-032). The exception is bounded and
/// deliberate: a model's table is created once, by an explicit call, and
/// the alternative — a caller-supplied write connection — is the very thing
/// the Write Actor exists to make impossible.
///
/// # Latency
///
/// One small transaction, but it queues like any other write: see §5.1.8.
pub async fn register_model(&self, model: &ModelName, dim: usize) -> Result<()> {
let model = model.clone();
self.high(|responder| HighPriCommand::RegisterModel {
model,
dim,
responder,
})
.await
}
/// Store or replace vectors for `model`, chunked (§5.9, D-011, D-048).
///
/// The write path for embeddings. Before 0.5.4 there was none:
/// [`crate::vector::upsert_embedding`] takes a raw connection, `read_conn`
/// is `query_only`, and the write connection lives inside the actor — so an
/// application could search vectors it had no way to store.
///
/// Low priority and chunked at [`chunk_rows::EMBEDDINGS`], because embedding
/// is bulk derived work: a 50,000-vector backfill must yield to an
/// interactive assertion at every chunk boundary. That constant is the
/// smallest of the four by a wide margin — DiskANN index maintenance makes an
/// embedding the most expensive row in the system (D-058). Atomic per chunk, not overall, which
/// is the same trade [`Database::bulk_import`] makes and is safer here than
/// there — an embedding is derived (Doctrine VII), so a partially written
/// batch is recoverable by re-embedding.
///
/// Fails with [`DbError::ModelNotRegistered`] if `model` has no table, and
/// [`DbError::DimMismatch`] if a vector's length is not the declared
/// dimension. The dimension is read from the schema once per chunk (D-037):
/// the crate keeps no registry of its own to fall out of date.
///
/// Chunked, so it returns [`BulkInterrupted`] and its `written` count on
/// failure (0.13.8, W7.6). A 50,000-vector backfill is the longest-running
/// write the crate has, which makes it the one most likely to be cancelled
/// — [`Self::upsert_embeddings_with`] is how.
pub async fn upsert_embeddings(
&self,
model: &ModelName,
rows: Vec<(String, Vec<f32>)>,
) -> BulkResult<usize> {
self.upsert_embeddings_with(model, rows, BulkControl::new())
.await
}
/// [`Self::upsert_embeddings`] with cancellation and progress (0.13.8,
/// W7.6).
pub async fn upsert_embeddings_with(
&self,
model: &ModelName,
rows: Vec<(String, Vec<f32>)>,
control: BulkControl,
) -> BulkResult<usize> {
self.low_chunked(rows, chunk_rows::EMBEDDINGS, control, |chunk, responder| {
LowPriCommand::UpsertEmbeddingChunk {
model: model.clone(),
chunk,
responder,
}
})
.await
}
/// Load embeddings **without the DiskANN index in the way**, then rebuild
/// it in one pass (D-276, plan §9.1).
///
/// Three actor turns in sequence: drop the index (`DROP INDEX IF EXISTS`,
/// µs-scale), load every row through the same chunked path
/// [`Self::upsert_embeddings`] uses, rebuild the index in one statement.
/// Measured on the reference box, medians of three, 2,000 vectors:
///
/// | dim | indexed (`upsert_embeddings`) | `bulk_embeddings` | one-pass build alone |
/// |---|---|---|---|
/// | 64 | 3.78 s | 2.62 s | 2.61 s |
/// | 256 | 31.0 s | 19.7 s | 19.7 s |
/// | 512 | 56.0 s | 39.0 s | 39.0 s |
///
/// At 5,000 × 256: 89.2 s indexed, 48.6 s here — **1.8×**. The build, not
/// the blob writes, is the cost: inserts without the index are flat at
/// ~5 µs/row at every dimension measured. The rest of the gap against an
/// HNSW writer is DiskANN-vs-HNSW build economics inside libSQL, engine-
/// side, and out of scope for a crate that does not fork its engine.
///
/// # The trade, stated
///
/// Between the drop and the rebuild, the model's vectors are not
/// searchable (`vector_top_k` reports the missing index) and **not
/// dimension-checked at the storage layer** — the index is that check
/// (D-037, `ddl::create_embeddings_index`). The crate-side check
/// ([`crate::vector::EmbeddingCodec::encode`]) still applies to everything
/// this method loads, and the rebuild restores the storage check at the
/// end; what is given up is the check on rows a *different* client inserts
/// during the window. That is the measured decision, opt-in by signature:
/// the plain path keeps the storage backstop at every instant.
///
/// **A failed or cancelled load still rebuilds.** The load's own failure —
/// [`BulkInterrupted`] with its `written` count — propagates only after
/// the rebuild has run, so a bulk load can never leave the file in the
/// disarmed state. An empty `rows` is a no-op: the index is not touched
/// for a load of nothing.
///
/// The rebuild turn is budget-exempt (atomic by necessity, same criterion
/// as `shadow_swap` — [`crate::CHUNK_BUDGET`]'s table), and its hold grows
/// with the corpus: ~10 ms/vector at dim 256, ~20 ms at 512. For a very
/// large backfill that hold is the price of the recipe; the alternative —
/// keep the index and pay the same total spread across rows — is what
/// `upsert_embeddings` already is.
pub async fn bulk_embeddings(
&self,
model: &ModelName,
rows: Vec<(String, Vec<f32>)>,
) -> BulkResult<usize> {
self.bulk_embeddings_with(model, rows, BulkControl::new())
.await
}
/// [`Self::bulk_embeddings`] with cancellation and progress, on the load
/// half. The drop and the rebuild are single statements; `progress` and
/// `cancel` bound the chunked load between them, exactly as they do for
/// [`Self::upsert_embeddings_with`].
pub async fn bulk_embeddings_with(
&self,
model: &ModelName,
rows: Vec<(String, Vec<f32>)>,
control: BulkControl,
) -> BulkResult<usize> {
// The table must exist before anything drops its index, and the first
// row's width is checked *before* the drop so a wholly wrong batch
// fails without ever disarming the storage check.
let dim = match crate::vector::declared_dimension(self.read_conn(), model).await {
Ok(dim) => dim,
Err(cause) => return Err(BulkInterrupted { written: 0, cause }),
};
if let Some((_, first)) = rows.first() {
if first.len() != dim {
return Err(BulkInterrupted {
written: 0,
cause: DbError::DimMismatch {
got: first.len(),
expected: dim,
model: model.to_string(),
},
});
}
} else {
return Ok(0);
}
if let Err(cause) = self
.low(move |responder| LowPriCommand::DropEmbeddingIndex {
model: model.clone(),
responder,
})
.await
{
// A drop that failed has left the index in place — the armed,
// searchable state — so stopping here is safe: the load never
// ran, and the file is exactly what it was.
return Err(BulkInterrupted { written: 0, cause });
}
let loaded = self
.low_chunked(rows, chunk_rows::EMBEDDINGS, control, |chunk, responder| {
LowPriCommand::UpsertEmbeddingChunk {
model: model.clone(),
chunk,
responder,
}
})
.await;
// Whether the load committed or was interrupted, the index goes back:
// the one state this method must never leave the file in is the state
// `drop_embedding_index` documents as unsearchable and unchecked.
let rebuilt = self
.low(move |responder| LowPriCommand::RebuildEmbeddingIndex {
model: model.clone(),
responder,
})
.await;
// The load's failure is the caller's answer; the rebuild's failure is
// worse and takes precedence, because a load that reports success
// beside a missing index is the silent half of the same defect. In the
// double-failure case the rebuild becomes the cause and the load's
// `written` count survives as the count.
match (loaded, rebuilt) {
(Ok(count), Ok(())) => Ok(count),
(Err(e), Ok(())) => Err(e),
(Ok(count), Err(e)) => Err(BulkInterrupted {
written: count,
cause: e,
}),
(Err(load), Err(rebuild)) => Err(BulkInterrupted {
written: load.written,
cause: rebuild,
}),
}
}
/// Reconstruct the concept-text search index from the ledger (§5.9, D-036).
///
/// The FTS index is derivative: D-036 promises every derivative table can be
/// rebuilt from the ledger tables, and this is that promise made callable
/// for `concepts_fts`. Needed after a restore that skipped the shadow
/// tables, or if the index is ever suspected of drifting from the text —
/// and, as a matter of policy, cheaper to run than to reason about.
///
/// The work is `INSERT INTO concepts_fts(concepts_fts) VALUES('rebuild')`,
/// which is FTS5's own operation over the content table, so this is not a
/// second implementation of the sync triggers that could disagree with them.
pub async fn rebuild_fts(&self) -> Result<()> {
self.low(|responder| LowPriCommand::RebuildFts { responder })
.await
}
/// Refresh the query planner's statistics (0.12.4, [D-149]).
///
/// Runs `ANALYZE`, which writes `sqlite_stat1`. **Before 0.12.4 nothing in
/// this crate ever did**, so the planner costed every query against SQLite's
/// built-in defaults — assume ~1M rows, assume each bound equality column
/// divides by ten. That estimate is structural: it depends on how many
/// columns a query binds, not on what the table contains.
///
/// Which is this schema's own worst defect restated. D-042, D-059 and D-064
/// are three occasions where *a covering index captured a query because it
/// contained the columns, not because it discriminated*, and two of the four
/// declared indices lead on the same column. Statistics are what let the
/// planner tell them apart by measurement instead of by shape.
///
/// # Cost, and why it is bounded
///
/// This is a write and it takes the write lock. `PRAGMA analysis_limit`
/// (set per connection, see [`ddl::ANALYSIS_LIMIT`]) caps the rows examined
/// per index. It is scheduled as low-priority work and will not preempt an
/// interactive assertion.
///
/// **The bound is a constant factor, not an independence** (0.12.23,
/// D-166). This rustdoc said the hold "scales with the number of indices —
/// four — and not with the size of `links_current`", which is measurably
/// wrong: the pragma is worth 3–4× and what remains still grows with the
/// table. Measured, `examples/analyze_hold.rs`: **5.26 ms at 10,000 edges,
/// 19.1 ms at 40,000**, against a 3 ms [`crate::CHUNK_BUDGET`].
///
/// So this call **misses the budget by ~6× on a moderately sized ledger**,
/// and [`crate::metrics::CommandKind::Analyze`] is deliberately not among
/// the budget-exempt kinds — `metrics().budget_violations()` names it. That
/// is the honest position: the work is low priority and preemptible between
/// commands, but it is one indivisible statement and cannot be chunked, so
/// the hold is what it is. Prefer [`optimize`], which does nothing when
/// nothing has moved.
///
/// **Since 0.13.24 the counter is this call and not also [`optimize`]**
/// (W10.5, [D-197]). The two shared `CommandKind::Analyze` until then, which
/// is why an `analyze` row in `budget_violations()` used to be unreadable:
/// it could have been an explicit call or a handle close.
///
/// # When to call it
///
/// After a bulk import, and after anything that changes a table's shape by
/// an order of magnitude. Prefer [`optimize`] for routine upkeep: it does
/// nothing when nothing has moved, and this does the work unconditionally.
///
/// Statistics are derived state in the sense Doctrine VI means it — deleting
/// `sqlite_stat1` costs plan quality and no information, and this call
/// rebuilds it.
///
/// [D-149]: ../docs/architecture/s13-decision-register.md#d-149
/// [`ddl::ANALYSIS_LIMIT`]: crate::schema::ddl::ANALYSIS_LIMIT
/// [`optimize`]: Database::optimize
pub async fn analyze(&self) -> Result<()> {
self.low(|responder| LowPriCommand::Analyze {
incremental: false,
responder,
})
.await
}
/// Re-analyse only what has gone stale (0.12.4, [D-149]).
///
/// `PRAGMA optimize`. SQLite tracks how far each table has drifted since its
/// last analysis and re-analyses only where it believes the statistics no
/// longer hold — so this is a no-op on an idle database and the full cost of
/// [`analyze`] on one that has changed completely.
///
/// That property is the whole point: it is safe to call on a schedule, where
/// [`analyze`] is not. `close()` runs it, so a process that opens, works and
/// closes keeps its statistics current without anybody arranging it.
///
/// # What it costs, measured, and the threshold it applies rather than takes
/// (0.13.24, W10.5, [D-197])
///
/// `examples/optimize_hold.rs`, on a 40,000-edge ledger: **10.7 ms the
/// first time on a database that has never been analysed** — there is
/// nothing incremental about the first call — and **90–220 µs every time
/// after**, well inside [`crate::CHUNK_BUDGET`].
///
/// **The staleness test is SQLite's and it is a ratio, not a row count.**
/// Measured by reading `sqlite_stat1` across the call rather than by timing
/// it: growth of 2× and 5× both left the statistics **untouched**, and
/// only at 25× did it re-analyse — for a 460 ms hold. So this is not a
/// cheaper `analyze()` and calling it after a bulk load is not a way to
/// refresh statistics the load invalidated: below the ratio it declines,
/// and above it it costs what [`analyze`] costs. It reports as
/// [`crate::metrics::CommandKind::Optimize`] since 0.13.24, which is what
/// makes those two outcomes distinguishable in the metrics at all.
///
/// [D-197]: ../docs/architecture/s13-decision-register.md#d-197
///
/// [D-149]: ../docs/architecture/s13-decision-register.md#d-149
/// [`analyze`]: Database::analyze
pub async fn optimize(&self) -> Result<()> {
self.low(|responder| LowPriCommand::Analyze {
incremental: true,
responder,
})
.await
}
// **There is deliberately no `verify_fts()` (§5.9, D-071).**
//
// `rebuild_fts` is the repair with no way to ask whether it is needed, and
// Wave 5 set out to add the missing half. FTS5 offers `'integrity-check'`,
// which looked like exactly the engine-provided answer this crate prefers.
// It is not: on libSQL 0.9.30 it verifies the index's *internal* consistency
// and not its agreement with the content table. Measured — after
// `'delete-all'` the index matches nothing where it matched ten rows, and
// both `'integrity-check'` and `'integrity-check', 0` still report success.
//
// A `verify_fts()` on that footing would answer "healthy" for an empty
// index, which is worse than having no method at all: it is the shape of
// defect AC, a function that looks like it checks something and does not.
// `an_emptied_fts_index_still_passes_integrity_check` pins the limitation so
// that if a later libSQL fixes it, the test fails and says so.
/// Write derived analytics results on the background channel, chunked
/// (§5.4, D-041).
///
/// Rows go to `analytics_annotations`, which has no log trigger, so nothing
/// written here reaches `transaction_log` and nothing here versions a
/// concept. Rerunning an algorithm replaces the previous pass rather than
/// recording that the world changed.
///
/// Low priority and chunked at up to [`chunk_rows::ANNOTATIONS`] — the
/// largest ceiling of the four, because this is the only bulk table carrying
/// no triggers at all
/// and its rows are correspondingly cheap (D-058) — so a 50,000-label Louvain
/// save yields to interactive writes at every chunk boundary and carries the
/// per-chunk fidelity boundary of §5.1.6 — a partially written pass is
/// recoverable by rerunning, which is the property that makes derived state
/// safe to write this way and assertions not.
///
/// Chunked, so it returns [`BulkInterrupted`] and its `written` count on
/// failure (0.13.8, W7.6); [`Self::write_analytics_annotations_with`] adds
/// cancellation and progress.
pub async fn write_analytics_annotations(
&self,
annotations: Vec<Annotation>,
) -> BulkResult<usize> {
self.write_analytics_annotations_with(annotations, BulkControl::new())
.await
}
/// [`Self::write_analytics_annotations`] with cancellation and progress
/// (0.13.8, W7.6).
pub async fn write_analytics_annotations_with(
&self,
annotations: Vec<Annotation>,
control: BulkControl,
) -> BulkResult<usize> {
self.low_chunked(
annotations,
chunk_rows::ANNOTATIONS,
control,
|chunk, responder| LowPriCommand::WriteAnalyticsChunk { chunk, responder },
)
.await
}
/// Move closed intervals and superseded log rows older than `cutoff` to the
/// cold database (§5.7, D-012).
pub async fn archive(&self, cutoff: &str) -> Result<ArchiveReport> {
let cutoff = timestamp::normalize(cutoff)?;
let archive_path = self.archive_path.clone();
self.low(|responder| LowPriCommand::Archive {
cutoff,
archive_path,
responder,
})
.await
}
/// Forget one lineage: move its whole ledger to the cold database and
/// remove the lineage record (0.14.13, §15.4, D-230).
///
/// The abandonment arm. A conversation tree discards most of what it grows,
/// and [`Self::archive`] cannot reclaim it: that arm is indexed by *time*,
/// so archiving an abandoned branch's recent history means archiving the
/// trunk's recent history with it.
///
/// **Everything the lineage holds moves in one transaction** — its `links`,
/// its `concepts`, its `transaction_log` entries and its `branches` row —
/// and afterwards the name is unknown: every read and write naming it
/// raises [`DbError::UnknownBranch`]. That is the design's whole shape, and
/// `temporal::archive::archive_branch` records why it has no smaller
/// version.
///
/// # It refuses more than it accepts, on purpose
///
/// - The trunk, and a name that is not registered
/// ([`DbError::UnknownBranch`]).
/// - A branch with **descendants**: they read through it, so archiving it
/// would delete rows they still believe.
/// - A branch whose **concepts another lineage's hot link names**. The road
/// map assumed an abandoned branch's rows were "a contiguous archivable
/// set by construction"; a concept is keyed by identity across the whole
/// ledger (D-214), so they are not, and this refusal is what makes them
/// contiguous in the cases it accepts.
///
/// All but the first return [`DbError::BranchNotArchivable`] with a reason.
///
/// The lineage record lands in `cold.branches` with an `archived_at`, so a
/// cold row's `branch_id` still resolves to something — in the cold file,
/// which is now the only place it does.
pub async fn archive_branch(&self, branch: crate::branch::BranchId) -> Result<ArchiveReport> {
let branch = branch.as_str().to_string();
let archive_path = self.archive_path.clone();
self.low(|responder| LowPriCommand::ArchiveBranch {
branch,
archive_path,
responder,
})
.await
}
/// Move the named concepts back from the cold database into the hot tables
/// (§2.3, C3).
///
/// Rehydration is a **physical move back, not a write**: it mints no
/// transaction-time facts and is invisible to both clocks. An id that is not
/// in the cold file is skipped rather than being an error — the caller
/// generally has a list from a cold-side query, and a partially-stale list is
/// the normal case rather than a mistake. The report says how many actually
/// moved.
///
/// See [`RehydrateReport::rowids_reassigned`] for the one way a rehydrated
/// row can differ from the row that was archived.
pub async fn rehydrate(&self, ids: &[&str]) -> Result<RehydrateReport> {
let ids: Vec<String> = ids.iter().map(|s| (*s).to_string()).collect();
let archive_path = self.archive_path.clone();
self.low(|responder| LowPriCommand::Rehydrate {
ids,
archive_path,
responder,
})
.await
}
/// Archive up to `cutoff` as a sequence of sessions, each covering at most
/// `window` of **transaction** time (T1.1, D-080).
///
/// `archive(cutoff)` is one transaction whose size is set by how long it has
/// been since the last one, which makes it the least bounded of the three
/// operations exempt from [`CHUNK_BUDGET`] — its hold is a function of
/// operational history rather than of anything a caller chose. This runs the
/// same work as *N* complete sessions, each with its own marker, horizon row
/// and rebuild, and returns one [`ArchiveReport`] per session in order.
///
/// # D-012 is satisfied per session, and that is what it requires
///
/// The atomicity D-012 demands is that copy-then-delete never be split — a
/// crash between the phases duplicates or loses rows. *N* small sessions
/// satisfy that exactly as one large one does. The obligation windowing adds
/// is that a partial run leave a coherent intermediate state, which it does:
/// each session commits a valid horizon, so a failure at window *k* leaves a
/// database archived up to boundary *k−1* and nothing in between. **The
/// sequence is not atomic and does not claim to be** — on error, the reports
/// for the sessions that did commit are lost with it, but their effect is
/// not, and re-running with the same `cutoff` completes the job.
///
/// # Each session is its own actor turn, and that is the entire point
///
/// This loop lives here, on the handle, rather than inside the actor's
/// `Archive` arm. Putting it there would have produced *N* small
/// transactions inside **one** hold, which shrinks the transaction and
/// changes the latency not at all: the actor is single-threaded, so nothing
/// else writes until its turn returns regardless of how many `COMMIT`s the
/// turn contains. Sending *N* commands returns the actor to its `select!`
/// between sessions, which is where an interactive assertion gets to jump
/// the queue — and it is high-priority, so it does.
///
/// The same reasoning is why [`Self::bulk_import`] chunks here and not
/// there, and it is the trap T1.2 names for `CREATE TABLE … AS SELECT`.
///
/// # Choosing a window
///
/// The bound is on *transaction* time, so the session count is set by how
/// far back the hot file goes, not by how much it holds. A window is
/// rejected rather than clamped if it would need more than
/// [`MAX_ARCHIVE_SESSIONS`] sessions — see [`DbError::ArchiveWindow`].
///
/// Windows containing nothing archivable are cheap but not free: each still
/// opens a transaction and writes a horizon row. What they no longer do is
/// re-project `links_current`, which `archive_session` now skips when its
/// `DELETE` removed no rows — without that, windowing costs *more* in total
/// than not windowing, because the repair term scales with the surviving
/// table and not with the batch (D-077).
pub async fn archive_windowed(
&self,
cutoff: &str,
window: std::time::Duration,
) -> Result<Vec<ArchiveReport>> {
let cutoff = timestamp::normalize(cutoff)?;
let boundaries = self.archive_boundaries(&cutoff, window).await?;
let mut reports = Vec::with_capacity(boundaries.len());
for boundary in boundaries {
let archive_path = self.archive_path.clone();
reports.push(
self.low(|responder| LowPriCommand::Archive {
cutoff: boundary,
archive_path,
responder,
})
.await?,
);
}
Ok(reports)
}
/// The cutoffs [`Self::archive_windowed`] will run, ascending, ending at
/// `cutoff` exactly.
///
/// Read on `read_conn`, not on the actor: this is two `MIN`s and the actor
/// has no reason to hold its lock for them.
///
/// The lower end comes from the data rather than from the clock. Stepping
/// from some fixed epoch would make the session count a function of the
/// calendar — a database opened yesterday would still be asked to archive
/// 1970 — whereas the oldest `recorded_at` actually present is the earliest
/// boundary that can contain anything.
async fn archive_boundaries(
&self,
cutoff: &str,
window: std::time::Duration,
) -> Result<Vec<String>> {
// A single session at `cutoff` is exactly `archive(cutoff)`, and it is
// the right answer for an empty hot file: it still writes the horizon
// row, so windowed and unwindowed runs leave the same observable state.
let Some(oldest) = self.oldest_hot_stamp(cutoff).await? else {
return Ok(vec![cutoff.to_string()]);
};
let start = timestamp::parse(&oldest)?;
let end = timestamp::parse(cutoff)?;
let Ok(span) = end.duration_since(start) else {
// Everything in the hot file is at or after the cutoff, so there is
// nothing in range to divide.
return Ok(vec![cutoff.to_string()]);
};
if window.is_zero() {
return Err(DbError::ArchiveWindow {
window,
reason: "a zero-length window never advances past the first boundary".into(),
});
}
// `div_ceil` on nanos: a span of 90 minutes in 60-minute windows is two
// sessions, not one. `as_nanos` is u128, so neither the division nor the
// span can overflow for any timestamp this crate can store.
let sessions = span.as_nanos().div_ceil(window.as_nanos());
if sessions > MAX_ARCHIVE_SESSIONS as u128 {
return Err(DbError::ArchiveWindow {
window,
reason: format!(
"a span of {span:?} would need {sessions} sessions (limit \
{MAX_ARCHIVE_SESSIONS}); widen the window"
),
});
}
let mut boundaries = Vec::with_capacity(sessions as usize);
for k in 1..sessions {
boundaries.push(timestamp::format(start + window * k as u32));
}
// The last boundary is `cutoff` itself and not `start + n*window`, which
// would overshoot and archive rows the caller excluded.
boundaries.push(cutoff.to_string());
Ok(boundaries)
}
/// Oldest `recorded_at` below `cutoff` in either hot table, or `None`.
async fn oldest_hot_stamp(&self, cutoff: &str) -> Result<Option<String>> {
let mut oldest: Option<String> = None;
for table in ["links", "transaction_log"] {
let found: Option<String> = self
.read_conn
.query(
&format!("SELECT MIN(recorded_at) FROM {table} WHERE recorded_at < ?1"),
libsql::params![cutoff],
)
.await?
.next()
.await?
.and_then(|row| row.get(0).ok());
if let Some(found) = found {
if oldest.as_ref().is_none_or(|o| found < *o) {
oldest = Some(found);
}
}
}
Ok(oldest)
}
/// Send a high-priority command and wait for its answer.
///
/// The two error mappings here are the whole reason this helper exists.
/// `send` failing means the actor is gone — `WriterUnavailable`. The
/// responder being dropped without an answer means the actor took the
/// command and never replied — `WriterDroppedResponder`, which is a bug in
/// the actor rather than a condition the caller can retry. Both variants
/// existed in `error.rs` from 0.4.5 and neither was ever constructed, so a
/// dead actor and a hung one were both just a caller waiting forever.
async fn high<T>(
&self,
make: impl FnOnce(oneshot::Sender<Result<T>>) -> HighPriCommand,
) -> Result<T> {
let (tx, rx) = oneshot::channel();
self.highpri_tx
.send(make(tx))
.await
.map_err(|_| DbError::WriterUnavailable)?;
rx.await.map_err(|_| DbError::WriterDroppedResponder)?
}
/// Send each chunk in turn and sum the counts — the shape all four bulk
/// paths share (T3.4, D-086).
///
/// # This is sequential on purpose, and the purpose is a measurement
///
/// T3.4 proposed pipelining: send *k* chunks ahead so the actor never finds
/// an empty queue. The reasoning is that awaiting each chunk before building
/// the next leaves the actor idle for a channel round trip every time, which
/// on a 1M-edge import is ~11,000 idle gaps.
///
/// Both halves of that are true and the conclusion does not follow. The gaps
/// are real; they are also **four orders of magnitude smaller than the work
/// they interrupt**. A tokio mpsc hop is sub-microsecond and a chunk takes
/// 13–21 ms. Implemented and swept at depths 1, 2, 4, 8 and 16 over 20K and
/// 100K edges: every cell landed within 1% of sequential, in both directions
/// — see `examples/pipeline_diag.rs`, which is kept precisely so this is not
/// re-proposed from the same reasoning.
///
/// So the pipelining was removed and the deduplication kept. It was not free
/// to hold: with chunks in flight, a failure at chunk `i` no longer leaves a
/// **prefix** committed, because `i+1 ..= i+k-1` were already sent and commit
/// anyway. D-011 promises "earlier chunks committed", and paying for that
/// with a weaker recovery story in exchange for nothing measurable is the
/// wrong trade.
///
/// Sending stops at the first error, so what commits is exactly the prefix
/// before the failure.
/// # The size is now measured, not assumed (0.12.0, W3)
///
/// Until 0.11.0 the caller pre-split into `chunks(chunk_rows::WHATEVER)` and
/// this loop sent what it was given. That made the constant *the* size, and
/// D-143 is the record of a constant fitted at one population being wrong at
/// another: all four D-088 shapes agreed the largest in-budget edge chunk was
/// **20** against a shipped 90, and 20 would itself have been wrong at 80,000
/// edges, because per-row cost on that path grows with `links_current`.
///
/// No row count can bound a duration on such a path, so the loop stopped
/// trying to pick one ahead of time. `ceiling` — still the path's
/// [`chunk_rows`] constant, with its derivation intact — is now the largest
/// size this will ever ask for, and each chunk's measured hold chooses the
/// next through `next_chunk_size`.
///
/// **Feedback, not preemption.** The chunk in flight always commits in full;
/// the SQLite write lock is not preemptible, so nothing here can shorten a
/// transaction already running. A batch of one chunk gets no protection at
/// all, and convergence costs one or two chunks — which is the price of the
/// bound being a duration rather than a promise.
///
/// The last chunk's outcome is discarded, there being no next chunk to size.
/// The chunk loop behind all four bulk paths.
///
/// **Every exit carries `written`** (0.13.8, W7.6, D-181). It used to
/// carry it only out of the success arm: the three error paths were `?` on
/// a [`DbError`], which discards the local, so a caller whose 20,000-row
/// import failed in the last chunk learned that it failed and not that
/// 19,000 rows were already in the database. The count was never expensive
/// to keep — it is right there, and the loop needs it anyway to size the
/// next chunk.
async fn low_chunked<T>(
&self,
items: Vec<T>,
ceiling: usize,
control: BulkControl,
make: impl Fn(Vec<T>, oneshot::Sender<Result<ChunkOutcome>>) -> LowPriCommand,
) -> BulkResult<usize> {
let total = items.len();
let mut items = items.into_iter();
let mut size = ceiling.max(1);
let mut written = 0usize;
loop {
let chunk: Vec<T> = items.by_ref().take(size).collect();
if chunk.is_empty() {
// Emptiness is checked before cancellation on purpose: a token
// raised after the last chunk committed is asking to stop work
// that is already done, and reporting that as a failure would
// make a race between the caller's two threads decide whether a
// complete import counts as one.
return Ok(written);
}
// Between chunks, never inside one. Nothing is rolled back and no
// transaction is interrupted -- the loop simply stops sending, and
// the prefix that committed is the same kind of prefix a failure
// would have left.
if control.is_cancelled() {
return Err(BulkInterrupted {
written,
cause: DbError::BulkCancelled,
});
}
let stop = |cause: DbError| BulkInterrupted { written, cause };
let (tx, rx) = oneshot::channel();
self.lowpri_tx
.send(make(chunk, tx))
.await
.map_err(|_| stop(DbError::WriterUnavailable))?;
let outcome = match rx.await {
Err(_) => return Err(stop(DbError::WriterDroppedResponder)),
Ok(Err(e)) => return Err(stop(e)),
Ok(Ok(outcome)) => outcome,
};
written += outcome.rows;
control.report(BulkProgress {
written,
total,
rows: outcome.rows,
held: outcome.held,
});
size = next_chunk_size(size, outcome.held, CHUNK_BUDGET, CHUNK_FLOOR, ceiling);
}
}
async fn low<T>(
&self,
make: impl FnOnce(oneshot::Sender<Result<T>>) -> LowPriCommand,
) -> Result<T> {
let (tx, rx) = oneshot::channel();
self.lowpri_tx
.send(make(tx))
.await
.map_err(|_| DbError::WriterUnavailable)?;
rx.await.map_err(|_| DbError::WriterDroppedResponder)?
}
/// Clean shutdown: stop the Write Actor, then write the final snapshot (§5.1.7).
///
/// Order matters. The snapshot is taken *after* the actor has stopped and
/// been joined, so no write can land between the fold and the file — the
/// anchor it records is the last thing that happened, not the last thing
/// that happened to be visible.
///
/// A failed snapshot is reported rather than swallowed. It is not a
/// durability loss — the ledger is in the WAL and the log replays without
/// it — but it means the next open starts from an older anchor, and a caller
/// that never hears about it cannot know why startup got slower.
///
/// **The cadence stops first (§5.5, D-053).** Both it and `write_final` end
/// by running retention over the snapshot directory, and retention deletes
/// files. Letting them overlap would mean one pass enumerating the directory
/// while the other removes from it — not a correctness problem for the
/// ledger, which is why the ordering is stated rather than locked, but a
/// source of spurious warnings and of a final anchor that could be deleted
/// by a cleanup that started before it existed. Stopping the cadence, then
/// the actor, then taking the snapshot leaves exactly one writer at each
/// step.
pub async fn close(mut self) -> Result<()> {
if let Some(stop) = self.cadence_stop.take() {
let _ = stop.send(true);
}
if let Some(handle) = self.cadence.take() {
let _ = handle.await;
}
// Top up the planner's statistics while the actor is still alive to do
// it (0.12.4, D-149). `PRAGMA optimize` re-analyses only what SQLite
// believes has gone stale, so on a database that did nothing this costs
// nothing, and on one that was just bulk-loaded it is the difference
// between the next process planning on measurements and planning on
// built-in guesses.
//
// **Deliberately not fatal.** A failure here costs plan quality on the
// next open and nothing else — no ledger state depends on it — and
// `close()` is where a caller learns whether their *writes* survived.
// Turning a stale-statistics problem into a failed close would bury that
// answer under a much less important one.
if let Err(e) = self.optimize().await {
tracing::warn!(
"PRAGMA optimize failed during close(): {e}. Statistics may be \
stale for the next process; call analyze() to rebuild them. \
Nothing else is affected."
);
}
let (tx, rx) = oneshot::channel();
let _ = self
.highpri_tx
.send(HighPriCommand::Shutdown { responder: tx })
.await;
let _ = rx.await;
// **The writer's exit status is propagated, not discarded (Wave 4.2).**
// It used to be `let _ = handle.await`, so an actor that had died closed
// "successfully" and the caller's last chance to learn that the write
// path was gone was spent silently.
//
// Through 0.13.3 this awaited a `JoinHandle<Result<()>>` and did
// `Ok(res) => res?`, which looked like two failure paths and was one:
// the actor's `Result` could not be `Err` (W7.3, D-177). What remains is
// the branch that can fire — the actor panicked or was aborted — mapped
// by `writer_exit`, which is tested against a real `JoinError`.
//
// Ordered before the final snapshot on purpose: a snapshot written after
// a dead writer records a state the caller has no reason to trust, and
// returning the error while also having written that file is worse than
// not writing it.
if let Some(handle) = self.writer.take() {
writer_exit(handle.await)?;
}
let ts = self.clock.now();
let archive = crate::temporal::archive::archive_present(&self.archive_path)
.then_some(self.archive_path.as_path());
snapshot::write_final(&self.read_conn, &self.snapshots_dir, &ts, archive).await?;
// Marks the handle closed so `Drop` knows not to complain.
self.closed = true;
Ok(())
}
}
/// Notes a missed `close()` at `warn!`, and deliberately does **not** assert.
///
/// **§7.3 offered option B — document `close()` as mandatory and `debug_assert`
/// in `Drop` — and Wave 4.2 implemented it, measured the consequence, and
/// reduced it to a warning.** The assert fired on roughly thirty tests on its
/// first run. That is the signal it was built to produce, and the right reading
/// of it was not "thirty tests are wrong".
///
/// What dropping actually costs is one final snapshot. Nothing else: every
/// public write method awaits its responder, so by the time a caller *can* drop
/// the handle, every write it issued has already committed; and the cadence stops
/// on its own, because `cadence_stop` is a `watch::Sender` whose drop signals the
/// task. A snapshot is derivative state under Doctrine VI — disposable,
/// reconstructible, and never the only copy of anything. Losing one makes the
/// next `reconstruct` fold from an older anchor, which is **slower, not wrong**.
///
/// A `debug_assert` aborts a test run. Spending that on a performance loss, in a
/// project whose own notes say a suite that fails for reasons unrelated to the
/// code under test trains people to ignore red, is the wrong trade — and paying
/// it in thirty places would have made `close()` look mandatory by ceremony
/// rather than by consequence. `close()` remains the right thing to call, and
/// the two reasons to call it are now stated where they can be acted on: the
/// snapshot, and the writer's `Result`, which only `close()` can return.
///
/// Option A ("abort the actor and log") stays rejected, for the reason it was
/// rejected twice before: `Drop` cannot await, so it cannot drain, and cleanup
/// that cannot clean up is worse than none — it looks like cleanup.
impl Drop for Database {
fn drop(&mut self) {
if !self.closed {
tracing::warn!(
"Database dropped without close(): the final snapshot was not written, \
so the next reconstruct folds from an older anchor, and the write \
actor's exit status was not checked. Prefer close().await."
);
}
}
}
/// A failure before the first chunk was sent, which committed nothing.
///
/// Normalisation runs over the whole batch up front, so its errors are the one
/// class the chunk loop never sees — and they are still [`BulkInterrupted`],
/// because a caller matching on one error type should not have to match on two
/// to find out that nothing landed (0.13.8, W7.6).
fn before_any_chunk(cause: DbError) -> BulkInterrupted {
BulkInterrupted { written: 0, cause }
}
fn normalize_all(edges: Vec<EdgeAssertion>) -> Result<Vec<EdgeAssertion>> {
edges.into_iter().map(EdgeAssertion::normalized).collect()
}
/// One `FULL` checkpoint for the numbers, then a `TRUNCATE` for the file.
///
/// # Why `TRUNCATE` and not a mode parameter
///
/// The four SQLite modes are not four things a caller of *this* crate wants.
/// `PASSIVE` is what the automatic checkpointer already runs on its own, so an
/// explicit `PASSIVE` asks for something that was going to happen anyway;
/// `RESTART` and `FULL` differ from `TRUNCATE` only in whether the WAL file is
/// left at its high-water size. The reason W5.2 exists is
/// [`Tuning::wal_autocheckpoint`] — a bulk importer turns the automatic
/// checkpointer off and calls this once at the end — and what that caller wants
/// is the WAL *gone*, not smaller than it was. So the mode is fixed and decided
/// here rather than pushed to the caller as a choice they would have to read
/// SQLite's documentation to make. If a mode ever needs selecting, that is an
/// additive method, not a change to this one.
///
/// # Why it is two pragmas, which is not the obvious implementation
///
/// **A successful `TRUNCATE` reports `busy=0, log=0, checkpointed=0`** — the
/// counts describe the WAL *after* the operation, and after a truncation there
/// is no WAL to describe. Measured, not inferred: on a 387-frame WAL, `PASSIVE`
/// returns `0, 387, 387` and `TRUNCATE` on the same file returns `0, 0, 0`. So
/// the single-pragma implementation returns a [`CheckpointReport`] whose two
/// counts are structurally zero on success, which makes the whole struct a
/// less useful `bool`.
///
/// `FULL` copies every frame back and reports what it moved; the `TRUNCATE`
/// that follows finds nothing left to copy and resets the file. The second pass
/// is close to free for exactly that reason — it is a file operation, not a
/// second copy. `busy` is the **union**: a checkpoint that was blocked in
/// either phase did not fully happen, and a caller about to copy the database
/// file elsewhere needs the pessimistic answer.
///
/// # They return rows, so they go through `query()`
///
/// The same libsql constraint the pragmas in `configure` document: `execute()`
/// rejects any statement that yields rows, and these yield the row that is the
/// entire point.
async fn run_checkpoint(conn: &libsql::Connection) -> Result<CheckpointReport> {
// The columns are `busy, log, checkpointed`. SQLite reports -1 for the two
// counts when the checkpoint could not run; clamped to 0 rather than
// surfaced as a signed count, because `busy` already carries "this did not
// happen" and a negative frame count is not a quantity anyone can use.
//
// A database not in WAL mode returns no row at all. `configure` puts every
// connection this crate opens into WAL, so that is unreachable here — but a
// zeroed report is a better failure than a panic if it stops being.
async fn one(conn: &libsql::Connection, sql: &str) -> Result<(bool, u64, u64)> {
let mut rows = conn.query(sql, ()).await?;
let Some(row) = rows.next().await? else {
return Ok((false, 0, 0));
};
let field = |i: i32| -> u64 { row.get::<i64>(i).unwrap_or(0).max(0) as u64 };
Ok((row.get::<i64>(0).unwrap_or(0) != 0, field(1), field(2)))
}
let (full_busy, _, moved) = one(conn, "PRAGMA wal_checkpoint(FULL)").await?;
let (trunc_busy, log_frames, _) = one(conn, "PRAGMA wal_checkpoint(TRUNCATE)").await?;
Ok(CheckpointReport {
busy: full_busy || trunc_busy,
log_frames,
checkpointed_frames: moved,
})
}
/// Pragmas that mean something on **any** connection, including one opened
/// `SQLITE_OPEN_READ_ONLY` (0.12.16, W5.5, D-159).
///
/// Both of these are per-connection state that a reader is subject to just as a
/// writer is. `busy_timeout` is the one that made this a finding:
/// [`Database::diagnostic_conn`] ran with SQLite's default of **0** — return
/// `SQLITE_BUSY` immediately — while every other connection in the process
/// waited 5 s, so the one surface whose job is to answer questions when the
/// typed path is already suspect was also the one most likely to fail with
/// "database is locked" under exactly the contention that prompted the
/// question.
/// Empty the slot unless what is in it is fit to hand to the next caller
/// (0.15.15, W15.5, [D-257]).
///
/// Shared by [`Database::diagnostic_conn`], which runs it on the way in, and
/// [`Database::scrub_diagnostic_conn`], which the Python binding runs on the way
/// out. One implementation because the two must agree: a connection the entry
/// path would have discarded is one the exit path must not leave sitting there.
async fn scrub(slot: &mut Option<libsql::Connection>) {
let Some(conn) = slot.as_ref() else {
return;
};
// A leaked `BEGIN` is the one that leaves this surface: it pins a WAL read
// snapshot, so later diagnostic reads answer from it and `checkpoint()`
// cannot truncate past it. Rolled back rather than reported, because the
// caller who would read the report is the one who did not do it.
let recovered = conn.is_autocommit() || conn.execute("ROLLBACK", ()).await.is_ok();
// A connection that will not roll back is not one to hand on, and neither
// is one whose own state cannot be read: either way the answer is a fresh
// connection, which costs 56.5 us on the call that dirtied it and nothing
// on any other.
if !recovered || diagnostic_is_dirty(conn).await.unwrap_or(true) {
*slot = None;
}
}
/// Is this cached diagnostic connection carrying state from an earlier caller
/// (0.15.15, W15.5, [D-257])?
///
/// Two questions — are there temp objects, is anything attached beyond `main`
/// and `temp` — asked as pragmas rather than as a query over
/// `temp.sqlite_master` and `pragma_database_list`. Same answers, and
/// `examples/diagnostic_hygiene_probe.rs` measures the pragma form at
/// **2.4 µs** against the query form's **7.8 µs**, on a call whose whole warm
/// cost is the `stat` in front of it. Detection is measured rather than
/// assumed: `temp.schema_version` goes 0 → 1 on `CREATE TEMP TABLE`, and
/// `database_list` 2 → 3 on `ATTACH`.
///
/// **What it cannot see is a `PRAGMA`**, and neither can any other cheap check:
/// SQLite does not enumerate connection-scoped pragma state. The crate restates
/// the two it sets instead — see the call to `configure_common` in
/// [`Database::diagnostic_conn`].
///
/// An `Err` here is not propagated by the caller: a connection whose own state
/// cannot be read is replaced rather than reported on.
///
/// [D-257]: ../../docs/architecture/s13-decision-register.md#d-257
async fn diagnostic_is_dirty(conn: &libsql::Connection) -> Result<bool> {
let mut rows = conn.query("PRAGMA temp.schema_version", ()).await?;
let temp_schema: i64 = match rows.next().await? {
Some(row) => row.get(0)?,
// No row at all is not a clean connection, it is an answer this
// function did not understand.
None => return Ok(true),
};
if temp_schema != 0 {
return Ok(true);
}
let mut rows = conn.query("PRAGMA database_list", ()).await?;
let mut databases = 0_usize;
while rows.next().await?.is_some() {
databases += 1;
}
// `main` and `temp`, always both, on a connection nobody has attached to.
Ok(databases > 2)
}
async fn configure_common(conn: &libsql::Connection, cache_size: Option<i32>) -> Result<()> {
// NOTE: `busy_timeout` returns its resulting value as a row, and libsql's
// `execute()` rejects any statement that yields rows ("Execute returned
// rows"). It must be issued through `query()`.
let _ = conn.query("PRAGMA busy_timeout = 5000", ()).await?;
// Per-connection, and split writer from reader since 0.12.15 (W5.4,
// D-158). `None` runs no pragma at all rather than restating SQLite's
// default, so the default remains SQLite's to change.
if let Some(pages) = cache_size {
conn.execute(&format!("PRAGMA cache_size = {pages}"), ())
.await?;
}
Ok(())
}
/// Pragmas that only mean anything where writes can happen (0.12.16, W5.5).
///
/// Not run on [`Database::diagnostic_conn`], and the reason is not tidiness:
/// `journal_mode = WAL` is a change to the *database file*, which a connection
/// opened `SQLITE_OPEN_READ_ONLY` cannot make. The rest —
/// `synchronous`, `foreign_keys`, `recursive_triggers`, and the `ANALYZE`
/// bound — govern how writes behave, and a connection that cannot write is not
/// governed by them.
///
/// The write connection and the two internal readers all still get these. The
/// internal readers are opened from the same read-write `libsql::Database`, so
/// the pragmas apply; leaving them out would be a behaviour change made for
/// symmetry, which is not a reason.
async fn configure_writable(conn: &libsql::Connection) -> Result<()> {
// Returns its resulting value as a row — see the note in `configure_common`.
let _ = conn.query("PRAGMA journal_mode = WAL", ()).await?;
conn.execute("PRAGMA synchronous = NORMAL", ()).await?;
conn.execute("PRAGMA foreign_keys = ON", ()).await?;
conn.execute("PRAGMA recursive_triggers = OFF", ()).await?;
// Bounds every `ANALYZE` this connection will ever run, explicit or
// triggered by `PRAGMA optimize` (D-149). Set here rather than around the
// call sites so the scheduled path is bounded too — that is the half that
// runs with nobody watching. Returns the previous limit as a row, so it goes
// through `query()` for the reason the note above gives.
let _ = conn.query(crate::schema::ddl::ANALYSIS_LIMIT, ()).await?;
Ok(())
}
/// Full pragma configuration, for a connection that can write.
async fn configure(
conn: libsql::Connection,
cache_size: Option<i32>,
) -> Result<libsql::Connection> {
configure_writable(&conn).await?;
configure_common(&conn, cache_size).await?;
Ok(conn)
}
/// Helper to derive the snapshot directory by convention: foo.db -> foo_snapshots/
fn derive_snapshots_dir(path: &Path) -> PathBuf {
let mut dir = path.to_path_buf();
let stem = path
.file_stem()
.and_then(|s| s.to_str())
.unwrap_or("macrame");
dir.set_file_name(format!("{stem}_snapshots"));
dir
}
/// Helper to derive archive database path by convention: foo.db -> foo_archive.db
fn derive_archive_path(path: &Path) -> PathBuf {
let mut archive = path.to_path_buf();
if let Some(stem) = path.file_stem().and_then(|s| s.to_str()) {
let ext = path.extension().and_then(|e| e.to_str()).unwrap_or("db");
archive.set_file_name(format!("{stem}_archive.{ext}"));
} else {
archive.set_extension("archive.db");
}
archive
}
/// Dedicated Write Actor event loop prioritizing high-priority UI requests over low-priority background work.
///
/// # The turn is the unit, not the statement (T1.4)
///
/// One iteration of this loop is one *hold*: the actor is single-threaded and
/// the SQLite write lock is not preemptible, so from the moment a command starts
/// executing until it returns, nothing else writes. That is the quantity
/// [`CHUNK_BUDGET`] bounds, and so it is the quantity
/// [`crate::metrics::ActorMetrics`] measures — deliberately around the whole
/// `execute` call rather than inside it. Timing the SQL alone would have
/// reported a bound that held while callers waited.
///
/// Queue depth is sampled *before* the `select!`, so it is the backlog the turn
/// found on arrival rather than the one it left behind.
///
/// # `biased` has no floor, and since 0.12.10 that is measured (W4.4, D-153)
///
/// `biased` makes the arms poll in declaration order, so high-priority work is
/// taken whenever any is ready. Nothing bounds how long that can continue:
/// sustained interactive traffic can hold the low tier off indefinitely, and
/// through 0.12.9 nothing in the crate could say whether it ever did.
/// `record_priority_choice` counts the turns where the choice went against
/// queued low-priority work, and the longest unbroken run of them, which is the
/// half that distinguishes "prioritised" from "starved".
///
/// **No forced yield is added here.** Whether one is needed is the question the
/// counter answers, and adding a policy now would be fixing a bound nobody has
/// observed being hit — the same mistake D-124 was retracted for.
///
/// # It returns nothing, and used to return a `Result` it could not fail
/// (0.13.4, W7.3, §3.5, [D-177])
///
/// The two exits are `LoopCtl::Break` from [`HighPriCommand::Shutdown`] and the
/// `else` arm when both channels are closed. Neither can fail, and neither
/// could before: every command's error goes back on that command's own
/// responder, where the caller who issued it can act on it. There was no third
/// thing for an actor-level `Err` to carry, and none was ever constructed.
///
/// A `Result` that is structurally always `Ok` is not free. It reads as a
/// failure path under review, so `close()`'s `res?` looked like it was doing
/// something, and the branch that actually fires — a **panicked** actor,
/// reported as a `JoinError` — sat beside it untested. That is the swap this
/// change makes: the unfireable branch is gone and the real one is pinned, in
/// [`writer_exit`].
async fn run_writer_actor(
conn: libsql::Connection,
clock: Arc<dyn Clock>,
mut highpri_rx: mpsc::Receiver<HighPriCommand>,
mut lowpri_rx: mpsc::Receiver<LowPriCommand>,
shared: Arc<ActorShared>,
) {
// Owned by the loop and lent to each command, which is the whole of A-3:
// the actor had a connection and no memory, so every turn re-established
// what the turn before it had just established (0.15.6, W14.3, D-248).
let mut state = ActorState::new();
loop {
// Read once and reused by both the depth sample and the starvation
// counter, so the two cannot disagree about what was queued when this
// turn went looking (W4.4, D-153).
let low_queued = lowpri_rx.len();
shared.metrics.record_turn(highpri_rx.len(), low_queued);
let ctl = tokio::select! {
biased;
Some(cmd) = highpri_rx.recv() => {
shared.metrics.record_priority_choice(true, low_queued);
let turn = Turn::start(cmd.kind(), &shared);
cmd.execute(&conn, &*clock, &turn, &mut state).await
}
Some(cmd) = lowpri_rx.recv() => {
shared.metrics.record_priority_choice(false, low_queued);
let turn = Turn::start(cmd.kind(), &shared);
cmd.execute(&conn, &*clock, &turn, &mut state).await
}
else => LoopCtl::Break,
};
if matches!(ctl, LoopCtl::Break) {
break;
}
}
}
/// Turn the write actor's join status into the error `close()` reports.
///
/// One line of mapping, given a name so it can be tested against a real
/// [`tokio::task::JoinError`]. Before 0.13.4 this was inline beside a `res?` on
/// an actor `Result` that could only ever be `Ok`, and the arrangement had the
/// coverage exactly backwards: the branch that cannot fire was plumbed through
/// two signatures, and the branch that does fire — the actor panicked, and the
/// caller's writes are going nowhere — had no test at all (W7.3, D-177).
///
/// Cancellation is folded in with panics deliberately. `JoinError` distinguishes
/// them, and nothing in the crate ever aborts this task, so a cancelled writer
/// means something outside the crate reached in and stopped it. That is not a
/// gentler condition than a panic and must not read as one.
fn writer_exit(joined: std::result::Result<(), tokio::task::JoinError>) -> Result<()> {
joined.map_err(|e| DbError::WriterStopped(format!("the write actor did not exit cleanly: {e}")))
}
/// One command's hold: the timer, its label, and the counters it reports to.
///
/// # The hold is recorded *before* the caller is answered, and it has to be
///
/// The obvious placement — time the whole `execute` call from the loop — is
/// wrong in a way that only shows up under test. Every arm of `execute` ends by
/// sending on a `oneshot`, which wakes the waiting caller; the actor then
/// returns to the loop and records. Those are two tasks, so a caller that awaits
/// its own write and immediately reads [`Database::metrics`] can be scheduled
/// first and see a turn count that does not include the write it just did.
///
/// Not a correctness bug in the ledger, and it would never have been noticed in
/// production — a dashboard sampling every few seconds cannot see the window.
/// It makes every test and diagnostic of the counters flaky, which is worse: the
/// instrumentation would have been *believed* while being wrong exactly when
/// someone tried to check it. `examples/bulk_atomic_diag.rs` was the thing that
/// caught it, reporting a 20,000-row batch as a 0 ms hold.
///
/// So `answer` records and then sends, in that order, and the ordering is the
/// method's whole reason to exist. What it costs is that the `oneshot::send`
/// itself falls outside the measurement, which is a few nanoseconds against a
/// turn measured in microseconds at best.
struct Turn<'a> {
kind: crate::metrics::CommandKind,
timer: crate::metrics::HoldTimer,
shared: &'a ActorShared,
}
/// State the actor owns and a `Turn` needs to reach.
///
/// `archive_epoch` is here rather than in [`crate::metrics::ActorMetrics`]
/// because it is **not** a metric: T1.2's shadow rebuild reads it to decide
/// whether its work is still valid, so it has to be present in every build, not
/// only under the `metrics` feature. Counting archives happens to be what both
/// want; only one of them is allowed to be compiled out.
///
/// `turns` is here for the same reason and serves the snapshot cadence — see
/// its own note.
#[derive(Default)]
struct ActorShared {
metrics: crate::metrics::ActorMetrics,
archive_epoch: std::sync::atomic::AtomicU64,
/// Commands this actor has answered `Ok` to (0.15.19, review C-19).
///
/// # What it is for, and why it is not a `seq_id`
///
/// `snapshot::run_cadence` used to run `SELECT MAX(seq_id), MAX(recorded_at)`
/// on **every tick**, five seconds apart by default, whether or not
/// anything had been written. Two aggregates on an idle database, for a
/// fact the actor already had: *nothing has happened*.
///
/// The review asked for a `watch<u64>` of the last committed `seq_id`. This
/// is the same idea one step cheaper, and the difference matters. The actor
/// does not currently know the `seq_id` its writes produced — the log rows
/// are written by triggers — so publishing one would mean adding a query to
/// **every write** in order to remove a query from an idle timer, which is
/// the wrong direction. A turn count needs no query at all: one relaxed
/// `fetch_add` on a path already doing a database round trip.
///
/// # Why this cannot change when a snapshot is written
///
/// The cadence skips its tick when the count has not moved since the last
/// one. That is sound because the implication runs the right way: if
/// `MAX(seq_id)` grew, some command committed, so some turn answered `Ok`,
/// so the count moved. The converse is not claimed and does not need to be
/// — a turn that answered `Ok` without writing a log row makes the cadence
/// do exactly the query it used to do every time. It over-counts, never
/// under-counts, and the tick it protects had nothing to do anyway.
///
/// `Relaxed` because nothing is ordered against it. The cadence reads a
/// number to compare with a number it read before; a value one tick stale
/// costs one deferred tick and no correctness, and the same is true of the
/// `archive_epoch` beside it.
turns: std::sync::atomic::AtomicU64,
}
impl crate::temporal::snapshot::CommittedTurns for ActorShared {
fn committed_turns(&self) -> u64 {
self.turns.load(std::sync::atomic::Ordering::Relaxed)
}
}
impl<'a> Turn<'a> {
fn start(kind: crate::metrics::CommandKind, shared: &'a ActorShared) -> Self {
Self {
kind,
timer: crate::metrics::HoldTimer::start(),
shared,
}
}
fn epoch(&self) -> u64 {
self.shared
.archive_epoch
.load(std::sync::atomic::Ordering::Relaxed)
}
/// Record that an archive session committed.
///
/// Bumped on **success only**: a failed archive rolls back, so it deletes
/// nothing and invalidates no shadow build.
fn archive_committed(&self) {
self.shared
.archive_epoch
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
/// Close the hold and hand the result back. Never the other way round.
///
/// The `let _ =` on the send is deliberate and predates this: a caller that
/// dropped its receiver — `tokio::time::timeout` around a write, which
/// [`Database`]'s write surface explicitly documents — is not an actor
/// error, and the command committed regardless.
fn answer<T>(&self, responder: oneshot::Sender<Result<T>>, res: Result<T>) {
self.shared
.metrics
.record_hold(self.kind, self.timer.elapsed());
if res.is_ok() {
self.turn_committed();
}
let _ = responder.send(res);
}
/// [`answer`](Self::answer) for a chunk: the same reading, handed back to the
/// caller as well as recorded (0.12.0, W1).
///
/// One `elapsed()` serves both, so the duration the chunk loop sizes against
/// is *the same number* the histogram shows — a controller and a dashboard
/// disagreeing about what a chunk cost would be a bad way to spend a
/// debugging session.
///
/// The record-then-send ordering documented on [`Turn`] is preserved, and
/// matters here for the same reason: the send wakes the caller, which may be
/// scheduled before this method returns.
fn answer_chunk(&self, responder: oneshot::Sender<Result<ChunkOutcome>>, res: Result<usize>) {
let held = self.timer.elapsed();
self.shared.metrics.record_hold(self.kind, held);
if res.is_ok() {
self.turn_committed();
}
let _ = responder.send(res.map(|rows| ChunkOutcome { rows, held }));
}
/// Record that a turn answered `Ok`, for [`ActorShared::turns`].
///
/// Called from both answer paths rather than from the loop, because those
/// are the two places that know the result. A command that returns an error
/// does not bump it: a failed write rolls back and the log is where it was.
fn turn_committed(&self) {
self.shared
.turns
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
}
const INSERT_LINK: &str = "INSERT INTO links \
(source_id, target_id, edge_type, valid_from, valid_to, weight, properties, \
recorded_at, branch_id) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)";
/// The parameter row for [`INSERT_LINK`], in one place since 0.14.8.
///
/// The single-edge path and the chunk path spelled these out separately, which
/// was survivable at eight and is not at nine: `branch_id` is the one parameter
/// whose omission is *silent* — the column defaults to `'main'`, so a path that
/// forgot it would write to the trunk and pass every test that did not fork.
/// [`concept_params`] has existed for this reason since D-056.
fn edge_params<'a>(edge: &'a EdgeAssertion, stamp: &'a str) -> [libsql::Value; 9] {
[
edge.source.as_str().into(),
edge.target.as_str().into(),
edge.edge_type.as_str().into(),
edge.valid_from.as_str().into(),
edge.valid_to.as_str().into(),
edge.weight.into(),
edge.properties.as_str().into(),
stamp.into(),
edge.branch_name().into(),
]
}
/// Shared by the single-concept write and the chunked one, so the two paths
/// cannot drift into upserting different column sets — and so the chunk has a
/// statement text it can prepare once (D-056).
const UPSERT_CONCEPT: &str = "INSERT INTO concepts \
(id, title, content, embedding_model, valid_from, valid_to, recorded_at, retired, \
branch_id) \
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) \
ON CONFLICT(id) DO UPDATE SET \
title = excluded.title, \
content = excluded.content, \
embedding_model = excluded.embedding_model, \
valid_from = excluded.valid_from, \
valid_to = excluded.valid_to, \
recorded_at = excluded.recorded_at, \
retired = excluded.retired";
// `branch_id` is deliberately **not** in that `DO UPDATE` list. The column is
// provenance and minting happened once (D-214), and
// `trg_concepts_branch_immutable` would abort an update that moved it — so
// listing it would turn every re-upsert of an inherited concept into a guard
// abort instead of the no-op it is. The insert arm carries it; the update arm
// leaves the row where it was minted.
/// The parameter row for [`UPSERT_CONCEPT`], in one place for the same reason.
fn concept_params<'a>(concept: &'a ConceptUpsert, stamp: &'a str) -> [libsql::Value; 9] {
[
concept.id.as_str().into(),
concept.title.as_str().into(),
concept.content.as_str().into(),
concept
.embedding_model
.as_deref()
.map_or(libsql::Value::Null, Into::into),
concept.valid_from.as_str().into(),
concept.valid_to.as_str().into(),
stamp.into(),
(concept.retired as i64).into(),
concept.branch_name().into(),
]
}
/// Check every lineage a write names, and decide which shape its guard takes.
///
/// **One function, two answers, one query per distinct lineage** — and it is
/// [`Lineages::shape`](crate::graph::lineage::Lineages::shape), the same function
/// the read path calls, for the same reason it calls it. A write naming a
/// branch that is not in `branches` has asked about something that does not
/// exist, and answering it by writing to the trunk is [D-069]'s failure in its
/// most expensive form: not a right-looking answer to a question that was not
/// asked, but a *durable* one.
///
/// Relying on the foreign key instead would refuse the write — `branch_id`
/// `REFERENCES branches(branch_id)` and the key is enforced — but it would
/// refuse it as an unqualified "FOREIGN KEY constraint failed" from inside a
/// rolled-back transaction, naming neither the column nor the branch. The same
/// argument [`classify`](crate::error::classify) makes for annotations and
/// edges, one table further along.
///
/// # Why a trunk write pays for it too
///
/// `None` resolves to `'main'` here rather than skipping the check, and that is
/// not tidiness. Once a second lineage can write, the *trunk's* overlap guard
/// is wrong in the other direction — it would be refused for overlapping a
/// branch's belief it cannot see — so the shape decision is one every write
/// needs, not one that branched writes need. On a database that has never
/// forked the answer is [`LineageShape::Trunk`] and the guard is the statement
/// it has always been.
///
/// # This was a query per name until 0.15.6
///
/// It ran [`crate::graph::lineage::Lineages::shape`] once per name and kept the
/// last answer — one round trip per write, for a table only this task writes.
/// [`ActorState`] holds `branches` instead, and [`Lineages::shape_of`] carries
/// what is left of this function's reasoning, including the part about the last
/// answer that stopped being true at 0.15.2.
///
/// [D-069]: ../../docs/architecture/s13-decision-register.md
async fn check_lineages(
state: &mut ActorState,
conn: &libsql::Connection,
names: &[&str],
) -> Result<LineageShape> {
state.shape_of(conn, names).await
}
/// The shape and the rows, for a caller that also has to resolve an ancestry.
///
/// [`check_lineages`] with the table it read handed back rather than dropped
/// (0.15.17). Both callers need it: the guard compiles a statement per lineage,
/// and the retirement binds one.
async fn check_lineages_with<'a>(
state: &'a mut ActorState,
conn: &libsql::Connection,
names: &[&str],
) -> Result<(LineageShape, &'a Lineages)> {
let lineages = state.lineages(conn).await?;
Ok((lineages.shape_of(names)?, lineages))
}
/// The distinct lineages a batch names, in first-seen order.
///
/// A `Vec` and a linear scan rather than a set: batches name one lineage in
/// every case this crate has, the bound is the number of *branches* and not the
/// number of rows, and a `BTreeSet` would allocate per batch to deduplicate a
/// list of length one.
fn distinct_branches(edges: &[EdgeAssertion]) -> Vec<&str> {
let mut out: Vec<&str> = Vec::with_capacity(1);
for edge in edges {
let name = edge.branch_name();
if !out.contains(&name) {
out.push(name);
}
}
if out.is_empty() {
out.push(crate::schema::ddl::MAIN_BRANCH);
}
out
}
/// The overlap guard's prepared statement, and which question it asks.
///
/// The two statements take different parameter counts and mean different things
/// by the rows they return, so pairing them with the shape here is what stops
/// [`check_prepared`] from having to be told twice.
struct OverlapGuard {
stmt: libsql::Statement,
shape: LineageShape,
/// The lineage this statement was compiled for. See [`Self::prepare`].
branch: String,
/// That lineage's ancestry, bound after the statement's own parameters.
/// Empty under both trunk shapes, which emit no `lineage` relation.
ancestry: Vec<Ancestor>,
}
impl OverlapGuard {
/// Prepare once per turn or per chunk, never per row (D-056, §8.8).
///
/// # One statement per *lineage* since 0.15.17 ([D-259])
///
/// The statement used to be a function of the shape alone: the recursive
/// `lineage` CTE derived the ancestry from the branch bound at `?5`, so one
/// compiled form answered for every lineage that shared a shape. A bound
/// ancestry is not derived from anything — it *is* the answer for one
/// reader — so the guard now carries the lineage it was compiled for and
/// the values that lineage binds.
///
/// The cost is bounded by [`distinct_branches`], which is a `Vec` because
/// every batch this crate has seen names one lineage. A chunk that names
/// two prepares two, which is the price of the resolution being correct for
/// both; the shape it would otherwise share is `Resolved`, since the two
/// trunk shapes each describe a database with exactly one lineage to name.
///
/// [D-259]: ../../docs/architecture/s13-decision-register.md#d-259
async fn prepare(
conn: &libsql::Connection,
shape: LineageShape,
lineages: &Lineages,
branch: &str,
) -> Result<Self> {
// Three shapes, three statements, one spelling (0.15.8, W13.3,
// D-250). Until that release the trunk had a hand-written constant and
// the other two shared the resolved form, which was exact for a root
// only because a root's ancestry is itself — so `Lineages::shape_of`
// could return either of them and nothing observable changed. It
// cannot now: the root gets a two-predicate lookup on the projection
// and a branch gets the four-CTE resolution, and D-248's C-24 repair
// is what decides which.
let ancestry = match shape {
LineageShape::Resolved => lineages.ancestry(branch),
_ => Vec::new(),
};
let sql = crate::graph::lineage::overlap_candidates_resolved(shape, &ancestry);
Ok(Self {
stmt: conn.prepare(&sql).await?,
shape,
branch: branch.to_string(),
ancestry,
})
}
/// Whether this guard answers for `branch` under `shape`.
fn answers_for(&self, shape: LineageShape, branch: &str) -> bool {
self.shape == shape && self.branch == branch
}
}
impl HighPriCommand {
/// The metrics label for this variant (T1.4).
///
/// Exhaustive for the same reason `execute` is: a new variant that silently
/// borrowed another's label would attribute its holds to the wrong command,
/// and the one question the counters exist to answer is *which* command
/// broke the budget.
fn kind(&self) -> crate::metrics::CommandKind {
use crate::metrics::CommandKind as K;
match self {
HighPriCommand::AssertEdge { .. } => K::AssertEdge,
HighPriCommand::RetireEdge { .. } => K::RetireEdge,
HighPriCommand::UpsertConcept { .. } => K::UpsertConcept,
HighPriCommand::WriteBulkAtomic { .. } => K::WriteBulkAtomic,
HighPriCommand::RebuildCurrent { .. } => K::RebuildCurrent,
HighPriCommand::RegisterModel { .. } => K::RegisterModel,
HighPriCommand::Fork { .. } => K::Fork,
HighPriCommand::Checkpoint { .. } => K::Checkpoint,
HighPriCommand::Shutdown { .. } => K::Shutdown,
}
}
/// Run one command and answer its caller.
///
/// Deliberately exhaustive — there is no `_` arm. The 0.4.5–0.5.4 actor
/// matched `Shutdown` and `AssertEdge` and sent everything else to
/// `_ => LoopCtl::Continue`, which **dropped the responder**: the caller's
/// `rx.await` resolved to a `RecvError` that no code mapped, so four of six
/// commands were indistinguishable from a hung database. An exhaustive match
/// makes that failure a compile error instead of a runtime silence, which is
/// why adding a variant should break this function.
async fn execute(
self,
conn: &libsql::Connection,
clock: &dyn Clock,
turn: &Turn<'_>,
state: &mut ActorState,
) -> LoopCtl {
match self {
HighPriCommand::Shutdown { responder } => {
turn.answer(responder, Ok(()));
return LoopCtl::Break;
}
HighPriCommand::Checkpoint { responder } => {
let res = run_checkpoint(conn).await;
turn.answer(responder, res);
}
HighPriCommand::AssertEdge { edge, responder } => {
let stamp = clock.now();
// Before the guard, because a write naming an unregistered
// lineage should be refused by name rather than by whatever the
// guard happens to find when it looks in the wrong place.
let shape = match check_lineages(state, conn, &[edge.branch_name()]).await {
Ok(shape) => shape,
Err(e) => {
turn.answer(responder, Err(e));
return LoopCtl::Continue;
}
};
if let Err(e) = reject_overlapping_interval(state, conn, &edge, shape).await {
turn.answer(responder, Err(e));
return LoopCtl::Continue;
}
// The statement is held across turns, so it is reset before it
// is bound rather than after it was stepped — `check_prepared`
// makes the same argument at more length.
let res = match state.insert_link(conn).await {
Err(e) => Err(e),
Ok(stmt) => {
stmt.reset();
match stmt.execute(edge_params(&edge, &stamp)).await {
Ok(_) => Ok(()),
Err(e) => Err(classify(
conn,
e,
WriteOp::Edge {
source_id: &edge.source,
target_id: &edge.target,
edge_type: &edge.edge_type,
},
)
.await),
}
}
};
turn.answer(responder, res);
}
HighPriCommand::RetireEdge {
source,
target,
edge_type,
valid_from,
valid_to,
branch,
responder,
} => {
let stamp = clock.now();
let name = branch
.as_ref()
.map_or(crate::schema::ddl::MAIN_BRANCH, |b| b.as_str());
let resolved = check_lineages_with(state, conn, &[name])
.await
.map(|(shape, l)| (shape, l.ancestry(name)));
let res = match resolved {
Ok((shape, ancestry)) => {
retire_edge(
conn,
&source,
&target,
&edge_type,
&valid_from,
&valid_to,
&stamp,
name,
shape,
&ancestry,
)
.await
}
Err(e) => Err(e),
};
turn.answer(responder, res);
}
HighPriCommand::UpsertConcept { concept, responder } => {
let stamp = clock.now();
let res = match check_lineages(state, conn, &[concept.branch_name()]).await {
Ok(_) => upsert_concept(conn, &concept, &stamp).await,
Err(e) => Err(e),
};
turn.answer(responder, res);
}
HighPriCommand::WriteBulkAtomic { edges, responder } => {
// One stamp for the whole batch (D-014): the rows were asserted
// by one act, and giving them different transaction times would
// invent an ordering the caller never expressed.
let stamp = clock.now();
let res = write_edges_atomic(state, conn, &edges, &stamp).await;
turn.answer(responder, res);
}
HighPriCommand::RebuildCurrent { responder } => {
turn.answer(responder, rebuild_current(conn).await);
}
HighPriCommand::RegisterModel {
model,
dim,
responder,
} => {
turn.answer(
responder,
crate::vector::register_model(conn, &model, dim).await,
);
}
HighPriCommand::Fork {
name,
parent,
responder,
} => {
// The same clock as every other write, and the same instant in
// both columns: `forked_at` is a transaction-time point in the
// parent's history, and the point this release can fork from is
// now. See `branch::Branch::created_at` for why they are two
// columns anyway.
let stamp = clock.now();
let res = crate::branch::fork(conn, &name, &parent, &stamp).await;
// `branches` has a row it did not have. Unconditional rather
// than `if res.is_ok()`: a fork that failed leaves the table
// as it was, so forgetting costs one query and asserting that
// it failed cleanly costs an argument (0.15.6, D-248).
state.forget_lineages();
turn.answer(responder, res);
}
}
LoopCtl::Continue
}
}
impl LowPriCommand {
/// The metrics label for this variant (T1.4). See [`HighPriCommand::kind`].
fn kind(&self) -> crate::metrics::CommandKind {
use crate::metrics::CommandKind as K;
match self {
LowPriCommand::WriteConceptsChunk { .. } => K::WriteConceptsChunk,
LowPriCommand::WriteAnalyticsChunk { .. } => K::WriteAnalyticsChunk,
LowPriCommand::UpsertEmbeddingChunk { .. } => K::UpsertEmbeddingChunk,
LowPriCommand::DropEmbeddingIndex { .. } => K::DropEmbeddingIndex,
LowPriCommand::RebuildEmbeddingIndex { .. } => K::RebuildEmbeddingIndex,
LowPriCommand::BulkImportChunk { .. } => K::BulkImportChunk,
LowPriCommand::LinksCurrentMirror { .. } => K::LinksCurrentMirror,
LowPriCommand::Archive { .. } => K::Archive,
// Its own counter since 0.12.9 (W4.3, D-152). It reported as
// `K::Archive` from 0.9.0 to 0.12.8 — the budget really is shared,
// but attribution is not budget, and an operator reading a long
// `archive` hold could not tell whether anything had been archived.
// What kept it folded was that a `CommandKind` variant was a
// breaking addition; `#[non_exhaustive]` (W4.2) removed that.
LowPriCommand::Rehydrate { .. } => K::Rehydrate,
// Its own counter from the day it shipped, which is the whole point
// of the paragraph above: `Rehydrate` spent four releases folded
// into `Archive` for a reason that was never good, and the cost of
// unfolding it was a rung's worth of care about declaration order.
LowPriCommand::ArchiveBranch { .. } => K::ArchiveBranch,
LowPriCommand::RebuildFts { .. } => K::RebuildFts,
// Two kinds out of one variant since 0.13.24 (W10.5, D-197). The
// command carries the flag; the counter has to carry it too, or the
// budget exemption for either half is decided about both (D-168).
LowPriCommand::Analyze { incremental, .. } => {
if *incremental {
K::Optimize
} else {
K::Analyze
}
}
// Two kinds out of one variant since 0.14.16 (W12.16, D-233),
// and for D-197's reason one line up: the command carries the step,
// so the counter has to carry it too, or the budget exemption for
// either half is decided about both. Here that is not hypothetical
// — the halves want opposite answers. The swap is over budget by
// construction and the fill chunks are meant to fit, so a merged
// kind's `over_budget` read `N(rebuilds) + regressions` and could
// not be decomposed.
LowPriCommand::ShadowRebuild { step, .. } => match step {
crate::integrity::ShadowStep::Swap { .. } => K::ShadowSwap,
crate::integrity::ShadowStep::Begin | crate::integrity::ShadowStep::Fill { .. } => {
K::ShadowRebuild
}
},
}
}
/// Run one background command and answer its caller.
///
/// Also exhaustive. The pre-0.5.4 version was a single `LoopCtl::Continue`
/// for *every* variant — every background write silently discarded, its
/// caller waiting forever.
async fn execute(
self,
conn: &libsql::Connection,
clock: &dyn Clock,
turn: &Turn<'_>,
state: &mut ActorState,
) -> LoopCtl {
match self {
LowPriCommand::BulkImportChunk { chunk, responder } => {
// A stamp per chunk, not per batch: the chunks commit
// separately, so a shared stamp would claim a simultaneity the
// storage does not have.
let stamp = clock.now();
turn.answer_chunk(
responder,
write_edges_atomic(state, conn, &chunk, &stamp).await,
);
}
LowPriCommand::LinksCurrentMirror { present, responder } => {
// Schema work on the actor's own connection, like
// RegisterModel and the embedding-index pair: the
// single-writer invariant is what keeps the mirror's window
// on one timeline with the chunks it wraps.
let sql = if present {
crate::schema::ddl::CREATE_LINKS_CURRENT_SYNC.to_string()
} else {
"DROP TRIGGER IF EXISTS trg_links_current_sync".to_string()
};
turn.answer(
responder,
conn.execute(&sql, ()).await.map(|_| ()).map_err(Into::into),
);
}
LowPriCommand::WriteConceptsChunk { chunk, responder } => {
let stamp = clock.now();
turn.answer_chunk(
responder,
write_concepts_atomic(state, conn, &chunk, &stamp).await,
);
}
LowPriCommand::WriteAnalyticsChunk { chunk, responder } => {
let stamp = clock.now();
turn.answer_chunk(
responder,
write_annotations_atomic(conn, &chunk, &stamp).await,
);
}
LowPriCommand::UpsertEmbeddingChunk {
model,
chunk,
responder,
} => {
// No clock reading: an embedding carries no timestamp on either
// axis. It is a derived artifact of a model applied to content
// (Doctrine VII), and the ledger already records when the
// content changed.
turn.answer_chunk(
responder,
crate::vector::search::upsert_embedding_chunk(conn, &model, &chunk).await,
);
}
LowPriCommand::DropEmbeddingIndex { model, responder } => {
// Schema work on the actor's own connection, like
// RegisterModel: the single-writer invariant is what keeps the
// drop and the load on one timeline.
turn.answer(
responder,
crate::vector::registry::drop_embedding_index(conn, &model).await,
);
}
LowPriCommand::RebuildEmbeddingIndex { model, responder } => {
// One statement, one turn, no smaller unit. Its hold is the
// one-pass DiskANN build's and it is budget-exempt by the same
// criterion as `ShadowSwap` — see the CommandKind doc.
turn.answer(
responder,
crate::vector::registry::rebuild_embedding_index(conn, &model).await,
);
}
LowPriCommand::Archive {
cutoff,
archive_path,
responder,
} => {
// The archive *time*, not the cutoff. `archive_horizon` records
// both and they are different facts — see `archive()` (Wave 4.5).
let archived_at = clock.now();
let res = archive(conn, &cutoff, &archived_at, &archive_path).await;
// The session attached and detached a second database. The
// statements are dropped rather than trusted to recompile —
// see [`ActorState::forget_statements`].
state.forget_statements();
// Before the answer, so a shadow rebuild that reads the epoch on
// its next turn cannot miss an archive that has already deleted
// rows out from under it (T1.2).
if res.is_ok() {
turn.archive_committed();
}
turn.answer(responder, res);
}
LowPriCommand::ArchiveBranch {
branch,
archive_path,
responder,
} => {
// The wall clock, recorded in `cold.branches.archived_at`: when
// the ledger stopped knowing about the lineage. Not a ledger
// fact and not on either of Doctrine II's timelines — nothing
// was asserted or retired here.
let archived_at = clock.now();
let res = crate::temporal::archive::archive_branch(
conn,
&branch,
&archived_at,
&archive_path,
)
.await;
// Both: this session attaches a second database *and* deletes
// the lineage's row from `branches`.
state.forget_everything();
// `Archive`'s reason exactly: a shadow rebuild reading the epoch
// on its next turn must not miss a session that has already
// deleted rows out from under it (T1.2).
if res.is_ok() {
turn.archive_committed();
}
turn.answer(responder, res);
}
LowPriCommand::Rehydrate {
ids,
archive_path,
responder,
} => {
let refs: Vec<&str> = ids.iter().map(String::as_str).collect();
let res = rehydrate(conn, &refs, &archive_path).await;
// Attaches, like the two archive sessions. It restores concepts
// and log rows, never a lineage — `cold.branches` is read by
// `archive_hint` and not written back — so the lineages stand.
state.forget_statements();
// Same reason as `Archive`: rehydration moves rows into `links`'
// parent table, so a shadow rebuild in flight must see the epoch
// move before the caller is answered (T1.2).
if res.is_ok() {
turn.archive_committed();
}
turn.answer(responder, res);
}
LowPriCommand::ShadowRebuild { step, responder } => {
use crate::integrity::{shadow, ShadowOutcome, ShadowStep};
let res = match step {
ShadowStep::Begin => {
shadow::begin(conn)
.await
.map(|build_start| ShadowOutcome::Started {
build_start,
epoch: turn.epoch(),
})
}
ShadowStep::Fill { after } => shadow::fill_chunk(conn, after.as_deref())
.await
.map(|last| ShadowOutcome::Filled { last }),
ShadowStep::Swap { build_start, epoch } => {
shadow::swap(conn, &build_start, epoch, turn.epoch())
.await
.map(|rows| ShadowOutcome::Swapped { rows })
}
};
turn.answer(responder, res);
}
LowPriCommand::RebuildFts { responder } => {
let res = conn
.execute(crate::schema::ddl::REBUILD_CONCEPTS_FTS, ())
.await
.map(|_| ())
.map_err(Into::into);
turn.answer(responder, res);
}
LowPriCommand::Analyze {
incremental,
responder,
} => {
// Both go through `query()`, not `execute()`. `PRAGMA optimize`
// yields rows, and libsql's `execute()` rejects any statement
// that does ("Execute returned rows") — the same trap
// `configure` documents. `ANALYZE` does not yield rows, but is
// issued the same way so the two arms cannot drift into needing
// different call shapes for no visible reason.
let sql = if incremental {
crate::schema::ddl::OPTIMIZE
} else {
crate::schema::ddl::ANALYZE
};
let res = conn.query(sql, ()).await.map(|_| ()).map_err(Into::into);
turn.answer(responder, res);
}
}
LoopCtl::Continue
}
}
/// Close an open interval by asserting its successor (Doctrine III).
///
/// Never an `UPDATE`. The replacement row copies weight and properties from
/// current belief and differs only in `valid_to` and `recorded_at`, so the
/// original assertion survives intact and `reconstruct` at an earlier instant
/// still sees the interval open — which is the entire point of a bitemporal
/// ledger.
// The first of these in the crate proper (0.14.8). All nine are the edge key,
// two stamps and the lineage — a struct to carry them would exist for one call
// site and would put a name between the caller and parameters it already spells
// out positionally at the only place it calls this.
#[allow(clippy::too_many_arguments)]
async fn retire_edge(
conn: &libsql::Connection,
source: &str,
target: &str,
edge_type: &str,
valid_from: &str,
valid_to: &str,
stamp: &str,
branch: &str,
shape: LineageShape,
ancestry: &[Ancestor],
) -> Result<()> {
// Shadow retirement: the row being closed may belong to an ancestor, and
// the row written carries *this* lineage's id. See
// `lineage::retire_from_resolved`.
//
// One statement for all three shapes since 0.15.8 (W13.3, D-250). The
// trunk had its own until then, kept apart on [`LineageShape`]'s ground —
// the resolved form was opaque to the planner and cost 3.0x where there
// was nothing to resolve (D-220). That ground is gone rather than
// overruled: a keyed `Trunk` resolution lowers to no CTEs at all, so the
// statement the lowering emits *is* the one this arm used to hold, with
// the lineage stamped rather than defaulted.
// The ancestry follows the seven, at `RETIRE_ANCESTRY_SLOT`.
let mut params: Vec<libsql::Value> = vec![
source.into(),
target.into(),
edge_type.into(),
valid_from.into(),
branch.into(),
valid_to.into(),
stamp.into(),
];
params.extend(crate::graph::lineage::ancestry_params(ancestry));
let affected = conn
.execute(
&crate::graph::lineage::retire_from_resolved(shape, ancestry),
params,
)
.await
.map_err(DbError::Engine)?;
if affected == 0 {
return Err(DbError::NotFound(format!(
"{source} -> {target} ({edge_type}) at {valid_from}"
)));
}
Ok(())
}
async fn upsert_concept(
conn: &libsql::Connection,
concept: &ConceptUpsert,
stamp: &str,
) -> Result<()> {
let res = conn
.execute(UPSERT_CONCEPT, concept_params(concept, stamp))
.await;
match res {
Ok(_) => Ok(()),
Err(e) => Err(classify(
conn,
e,
WriteOp::Concept {
id: &concept.id,
recorded_at: stamp,
branch: concept.branch_name(),
},
)
.await),
}
}
/// Whether this pair is the storage layer's case rather than this guard's.
///
/// Two **open** intervals overlap — they share every instant from the later
/// start onwards — so a naive overlap check reports them, and reporting them
/// here would leave `DbError::SingleOpenViolation` constructible by nothing.
/// That variant is the more specific error, it is enforced by
/// `trg_links_single_open` rather than by this function, and its field names
/// were ratified in §1.2. Shadowing it with a general one would be defect Q's
/// shape reintroduced by a fix: a typed error that no code path can produce.
///
/// So the two guards partition the space rather than overlapping it. Both open
/// belongs to the trigger. Everything else — open against closed, closed against
/// closed — is unguarded at the storage layer and belongs here. That the split
/// is exactly the trigger's `WHEN` clause is not a coincidence; it is the
/// definition of what was missing.
fn defer_to_single_open(proposed: &Interval, existing: &Interval) -> bool {
proposed.is_open() && existing.is_open()
}
/// Refuse an assertion whose valid-time interval overlaps one already recorded
/// for the same `(source, target, edge_type)` — **defect AA, D-060**.
///
/// `trg_links_single_open` fires only `WHEN NEW.valid_to = '9999-…'`, so it
/// guards the open sentinel and nothing else. Two *closed* intervals that
/// overlap were accepted without complaint, and `query_as_of_edges` at an
/// instant inside both returned one relationship as two edges.
///
/// **This runs in the write actor, which is what makes it sound.** The obvious
/// place is `EdgeAssertion::normalized`, and it cannot go there — `normalized`
/// is a pure function with no connection, and doing the read at the API boundary
/// instead would leave a check-then-write race between the read and the actor's
/// insert. Inside the actor there is one writer by construction (D-014), and for
/// the batch paths this runs inside the same transaction as the insert, so the
/// window does not exist rather than being small.
///
/// **What it does not cover, and §4.2 now says so:** raw SQL against the same
/// file. The storage layer permits what this API refuses, which is the honest
/// cost of not putting the check in a trigger. The alternative was a second
/// index probe inside `trg_links_single_open` on every insert — on the path
/// D-059 has just finished making fast — for a guarantee that only holds against
/// callers who were going through the actor anyway.
///
/// `valid_from <> ?4` excludes the row being re-asserted. Re-assertion at the
/// same `valid_from` is Doctrine III's ordinary case — a new belief about the
/// same interval — and is settled by the primary key and the single-open
/// trigger, not here.
/// The single-assertion path holds one statement across turns (0.15.6, D-248);
/// the batch path prepares one inside its own transaction and calls
/// [`check_prepared`] per row.
async fn reject_overlapping_interval(
state: &mut ActorState,
conn: &libsql::Connection,
edge: &EdgeAssertion,
shape: LineageShape,
) -> Result<()> {
let branch = edge.branch_name().to_string();
check_prepared(state.guard(conn, shape, &branch).await?, edge).await
}
/// The guard's body, against a statement the caller has already prepared.
///
/// **Split out because preparing per row was worth 10.4 ms on a 90-edge chunk**
/// (§8.8) — the same defect D-056 and D-057 diagnosed and fixed for
/// `INSERT_LINK`, reintroduced by the Wave 2 guard that was written beside it.
/// Measured with and without the guard, on a 2,000-edge hub: 8.65 ms → 19.25 ms,
/// and *identical* with and without `idx_lc_open_interval`, which is what
/// identified preparation rather than a scan as the cost. A guard that reads an
/// index correctly and prepares its statement 90 times is indistinguishable, at
/// the call site, from one that scans.
///
/// `reset()` between rows is not optional: libsql binds and steps without
/// resetting, so a reused statement must be returned to its initial state.
///
/// # And once more on the way out (0.15.6, W14.3)
///
/// The reset used to be enough at the top, because the guard was compiled per
/// call or per chunk and dropped where it was made — the drop finalized it, and
/// SQLite's objection to a live statement never came up. [`ActorState`] holds
/// this one across turns, and the loop below can leave a cursor open on it: the
/// overlap arm returns from inside the `while`. A statement left mid-scan is
/// what makes SQLite refuse to end a transaction, so the next `Archive` — not
/// the next assertion — would be the thing that failed, a command and a
/// diagnosis apart from the code that caused it. So the scan is a function of
/// its own, and the statement is reset on both ways out of it.
async fn check_prepared(guard: &OverlapGuard, edge: &EdgeAssertion) -> Result<()> {
let proposed = Interval::new(edge.valid_from.clone(), edge.valid_to.clone());
guard.stmt.reset();
// The resolved form takes a fifth parameter, the writing lineage, and
// returns what that lineage can see; the trunk form takes four and returns
// the table. Binding five to the trunk statement would be an error from
// libsql rather than a wrong answer, which is the failure mode to prefer.
let mut rows = match guard.shape {
LineageShape::Trunk => {
guard
.stmt
.query(libsql::params![
edge.source.as_str(),
edge.target.as_str(),
edge.edge_type.as_str(),
edge.valid_from.as_str()
])
.await?
}
LineageShape::Resolved | LineageShape::TrunkOnForked => {
// The ancestry follows the five, at `GUARD_ANCESTRY_SLOT`, and is
// empty for `TrunkOnForked` — a root emits no `lineage` relation.
let mut params: Vec<libsql::Value> = vec![
edge.source.as_str().into(),
edge.target.as_str().into(),
edge.edge_type.as_str().into(),
edge.valid_from.as_str().into(),
edge.branch_name().into(),
];
params.extend(crate::graph::lineage::ancestry_params(&guard.ancestry));
guard.stmt.query(params).await?
}
};
let verdict = scan_candidates(&mut rows, &proposed, edge).await;
drop(rows);
guard.stmt.reset();
verdict
}
/// The guard's loop, over candidates the statement has already produced.
///
/// Split from [`check_prepared`] so that the statement is reset on both exits
/// from it, including the one that returns an overlap.
async fn scan_candidates(
rows: &mut libsql::Rows,
proposed: &Interval,
edge: &EdgeAssertion,
) -> Result<()> {
while let Some(row) = rows.next().await? {
let existing = Interval::new(row.get::<String>(0)?, row.get::<String>(1)?);
if defer_to_single_open(proposed, &existing) {
continue;
}
if proposed.overlaps(&existing) {
return Err(DbError::OverlappingInterval {
overlap: Box::new(crate::error::Overlap {
source_id: edge.source.clone(),
target_id: edge.target.clone(),
edge_type: edge.edge_type.clone(),
valid_from: edge.valid_from.clone(),
valid_to: edge.valid_to.clone(),
existing_from: existing.valid_from,
existing_to: existing.valid_to,
// This guard reads committed rows, so the interval it names
// is one the caller can go and look at (D-180).
within_batch: false,
}),
});
}
}
Ok(())
}
/// The same guard applied *within* a batch, before any of it is written.
///
/// The database check cannot see rows that are not in the database yet, so a
/// batch carrying two overlapping intervals for one relationship would pass
/// every per-row check and commit the overlap in one transaction.
///
/// # Sorted and swept rather than compared pairwise (0.13.6, W7.5, D-179)
///
/// This used to compare every pair. At [`chunk_rows::EDGES`] = 90 that is
/// nothing, and the chunked paths are the only ones where 90 is the bound —
/// [`Database::write_bulk_atomic`] is exempt from [`CHUNK_BUDGET`] by contract,
/// so its batch is whatever the caller passed, and the quadratic term is what
/// made 20,000 corrections to one relationship's history cost seconds rather
/// than milliseconds. Sorting by `(source, target, edge_type, valid_from)` and
/// sweeping costs `n log n` and changes nothing a caller can observe except the
/// wait.
///
/// **Adjacent pairs are not sufficient, and that is the whole difficulty.** For
/// plain intervals they would be: sort by start, and if any two overlap then
/// some neighbouring two overlap. That proof needs every pair to be *eligible*,
/// and here two are not — identical `valid_from` is re-assertion rather than
/// overlap, and two open intervals belong to `trg_links_single_open`. Skip an
/// adjacent pair for either reason and a real overlap can hide behind it:
/// `[5,20)`, `[5,6)`, `[7,8)` has the first pair skipped for equal `valid_from`
/// and the second not overlapping, while `[5,20)` and `[7,8)` overlap plainly.
/// So the sweep carries the widest `valid_to` reached so far instead of looking
/// only backwards one step, and carries a second one restricted to closed
/// intervals — because an open predecessor is excluded for an open candidate
/// and eligible for a closed one, which are different questions with different
/// answers.
///
/// Equal `valid_from` is handled by advancing in runs: everything with the same
/// start is checked against the maxima, and only then folded into them, so the
/// members of a run never see each other.
///
/// The report names the *earlier* interval as the existing one, which is the
/// pairwise version's input order only by accident. Within a batch neither is
/// older in transaction time — they arrive under one stamp — so valid-time order
/// is the only ordering that means anything, and it is the one a reader will
/// assume the words carry.
fn reject_overlaps_within(edges: &[EdgeAssertion]) -> Result<()> {
// Indices, not the edges. The batch is borrowed and its order is the order
// the rows are written in; sorting it would either clone it or reorder the
// caller's data, which is `estimated_bulk_hold`'s reason for grouping too.
let mut order: Vec<u32> = (0..edges.len() as u32).collect();
order.sort_unstable_by(|&i, &j| {
let a = &edges[i as usize];
let b = &edges[j as usize];
(
&a.source,
&a.target,
&a.edge_type,
a.branch_name(),
&a.valid_from,
)
.cmp(&(
&b.source,
&b.target,
&b.edge_type,
b.branch_name(),
&b.valid_from,
))
});
fn key(e: &EdgeAssertion) -> (&str, &str, &str, &str) {
(
e.source.as_str(),
e.target.as_str(),
e.edge_type.as_str(),
e.branch_name(),
)
}
let at = |k: usize| &edges[order[k] as usize];
let mut group = 0;
while group < order.len() {
let mut group_end = group + 1;
while group_end < order.len() && key(at(group_end)) == key(at(group)) {
group_end += 1;
}
// The furthest `valid_to` reached by anything already swept in this key
// group, and the edge it came from so the error can name it. The second
// one ignores open intervals: an open candidate may not be compared
// against an open predecessor, and the sentinel would otherwise win the
// maximum every time and make every such pair look like an overlap.
let mut widest: Option<&EdgeAssertion> = None;
let mut widest_closed: Option<&EdgeAssertion> = None;
let mut run = group;
while run < group_end {
let mut run_end = run + 1;
while run_end < group_end && at(run_end).valid_from == at(run).valid_from {
run_end += 1;
}
for k in run..run_end {
let e = at(k);
let existing = if e.valid_to == timestamp::OPEN_SENTINEL {
widest_closed
} else {
widest
};
let Some(p) = existing else { continue };
// `Interval::overlaps` is `max(from) < min(to)`, and the sort
// has already settled the max: `p.valid_from <= e.valid_from`.
// What is left is the same predicate with the maximum resolved,
// and it is written out rather than allocating two `Interval`s
// per row to ask the same question.
if e.valid_from < p.valid_to && e.valid_from < e.valid_to {
return Err(DbError::OverlappingInterval {
overlap: Box::new(crate::error::Overlap {
source_id: e.source.clone(),
target_id: e.target.clone(),
edge_type: e.edge_type.clone(),
valid_from: e.valid_from.clone(),
valid_to: e.valid_to.clone(),
existing_from: p.valid_from.clone(),
existing_to: p.valid_to.clone(),
// Nothing here is in the database, and the batch is
// refused whole, so nothing here ever will be. The
// message has to say so (D-180).
within_batch: true,
}),
});
}
}
for k in run..run_end {
let e = at(k);
if widest.is_none_or(|w| e.valid_to > w.valid_to) {
widest = Some(e);
}
if e.valid_to != timestamp::OPEN_SENTINEL
&& widest_closed.is_none_or(|w| e.valid_to > w.valid_to)
{
widest_closed = Some(e);
}
}
run = run_end;
}
group = group_end;
}
Ok(())
}
/// Write every edge or none, under a single stamp.
///
/// **The statement is prepared once for the whole chunk (§9, D-056).** It used to
/// be `tx.execute(INSERT_LINK, …)` per row, which re-prepares on every call — and
/// `links` carries two triggers, so each preparation compiles their bodies along
/// with the insert.
///
/// Measured at 500 rows: **≈62 ms → ≈37 ms, a 41% saving.** Preparation was a
/// large cost and *not* the dominant one, which the first guess had it as. The
/// residual is the triggers themselves: the same 500 rows with
/// `trg_links_log_insert` and `trg_links_current_sync` dropped commit in **2.96
/// ms**, so trigger amplification is ~92% of what remains. There is no further
/// win available here without changing what the ledger records, and Doctrine IV
/// is what says it must be recorded. See D-056 for what that implies about §9's
/// ≤ 3 ms budget — briefly, 2.96 ms *is* the un-amplified figure, so the budget
/// appears to have been set without the amplification its own preamble says is
/// included.
///
/// `reset()` between rows is not optional: libsql's `execute` binds and steps
/// without resetting, so a reused statement must be returned to its initial state
/// or the second row steps a completed statement.
async fn write_edges_atomic(
state: &mut ActorState,
conn: &libsql::Connection,
edges: &[EdgeAssertion],
stamp: &str,
) -> Result<usize> {
if edges.is_empty() {
return Ok(0);
}
// Before the transaction opens: a batch that contradicts itself is refused
// without taking the write lock at all (D-060), and a batch naming a
// lineage that does not exist is refused before it can take the lock at all
// (0.14.8).
reject_overlaps_within(edges)?;
let branches = distinct_branches(edges);
let (shape, lineages) = check_lineages_with(state, conn, &branches).await?;
let lineages = lineages.clone();
let tx = conn
.transaction_with_behavior(libsql::TransactionBehavior::Immediate)
.await?;
// Inside the transaction, so the rows this checks against cannot change
// between the check and the insert.
// One preparation for the whole chunk, not one per row — see
// `check_prepared`, and D-056 for the same lesson learned on `INSERT_LINK`.
// One per lineage the chunk names, not one per row — see
// `OverlapGuard::prepare` for why the shape alone stopped being enough, and
// `distinct_branches` for why this is a `Vec` of length one in every batch
// this crate has seen.
let mut guards = Vec::with_capacity(branches.len());
for name in &branches {
guards.push(OverlapGuard::prepare(&tx, shape, &lineages, name).await?);
}
for edge in edges {
let guard = guards
.iter()
.find(|g| g.answers_for(shape, edge.branch_name()))
.expect("a guard per distinct branch, and the row names one of them");
if let Err(e) = check_prepared(guard, edge).await {
// Released before the rollback: a live statement on the connection
// is what makes SQLite refuse to end a transaction.
drop(guards);
let _ = tx.rollback().await;
return Err(e);
}
}
drop(guards);
let stmt = tx.prepare(INSERT_LINK).await?;
for edge in edges {
stmt.reset();
let res = stmt.execute(edge_params(edge, stamp)).await;
if let Err(e) = res {
let typed = classify(
&tx,
e,
WriteOp::Edge {
source_id: &edge.source,
target_id: &edge.target,
edge_type: &edge.edge_type,
},
)
.await;
// Released before the rollback: a live statement on the connection
// is exactly what makes SQLite refuse to end a transaction.
drop(stmt);
let _ = tx.rollback().await;
return Err(typed);
}
}
drop(stmt);
tx.commit().await?;
Ok(edges.len())
}
/// Write every concept or none, under a single stamp.
/// Upsert one chunk of derived annotations in a single transaction (D-041).
///
/// `stamp` is the actor's clock reading, exactly as for every other chunk — but
/// it lands in `computed_at`, not in a `recorded_at`, and the difference is not
/// cosmetic. `recorded_at` is the transaction-time axis and is subject to
/// Doctrine II and the monotonicity guard; `computed_at` is a note about when a
/// derivation last ran, on a table the ledger does not see. Rerunning an
/// algorithm therefore replaces the row and advances the note, rather than
/// versioning a concept the world did not change.
///
/// # Failures name the concept (0.13.3, W7.2, D-176)
///
/// This was the one write path in the crate that returned
/// [`DbError::Engine`] raw, and the omission looked harmless: the table
/// carries no triggers, so none of [`crate::error::AbortKind`]'s guards can
/// fire on it and [`classify`] would have returned the same raw error it was
/// given. What that reasoning missed is the foreign key onto `concepts`, which
/// the engine enforces itself. Annotating a concept that does not exist is the
/// one failure a caller can cause here, and it reported as
/// `FOREIGN KEY constraint failed` with no row named — out of a chunk of up to
/// [`chunk_rows::ANNOTATIONS`].
///
/// It now goes through [`classify`] with [`WriteOp::Annotation`] like every
/// other write, and a missing concept returns [`DbError::NotFound`] carrying
/// its id.
async fn write_annotations_atomic(
conn: &libsql::Connection,
annotations: &[Annotation],
stamp: &str,
) -> Result<usize> {
if annotations.is_empty() {
return Ok(0);
}
let tx = conn
.transaction_with_behavior(libsql::TransactionBehavior::Immediate)
.await?;
let stmt = tx
.prepare(
"INSERT INTO analytics_annotations (concept_id, label, value, computed_at) \
VALUES (?1, ?2, ?3, ?4) \
ON CONFLICT(concept_id, label) DO UPDATE SET \
value = excluded.value, computed_at = excluded.computed_at",
)
.await?;
for a in annotations {
stmt.reset();
let res = stmt
.execute(libsql::params![
a.concept_id.as_str(),
a.label.as_str(),
a.value.as_str(),
stamp
])
.await;
if let Err(e) = res {
let typed = classify(
&tx,
e,
WriteOp::Annotation {
concept_id: &a.concept_id,
},
)
.await;
drop(stmt);
let _ = tx.rollback().await;
return Err(typed);
}
}
drop(stmt);
tx.commit().await?;
Ok(annotations.len())
}
async fn write_concepts_atomic(
state: &mut ActorState,
conn: &libsql::Connection,
concepts: &[ConceptUpsert],
stamp: &str,
) -> Result<usize> {
if concepts.is_empty() {
return Ok(0);
}
// Named lineages, before the write lock — `check_lineages`' reason, and the
// shape it also returns is unused here because `concepts` is keyed by
// identity and has no resolution to do (see `ConceptUpsert::branch`).
let mut named: Vec<&str> = Vec::with_capacity(1);
for concept in concepts {
let name = concept.branch_name();
if !named.contains(&name) {
named.push(name);
}
}
check_lineages(state, conn, &named).await?;
let tx = conn
.transaction_with_behavior(libsql::TransactionBehavior::Immediate)
.await?;
// Prepared once, like the edge chunk (D-056). This no longer routes through
// [`upsert_concept`] — that function prepares per call by construction — but
// it shares that function's statement text and parameter row, so the two
// cannot upsert different columns.
let stmt = tx.prepare(UPSERT_CONCEPT).await?;
for concept in concepts {
stmt.reset();
let res = stmt.execute(concept_params(concept, stamp)).await;
if let Err(e) = res {
let typed = classify(
&tx,
e,
WriteOp::Concept {
id: &concept.id,
recorded_at: stamp,
branch: concept.branch_name(),
},
)
.await;
drop(stmt);
let _ = tx.rollback().await;
return Err(typed);
}
}
drop(stmt);
tx.commit().await?;
Ok(concepts.len())
}
#[cfg(test)]
mod lineage_cache {
//! [`Lineages::shape_of`] against every shape combination (0.15.6, W14.3).
//!
//! Unit tests rather than a write through the actor, because the case this
//! function exists for **cannot be observed from outside**: where a batch
//! names lineages of different shapes, both of them currently compile the
//! same overlap statement, so a wrong choice between them is invisible
//! until W13.3 gives the guard a third lowering. A behavioural test would
//! pass on the code this replaces and on the code that replaces it, and
//! would go on passing through the release that made it matter.
use super::*;
/// `(name, is_root)`, kept as the fixture's spelling because that is what
/// these tests are about — the shape, not the ancestry. A root is a row
/// with no parent and no fork point, which is the pairing the `branches`
/// CHECK enforces; a non-root is given both, since a row with one and not
/// the other is not a state the schema permits.
fn lineages(rows: &[(&str, bool)]) -> Lineages {
Lineages {
rows: rows
.iter()
.map(|(id, root)| crate::graph::lineage::BranchRow {
id: (*id).into(),
parent: (!root).then(|| "main".to_string()),
forked_at: (!root).then(|| "2026-01-01T00:00:00.000000Z".to_string()),
})
.collect(),
}
}
fn trunk_only() -> Lineages {
lineages(&[("main", true)])
}
fn forked() -> Lineages {
lineages(&[("main", true), ("alt", false), ("other", false)])
}
/// One lineage is the trunk, whoever asks.
#[test]
fn one_lineage_is_the_trunk() {
assert_eq!(
trunk_only().shape_of(&["main"]).unwrap(),
LineageShape::Trunk
);
}
/// A root on a forked database is `TrunkOnForked`; anything else resolves.
#[test]
fn the_shape_reads_the_name_and_not_only_the_count() {
let l = forked();
assert_eq!(
l.shape_of(&["main"]).unwrap(),
LineageShape::TrunkOnForked,
"a root has no ancestors and D-244 emits that reduction directly"
);
assert_eq!(l.shape_of(&["alt"]).unwrap(), LineageShape::Resolved);
}
/// The case review C-24 is about: names of two different shapes.
///
/// Pinned in **both orders**, which is the whole content of the finding.
/// The loop this replaces returned whichever came last, so a batch naming
/// `[main, alt]` and one naming `[alt, main]` disagreed about their own
/// shape while describing the same set of lineages.
#[test]
fn a_batch_of_mixed_shapes_resolves_rather_than_taking_the_last_name() {
let l = forked();
assert_eq!(
l.shape_of(&["main", "alt"]).unwrap(),
LineageShape::Resolved,
"the resolved form is exact for a root as well, which is the \
argument OverlapGuard::prepare already makes"
);
assert_eq!(
l.shape_of(&["alt", "main"]).unwrap(),
LineageShape::Resolved,
"and it must not depend on the order the batch happened to name them"
);
}
/// Agreement is kept rather than widened: two forks are still `Resolved`,
/// and two mentions of one root are still `TrunkOnForked`.
#[test]
fn names_that_agree_keep_their_shape() {
let l = forked();
assert_eq!(
l.shape_of(&["alt", "other"]).unwrap(),
LineageShape::Resolved
);
assert_eq!(
l.shape_of(&["main", "main"]).unwrap(),
LineageShape::TrunkOnForked
);
}
/// Existence is checked for **every** name, not only the one that decides.
///
/// The check is why the loop existed at all, and the shape was the thing it
/// returned on the way past. A version that stopped at the first name would
/// let a batch name a lineage that does not exist and be refused later by
/// the foreign key, from inside a rolled-back transaction, naming neither
/// the column nor the branch — `check_lineages`' opening paragraph.
#[test]
fn an_unknown_name_is_refused_wherever_it_sits() {
let l = forked();
for names in [
["ghost", "main"].as_slice(),
["main", "ghost"].as_slice(),
["main", "ghost", "alt"].as_slice(),
] {
match l.shape_of(names) {
Err(DbError::UnknownBranch(name)) => assert_eq!(name, "ghost"),
other => panic!("expected UnknownBranch for {names:?}, got {other:?}"),
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn edge(target: &str, micros: usize) -> EdgeAssertion {
EdgeAssertion::new("src", target, "LINKS")
.valid_from(format!("2026-01-01T00:00:00.{micros:06}Z"))
.valid_to(format!("2026-01-01T00:00:00.{:06}Z", micros + 1))
}
/// The estimate must **no longer** depend on the batch's shape (0.13.6).
///
/// Its dependence on shape was correct for as long as the guard was
/// quadratic and the constant differed 16× between the two paths through
/// its inner loop. W7.5 removed that term, and measurement agrees: 1.94 s
/// and 2.22 s for the two 20,000-edge batches that used to differ by 7×.
/// A model that kept predicting a 7× spread would now be wrong in the
/// *expensive* direction — warning loudly about a batch that is fine.
#[test]
fn two_batches_of_one_size_are_predicted_alike() {
const N: usize = 20_000;
let fanout: Vec<_> = (0..N).map(|i| edge(&format!("t{i:07}"), i)).collect();
let history: Vec<_> = (0..N).map(|i| edge("t0", i)).collect();
assert_eq!(
estimated_bulk_hold(&fanout),
estimated_bulk_hold(&history),
"the guard no longer reads the batch's shape, so neither may this"
);
}
/// Measured on libSQL 0.9.30 after W7.5: 1.94 s and 2.22 s for those two
/// batches, against 2.6 s and 18.1 s before it. This pins that the model
/// still tracks them — a coefficient edited without re-measuring fails here.
#[test]
fn the_estimate_matches_what_was_measured() {
const N: usize = 20_000;
let fanout: Vec<_> = (0..N).map(|i| edge(&format!("t{i:07}"), i)).collect();
let history: Vec<_> = (0..N).map(|i| edge("t0", i)).collect();
for (batch, measured_ms, label) in
[(fanout, 1_936u128, "fanout"), (history, 2_220, "history")]
{
let predicted = estimated_bulk_hold(&batch).as_millis();
let ratio = predicted as f64 / measured_ms as f64;
assert!(
(0.8..1.25).contains(&ratio),
"{label}: predicted {predicted} ms against a measured \
{measured_ms} ms ({ratio:.2}x). Re-run \
examples/bulk_atomic_diag.rs before changing the coefficients."
);
}
}
/// `ilog2` panics on zero, and an empty batch is the caller asking whether
/// a batch they have not built yet would be slow.
#[test]
fn an_empty_batch_estimates_nothing_rather_than_panicking() {
assert_eq!(estimated_bulk_hold(&[]), std::time::Duration::ZERO);
let one = [edge("t0", 0)];
assert_eq!(
estimated_bulk_hold(&one),
std::time::Duration::from_nanos(7_400)
);
}
/// The model is used as a threshold test, so it must not go backwards.
#[test]
fn a_bigger_batch_never_predicts_a_shorter_hold() {
let mut last = std::time::Duration::ZERO;
for n in [1usize, 2, 3, 7, 8, 100, 511, 512, 513, 5_000, 20_000] {
let batch: Vec<_> = (0..n).map(|i| edge(&format!("t{i:07}"), i)).collect();
let now = estimated_bulk_hold(&batch);
assert!(now >= last, "{n} rows predicts {now:?} after {last:?}");
last = now;
}
}
/// The warning threshold sits well above the bound this path is exempt from.
///
/// Warning at `CHUNK_BUDGET` would fire on batches working exactly as
/// designed — the exemption is a contract (D-014), not a failure — and a
/// warning that fires on correct behaviour gets filtered out, taking the
/// 18-second case with it.
#[test]
fn the_warning_threshold_is_not_the_chunk_budget() {
assert!(BULK_ATOMIC_WARN_HOLD > CHUNK_BUDGET * 10);
}
// -----------------------------------------------------------------------
// reject_overlaps_within — sorted and swept (0.13.6, W7.5, D-179)
//
// The pairwise version was obviously correct and too slow; this one is
// neither, so what follows pins the cases where the obvious fix is wrong
// rather than only the cases the guard already caught.
// -----------------------------------------------------------------------
/// An edge over an explicit interval, all four key columns spelled out.
fn span(target: &str, edge_type: &str, from: usize, to: Option<usize>) -> EdgeAssertion {
let stamp = |n: usize| format!("2026-01-01T00:00:00.{n:06}Z");
EdgeAssertion::new("src", target, edge_type)
.valid_from(stamp(from))
.valid_to(to.map_or_else(|| timestamp::OPEN_SENTINEL.to_string(), stamp))
}
fn closed(from: usize, to: usize) -> EdgeAssertion {
span("t0", "LINKS", from, Some(to))
}
fn open_at(from: usize) -> EdgeAssertion {
span("t0", "LINKS", from, None)
}
/// The case that makes adjacent pairs insufficient.
///
/// Sort by start and any overlap shows up between neighbours — but only if
/// every neighbouring pair is eligible to be checked. `[5,20)` and `[5,6)`
/// are not: identical `valid_from` is re-assertion. Skip them, and `[5,6)`
/// against `[7,8)` is a clean gap, and the plain overlap between `[5,20)`
/// and `[7,8)` never gets looked at.
#[test]
fn an_overlap_hidden_behind_an_equal_valid_from_is_still_found() {
let batch = vec![closed(5, 20), closed(5, 6), closed(7, 8)];
assert!(matches!(
reject_overlaps_within(&batch),
Err(DbError::OverlappingInterval { .. })
));
}
/// The same trap in the other direction: skipped for being open.
///
/// Two open intervals are `trg_links_single_open`'s case and are passed
/// over here. A running maximum that counted them would take the sentinel
/// as the widest reach and report every later open interval as overlapping
/// it — inventing an error rather than missing one, which is why the sweep
/// carries a second maximum restricted to closed intervals.
#[test]
fn two_open_intervals_are_left_to_the_trigger() {
let batch = vec![closed(1, 5), open_at(10), open_at(20)];
assert!(reject_overlaps_within(&batch).is_ok());
}
/// An open interval still overlaps a closed one that reaches past its start.
#[test]
fn an_open_interval_over_a_closed_one_is_an_overlap() {
let batch = vec![closed(1, 50), open_at(10)];
assert!(matches!(
reject_overlaps_within(&batch),
Err(DbError::OverlappingInterval { .. })
));
}
/// Same `valid_from`, different `valid_to`: a batch correcting itself.
///
/// Last writer wins by `seq_id`, exactly as it does across batches. The
/// guard has no opinion.
#[test]
fn equal_valid_from_is_re_assertion_not_overlap() {
let batch = vec![closed(5, 20), closed(5, 6), closed(5, 900)];
assert!(reject_overlaps_within(&batch).is_ok());
}
/// Grouping is what makes the sweep sound, so it is pinned rather than read.
#[test]
fn edges_with_different_keys_do_not_see_each_other() {
let batch = vec![
span("t0", "LINKS", 1, Some(50)),
span("t1", "LINKS", 10, Some(60)),
span("t0", "CITES", 10, Some(60)),
span("t0", "LINKS", 50, Some(60)),
];
assert!(reject_overlaps_within(&batch).is_ok());
}
/// Which of the two the report calls *existing* (0.13.6).
///
/// Neither is older in transaction time — a batch lands under one stamp —
/// so the pairwise version's answer was its input order, which means
/// nothing. Valid-time order is the only ordering the two intervals have.
#[test]
fn the_report_names_the_earlier_interval_as_the_existing_one() {
let batch = vec![closed(7, 8), closed(5, 20)];
let Err(DbError::OverlappingInterval { overlap }) = reject_overlaps_within(&batch) else {
panic!("the batch overlaps itself");
};
assert!(overlap.valid_from.ends_with(".000007Z"), "{overlap:?}");
assert!(overlap.existing_from.ends_with(".000005Z"), "{overlap:?}");
}
/// A guard whose answer depended on the caller's ordering would be a worse
/// guard than the one it replaced, and sorting is exactly the change that
/// could introduce that.
#[test]
fn the_answer_does_not_depend_on_the_order_the_caller_passed() {
let mut batch = vec![closed(5, 20), closed(5, 6), closed(7, 8)];
batch.reverse();
assert!(reject_overlaps_within(&batch).is_err());
let mut clean = vec![closed(1, 5), closed(5, 6), closed(7, 8), open_at(8)];
clean.reverse();
assert!(reject_overlaps_within(&clean).is_ok());
}
/// §2.6, and the reason the rewrite happened rather than the doc alone.
///
/// Every edge shares a key, so the old loop reached `Interval::overlaps`
/// on all n(n−1)/2 pairs — 50 million of them here, which is seconds even
/// in release and considerably worse in the debug profile this runs under.
/// The bound is loose on purpose: it is an order of magnitude, not a
/// benchmark, and the only thing it can fail on is the quadratic term
/// coming back.
#[test]
fn one_relationships_whole_history_is_no_longer_quadratic() {
const N: usize = 10_000;
let batch: Vec<_> = (0..N).map(|i| closed(i * 2, i * 2 + 1)).collect();
let started = std::time::Instant::now();
assert!(reject_overlaps_within(&batch).is_ok());
let took = started.elapsed();
assert!(
took < std::time::Duration::from_secs(2),
"{N} same-key edges took {took:?} in the guard"
);
}
// -----------------------------------------------------------------------
// next_chunk_size — the control law (0.12.0, W2)
//
// All of these run without a database, a clock or an actor, which is why
// W2 comes before W3: the loop that will use this function can only be
// tested against a real write, and the properties below cannot be observed
// there without also observing the machine.
// -----------------------------------------------------------------------
use std::time::Duration;
/// `next_chunk_size` with the shipped budget and floor.
fn step_to(current: usize, held_ms: f64, ceiling: usize) -> usize {
next_chunk_size(
current,
Duration::from_nanos((held_ms * 1_000_000.0) as u64),
CHUNK_BUDGET,
CHUNK_FLOOR,
ceiling,
)
}
/// The edge path, which is every test here that does not say otherwise.
fn step(current: usize, held_ms: f64) -> usize {
step_to(current, held_ms, chunk_rows::EDGES)
}
/// Iterate the law against a machine that costs `per_row_us` per row plus a
/// fixed `overhead_ms` per transaction — the two-term model D-142 measured.
fn converge(
start: usize,
per_row_us: f64,
overhead_ms: f64,
ceiling: usize,
steps: usize,
) -> Vec<usize> {
let mut size = start;
(0..steps)
.map(|_| {
let held = overhead_ms + per_row_us * size as f64 / 1000.0;
size = step_to(size, held, ceiling);
size
})
.collect()
}
/// The reason the shrink is proportional rather than a halving: at 4× over
/// budget, halving needs three steps and every one of them is a latency
/// miss a caller can feel.
///
/// Run on the annotations path, because it is the only one whose ceiling
/// leaves room to start far above a size that is reachable — on the edge
/// path a 4× miss lands under [`CHUNK_FLOOR`], which is a different test.
#[test]
fn a_chunk_far_over_budget_converges_from_above_in_at_most_two_steps() {
const CEILING: usize = chunk_rows::ANNOTATIONS;
let (per_row_us, overhead_ms) = (20.0, 0.05);
let held = |n: usize| overhead_ms + per_row_us * n as f64 / 1000.0;
assert!(
held(CEILING) > 4.0 * 3.0,
"the start is not far over budget"
);
let trace = converge(CEILING, per_row_us, overhead_ms, CEILING, 4);
let first_in_budget = trace
.iter()
.position(|&n| held(n) <= 3.0)
.expect("never reached the budget");
assert!(
first_in_budget <= 1,
"took {} steps to get under budget: {trace:?}",
first_in_budget + 1
);
}
/// Growth is additive, so a size that is merely comfortable cannot leap the
/// ceiling — and cannot overshoot the budget by more than a quarter.
#[test]
fn growth_is_slow_and_shrinking_is_fast() {
let grown = step(40, 1.0);
assert!(
(41..=50).contains(&grown),
"40 rows at 1 ms should grow by about a quarter, got {grown}"
);
let shrunk = step(90, 9.0);
assert!(
shrunk <= 40,
"90 rows at 3x the budget should shrink proportionally, got {shrunk}"
);
}
/// The dead band. Between `budget / 2` and `budget` the size is right and
/// moving it only costs a re-measurement; without this the law oscillates
/// across the bound forever.
#[test]
fn a_chunk_inside_the_band_is_left_alone() {
for held_ms in [1.6, 2.0, 2.5, 2.9, 3.0] {
assert_eq!(step(60, held_ms), 60, "moved at {held_ms} ms");
}
assert_ne!(step(60, 1.4), 60, "did not grow at well under half budget");
}
/// Both clamps, and the floor's violation stated as a test rather than only
/// as a comment: a populated table drives this to `CHUNK_FLOOR` and holds it
/// there **over budget**, which is [`CHUNK_FLOOR`]'s documented trade.
#[test]
fn the_floor_and_the_ceiling_both_hold() {
// 118 µs/row + 0.03 ms fixed — the populated arm, where 35 rows is
// ~4.1 ms and no size in range meets the bound.
let trace = converge(chunk_rows::EDGES, 118.0, 0.03, chunk_rows::EDGES, 8);
assert!(
trace.iter().all(|&n| n >= CHUNK_FLOOR),
"fell through the floor: {trace:?}"
);
assert_eq!(*trace.last().unwrap(), CHUNK_FLOOR, "settled off the floor");
// A free machine cannot grow past the path's constant.
let fast = converge(CHUNK_FLOOR, 1.0, 0.01, chunk_rows::EDGES, 40);
assert_eq!(*fast.last().unwrap(), chunk_rows::EDGES);
assert!(fast.iter().all(|&n| n <= chunk_rows::EDGES));
}
/// Zero is the one answer that cannot be recovered from: a loop asked for
/// chunks of no rows makes no progress and never finishes. Degenerate
/// inputs included, since `held` is a measurement and measurements arrive
/// from a machine under load.
#[test]
fn the_law_never_returns_zero() {
let cases = [
(0usize, Duration::ZERO),
(0, Duration::from_secs(60)),
(1, Duration::from_secs(60)),
(90, Duration::from_secs(3600)),
(usize::MAX, Duration::from_nanos(1)),
(1, Duration::ZERO),
];
for (current, held) in cases {
for (floor, ceiling) in [(35, 90), (1, 1), (0, 0), (90, 35)] {
let n = next_chunk_size(current, held, CHUNK_BUDGET, floor, ceiling);
assert!(
n > 0,
"returned 0 for current={current}, held={held:?}, \
floor={floor}, ceiling={ceiling}"
);
}
}
}
/// A zero budget is not a configuration anyone should reach, but it is one
/// division away from a panic, so it is pinned.
#[test]
fn a_zero_budget_shrinks_to_the_floor_rather_than_dividing_by_it() {
assert_eq!(
next_chunk_size(90, Duration::from_millis(1), Duration::ZERO, 35, 90),
35
);
}
/// A panicked write actor is reported, and the report says so (W7.3, D-177).
///
/// This is the branch `close()` actually has. Through 0.13.3 it sat beside
/// `Ok(res) => res?` on an actor `Result` that could never be `Err`, and the
/// pair looked like two failure paths under review — so the one that cannot
/// fire was carried through two signatures and the one that can had no test.
///
/// The `JoinError` is real rather than mocked: `JoinError` has no public
/// constructor, and one built by hand would pin the mapping against a value
/// tokio does not produce.
#[tokio::test]
async fn a_writer_that_panicked_is_reported_by_close() {
// Swallow the panic's own output. The task is *meant* to panic, and a
// backtrace in a green suite trains people to skim it.
let prev = std::panic::take_hook();
std::panic::set_hook(Box::new(|_| {}));
let handle = tokio::spawn(async { panic!("the write connection is gone") });
let joined = handle.await;
std::panic::set_hook(prev);
assert!(
joined.is_err(),
"the task must have panicked for this to test anything"
);
match writer_exit(joined) {
Err(DbError::WriterStopped(reason)) => {
assert!(
reason.contains("did not exit cleanly"),
"the message must say what happened: {reason}"
);
}
other => panic!("a panicked actor must be WriterStopped, got {other:?}"),
}
}
/// An actor that ran to completion closes clean.
///
/// The other half, and the one that must not acquire a failure mode by
/// accident: `run_writer_actor` returns `()`, so the only way this can start
/// reporting an error is if someone gives the actor a `Result` again.
#[tokio::test]
async fn a_writer_that_finished_normally_closes_clean() {
let handle = tokio::spawn(async {});
assert!(writer_exit(handle.await).is_ok());
}
/// A token is a handle to one flag, not a value that is copied (0.13.8,
/// W7.6). The clone the caller keeps and the clone the import holds have to
/// be the same flag, or `cancel()` reaches nothing.
#[test]
fn a_cloned_token_cancels_the_original() {
let token = CancelToken::new();
let held_by_the_import = token.clone();
assert!(!held_by_the_import.is_cancelled());
token.cancel();
assert!(held_by_the_import.is_cancelled());
// And it stays cancelled: there is no un-cancel, deliberately, because
// a token that could be reset would let a second import inherit a
// decision made about the first.
token.cancel();
assert!(held_by_the_import.is_cancelled());
}
/// The default control is the one the plain bulk methods pass, and it must
/// never stop a write.
#[test]
fn the_default_control_neither_cancels_nor_reports() {
let control = BulkControl::new();
assert!(!control.is_cancelled());
// No callback, so this is a no-op rather than a panic on an `unwrap`.
control.report(BulkProgress {
written: 1,
total: 1,
rows: 1,
held: std::time::Duration::ZERO,
});
}
/// The callback receives what it was promised, once per call to `report`.
#[test]
fn progress_reaches_the_callback_unchanged() {
let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
let control = BulkControl::new().on_progress({
let seen = Arc::clone(&seen);
move |p| seen.lock().unwrap().push(p)
});
let sample = BulkProgress {
written: 180,
total: 900,
rows: 90,
held: std::time::Duration::from_millis(12),
};
control.report(sample);
assert_eq!(*seen.lock().unwrap(), vec![sample]);
}
}