Skip to main content

khive_db/
pool.rs

1//! Connection pool for SQLite: one exclusive writer, N concurrent readers.
2use crossbeam_queue::ArrayQueue;
3use parking_lot::Mutex;
4use rusqlite::{Connection, OpenFlags};
5use std::fs;
6use std::ops::{Deref, DerefMut};
7use std::path::{Path, PathBuf};
8use std::sync::{Arc, OnceLock};
9use std::thread;
10use std::time::{Duration, Instant};
11
12use crate::error::SqliteError;
13use crate::writer_task::WriterTaskHandle;
14use khive_storage::error::StorageError;
15use khive_storage::tx_registry::{DbIdentity, TxOrigin};
16
17const CACHE_SIZE_KIB: &str = "-65536";
18const MMAP_SIZE_BYTES: &str = "1073741824";
19const DEFAULT_READER_CAP: usize = 8;
20
21const DEFAULT_WAL_AUTOCHECKPOINT_PAGES: u32 = 4000;
22const DEFAULT_JOURNAL_SIZE_LIMIT_BYTES: i64 = 67_108_864; // 64 MiB
23const DEFAULT_WRITE_QUEUE_CAPACITY: usize = 256;
24
25/// Configuration for the connection pool.
26#[derive(Clone, Debug)]
27pub struct PoolConfig {
28    /// Database path. None = in-memory (pool degrades to single connection).
29    pub path: Option<PathBuf>,
30    /// Number of reader connections (default: min(num_cpus, 8)).
31    pub max_readers: usize,
32    /// WAL mode (must be true for pooling to work; default: true).
33    pub wal_mode: bool,
34    /// Busy timeout per connection (default: 30s).
35    ///
36    /// Overridable via `KHIVE_BUSY_TIMEOUT_SECS`.
37    pub busy_timeout: Duration,
38    /// Time to wait for a reader connection before returning an error (default: 5s).
39    ///
40    /// Overridable via `KHIVE_CHECKOUT_TIMEOUT_SECS`.
41    pub checkout_timeout: Duration,
42    /// Number of WAL pages that triggers an automatic checkpoint.
43    ///
44    /// Maps to `PRAGMA wal_autocheckpoint`. The default (4000 pages, ~16 MiB
45    /// at SQLite's default 4 KiB page size) matches the pre-config behaviour.
46    ///
47    /// Overridable via `KHIVE_WAL_AUTOCHECKPOINT_PAGES`.
48    pub wal_autocheckpoint_pages: u32,
49    /// Maximum WAL journal size in bytes before SQLite resets the WAL.
50    ///
51    /// Maps to `PRAGMA journal_size_limit`. Default: 64 MiB.
52    ///
53    /// Overridable via `KHIVE_JOURNAL_SIZE_LIMIT_BYTES`.
54    pub journal_size_limit_bytes: i64,
55    /// Open the database read-only (default: false).
56    ///
57    /// When true, the pool's writer connection is opened with
58    /// `SQLITE_OPEN_READ_ONLY` (no `SQLITE_OPEN_CREATE`, so a missing path is
59    /// rejected instead of created) and `PRAGMA query_only = ON` is set on
60    /// every connection that can execute SQL. Reader connections are already
61    /// opened read-only regardless of this flag.
62    pub read_only: bool,
63    /// Route migrated store write paths through the single-writer
64    /// `WriterTask` channel (ADR-067 Component A) instead of the legacy
65    /// per-call pool-mutex/standalone-connection path. Off by default.
66    ///
67    /// Slice 1 wires exactly one path (`SqlEntityStore::upsert_entities`)
68    /// behind this flag; enabling it does not yet claim ADR-067's
69    /// single-writer guarantee — other write paths still open their own
70    /// writers until later slices migrate them.
71    ///
72    /// Overridable via `KHIVE_WRITE_QUEUE` (`"1"` or `"true"`,
73    /// case-insensitive, enables it; anything else, or unset, leaves it off).
74    pub write_queue_enabled: bool,
75    /// Bounded channel capacity for the `WriterTask` write queue.
76    ///
77    /// Overridable via `KHIVE_WRITE_QUEUE_CAPACITY`. Default: 256 pending
78    /// operations (ADR-067 Component A recommended default).
79    pub write_queue_capacity: usize,
80}
81
82impl Default for PoolConfig {
83    fn default() -> Self {
84        Self {
85            path: None,
86            max_readers: std::thread::available_parallelism()
87                .map(|n| n.get())
88                .unwrap_or(1)
89                .clamp(1, DEFAULT_READER_CAP),
90            wal_mode: true,
91            busy_timeout: Duration::from_secs(
92                std::env::var("KHIVE_BUSY_TIMEOUT_SECS")
93                    .ok()
94                    .and_then(|v| v.parse::<u64>().ok())
95                    .unwrap_or(30),
96            ),
97            checkout_timeout: Duration::from_secs(
98                std::env::var("KHIVE_CHECKOUT_TIMEOUT_SECS")
99                    .ok()
100                    .and_then(|v| v.parse::<u64>().ok())
101                    .unwrap_or(5),
102            ),
103            wal_autocheckpoint_pages: std::env::var("KHIVE_WAL_AUTOCHECKPOINT_PAGES")
104                .ok()
105                .and_then(|v| v.parse::<u32>().ok())
106                .unwrap_or(DEFAULT_WAL_AUTOCHECKPOINT_PAGES),
107            journal_size_limit_bytes: std::env::var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES")
108                .ok()
109                .and_then(|v| v.parse::<i64>().ok())
110                .unwrap_or(DEFAULT_JOURNAL_SIZE_LIMIT_BYTES),
111            read_only: false,
112            write_queue_enabled: std::env::var("KHIVE_WRITE_QUEUE")
113                .map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
114                .unwrap_or(false),
115            write_queue_capacity: std::env::var("KHIVE_WRITE_QUEUE_CAPACITY")
116                .ok()
117                .and_then(|v| v.parse::<usize>().ok())
118                .filter(|&n| n > 0)
119                .unwrap_or(DEFAULT_WRITE_QUEUE_CAPACITY),
120        }
121    }
122}
123
124/// A read-write connection pool for SQLite.
125///
126/// Architecture:
127/// - 1 writer connection protected by a Mutex (exclusive access)
128/// - N reader connections in a lock-free queue (concurrent access)
129/// - All connections share the same database file in WAL mode
130///
131/// For in-memory databases, or when WAL mode is disabled/unavailable, the pool
132/// degrades to single-connection mode and routes all operations through the
133/// writer connection.
134pub struct ConnectionPool {
135    writer: Arc<Mutex<Connection>>,
136    readers: ArrayQueue<Connection>,
137    max_readers: usize,
138    config: PoolConfig,
139    /// The pool-wide ADR-067 Component A writer task, spawned lazily and at
140    /// most once per pool (per DB file) via [`Self::writer_task_handle`] —
141    /// see that method's doc comment for why this lives here rather than on
142    /// each store.
143    writer_task: OnceLock<Option<WriterTaskHandle>>,
144    /// This pool's ADR-091 backend-scoped attribution origin, minted exactly
145    /// once at construction (see [`mint_db_identity`]): `Database(_)` for a
146    /// file-backed pool, `Memory` for an in-memory pool. Every
147    /// `tx_registry::register_scoped` call site in this crate reaches its
148    /// origin through [`Self::origin`] rather than re-deriving it.
149    origin: TxOrigin,
150    /// The canonical path `origin`'s `DbIdentity` was minted from, `None` for
151    /// an in-memory pool. `DbIdentity` is deliberately opaque (no path
152    /// accessor) — filesystem consumers that need the actual path (sidecar
153    /// derivation) use this, the same canonical value the identity was
154    /// minted from, via [`Self::canonical_path`].
155    identity_path: Option<PathBuf>,
156    /// Test-only instrumentation: counts how many times the writer-task
157    /// init closure actually ran. Must never exceed 1 per pool no matter how
158    /// many stores are constructed over it — that is the invariant
159    /// `OnceLock::get_or_init` exists to guarantee, and what
160    /// `pool.rs`'s and `entity_tests.rs`'s one-writer-per-pool tests assert.
161    #[cfg(test)]
162    writer_task_spawn_count: std::sync::atomic::AtomicUsize,
163}
164
165enum ReaderLease<'pool> {
166    Pooled(Connection),
167    Shared(parking_lot::MutexGuard<'pool, Connection>),
168}
169
170/// A reader connection checked out from the pool.
171/// Returns the connection to the pool on drop.
172pub struct ReaderGuard<'pool> {
173    lease: Option<ReaderLease<'pool>>,
174    pool: &'pool ConnectionPool,
175}
176
177impl<'pool> ReaderGuard<'pool> {
178    /// Access the connection.
179    pub fn conn(&self) -> &Connection {
180        match self
181            .lease
182            .as_ref()
183            .expect("reader guard missing connection")
184        {
185            ReaderLease::Pooled(conn) => conn,
186            ReaderLease::Shared(guard) => guard,
187        }
188    }
189}
190
191impl<'pool> Deref for ReaderGuard<'pool> {
192    type Target = Connection;
193
194    fn deref(&self) -> &Self::Target {
195        self.conn()
196    }
197}
198
199impl<'pool> Drop for ReaderGuard<'pool> {
200    fn drop(&mut self) {
201        let Some(lease) = self.lease.take() else {
202            return;
203        };
204
205        match lease {
206            ReaderLease::Pooled(conn) => self.pool.return_reader(conn),
207            ReaderLease::Shared(_guard) => {}
208        }
209    }
210}
211
212/// A writer connection checked out from the pool.
213/// The Mutex ensures only one writer at a time.
214pub struct WriterGuard<'pool> {
215    guard: parking_lot::MutexGuard<'pool, Connection>,
216    /// The origin (ADR-091 backend-scoped attribution) of the pool this
217    /// guard was checked out from, carried so `transaction` can register its
218    /// span with the correct origin without holding a `&ConnectionPool`.
219    origin: TxOrigin,
220}
221
222impl<'pool> WriterGuard<'pool> {
223    /// Returns a shared reference to the underlying connection.
224    pub fn conn(&self) -> &Connection {
225        &self.guard
226    }
227
228    /// Returns a mutable reference to the underlying connection.
229    pub fn conn_mut(&mut self) -> &mut Connection {
230        &mut self.guard
231    }
232
233    /// Execute a write transaction.
234    /// Wraps the closure in BEGIN IMMEDIATE ... COMMIT.
235    pub fn transaction<F, R>(&self, f: F) -> Result<R, SqliteError>
236    where
237        F: FnOnce(&Connection) -> Result<R, SqliteError>,
238    {
239        self.guard.execute_batch("BEGIN IMMEDIATE")?;
240        let _tx_handle = khive_storage::tx_registry::register_scoped(
241            Some("writer_guard_tx".to_string()),
242            self.origin.clone(),
243        );
244
245        match f(&self.guard) {
246            Ok(result) => {
247                if let Err(err) = self.guard.execute_batch("COMMIT") {
248                    let _ = self.guard.execute_batch("ROLLBACK");
249                    return Err(err.into());
250                }
251                Ok(result)
252            }
253            Err(err) => {
254                let _ = self.guard.execute_batch("ROLLBACK");
255                Err(err)
256            }
257        }
258    }
259}
260
261impl<'pool> Deref for WriterGuard<'pool> {
262    type Target = Connection;
263
264    fn deref(&self) -> &Self::Target {
265        self.conn()
266    }
267}
268
269impl<'pool> DerefMut for WriterGuard<'pool> {
270    fn deref_mut(&mut self) -> &mut Self::Target {
271        self.conn_mut()
272    }
273}
274
275impl ConnectionPool {
276    /// Create a new connection pool.
277    ///
278    /// Opens 1 writer + N reader connections to the same database when pooling
279    /// is enabled. All connections are configured consistently (busy timeout,
280    /// foreign keys, cache, mmap, temp store). For in-memory databases, or when
281    /// WAL is disabled or unavailable, the pool falls back to single-connection
282    /// mode.
283    pub fn new(config: PoolConfig) -> Result<Self, SqliteError> {
284        let writer = open_writer_connection(&config)?;
285        let wal_enabled = configure_writer_connection(&writer, &config)?;
286        let max_readers = effective_reader_count(&config, wal_enabled);
287
288        let readers = ArrayQueue::new(max_readers.max(1));
289
290        let (origin, identity_path) = match config.path.as_ref() {
291            Some(path) => {
292                let (identity, canonical) = mint_db_identity(path)?;
293                (TxOrigin::Database(identity), Some(canonical))
294            }
295            None => (TxOrigin::Memory, None),
296        };
297
298        let pool = Self {
299            writer: Arc::new(Mutex::new(writer)),
300            readers,
301            max_readers,
302            config,
303            writer_task: OnceLock::new(),
304            origin,
305            identity_path,
306            #[cfg(test)]
307            writer_task_spawn_count: std::sync::atomic::AtomicUsize::new(0),
308        };
309
310        for _ in 0..pool.max_readers {
311            let conn = pool.open_reader_connection()?;
312            pool.readers
313                .push(conn)
314                .expect("reader queue must have capacity during pool initialization");
315        }
316
317        Ok(pool)
318    }
319
320    /// Check out a reader connection.
321    ///
322    /// Tries to pop from the lock-free queue. If empty, spins briefly then
323    /// waits with exponential backoff up to `checkout_timeout`.
324    ///
325    /// # Deadlock Warning
326    ///
327    /// In degraded mode (WAL unavailable, `max_readers == 0`), this method locks
328    /// the writer mutex. If the calling thread already holds a [`WriterGuard`],
329    /// this will deadlock (parking_lot `Mutex` is not reentrant). Never call
330    /// `reader()` while holding a `WriterGuard` on the same pool.
331    pub fn reader(&self) -> Result<ReaderGuard<'_>, SqliteError> {
332        if self.max_readers == 0 {
333            return Ok(ReaderGuard {
334                lease: Some(ReaderLease::Shared(self.writer.lock())),
335                pool: self,
336            });
337        }
338
339        let started = Instant::now();
340        let mut attempt = 0u32;
341
342        loop {
343            if let Some(conn) = self.readers.pop() {
344                return Ok(ReaderGuard {
345                    lease: Some(ReaderLease::Pooled(conn)),
346                    pool: self,
347                });
348            }
349
350            if started.elapsed() >= self.config.checkout_timeout {
351                return Err(pool_exhausted_error(
352                    self.config.checkout_timeout,
353                    self.max_readers,
354                ));
355            }
356
357            match attempt {
358                0..=7 => {
359                    let spins = 1usize << attempt;
360                    for _ in 0..spins {
361                        std::hint::spin_loop();
362                    }
363                }
364                8..=15 => thread::yield_now(),
365                _ => {
366                    let remaining = self
367                        .config
368                        .checkout_timeout
369                        .saturating_sub(started.elapsed());
370                    let sleep = Duration::from_micros(50 * (1u64 << (attempt - 16).min(6)));
371                    thread::sleep(sleep.min(remaining).min(Duration::from_millis(2)));
372                }
373            }
374
375            attempt = attempt.saturating_add(1);
376        }
377    }
378
379    /// Check out the writer connection.
380    ///
381    /// Waits up to `checkout_timeout` for the writer Mutex and returns
382    /// `Err(SqliteError::InvalidData)` if the timeout is exceeded.
383    pub fn writer(&self) -> Result<WriterGuard<'_>, SqliteError> {
384        let guard = self
385            .writer
386            .try_lock_for(self.config.checkout_timeout)
387            .ok_or_else(|| {
388                SqliteError::InvalidData(format!(
389                    "timed out after {:?} waiting for sqlite writer connection",
390                    self.config.checkout_timeout
391                ))
392            })?;
393        Ok(WriterGuard {
394            guard,
395            origin: self.origin(),
396        })
397    }
398
399    /// Non-panicking writer checkout.
400    ///
401    /// Returns `Err` on timeout instead of panicking. Use this in request
402    /// handlers where a 500 is preferable to crashing the process.
403    pub fn try_writer(&self) -> Result<WriterGuard<'_>, SqliteError> {
404        self.writer()
405    }
406
407    /// Zero-wait writer checkout for background tasks.
408    ///
409    /// Uses `try_lock()` (no timeout, no spin) — returns `Err` immediately when
410    /// any other caller holds the writer Mutex. Background tasks (e.g. the WAL
411    /// checkpoint task) MUST use this instead of `try_writer` so that a busy
412    /// writer causes the background task to skip its current tick rather than
413    /// stalling for up to `checkout_timeout` (default 5s) while write traffic
414    /// is in progress.
415    pub fn try_writer_nowait(&self) -> Result<WriterGuard<'_>, SqliteError> {
416        let guard = self.writer.try_lock().ok_or_else(|| {
417            SqliteError::InvalidData(
418                "writer connection busy (checkpoint skipped this tick)".to_string(),
419            )
420        })?;
421        Ok(WriterGuard {
422            guard,
423            origin: self.origin(),
424        })
425    }
426
427    /// Get the current number of available reader connections.
428    pub fn available_readers(&self) -> usize {
429        self.readers.len()
430    }
431
432    /// Get the total number of reader connections in the pool.
433    pub fn max_readers(&self) -> usize {
434        self.max_readers
435    }
436
437    /// Return the pool configuration.
438    pub fn config(&self) -> &PoolConfig {
439        &self.config
440    }
441
442    /// This pool's ADR-091 backend-scoped attribution origin (ADR-091,
443    /// backend-scoped WAL-pin attribution design note): `Database(_)` for a
444    /// file-backed pool, `Memory` for an in-memory pool. Every
445    /// `tx_registry::register_scoped` call site threaded in this crate
446    /// passes this value as the span's origin.
447    pub fn origin(&self) -> TxOrigin {
448        self.origin.clone()
449    }
450
451    /// The canonical path this pool's `origin()` identity was minted from,
452    /// `None` for an in-memory pool. `DbIdentity` has no path accessor by
453    /// design; sidecar derivation and other filesystem consumers use this —
454    /// the same canonical value the identity was minted from — instead of
455    /// re-deriving a path from the raw configured one.
456    pub fn canonical_path(&self) -> Option<&Path> {
457        self.identity_path.as_deref()
458    }
459
460    /// Return the pool-wide ADR-067 Component A writer task, spawning it
461    /// lazily on first access if `PoolConfig::write_queue_enabled` is set.
462    /// Exactly one writer task exists per `ConnectionPool` (per DB file); see
463    /// crates/khive-db/docs/api/pool.md#connectionpoolwriter_task_handle--single-writer-task-rationale
464    /// for why a per-store writer task would defeat the single-writer
465    /// guarantee.
466    ///
467    /// Returns `Ok(None)` if the flag is off, or if the writer task failed to
468    /// spawn for a reason other than a missing runtime (for example, an
469    /// in-memory pool has no standalone-connection support) — callers fall
470    /// back to the legacy pool-mutex write path in either case. A spawn
471    /// failure is logged once here (at first access), not once per store.
472    ///
473    /// Returns `Err(StorageError::WriterTaskNoRuntime)` instead of panicking
474    /// when `write_queue_enabled` is set but this is the first access and no
475    /// Tokio runtime is available on the calling thread (checked via
476    /// [`tokio::runtime::Handle::try_current`]) — spawning the writer task
477    /// requires `tokio::spawn`, which panics outside a runtime. Callers that
478    /// already treat a missing writer task as best-effort (construction-time
479    /// degrade to the legacy path, matching slice 1's documented policy) can
480    /// collapse this into `None` with `.ok().flatten()`; callers that need to
481    /// fail loud on a genuine misconfiguration (write queue requested but no
482    /// runtime to run it on) can propagate the `Err` directly.
483    pub fn writer_task_handle(&self) -> Result<Option<WriterTaskHandle>, StorageError> {
484        if !self.config.write_queue_enabled {
485            return Ok(None);
486        }
487        // Fast path: already resolved (spawned, degraded, or off) by an
488        // earlier call — no need to re-check the runtime.
489        if let Some(existing) = self.writer_task.get() {
490            return Ok(existing.clone());
491        }
492        // Not yet initialized and the flag is on: spawning requires
493        // `tokio::spawn`, which panics outside a runtime context. Check
494        // first and fail loud with a typed error instead.
495        if tokio::runtime::Handle::try_current().is_err() {
496            return Err(StorageError::WriterTaskNoRuntime);
497        }
498        Ok(self
499            .writer_task
500            .get_or_init(|| {
501                #[cfg(test)]
502                self.writer_task_spawn_count
503                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
504
505                match crate::writer_task::spawn(self, self.config.write_queue_capacity) {
506                    Ok(handle) => Some(handle),
507                    Err(e) => {
508                        tracing::warn!(
509                            error = %e,
510                            "KHIVE_WRITE_QUEUE=1 but the writer task failed to spawn; \
511                             writes fall back to the pool-mutex path"
512                        );
513                        None
514                    }
515                }
516            })
517            .clone())
518    }
519
520    /// Test-only: how many times the writer-task init closure actually ran.
521    /// Must be at most 1 for the pool's whole lifetime, regardless of how
522    /// many times [`Self::writer_task_handle`] is called or how many stores
523    /// are constructed over this pool.
524    #[cfg(test)]
525    pub(crate) fn writer_task_spawn_count(&self) -> usize {
526        self.writer_task_spawn_count
527            .load(std::sync::atomic::Ordering::SeqCst)
528    }
529
530    /// Compatibility method: returns the writer connection wrapped in `Arc<Mutex>`.
531    ///
532    /// WARNING: This exists only for backward compatibility with code that
533    /// calls `store.conn()`. New code should use `reader()` and `writer()`.
534    pub fn legacy_conn(&self) -> Arc<Mutex<Connection>> {
535        Arc::clone(&self.writer)
536    }
537
538    fn open_reader_connection(&self) -> Result<Connection, SqliteError> {
539        let path = self
540            .config
541            .path
542            .as_ref()
543            .expect("reader connections require a file-backed database");
544        open_reader_connection(path, &self.config)
545    }
546
547    /// Open a standalone read-write connection to the same file-backed database.
548    ///
549    /// Stores whose trait methods take `Send + 'static` closures (executed via
550    /// `spawn_blocking`) cannot hold the pooled `WriterGuard`'s `MutexGuard`
551    /// across the call — it opens an independent connection instead. This
552    /// must still honor `PoolConfig::read_only`: opening
553    /// `SQLITE_OPEN_READ_WRITE` unconditionally here would let a read-only
554    /// backend's graph/event/text stores bypass the flag that the pooled
555    /// writer enforces via `query_only`.
556    pub fn open_standalone_writer(&self) -> Result<Connection, SqliteError> {
557        let path = self.config.path.as_ref().ok_or_else(|| {
558            SqliteError::InvalidData(
559                "in-memory databases do not support standalone connections".to_string(),
560            )
561        })?;
562
563        if self.config.read_only {
564            return Err(SqliteError::InvalidData(
565                "database is read-only: standalone write connections are not permitted".to_string(),
566            ));
567        }
568
569        let conn = Connection::open_with_flags(
570            path,
571            OpenFlags::SQLITE_OPEN_READ_WRITE
572                | OpenFlags::SQLITE_OPEN_NO_MUTEX
573                | OpenFlags::SQLITE_OPEN_URI,
574        )?;
575        conn.busy_timeout(self.config.busy_timeout)?;
576        conn.pragma_update(None, "foreign_keys", "ON")?;
577        conn.pragma_update(None, "synchronous", "NORMAL")?;
578        Ok(conn)
579    }
580
581    /// Open a standalone read-only connection to the same file-backed database.
582    ///
583    /// Companion to `open_standalone_writer` for stores that also need an
584    /// independent reader connection outside the pooled reader queue.
585    pub fn open_standalone_reader(&self) -> Result<Connection, SqliteError> {
586        let path = self.config.path.as_ref().ok_or_else(|| {
587            SqliteError::InvalidData(
588                "in-memory databases do not support standalone connections".to_string(),
589            )
590        })?;
591
592        let conn = Connection::open_with_flags(
593            path,
594            OpenFlags::SQLITE_OPEN_READ_ONLY
595                | OpenFlags::SQLITE_OPEN_NO_MUTEX
596                | OpenFlags::SQLITE_OPEN_URI,
597        )?;
598        conn.busy_timeout(self.config.busy_timeout)?;
599        conn.pragma_update(None, "foreign_keys", "ON")?;
600        conn.pragma_update(None, "synchronous", "NORMAL")?;
601        Ok(conn)
602    }
603
604    fn return_reader(&self, conn: Connection) {
605        if self.max_readers == 0 {
606            return;
607        }
608
609        let conn = if reset_reader_connection(&conn) && reader_connection_is_healthy(&conn) {
610            Some(conn)
611        } else {
612            close_connection_quietly(conn);
613            self.open_reader_connection().ok()
614        };
615
616        if let Some(conn) = conn {
617            if let Err(conn) = self.readers.push(conn) {
618                eprintln!(
619                    "[sqlite-pool] reader pool queue full, discarding replacement connection"
620                );
621                close_connection_quietly(conn);
622            }
623        }
624    }
625}
626
627/// Bound on the final-component symlink chain [`resolve_symlink_chain`]
628/// follows before failing loud, mirroring the OS's own loop limit (e.g.
629/// Linux/macOS `ELOOP`, commonly 40 hops) rather than looping forever on a
630/// cycle.
631const MAX_SYMLINK_DEPTH: u32 = 40;
632
633/// Mint the canonical [`DbIdentity`] for a configured database path.
634///
635/// The sole minting point (ADR-091 backend-scoped attribution design note):
636/// `tx_registry` origin threading and `sidecar_dir_for` re-keying both
637/// consume this function's output rather than re-deriving it. Operationally
638/// three steps:
639///
640/// 1. A relative configured path is resolved against the process's current
641///    directory BEFORE any canonicalization — a bare file name has an empty
642///    parent, and canonicalizing an empty path fails.
643/// 2. If the resolved path exists, canonicalize the full path: this
644///    resolves symlinks at every level, including a symlink at the
645///    database-file level itself (a `link.sqlite` pointing at the real file
646///    mints the target's identity).
647/// 3. If the resolved path does not yet exist (first open), a dangling
648///    file-level symlink is a valid first-open state — SQLite creates the
649///    target through the link on first write, and minting the link's own
650///    name would diverge from a later opener using the target path
651///    directly. The final-component symlink chain is followed to its
652///    ultimate target first (bounded, see [`MAX_SYMLINK_DEPTH`]), then that
653///    target's PARENT directory is canonicalized and the file name is
654///    appended unchanged — the same pattern `FsBlobStore` uses for its
655///    root-keyed write locks (`stores/blob.rs::write_lock_for_root`), and
656///    for the same reason: `Path::canonicalize` requires an existing path.
657///
658/// A resolved target whose parent directory does not exist fails minting
659/// exactly as the subsequent database open itself would fail.
660///
661/// Returns the minted [`DbIdentity`] alongside the canonical [`PathBuf`] it
662/// was built from — `DbIdentity` has no path accessor by design, so callers
663/// that need the filesystem path (sidecar derivation) keep this pairing
664/// rather than re-deriving it from the raw configured path.
665fn mint_db_identity(configured_path: &Path) -> Result<(DbIdentity, PathBuf), SqliteError> {
666    let absolute = if configured_path.is_absolute() {
667        configured_path.to_path_buf()
668    } else {
669        let cwd = std::env::current_dir().map_err(|e| {
670            SqliteError::InvalidData(format!(
671                "cannot mint database identity for {configured_path:?}: failed to resolve the \
672                 process current directory: {e}"
673            ))
674        })?;
675        cwd.join(configured_path)
676    };
677
678    if absolute.exists() {
679        let canonical = absolute.canonicalize().map_err(|e| {
680            SqliteError::InvalidData(format!(
681                "cannot mint database identity: failed to canonicalize existing path \
682                 {absolute:?}: {e}"
683            ))
684        })?;
685        return Ok((
686            DbIdentity::new(canonical.clone().into_os_string()),
687            canonical,
688        ));
689    }
690
691    let resolved_target = resolve_symlink_chain(&absolute)?;
692    let parent = resolved_target.parent().ok_or_else(|| {
693        SqliteError::InvalidData(format!(
694            "cannot mint database identity for {resolved_target:?}: path has no parent \
695             directory"
696        ))
697    })?;
698    let file_name = resolved_target.file_name().ok_or_else(|| {
699        SqliteError::InvalidData(format!(
700            "cannot mint database identity for {resolved_target:?}: path has no file name"
701        ))
702    })?;
703    let canonical_parent = parent.canonicalize().map_err(|e| {
704        SqliteError::InvalidData(format!(
705            "cannot mint database identity: parent directory {parent:?} of first-open path \
706             {resolved_target:?} does not exist or is inaccessible: {e}"
707        ))
708    })?;
709    let mut identity_path = canonical_parent;
710    identity_path.push(file_name);
711    Ok((
712        DbIdentity::new(identity_path.clone().into_os_string()),
713        identity_path,
714    ))
715}
716
717/// Follow a (possibly dangling) final-component symlink chain to its
718/// ultimate target, bounded at [`MAX_SYMLINK_DEPTH`] hops. A path that is
719/// not itself a symlink — including one that does not exist at all —
720/// returns unchanged on the first iteration; this is the common case, a
721/// first-open path with no symlink involved.
722fn resolve_symlink_chain(path: &Path) -> Result<PathBuf, SqliteError> {
723    let mut current = path.to_path_buf();
724    for _ in 0..MAX_SYMLINK_DEPTH {
725        match fs::symlink_metadata(&current) {
726            Ok(meta) if meta.file_type().is_symlink() => {
727                let target = fs::read_link(&current).map_err(|e| {
728                    SqliteError::InvalidData(format!(
729                        "cannot mint database identity: failed to read symlink {current:?}: {e}"
730                    ))
731                })?;
732                current = if target.is_absolute() {
733                    target
734                } else {
735                    match current.parent() {
736                        Some(parent) => parent.join(&target),
737                        None => target,
738                    }
739                };
740            }
741            _ => return Ok(current),
742        }
743    }
744    Err(SqliteError::InvalidData(format!(
745        "cannot mint database identity for {path:?}: symlink chain exceeds \
746         {MAX_SYMLINK_DEPTH} levels"
747    )))
748}
749
750fn effective_reader_count(config: &PoolConfig, wal_enabled: bool) -> usize {
751    if config.path.is_some() && config.wal_mode && wal_enabled {
752        config.max_readers
753    } else {
754        0
755    }
756}
757
758fn open_writer_connection(config: &PoolConfig) -> Result<Connection, SqliteError> {
759    match config.path.as_ref() {
760        Some(path) => {
761            let flags = if config.read_only {
762                writer_read_only_open_flags()
763            } else {
764                writer_open_flags()
765            };
766            Connection::open_with_flags(path, flags).map_err(Into::into)
767        }
768        None => Connection::open_in_memory().map_err(Into::into),
769    }
770}
771
772fn open_reader_connection(path: &Path, config: &PoolConfig) -> Result<Connection, SqliteError> {
773    let conn = Connection::open_with_flags(path, reader_open_flags())?;
774    configure_reader_connection(&conn, config)?;
775    Ok(conn)
776}
777
778fn writer_open_flags() -> OpenFlags {
779    OpenFlags::SQLITE_OPEN_READ_WRITE
780        | OpenFlags::SQLITE_OPEN_CREATE
781        | OpenFlags::SQLITE_OPEN_URI
782        | OpenFlags::SQLITE_OPEN_NO_MUTEX
783}
784
785/// Read-only writer-slot open flags: no `SQLITE_OPEN_CREATE`, so a missing
786/// path is rejected rather than silently created.
787fn writer_read_only_open_flags() -> OpenFlags {
788    OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX
789}
790
791fn reader_open_flags() -> OpenFlags {
792    OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX
793}
794
795fn configure_writer_connection(
796    conn: &Connection,
797    config: &PoolConfig,
798) -> Result<bool, SqliteError> {
799    if config.read_only {
800        // Read-only writer slot: skip write-intent PRAGMAs (journal_mode,
801        // wal_autocheckpoint, journal_size_limit all require write access to
802        // change) and lock the connection down with query_only instead.
803        conn.pragma_update(None, "foreign_keys", "ON")?;
804        conn.busy_timeout(config.busy_timeout)?;
805        conn.pragma_update(None, "cache_size", CACHE_SIZE_KIB)?;
806        conn.pragma_update(None, "mmap_size", MMAP_SIZE_BYTES)?;
807        conn.pragma_update(None, "temp_store", "MEMORY")?;
808        conn.pragma_update(None, "query_only", "ON")?;
809
810        let wal_enabled =
811            config.wal_mode && current_journal_mode(conn)?.eq_ignore_ascii_case("wal");
812        return Ok(wal_enabled);
813    }
814
815    let wants_wal = config.path.is_some() && config.wal_mode;
816
817    if wants_wal {
818        conn.pragma_update(None, "journal_mode", "WAL")?;
819    }
820
821    conn.pragma_update(None, "synchronous", "NORMAL")?;
822    conn.pragma_update(None, "foreign_keys", "ON")?;
823    conn.busy_timeout(config.busy_timeout)?;
824    conn.pragma_update(None, "cache_size", CACHE_SIZE_KIB)?;
825    conn.pragma_update(None, "mmap_size", MMAP_SIZE_BYTES)?;
826    conn.pragma_update(None, "temp_store", "MEMORY")?;
827
828    let wal_enabled = wants_wal && current_journal_mode(conn)?.eq_ignore_ascii_case("wal");
829
830    if wal_enabled {
831        conn.pragma_update(None, "wal_autocheckpoint", config.wal_autocheckpoint_pages)?;
832        conn.pragma_update(None, "journal_size_limit", config.journal_size_limit_bytes)?;
833    }
834
835    Ok(wal_enabled)
836}
837
838fn configure_reader_connection(conn: &Connection, config: &PoolConfig) -> Result<(), SqliteError> {
839    conn.pragma_update(None, "foreign_keys", "ON")?;
840    conn.busy_timeout(config.busy_timeout)?;
841    conn.pragma_update(None, "cache_size", CACHE_SIZE_KIB)?;
842    conn.pragma_update(None, "mmap_size", MMAP_SIZE_BYTES)?;
843    conn.pragma_update(None, "temp_store", "MEMORY")?;
844    Ok(())
845}
846
847fn current_journal_mode(conn: &Connection) -> Result<String, SqliteError> {
848    conn.pragma_query_value(None, "journal_mode", |row| row.get::<_, String>(0))
849        .map(|mode| mode.to_ascii_lowercase())
850        .map_err(Into::into)
851}
852
853fn reset_reader_connection(conn: &Connection) -> bool {
854    if conn.is_autocommit() {
855        return true;
856    }
857
858    match conn.execute_batch("ROLLBACK") {
859        Ok(()) => conn.is_autocommit(),
860        Err(rusqlite::Error::SqliteFailure(err, _)) => {
861            if matches!(
862                err.code,
863                rusqlite::ErrorCode::CannotOpen
864                    | rusqlite::ErrorCode::DatabaseCorrupt
865                    | rusqlite::ErrorCode::NotADatabase
866                    | rusqlite::ErrorCode::DiskFull
867            ) {
868                return false;
869            }
870            conn.is_autocommit()
871        }
872        Err(_) => false,
873    }
874}
875
876fn reader_connection_is_healthy(conn: &Connection) -> bool {
877    match conn.query_row("SELECT 1", [], |row| row.get::<_, i64>(0)) {
878        Ok(_) => true,
879        Err(rusqlite::Error::SqliteFailure(err, _)) => !matches!(
880            err.code,
881            rusqlite::ErrorCode::CannotOpen
882                | rusqlite::ErrorCode::NotADatabase
883                | rusqlite::ErrorCode::DatabaseCorrupt
884                | rusqlite::ErrorCode::PermissionDenied
885                | rusqlite::ErrorCode::SystemIoFailure
886        ),
887        Err(_) => true,
888    }
889}
890
891fn close_connection_quietly(conn: Connection) {
892    match conn.close() {
893        Ok(()) => {}
894        Err((conn, _)) => drop(conn),
895    }
896}
897
898fn pool_exhausted_error(timeout: Duration, max_readers: usize) -> SqliteError {
899    rusqlite::Error::SqliteFailure(
900        rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY),
901        Some(format!(
902            "Pool exhausted: no reader available after {timeout:?} (max_readers={max_readers})"
903        )),
904    )
905    .into()
906}
907
908#[cfg(test)]
909mod tests {
910    use super::*;
911    use serial_test::serial;
912
913    /// Restores the process CWD on drop — including on panic — so a mid-test
914    /// assertion failure (or an unexpected panic from the code under test)
915    /// can never leave the process chdir'd into a `tempfile::tempdir()` that
916    /// unwinds out from under every later test sharing this process.
917    struct CwdGuard {
918        original: PathBuf,
919    }
920
921    impl CwdGuard {
922        fn enter(dir: &Path) -> Self {
923            let original = std::env::current_dir().unwrap();
924            std::env::set_current_dir(dir).unwrap();
925            Self { original }
926        }
927    }
928
929    impl Drop for CwdGuard {
930        fn drop(&mut self) {
931            let _ = std::env::set_current_dir(&self.original);
932        }
933    }
934
935    #[test]
936    #[serial]
937    fn pool_config_default_values_match_constants() {
938        // Ensure defaults are not accidentally changed.
939        let cfg = PoolConfig::default();
940        assert_eq!(
941            cfg.wal_autocheckpoint_pages,
942            DEFAULT_WAL_AUTOCHECKPOINT_PAGES
943        );
944        assert_eq!(
945            cfg.journal_size_limit_bytes,
946            DEFAULT_JOURNAL_SIZE_LIMIT_BYTES
947        );
948        assert_eq!(cfg.busy_timeout, Duration::from_secs(30));
949        assert_eq!(cfg.checkout_timeout, Duration::from_secs(5));
950    }
951
952    #[test]
953    #[serial]
954    fn pool_config_env_override_wal_autocheckpoint() {
955        std::env::set_var("KHIVE_WAL_AUTOCHECKPOINT_PAGES", "8000");
956        let cfg = PoolConfig::default();
957        std::env::remove_var("KHIVE_WAL_AUTOCHECKPOINT_PAGES");
958        assert_eq!(cfg.wal_autocheckpoint_pages, 8000);
959    }
960
961    #[test]
962    #[serial]
963    fn pool_config_env_override_journal_size_limit() {
964        std::env::set_var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES", "134217728");
965        let cfg = PoolConfig::default();
966        std::env::remove_var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES");
967        assert_eq!(cfg.journal_size_limit_bytes, 134_217_728);
968    }
969
970    #[test]
971    #[serial]
972    fn pool_config_env_override_busy_timeout() {
973        std::env::set_var("KHIVE_BUSY_TIMEOUT_SECS", "60");
974        let cfg = PoolConfig::default();
975        std::env::remove_var("KHIVE_BUSY_TIMEOUT_SECS");
976        assert_eq!(cfg.busy_timeout, Duration::from_secs(60));
977    }
978
979    #[test]
980    #[serial]
981    fn pool_config_env_override_checkout_timeout() {
982        std::env::set_var("KHIVE_CHECKOUT_TIMEOUT_SECS", "10");
983        let cfg = PoolConfig::default();
984        std::env::remove_var("KHIVE_CHECKOUT_TIMEOUT_SECS");
985        assert_eq!(cfg.checkout_timeout, Duration::from_secs(10));
986    }
987
988    #[test]
989    #[serial]
990    fn pool_config_write_queue_defaults_off() {
991        let cfg = PoolConfig::default();
992        assert!(!cfg.write_queue_enabled);
993        assert_eq!(cfg.write_queue_capacity, DEFAULT_WRITE_QUEUE_CAPACITY);
994    }
995
996    #[test]
997    #[serial]
998    fn pool_config_env_override_write_queue_enabled() {
999        std::env::set_var("KHIVE_WRITE_QUEUE", "1");
1000        let cfg = PoolConfig::default();
1001        std::env::remove_var("KHIVE_WRITE_QUEUE");
1002        assert!(cfg.write_queue_enabled);
1003    }
1004
1005    #[test]
1006    #[serial]
1007    fn pool_config_env_override_write_queue_enabled_accepts_true_case_insensitive() {
1008        std::env::set_var("KHIVE_WRITE_QUEUE", "True");
1009        let cfg = PoolConfig::default();
1010        std::env::remove_var("KHIVE_WRITE_QUEUE");
1011        assert!(cfg.write_queue_enabled);
1012    }
1013
1014    #[test]
1015    #[serial]
1016    fn pool_config_env_override_write_queue_capacity() {
1017        std::env::set_var("KHIVE_WRITE_QUEUE_CAPACITY", "64");
1018        let cfg = PoolConfig::default();
1019        std::env::remove_var("KHIVE_WRITE_QUEUE_CAPACITY");
1020        assert_eq!(cfg.write_queue_capacity, 64);
1021    }
1022
1023    #[test]
1024    #[serial]
1025    fn pool_config_env_invalid_write_queue_capacity_falls_back_to_default() {
1026        std::env::set_var("KHIVE_WRITE_QUEUE_CAPACITY", "0");
1027        let cfg = PoolConfig::default();
1028        std::env::remove_var("KHIVE_WRITE_QUEUE_CAPACITY");
1029        assert_eq!(cfg.write_queue_capacity, DEFAULT_WRITE_QUEUE_CAPACITY);
1030    }
1031
1032    #[test]
1033    #[serial]
1034    fn pool_config_env_invalid_falls_back_to_default() {
1035        std::env::set_var("KHIVE_WAL_AUTOCHECKPOINT_PAGES", "not_a_number");
1036        std::env::set_var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES", "");
1037        let cfg = PoolConfig::default();
1038        std::env::remove_var("KHIVE_WAL_AUTOCHECKPOINT_PAGES");
1039        std::env::remove_var("KHIVE_JOURNAL_SIZE_LIMIT_BYTES");
1040        assert_eq!(
1041            cfg.wal_autocheckpoint_pages,
1042            DEFAULT_WAL_AUTOCHECKPOINT_PAGES
1043        );
1044        assert_eq!(
1045            cfg.journal_size_limit_bytes,
1046            DEFAULT_JOURNAL_SIZE_LIMIT_BYTES
1047        );
1048    }
1049
1050    #[test]
1051    fn file_backed_pool_opens_successfully() {
1052        let dir = tempfile::tempdir().unwrap();
1053        let path = dir.path().join("test_pool.db");
1054        let cfg = PoolConfig {
1055            path: Some(path.clone()),
1056            ..PoolConfig::default()
1057        };
1058        let pool = ConnectionPool::new(cfg).expect("file-backed pool should open");
1059        assert!(path.exists());
1060        assert!(pool.max_readers() > 0);
1061    }
1062
1063    #[test]
1064    fn in_memory_pool_degrades_to_single_connection() {
1065        let cfg = PoolConfig {
1066            path: None,
1067            ..PoolConfig::default()
1068        };
1069        let pool = ConnectionPool::new(cfg).expect("in-memory pool should open");
1070        assert_eq!(pool.max_readers(), 0);
1071    }
1072
1073    #[test]
1074    fn writer_checkout_and_release_works() {
1075        let cfg = PoolConfig {
1076            path: None,
1077            ..PoolConfig::default()
1078        };
1079        let pool = ConnectionPool::new(cfg).unwrap();
1080        {
1081            let _writer = pool.writer().expect("writer checkout should succeed");
1082        }
1083        // After drop, writer should be re-acquirable.
1084        let _writer2 = pool
1085            .writer()
1086            .expect("second writer checkout should succeed");
1087    }
1088
1089    /// ADR-091 Plank 0: `WriterGuard::transaction` registers/deregisters a
1090    /// tx_registry entry around the closure. See
1091    /// crates/khive-db/docs/api/pool.md#writer_guard_transaction_registers_during_closure_only
1092    #[test]
1093    #[serial(tx_registry)]
1094    fn writer_guard_transaction_registers_during_closure_only() {
1095        let cfg = PoolConfig {
1096            path: None,
1097            ..PoolConfig::default()
1098        };
1099        let pool = ConnectionPool::new(cfg).unwrap();
1100        let guard = pool.writer().unwrap();
1101
1102        let mut seen_during_closure = false;
1103        let result: Result<(), SqliteError> = guard.transaction(|_conn| {
1104            seen_during_closure = khive_storage::tx_registry::snapshot()
1105                .iter()
1106                .any(|(_, label)| label.as_deref() == Some("writer_guard_tx"));
1107            Ok(())
1108        });
1109        result.expect("transaction should commit");
1110
1111        assert!(
1112            seen_during_closure,
1113            "expected a writer_guard_tx entry visible inside the closure"
1114        );
1115        assert!(
1116            !khive_storage::tx_registry::snapshot()
1117                .iter()
1118                .any(|(_, label)| label.as_deref() == Some("writer_guard_tx")),
1119            "expected the entry to be gone after the transaction completes"
1120        );
1121    }
1122
1123    /// ADR-067 Component A: `writer_task_handle` must fail loud (typed
1124    /// error, not panic) with no Tokio runtime available. See
1125    /// crates/khive-db/docs/api/pool.md#writer_task_handle_fails_loud_without_tokio_runtime
1126    #[test]
1127    fn writer_task_handle_fails_loud_without_tokio_runtime() {
1128        let dir = tempfile::tempdir().unwrap();
1129        let path = dir.path().join("writer_task_no_runtime.db");
1130        let cfg = PoolConfig {
1131            path: Some(path),
1132            write_queue_enabled: true,
1133            ..PoolConfig::default()
1134        };
1135        let pool = ConnectionPool::new(cfg).expect("file-backed pool should open");
1136
1137        let result = pool.writer_task_handle();
1138
1139        assert!(
1140            matches!(result, Err(StorageError::WriterTaskNoRuntime)),
1141            "expected Err(StorageError::WriterTaskNoRuntime) outside a Tokio \
1142             runtime, got {result:?}"
1143        );
1144        assert_eq!(
1145            pool.writer_task_spawn_count(),
1146            0,
1147            "the guard must reject before ever attempting tokio::spawn"
1148        );
1149    }
1150
1151    /// ADR-091 backend-scoped attribution: the real path, a directory
1152    /// symlink, a file-level symlink, a relative spelling, and a bare file
1153    /// name (resolved against the current directory) must all mint an
1154    /// identical `DbIdentity` and canonical path for the same database.
1155    #[test]
1156    #[serial(pool_cwd)]
1157    fn mint_db_identity_alias_convergence() {
1158        let dir = tempfile::tempdir().unwrap();
1159        let real_dir = dir.path().join("real");
1160        fs::create_dir(&real_dir).unwrap();
1161        let db_path = real_dir.join("khive.db");
1162        fs::write(&db_path, b"").unwrap();
1163
1164        let dir_symlink = dir.path().join("dir_link");
1165        let file_symlink = dir.path().join("file_link.db");
1166        #[cfg(unix)]
1167        {
1168            std::os::unix::fs::symlink(&real_dir, &dir_symlink).unwrap();
1169            std::os::unix::fs::symlink(&db_path, &file_symlink).unwrap();
1170        }
1171
1172        let (via_real, canonical_real) = mint_db_identity(&db_path).unwrap();
1173
1174        // Relative spelling: resolved against the process CWD (step 1).
1175        let relative_result = {
1176            let _cwd = CwdGuard::enter(&real_dir);
1177            mint_db_identity(&PathBuf::from("khive.db"))
1178        };
1179        let (via_relative, canonical_relative) = relative_result.unwrap();
1180        assert_eq!(canonical_real, canonical_relative);
1181        assert_eq!(via_real, via_relative);
1182
1183        #[cfg(unix)]
1184        {
1185            let (via_dir_symlink, canonical_dir_symlink) =
1186                mint_db_identity(&dir_symlink.join("khive.db")).unwrap();
1187            assert_eq!(canonical_real, canonical_dir_symlink);
1188            assert_eq!(via_real, via_dir_symlink);
1189
1190            let (via_file_symlink, canonical_file_symlink) =
1191                mint_db_identity(&file_symlink).unwrap();
1192            assert_eq!(canonical_real, canonical_file_symlink);
1193            assert_eq!(via_real, via_file_symlink);
1194        }
1195
1196        // Bare file name: resolved against the current directory (step 1).
1197        let bare_name_result = {
1198            let _cwd = CwdGuard::enter(&real_dir);
1199            mint_db_identity(&PathBuf::from("khive.db"))
1200        };
1201        let (via_bare_name, canonical_bare_name) = bare_name_result.unwrap();
1202        assert_eq!(canonical_real, canonical_bare_name);
1203        assert_eq!(via_real, via_bare_name);
1204    }
1205
1206    /// ADR-091 backend-scoped attribution: `DbIdentity`/canonical-path
1207    /// equality across alias spellings (proven above by
1208    /// `mint_db_identity_alias_convergence`) does not by itself prove the
1209    /// walpin sidecar re-key — `sidecar_dir_for` is a separate, purely
1210    /// lexical derivation (`walpin::sidecar_dir_for`) that must be fed the
1211    /// *minted* canonical path, never the raw configured one. This test
1212    /// opens a real `ConnectionPool` (not the private `mint_db_identity` free
1213    /// function) through each alias spelling and asserts
1214    /// `sidecar_dir_for(pool.canonical_path())` converges to one directory —
1215    /// exercising the actual `ConnectionPool::new` → `canonical_path()` wiring
1216    /// every sidecar consumer (`checkpoint.rs`) reads from.
1217    #[test]
1218    #[serial(pool_cwd)]
1219    fn sidecar_dir_for_alias_convergence() {
1220        let dir = tempfile::tempdir().unwrap();
1221        let real_dir = dir.path().join("real");
1222        fs::create_dir(&real_dir).unwrap();
1223        let db_path = real_dir.join("khive.db");
1224        fs::write(&db_path, b"").unwrap();
1225
1226        let dir_symlink = dir.path().join("dir_link");
1227        let file_symlink = dir.path().join("file_link.db");
1228        #[cfg(unix)]
1229        {
1230            std::os::unix::fs::symlink(&real_dir, &dir_symlink).unwrap();
1231            std::os::unix::fs::symlink(&db_path, &file_symlink).unwrap();
1232        }
1233
1234        let pool_for = |path: &Path| -> Arc<ConnectionPool> {
1235            let cfg = PoolConfig {
1236                path: Some(path.to_path_buf()),
1237                ..PoolConfig::default()
1238            };
1239            Arc::new(ConnectionPool::new(cfg).expect("file-backed pool should open"))
1240        };
1241        let sidecar_of = |pool: &ConnectionPool| -> PathBuf {
1242            crate::walpin::sidecar_dir_for(pool.canonical_path().expect("file-backed pool"))
1243        };
1244
1245        let via_real = pool_for(&db_path);
1246        let sidecar_real = sidecar_of(&via_real);
1247
1248        let via_relative = {
1249            let _cwd = CwdGuard::enter(&real_dir);
1250            pool_for(Path::new("khive.db"))
1251        };
1252        assert_eq!(
1253            sidecar_real,
1254            sidecar_of(&via_relative),
1255            "a relative spelling of the same database must derive the same sidecar directory"
1256        );
1257
1258        #[cfg(unix)]
1259        {
1260            let via_dir_symlink = pool_for(&dir_symlink.join("khive.db"));
1261            assert_eq!(
1262                sidecar_real,
1263                sidecar_of(&via_dir_symlink),
1264                "opening through a directory symlink must derive the same sidecar directory"
1265            );
1266
1267            let via_file_symlink = pool_for(&file_symlink);
1268            assert_eq!(
1269                sidecar_real,
1270                sidecar_of(&via_file_symlink),
1271                "opening through a file-level symlink must derive the same sidecar directory"
1272            );
1273        }
1274
1275        let via_bare_name = {
1276            let _cwd = CwdGuard::enter(&real_dir);
1277            pool_for(Path::new("khive.db"))
1278        };
1279        assert_eq!(
1280            sidecar_real,
1281            sidecar_of(&via_bare_name),
1282            "a bare file name resolved against the current directory must derive the same \
1283             sidecar directory"
1284        );
1285    }
1286
1287    /// ADR-091 backend-scoped attribution: opening via a file-level symlink
1288    /// whose target does not exist yet (a valid first-open state), then
1289    /// after the target is created, opening via the target path directly,
1290    /// must mint identical `DbIdentity` values — the first-open path
1291    /// resolves the final component before canonicalizing the parent.
1292    #[cfg(unix)]
1293    #[test]
1294    fn mint_db_identity_dangling_symlink_first_open_convergence() {
1295        let dir = tempfile::tempdir().unwrap();
1296        let target = dir.path().join("target.db");
1297        let link = dir.path().join("link.db");
1298        std::os::unix::fs::symlink(&target, &link).unwrap();
1299        assert!(!target.exists(), "target must not exist yet (dangling)");
1300
1301        let (via_dangling_link, canonical_via_link) = mint_db_identity(&link).unwrap();
1302
1303        // Now create the target (as SQLite would on first write) and mint
1304        // again directly against the target path.
1305        fs::write(&target, b"").unwrap();
1306        let (via_target, canonical_via_target) = mint_db_identity(&target).unwrap();
1307
1308        assert_eq!(canonical_via_link, canonical_via_target);
1309        assert_eq!(via_dangling_link, via_target);
1310    }
1311
1312    /// A resolved target whose parent directory does not exist must fail
1313    /// minting exactly as the subsequent database open itself would fail.
1314    #[test]
1315    fn mint_db_identity_missing_parent_fails() {
1316        let dir = tempfile::tempdir().unwrap();
1317        let missing = dir.path().join("nonexistent_subdir").join("khive.db");
1318        let result = mint_db_identity(&missing);
1319        assert!(
1320            result.is_err(),
1321            "minting must fail when the parent directory does not exist"
1322        );
1323    }
1324
1325    /// Non-UTF-8 database paths (Unix) must round-trip through
1326    /// `DbIdentity`/canonicalization without loss.
1327    #[cfg(unix)]
1328    #[test]
1329    fn mint_db_identity_non_utf8_path_round_trips() {
1330        use std::ffi::OsStr;
1331        use std::os::unix::ffi::OsStrExt;
1332
1333        let dir = tempfile::tempdir().unwrap();
1334        // 0xFF is not valid UTF-8 as a standalone byte.
1335        let raw_name = OsStr::from_bytes(b"khive-\xffdb.sqlite");
1336        let db_path = dir.path().join(raw_name);
1337        // Some Unix filesystems (notably macOS's APFS) reject non-UTF-8
1338        // names outright at the syscall level — that is a filesystem
1339        // limitation, not a `mint_db_identity` bug, so skip rather than
1340        // fail where the underlying `write` itself cannot succeed.
1341        if let Err(e) = fs::write(&db_path, b"") {
1342            eprintln!(
1343                "skipping mint_db_identity_non_utf8_path_round_trips: filesystem rejected a \
1344                 non-UTF-8 file name ({e}); this platform's filesystem does not support the \
1345                 case under test"
1346            );
1347            return;
1348        }
1349
1350        let (identity, canonical) = mint_db_identity(&db_path).unwrap();
1351        assert_eq!(canonical.file_name().unwrap(), raw_name);
1352
1353        let (identity_again, canonical_again) = mint_db_identity(&db_path).unwrap();
1354        assert_eq!(identity, identity_again);
1355        assert_eq!(canonical, canonical_again);
1356    }
1357}