8000 minigraf/src/db.rs at v2.0.2 · project-minigraf/minigraf · GitHub
[go: up one dir, main page]
Skip to content

db.rs

Latest commit

 

History

History
2422 lines (2125 loc) · 91.9 KB

db.rs

File metadata and controls

2422 lines (2125 loc) · 91.9 KB
Raw
//! Public `Minigraf` facade with `WriteTransaction` and `OpenOptions`.
//!
//! This module provides the primary user-facing API for Minigraf:
//! - `Minigraf::open()` / `Minigraf::in_memory()` for database creation
//! - `Minigraf::execute()` for implicit (self-contained) transactions
//! - `Minigraf::begin_write()` / `WriteTransaction` for explicit transactions
//! - `Minigraf::checkpoint()` for manual WAL compaction
use crate::graph::types::{Fact, TxId, VALID_TIME_FOREVER};
/// Sentinel value used in `materialize_transaction` to signal "no explicit `valid_from`
/// was provided; use the transaction timestamp at commit time."
///
/// `i64::MIN` is chosen because it is not a representable Unix millisecond timestamp
/// in any practical context, avoiding the collision that `0` would have with the Unix
/// epoch (1970-01-01T00:00:00Z), which is a legitimate `valid_from` value.
pub(crate) const VALID_FROM_USE_TX_TIME: i64 = i64::MIN;
use crate::error::{ErrorCode, MinigrafError, bail_coded, err_coded};
use crate::graph::FactStorage;
use crate::graph::types::Value;
use crate::query::datalog::evaluator::DEFAULT_MAX_DERIVED_FACTS;
use crate::query::datalog::evaluator::DEFAULT_MAX_RESULTS;
use crate::query::datalog::executor::DatalogExecutor;
use crate::query::datalog::executor::QueryResult;
use crate::query::datalog::functions::{
AggImpl, AggregateDesc, FunctionRegistry, PredicateDesc, UdfFinaliseFn, UdfOps, UdfStepFn,
};
use crate::query::datalog::parser::parse_datalog_command;
use crate::query::datalog::prepared::reject_unbound_slots;
use crate::query::datalog::rules::RuleRegistry;
use crate::query::datalog::types::{AttributeSpec, DatalogCommand, Transaction};
use crate::storage::backend::MemoryBackend;
#[cfg(not(target_arch = "wasm32"))]
use crate::storage::backend::file::FileBackend;
use crate::storage::persistent_facts::PersistentFactStorage;
#[cfg(not(target_arch = "wasm32"))]
use crate::wal::WalWriter;
use anyhow::Result;
use std::any::Any;
#[cfg(not(target_arch = "wasm32"))]
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard, RwLock};
// ─── Thread-local reentrant-write detection ─────────────────────────────────
thread_local! {
/// Set to `true` while a `WriteTransaction` is active on this thread.
/// Prevents same-thread deadlock when `db.execute()` is called while
/// a `WriteTransaction` is in progress.
static WRITE_TX_ACTIVE: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}
fn set_write_tx_active(val: bool) {
WRITE_TX_ACTIVE.with(|f| f.set(val));
}
fn is_write_tx_active() -> bool {
WRITE_TX_ACTIVE.with(|f| f.get())
}
// ─── Page cache sizing ─────────────────────────────────────────────────────────
/// Page cache capacity to use for an in-memory (`MemoryBackend`) database.
///
/// `MemoryBackend` serves every read and write directly from RAM and never
/// consults the page cache, so `opts.page_cache_size` — which only affects
/// file-backed databases — is ignored here and zero capacity is always used.
fn memory_page_cache_capacity(_opts: &OpenOptions) -> usize {
0
}
// ─── SyncMode ────────────────────────────────────────────────────────────────
/// Controls how aggressively WAL writes are flushed to disk.
///
/// Independent of [`Minigraf::checkpoint`], which always fsyncs the main
/// `.graph` file regardless of this setting — see [`OpenOptions::synchronous`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SyncMode {
/// `fdatasync` after every WAL write. Default; matches Minigraf's
/// behavior before this option existed — every committed `transact`/
/// `retract` is durable immediately.
Full,
/// No per-write flush. The WAL entry is still `write_all()`'d — safe
/// across an ordinary process crash — but not forced to disk until the
/// next checkpoint (auto-threshold, explicit `checkpoint()`, or clean
/// close), which fsyncs the main file directly rather than the WAL —
/// the WAL is simply deleted once the main file holds the same data.
/// Data written since the last checkpoint is lost only on OS crash or
/// power loss, not process death. Intended for bulk loaders that can
/// safely re-run from a checkpoint watermark on failure.
///
/// Combined with [`OpenOptions::wal_checkpoint_threshold`] set to
/// `usize::MAX`, nothing is ever forced to disk for the handle's
/// lifetime — no per-write flush, no auto-checkpoint, and no
/// close-time checkpoint (the same sentinel also suppresses that).
/// Callers wanting a bounded durability window should keep an explicit
/// `checkpoint()` call or a finite threshold.
Normal,
}
// ─── OpenOptions ─────────────────────────────────────────────────────────────
/// Configuration options for opening a `Minigraf` database.
///
/// `#[non_exhaustive]`: construct via [`OpenOptions::new`]/[`OpenOptions::default`]
/// plus the builder methods, or `..Default::default()`, not a full struct literal.
/// This is the third field added since 1.0 (`allow_unlocked`, `synchronous`); marking
/// it non-exhaustive now means the next one won't be a breaking change.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct OpenOptions {
/// Number of WAL entries committed before an automatic checkpoint is triggered.
///
/// Defaults to 1000. Lower values mean more frequent checkpoints (smaller WAL,
/// more I/O). Higher values mean less frequent checkpoints (larger WAL, less I/O).
///
/// A checkpoint's cost grows with the total size of the database (see
/// [`Minigraf::checkpoint`]), while durability does not depend on it: the WAL
/// is crash-durable. Raise this for write-heavy workloads on large graphs.
pub wal_checkpoint_threshold: usize,
/// Number of pages to hold in the LRU page cache. Default: 256 (= 1MB at 4KB pages).
///
/// **File-backed databases only.** In-memory databases (`Minigraf::in_memory`,
/// `open_memory`) ignore this option — `MemoryBackend` serves every read and
/// write directly from RAM and never consults the page cache.
pub page_cache_size: usize,
/// Maximum facts that can be derived per recursive rule iteration.
/// Defaults to 1_000_000. Use to prevent runaway recursive rules.
pub max_derived_facts: usize,
/// Maximum total query results. Defaults to 1_000_000.
pub max_results: usize,
/// Open even when the filesystem cannot lock files at all.
///
/// Defaults to `false`, which refuses to open rather than run without the
/// protection the lock provides. Set this only when you know the database
/// has a single writer by other means — for example an NFSv3 mount with no
/// `lockd`, where locking is unavailable rather than merely contended.
///
/// This does **not** override a lock that another handle actually holds.
/// Opening a database twice is refused regardless of this setting.
pub allow_unlocked: bool,
/// Controls WAL write durability. See [`SyncMode`]. Defaults to `SyncMode::Full`.
///
/// The other durability/throughput lever is batching: `begin_write()` +
/// several `execute()` calls + one `commit()` collapses every staged
/// fact into a single WAL write and a single fsync, regardless of this
/// setting.
///
/// No effect on in-memory databases, which have no WAL.
pub synchronous: SyncMode,
}
impl Default for OpenOptions {
fn default() -> Self {
OpenOptions {
wal_checkpoint_threshold: 1000,
page_cache_size: 256,
max_derived_facts: DEFAULT_MAX_DERIVED_FACTS,
max_results: DEFAULT_MAX_RESULTS,
allow_unlocked: false,
synchronous: SyncMode::Full,
}
}
}
impl OpenOptions {
/// Create a new `OpenOptions` with default settings.
pub fn new() -> Self {
Self::default()
}
/// Set the number of WAL entries committed before an automatic checkpoint
/// is triggered. See [`OpenOptions::wal_checkpoint_threshold`].
#[must_use]
pub fn wal_checkpoint_threshold(mut self, n: usize) -> Self {
self.wal_checkpoint_threshold = n;
self
}
/// Set the number of pages to hold in the LRU page cache.
///
/// Each page is 4KB, so the default of 256 pages uses ~1MB of memory.
///
/// **File-backed databases only** — ignored when opening an in-memory
/// database (see [`OpenOptions::page_cache_size`]).
pub fn page_cache_size(mut self, size: usize) -> Self {
self.page_cache_size = size;
self
}
/// Set the maximum facts that can be derived per recursive rule iteration.
///
/// Defaults to 1_000_000. Use lower values to prevent runaway recursive rules
/// from consuming excessive memory. Can be overridden per-query using
/// `:max-derived-facts N` in the query vector.
pub fn max_derived_facts(mut self, n: usize) -> Self {
self.max_derived_facts = n;
self
}
/// Set the maximum total query results.
///
/// Defaults to 1_000_000. Use lower values to limit result set size.
pub fn max_results(mut self, n: usize) -> Self {
self.max_results = n;
self
}
/// Open even when the filesystem cannot lock files. See
/// [`OpenOptions::allow_unlocked`] for when this is appropriate.
#[must_use]
pub fn allow_unlocked(mut self, allow: bool) -> Self {
self.allow_unlocked = allow;
self
}
/// Set the WAL durability mode. See [`SyncMode`].
#[must_use]
pub fn synchronous(mut self, mode: SyncMode) -> Self {
self.synchronous = mode;
self
}
/// Set the path for a file-backed database.
#[cfg(not(target_arch = "wasm32"))]
pub fn path(self, path: impl AsRef<Path>) -> OpenOptionsWithPath {
OpenOptionsWithPath {
opts: self,
path: path.as_ref().to_path_buf(),
}
}
/// Open an in-memory (non-persistent) database.
///
/// Uses the options set on the builder. WAL-related options are ignored.
///
/// # Errors
///
/// Returns an error if the in-memory storage backend fails to initialise.
pub fn open_memory(self) -> Result<Minigraf, MinigrafError> {
Minigraf::in_memory_with_options(self)
}
}
/// `OpenOptions` combined with a file path, ready to open.
#[cfg(not(target_arch = "wasm32"))]
pub struct OpenOptionsWithPath {
opts: OpenOptions,
path: PathBuf,
}
#[cfg(not(target_arch = "wasm32"))]
impl OpenOptionsWithPath {
/// Open or create the file-backed database.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, the header is corrupt,
/// or WAL replay fails.
pub fn open(self) -> Result<Minigraf, MinigrafError> {
Minigraf::open_with_options(self.path, self.opts)
}
}
// ─── WriteContext ─────────────────────────────────────────────────────────────
/// Internal write context: distinguishes in-memory from file-backed databases.
enum WriteContext {
/// In-memory database: no WAL, no persistence.
Memory,
/// File-backed database: has a WAL sidecar and a persistent storage layer.
#[cfg(not(target_arch = "wasm32"))]
File {
pfs: PersistentFactStorage<FileBackend>,
/// WAL writer. `None` after a checkpoint until the next write.
wal: Option<WalWriter>,
db_path: PathBuf,
/// Count of WAL entries written since the last checkpoint (or since open).
wal_entry_count: usize,
},
}
// ─── Inner ────────────────────────────────────────────────────────────────────
struct Inner {
/// The shared in-memory fact store. Cloning is cheap (Arc-based).
fact_storage: FactStorage,
/// Shared rule registry, persists across all `execute()` calls.
rules: Arc<RwLock<RuleRegistry>>,
/// Function registry for aggregates and window functions.
/// `RwLock` is used in anticipation of the 7.7b `register_aggregate`/`register_predicate` mutation API.
functions: Arc<RwLock<FunctionRegistry>>,
/// Serialises all writes. Holds `WriteContext` which contains the PFS/WAL
/// for file-backed databases.
write_lock: Mutex<WriteContext>,
/// Configuration options.
options: OpenOptions,
}
impl Drop for Inner {
fn drop(&mut self) {
// On clean close, perform a best-effort checkpoint to reduce WAL size.
// Errors are silently ignored (can't propagate from Drop).
// Skip if wal_checkpoint_threshold is usize::MAX — that sentinel suppresses
// all checkpointing (used by benchmarks to keep WAL entries pending).
if self.options.wal_checkpoint_threshold == usize::MAX {
return;
}
if let Ok(mut ctx) = self.write_lock.lock() {
let _ = Minigraf::do_checkpoint(&self.fact_storage, &mut ctx);
}
}
}
// ─── Fact size validation (moved to WAL serialization) ────────────────────────
// ─── Minigraf ─────────────────────────────────────────────────────────────────
/// The primary embedded graph database handle.
///
/// `Minigraf` is cheap to clone — all clones share the same underlying database.
///
/// # File-backed usage
/// ```no_run
/// # #[cfg(not(target_arch = "wasm32"))] {
/// use minigraf::db::Minigraf;
///
/// let db = Minigraf::open("mydb.graph").unwrap();
/// db.execute(r#"(transact [[:alice :person/name "Alice"]])"#).unwrap();
/// # }
/// ```
///
/// # In-memory usage
/// ```
/// use minigraf::db::Minigraf;
///
/// let db = Minigraf::in_memory().unwrap();
/// db.execute(r#"(transact [[:alice :person/name "Alice"]])"#).unwrap();
/// ```
///
/// # Fact Size Limit (file-backed databases only)
///
/// Each fact persisted to a `.graph` file must serialise to at most
/// 4 080 bytes (the per-page capacity limit).
///
/// In practice, `Value::String` content is limited to roughly **3 900–4 000 bytes**
/// depending on entity and attribute name lengths.
///
/// Facts that exceed this limit are rejected at insertion time with a descriptive
/// error. This check does **not** apply to `Minigraf::in_memory()`.
///
/// ## Workarounds for large payloads
///
/// - **External blob reference** — store the payload in a file or object store
/// and record its path, URL, or content-addressed hash as a `Value::String`:
/// ```text
/// (transact [[:doc123 :blob/sha256 "a3f5c9..."]])
/// ```
/// - **Entity decomposition** — split large values across multiple facts using
/// a continuation-entity pattern.
/// - **In-memory database** — `Minigraf::in_memory()` has no fact size limit
/// and is suitable for workloads that do not require persistence.
#[derive(Clone)]
pub struct Minigraf {
inner: Arc<Inner>,
}
impl Minigraf {
// ── Constructors ─────────────────────────────────────────────────────────
/// Open or create a file-backed database with default options.
///
/// A sidecar WAL file (`<path>.wal`) is created alongside the main file.
/// Any existing WAL from a previous crash is replayed automatically.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, the header is corrupt,
/// or WAL replay fails.
#[cfg(not(target_arch = "wasm32"))]
pub fn open(path: impl AsRef<Path>) -> Result<Self, MinigrafError> {
Self::open_with_options(path, OpenOptions::default())
}
/// Open or create a file-backed database with custom options.
///
/// # Errors
///
/// Returns an error if the file cannot be opened, the header is corrupt,
/// or WAL replay fails.
#[cfg(not(target_arch = "wasm32"))]
pub fn open_with_options(
path: impl AsRef<Path>,
opts: OpenOptions,
) -> Result<Self, MinigrafError> {
Self::open_with_options_inner(path, opts).map_err(MinigrafError::from)
}
#[cfg(not(target_arch = "wasm32"))]
fn open_with_options_inner(path: impl AsRef<Path>, opts: OpenOptions) -> Result<Self> {
let db_path = path.as_ref().to_path_buf();
// Open the main .graph file
let backend = FileBackend::open_with(&db_path, opts.allow_unlocked)?;
let pfs = PersistentFactStorage::new(backend, opts.page_cache_size)?;
// Share the fact storage
let fact_storage = pfs.storage().clone();
// Derive WAL path: "<db_path>.wal"
let wal_path = Self::wal_path_for(&db_path);
// Replay any existing WAL entries before opening the writer
let wal_entry_count = Self::replay_wal(&wal_path, &fact_storage, &pfs)?;
// Open the WAL writer only if the WAL file already exists from a previous session.
// Otherwise, create it lazily on the first write.
let wal = if wal_path.exists() {
Some(WalWriter::open_or_create(&wal_path, opts.synchronous)?)
} else {
None
};
let ctx = WriteContext::File {
pfs,
wal,
db_path,
wal_entry_count,
};
Ok(Minigraf {
inner: Arc::new(Inner {
fact_storage,
rules: Arc::new(RwLock::new(RuleRegistry::new())),
functions: Arc::new(RwLock::new(FunctionRegistry::with_builtins())),
write_lock: Mutex::new(ctx),
options: opts,
}),
})
}
/// Create an in-memory database (no WAL, no persistence). Suitable for tests and REPL.
///
/// # Errors
///
/// Returns an error if the in-memory storage backend fails to initialise.
pub fn in_memory() -> Result<Self, MinigrafError> {
Self::in_memory_with_options(OpenOptions::default())
}
/// Create an in-memory database with custom options.
///
/// Note: WAL-related options are ignored for in-memory databases, as is
/// `page_cache_size` (see [`OpenOptions::page_cache_size`]).
///
/// # Errors
///
/// Returns an error if the in-memory storage backend fails to initialise.
pub fn in_memory_with_options(opts: OpenOptions) -> Result<Self, MinigrafError> {
Self::in_memory_with_options_inner(opts).map_err(MinigrafError::from)
}
fn in_memory_with_options_inner(opts: OpenOptions) -> Result<Self> {
let backend = MemoryBackend::new();
let pfs = PersistentFactStorage::new(backend, memory_page_cache_capacity(&opts))?;
let fact_storage = pfs.storage().clone();
// For in-memory databases we don't need the PFS beyond initialisation;
// we just use the shared FactStorage directly.
drop(pfs);
Ok(Minigraf {
inner: Arc::new(Inner {
fact_storage,
rules: Arc::new(RwLock::new(RuleRegistry::new())),
functions: Arc::new(RwLock::new(FunctionRegistry::with_builtins())),
write_lock: Mutex::new(WriteContext::Memory),
options: opts,
}),
})
}
// ── WAL replay helper ────────────────────────────────────────────────────
/// Replay any WAL entries that are newer than the main file's checkpoint.
///
/// Returns the number of entries replayed (used to seed `wal_entry_count`).
#[cfg(not(target_arch = "wasm32"))]
fn replay_wal(
wal_path: &Path,
fact_storage: &FactStorage,
pfs: &PersistentFactStorage<FileBackend>,
) -> Result<usize> {
if !wal_path.exists() {
return Ok(0);
}
let mut reader = crate::wal::WalReader::open(wal_path)?;
let entries = reader.read_entries()?;
let last_checkpointed = pfs.last_checkpointed_tx_count();
let mut replayed = 0;
for entry in &entries {
if entry.tx_count <= last_checkpointed {
// Already present in the main file; skip.
continue;
}
for fact in &entry.facts {
let _ = fact_storage.load_fact(fact.clone())?;
}
#[allow(clippy::arithmetic_side_effects)]
{
replayed += 1;
}
}
// Re-synchronise tx_counter to the maximum tx_count across all facts
fact_storage.restore_tx_counter()?;
Ok(replayed)
}
// ── Execute ──────────────────────────────────────────────────────────────
/// Execute a Datalog command as a self-contained implicit transaction.
///
/// For file-backed databases, the WAL entry is written **before** facts are
/// applied to the in-memory store. A successful return means the facts are in
/// both the WAL and the in-memory store; a crash after this call returns will
/// replay the facts on next open.
///
/// If the WAL write fails, an error is returned and the in-memory store is
/// left unchanged. The database remains consistent for subsequent in-process
/// operations.
///
/// Returns `Err` if called from the same thread that holds an active
/// `WriteTransaction` (use `tx.execute()` instead).
///
/// # Errors
///
/// Returns an error if:
/// - A `WriteTransaction` is already active on **this thread** (deadlock prevention).
/// - Parsing fails.
/// - Execution fails.
/// - WAL write fails (file-backed databases).
pub fn execute(&self, input: &str) -> Result<QueryResult, MinigrafError> {
self.execute_inner(input).map_err(MinigrafError::from)
}
fn execute_inner(&self, input: &str) -> Result<QueryResult> {
// Detect same-thread reentrant write (would deadlock on the Mutex).
if is_write_tx_active() {
bail_coded!(ErrorCode::Int001);
}
let cmd = parse_datalog_command(input)?;
if let DatalogCommand::Query(q) = &cmd {
reject_unbound_slots(q)?;
}
// Determine if this is a read-only command (query only).
// Rule registration is treated as a write because it mutates the shared RuleRegistry.
let is_write = matches!(
cmd,
DatalogCommand::Transact(_) | DatalogCommand::Retract(_) | DatalogCommand::Rule(_)
);
if is_write {
let mut ctx = self
.inner
.write_lock
.lock()
.map_err(|_| err_coded!(ErrorCode::Api001))?;
// Handle write commands with correct WAL-first ordering for Transact/Retract:
// 1. Materialize facts (no storage mutation yet)
// 2. Allocate tx_count + tx_id and stamp facts
// 3. Write WAL entry FIRST — if this fails, FactStorage is unchanged
// 4. Apply facts to shared FactStorage
// For Rule registration: acquire lock to serialize rule changes, no WAL needed
let (stamped, is_retract) = match &cmd {
DatalogCommand::Transact(tx) => (Minigraf::materialize_transaction(tx)?, false),
DatalogCommand::Retract(tx) => (Minigraf::materialize_retraction(tx)?, true),
DatalogCommand::Rule(_) => {
// Rule registration: execute while holding write_lock to serialize
// rule registry changes (no WAL needed for rules)
let executor = DatalogExecutor::new_with_rules_and_functions(
self.inner.fact_storage.clone(),
self.inner.rules.clone(),
self.inner.functions.clone(),
);
return executor.execute(cmd);
}
_ => return Err(err_coded!(ErrorCode::Api002)),
};
let tx_count = self.inner.fact_storage.allocate_tx_count();
let tx_id = crate::graph::types::tx_id_now();
let stamped: Vec<Fact> = stamped
.into_iter()
.map(|mut f| {
f.tx_id = tx_id;
f.tx_count = tx_count;
// Fix valid_from if it was left as the sentinel
if f.asserted && f.valid_from == VALID_FROM_USE_TX_TIME {
f.valid_from = tx_id.cast_signed();
}
f
})
.collect();
// WAL write includes size validation — no separate check needed.
// Write WAL BEFORE applying to shared FactStorage.
// If this fails, FactStorage is still unchanged — clean rollback.
let should_checkpoint = WriteTransaction::wal_write_stamped_batch(
&mut ctx,
&self.inner.options,
tx_count,
&stamped,
)?;
// WAL succeeded — now apply facts to shared FactStorage.
for fact in &stamped {
let _ = self.inner.fact_storage.load_fact(fact.clone())?;
}
// Trigger auto-checkpoint AFTER facts are in FactStorage so the
// checkpoint captures the newly written facts.
if should_checkpoint {
Minigraf::do_checkpoint(&self.inner.fact_storage, &mut ctx)?;
}
// Return the same QueryResult the executor would have returned.
if is_retract {
Ok(QueryResult::Retracted(tx_id))
} else {
Ok(QueryResult::Transacted(tx_id))
}
} else {
// Read-only: no lock needed
let mut executor = DatalogExecutor::new_with_rules_and_functions(
self.inner.fact_storage.clone(),
self.inner.rules.clone(),
self.inner.functions.clone(),
);
executor.set_limits(
self.inner.options.max_derived_facts,
self.inner.options.max_results,
);
executor.execute(cmd)
}
}
// ── Explicit transaction ──────────────────────────────────────────────────
/// Begin an explicit write transaction.
///
/// Acquires the write lock; held until `commit()`, `rollback()`, or drop.
///
/// # Errors
///
/// Returns an error if a `WriteTransaction` is already active on **this thread**.
pub fn begin_write(&self) -> Result<WriteTransaction<'_>, MinigrafError> {
self.begin_write_inner().map_err(MinigrafError::from)
}
fn begin_write_inner(&self) -> Result<WriteTransaction<'_>> {
if is_write_tx_active() {
bail_coded!(ErrorCode::Int001);
}
let guard = self
.inner
.write_lock
.lock()
.map_err(|_| err_coded!(ErrorCode::Api001))?;
// Set flag only after successfully acquiring the lock
set_write_tx_active(true);
Ok(WriteTransaction {
guard,
inner: &self.inner,
pending_facts: Vec::new(),
next_pending_tx_count: self.inner.fact_storage.current_tx_count().saturating_add(1),
next_pending_tx_id: crate::graph::types::tx_id_now(),
committed: false,
})
}
// ── Checkpoint ───────────────────────────────────────────────────────────
/// Manually trigger a checkpoint: flush all in-memory facts to the main file
/// and delete the WAL sidecar.
///
/// A checkpoint is **not** a durability boundary: committed writes are already
/// crash-durable in the `<db>.wal` sidecar and are replayed on the next open.
/// Checkpointing less often only makes that replay (reopen) slower.
///
/// Cost: every checkpoint rewrites the four covering indexes after the fact
/// pages, so it copies pages in proportion to the total index size; only the
/// index leaves that receive new entries are decoded and re-encoded. Prefer
/// checkpointing on a size or time budget over once per logical unit of work.
///
/// No-op for in-memory databases.
///
/// # Errors
///
/// Returns an error if the write lock is poisoned or the checkpoint I/O fails.
pub fn checkpoint(&self) -> Result<(), MinigrafError> {
self.checkpoint_inner().map_err(MinigrafError::from)
}
fn checkpoint_inner(&self) -> Result<()> {
let mut ctx = self
.inner
.write_lock
.lock()
.map_err(|_| err_coded!(ErrorCode::Api001))?;
Self::do_checkpoint(&self.inner.fact_storage, &mut ctx)
}
/// Returns the current monotonic transaction counter.
///
/// This is the value that `:as-of N` compares against. After a successful
/// [`execute`](Self::execute) that returns [`QueryResult::Transacted`] or
/// [`QueryResult::Retracted`], this reflects the count of that transaction.
///
/// Starts at 0 for a new database and increments once per `transact`/`retract` call,
/// regardless of how many facts the batch contains.
pub fn current_tx_count(&self) -> u64 {
self.inner.fact_storage.current_tx_count()
}
/// Internal checkpoint logic (operates on an already-held write-lock guard).
fn do_checkpoint(_fact_storage: &FactStorage, ctx: &mut WriteContext) -> Result<()> {
match ctx {
WriteContext::Memory => {
// No-op for in-memory databases.
}
#[cfg(not(target_arch = "wasm32"))]
WriteContext::File {
pfs,
wal,
db_path,
wal_entry_count,
} => {
// Skip checkpoint if nothing to flush.
//
// `wal_entry_count` is non-zero when this handle has made writes *or*
// replayed WAL entries on open (crash-recovery path). `pfs.is_dirty()`
// catches any facts marked dirty via the normal write path.
//
// Kernel file locking refuses a second writer (another process, or a
// second handle in this one — #304/#314). The guard still matters where
// the filesystem cannot lock (e.g. NFSv3 without lockd, some FUSE
// mounts) and the caller set `allow_unlocked`.
if *wal_entry_count == 0 && !pfs.is_dirty() {
return Ok(());
}
// `force_dirty` is needed for the WAL-replay case: facts were loaded
// into memory during `replay_wal` but `pfs.dirty` was not set because
// no write path was exercised. Without it `save()` would no-op and
// the replayed facts would never reach the main file.
pfs.force_dirty();
pfs.save()?;
// Derive WAL path and delete the sidecar.
let wal_path = Self::wal_path_for(db_path);
// Close the WAL writer (drop it) before deleting the file.
*wal = None;
if wal_path.exists() {
WalWriter::delete_file(&wal_path)?;
}
// WAL writer will be recreated lazily on the next write.
*wal_entry_count = 0;
}
}
Ok(())
}
/// Parse and plan a query once; bind slots (`$name`) are left unresolved.
///
/// Returns a [`crate::query::datalog::prepared::PreparedQuery`] that can be executed
/// many times with different bind values via
/// [`crate::query::datalog::prepared::PreparedQuery::execute`].
///
/// # Errors
/// - Parse failure.
/// - A bind slot appears in an attribute position (rejected at prepare time).
/// - The command is not a `(query ...)` — `transact`, `retract`, and `rule`
/// are not preparable.
pub fn prepare(
&self,
query_str: &str,
) -> Result<crate::query::datalog::prepared::PreparedQuery, MinigrafError> {
self.prepare_inner(query_str).map_err(MinigrafError::from)
}
fn prepare_inner(
&self,
query_str: &str,
) -> Result<crate::query::datalog::prepared::PreparedQuery> {
use crate::query::datalog::prepared::prepare_query;
let cmd = parse_datalog_command(query_str)?;
let query = match cmd {
DatalogCommand::Query(q) => q,
DatalogCommand::Transact(_) => {
bail_coded!(ErrorCode::Api005);
}
DatalogCommand::Retract(_) => {
bail_coded!(ErrorCode::Api006);
}
DatalogCommand::Rule(_) => {
bail_coded!(ErrorCode::Api007);
}
};
prepare_query(
query,
self.inner.fact_storage.clone(),
self.inner.rules.clone(),
self.inner.functions.clone(),
)
}
/// Return an interactive REPL that reads commands from stdin.
///
/// The REPL borrows the database for the duration of the session.
/// Call [`crate::repl::Repl::run`] to start the interactive loop.
///
/// # Example
/// ```no_run
/// # use minigraf::Minigraf;
/// let db = Minigraf::in_memory().unwrap();
/// db.repl().run();
/// ```
pub fn repl(&self) -> crate::repl::Repl<'_> {
crate::repl::Repl::new(self)
}
/// Compute the WAL sidecar path for a given database path.
#[cfg(not(target_arch = "wasm32"))]
fn wal_path_for(db_path: &Path) -> PathBuf {
let mut p = db_path.to_path_buf();
let name = p
.file_name()
.map(|n| {
let mut s = n.to_os_string();
s.push(".wal");
s
})
.unwrap_or_else(|| std::ffi::OsString::from("db.graph.wal"));
p.set_file_name(name);
p
}
// ── Materialize helpers ───────────────────────────────────────────────────
/// Convert a `Transaction` into a list of assertion `Fact`s (tx_id and tx_count
/// are set to 0 as placeholders; they are assigned at commit time).
pub(crate) fn materialize_transaction(tx: &Transaction) -> Result<Vec<Fact>> {
use crate::query::datalog::matcher::{edn_to_entity_id, edn_to_value};
use crate::query::datalog::types::EdnValue;
let tx_valid_from = tx.valid_from;
let tx_valid_to = tx.valid_to;
let mut facts = Vec::new();
for pattern in &tx.facts {
let entity =
edn_to_entity_id(&pattern.entity).map_err(|e| err_coded!(ErrorCode::Int002, e))?;
let attr = match &pattern.attribute {
AttributeSpec::Real(EdnValue::Keyword(k)) => k.clone(),
AttributeSpec::Real(_) => bail_coded!(ErrorCode::Api003),
AttributeSpec::Pseudo(_) => bail_coded!(ErrorCode::Api004),
};
let value =
edn_to_value(&pattern.value).map_err(|e| err_coded!(ErrorCode::Int003, e))?;
let valid_from = pattern
.valid_from
.or(tx_valid_from)
.unwrap_or(VALID_FROM_USE_TX_TIME);
let valid_to = pattern
.valid_to
.or(tx_valid_to)
.unwrap_or(VALID_TIME_FOREVER);
facts.push(Fact::with_valid_time(
entity, attr, value, 0, 0, valid_from, valid_to,
));
}
Ok(facts)
}
/// Convert a `Transaction` into a list of retraction `Fact`s.
pub(crate) fn materialize_retraction(tx: &Transaction) -> Result<Vec<Fact>> {
use crate::query::datalog::matcher::{edn_to_entity_id, edn_to_value};
use crate::query::datalog::types::EdnValue;
let mut facts = Vec::new();
for pattern in &tx.facts {
let entity =
edn_to_entity_id(&pattern.entity).map_err(|e| err_coded!(ErrorCode::Int002, e))?;
let attr = match &pattern.attribute {
AttributeSpec::Real(EdnValue::Keyword(k)) => k.clone(),
AttributeSpec::Real(_) => bail_coded!(ErrorCode::Api003),
AttributeSpec::Pseudo(_) => bail_coded!(ErrorCode::Api004),
};
let value =
edn_to_value(&pattern.value).map_err(|e| err_coded!(ErrorCode::Int003, e))?;
let mut f = Fact::retract(entity, attr, value, 0);
f.tx_count = 0;
facts.push(f);
}
Ok(facts)
}
// ── UDF registration ─────────────────────────────────────────────────────
/// Register a custom aggregate function.
///
/// `Acc` is any `Send + 'static` type that serves as the accumulator.
/// It is type-erased internally. The function is usable in both `:find`
/// grouping position and `:over` (window) position.
///
/// # Errors
///
/// Returns an error if `name` is already registered (built-in or UDF)
/// or the function registry lock is poisoned.
///
/// # Example
/// ```
/// # use minigraf::db::Minigraf;
/// # use minigraf::Value;
/// let db = Minigraf::in_memory().unwrap();
/// db.register_aggregate(
/// "mysum",
/// || 0i64,
/// |acc: &mut i64, v: &Value| { if let Value::Integer(i) = v { *acc += i; } },
/// |acc: &i64, _n: usize| Value::Integer(*acc),
/// ).unwrap();
/// ```
pub fn register_aggregate<Acc>(
&self,
name: &str,
init: impl Fn() -> Acc + Send + Sync + 'static,
step: impl Fn(&mut Acc, &Value) + Send + Sync + 'static,
finalise: impl Fn(&Acc, usize) -> Value + Send + Sync + 'static,
) -> Result<(), MinigrafError>
where
Acc: Any + Send + 'static,
{
self.register_aggregate_inner(name, init, step, finalise)
.map_err(MinigrafError::from)
}
fn register_aggregate_inner<Acc>(
&self,
name: &str,
init: impl Fn() -> Acc + Send + Sync + 'static,
step: impl Fn(&mut Acc, &Value) + Send + Sync + 'static,
finalise: impl Fn(&Acc, usize) -> Value + Send + Sync + 'static,
) -> Result<()>
where
Acc: Any + Send + 'static,
{
let init_boxed: Arc<dyn Fn() -> Box<dyn Any + Send> + Send + Sync> =
Arc::new(move || Box::new(init()) as Box<dyn Any + Send>);
let step_boxed: UdfStepFn = Arc::new(move |acc, v| {
// `init_boxed` always creates `Box<Acc>`, so downcast is infallible.
if let Some(typed_acc) = acc.downcast_mut::<Acc>() {
step(typed_acc, v);
}
});
let finalise_boxed: UdfFinaliseFn = Arc::new(move |acc, n| {
acc.downcast_ref::<Acc>()
.map(|typed_acc| finalise(typed_acc, n))
.unwrap_or(Value::Null)
});
let desc = AggregateDesc {
impl_: AggImpl::Udf(UdfOps {
init: init_boxed,
step: step_boxed,
finalise: finalise_boxed,
}),
View remainder of file in raw view
0