fix(db): serialize SQLite writes to eliminate 'database is locked' (v1.1.3)

CHANGELOG.md

@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.1.3] - 2026-06-01
+
+### Fixed
+
+- **Serialize SQLite writes to eliminate `database is locked` errors and dropped log batches.**
+  cortex ran an r2d2 pool of 4 write-capable connections while 6 writer subsystems
+  (syslog/docker ingest, heartbeat, notifications, AI index, retention maintenance)
+  raced SQLite's single write lock, exceeding `busy_timeout` (~1000+ lock errors/day and
+  silently discarded log batches). Added a process-wide reentrant write-serialization
+  lock (`db::write_lock()`) acquired by every write transaction and standalone mutating
+  statement (purge DELETEs, VACUUM, incremental_vacuum, merge). Reads stay concurrent on
+  the pool (WAL); PASSIVE checkpoints are left unguarded.
+
+
 ## [1.1.2] - 2026-05-31
 
 ### Fixed

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "cortex"
-version = "1.1.2"
+version = "1.1.3"
 edition = "2021"
 rust-version = "1.86"
 description = "Homelab intelligence platform — syslog/OTLP/Docker log aggregation, fleet awareness, and AI agent coordination over MCP, CLI, and HTTP"
@@ -27,6 +27,7 @@ tower-http = { version = "0.6", features = ["cors", "limit", "trace"] }
 
 # SQLite
 rusqlite = { version = "0.39", features = ["bundled", "vtab", "backup"] }
+parking_lot = "0.12"
 r2d2 = "0.8"
 r2d2_sqlite = "0.33"
 scheduled-thread-pool = "0.2"

mcpb/manifest.json

@@ -3,7 +3,7 @@
   "manifest_version": "0.4",
   "name": "cortex",
   "display_name": "Cortex",
-  "version": "1.1.2",
+  "version": "1.1.3",
   "description": "Query local cortex SQLite logs through a bundled stdio MCP server.",
   "long_description": "cortex packages the existing cortex stdio entrypoint as a local MCP Bundle. It is query-only: it reads the configured SQLite database and does not start syslog listeners, HTTP servers, Docker Compose, REST, or deploy flows.",
   "author": {

server.json

@@ -7,7 +7,7 @@
     "url": "https://github.com/jmagar/cortex",
     "source": "github"
   },
-  "version": "1.1.2",
+  "version": "1.1.3",
   "packages": [
     {
       "registryType": "oci",

src/app/error_detection/scanner.rs

@@ -183,6 +183,7 @@ pub(crate) fn process_chunk(
     }
 
     // --- Write signatures and windows in a single transaction ---
+    let _write_guard = crate::db::write_lock();
     let tx = conn.transaction()?;
 
     for (hash, group) in &groups {

src/app/service.rs

@@ -2059,6 +2059,7 @@ impl CortexService {
         let hash_clone = hash.clone();
         self.run_db("ack_error.commit", move |pool| {
             let mut conn = pool.get()?;
+            let _write_guard = crate::db::write_lock();
             let tx = conn.transaction()?;
             crate::db::error_signatures::record_ack_event(
                 &tx,
@@ -2127,6 +2128,7 @@ impl CortexService {
         let hash_clone = hash.clone();
         self.run_db("unack_error.commit", move |pool| {
             let mut conn = pool.get()?;
+            let _write_guard = crate::db::write_lock();
             let tx = conn.transaction()?;
             crate::db::error_signatures::record_ack_event(
                 &tx,

src/db.rs

@@ -46,7 +46,7 @@ pub use models::{
 pub use models::{StorageBudgetState, StorageEnforcementOutcome, StorageMetrics, StorageRecovery};
 pub use pool::{
     backfill_inventory_stats, init_pool, inventory_backfill_complete, read_schema_version_info,
-    read_schema_version_info_conn, DbPool, SchemaVersionInfo, KNOWN_SCHEMA_VERSION,
+    read_schema_version_info_conn, write_lock, DbPool, SchemaVersionInfo, KNOWN_SCHEMA_VERSION,
 };
 pub use queries::{
     ai_session_rollup_status, ask_history_sessions, get_error_summary, get_stats,

src/db/ingest.rs

@@ -33,6 +33,7 @@ pub fn insert_logs_batch(pool: &DbPool, entries: &[LogBatchEntry]) -> Result<usi
 
 fn insert_logs_batch_once(pool: &DbPool, entries: &[LogBatchEntry]) -> Result<usize> {
     let mut conn = pool.get()?;
+    let _write_guard = crate::db::write_lock();
     let tx = conn.transaction()?;
     insert_logs_batch_in_tx(&tx, entries)?;
     tx.commit()?;

src/db/maintenance.rs

@@ -52,12 +52,14 @@ pub fn db_wal_checkpoint(pool: &DbPool, mode: &str) -> Result<(i64, i64, i64)> {
 
 pub fn db_incremental_vacuum(pool: &DbPool, pages: u32) -> Result<()> {
     let conn = pool.get()?;
+    let _write_guard = crate::db::write_lock();
     conn.execute_batch(&format!("PRAGMA incremental_vacuum({pages});"))?;
     Ok(())
 }
 
 pub fn db_full_vacuum(pool: &DbPool) -> Result<()> {
     let conn = pool.get()?;
+    let _write_guard = crate::db::write_lock();
     conn.execute_batch("VACUUM;")?;
     Ok(())
 }
@@ -303,6 +305,7 @@ fn fts_incremental_merge(pool: &DbPool, deleted_rows: usize, merge_pages: u32) {
     for i in 0..iterations {
         match pool.get() {
             Ok(conn) => {
+                let _write_guard = crate::db::write_lock();
                 match conn.execute_batch(&merge_stmt) {
                     Ok(()) => {
                         consecutive_failures = 0;
@@ -406,6 +409,7 @@ pub fn purge_old_logs(pool: &DbPool, retention_days: u32, fts_merge_pages: u32)
     let mut total_deleted: usize = 0;
     loop {
         let conn = pool.get()?;
+        let _write_guard = crate::db::write_lock();
         let chunk = conn.execute(
             "DELETE FROM logs WHERE id IN (
                  SELECT id FROM logs
@@ -522,6 +526,7 @@ pub fn purge_by_tag_window(
     let mut total_deleted: usize = 0;
     loop {
         let conn = pool.get()?;
+        let _write_guard = crate::db::write_lock();
         let chunk = conn.execute(
             "DELETE FROM logs WHERE id IN (
                  SELECT id FROM logs
@@ -628,6 +633,7 @@ fn delete_oldest_logs_chunk(pool: &DbPool, chunk_size: usize) -> Result<DeletedC
 
     // Delete the oldest chunk using a subquery — O(1) SQL string size regardless
     // of chunk_size, no expression depth issues.
+    let _write_guard = crate::db::write_lock();
     let deleted_rows = conn.execute(
         "DELETE FROM logs \
          WHERE id IN (SELECT id FROM logs ORDER BY received_at ASC, id ASC LIMIT ?1)",
@@ -662,6 +668,7 @@ fn delete_heartbeat_chunk_where(
     chunk_size: usize,
 ) -> Result<usize> {
     let mut conn = pool.get()?;
+    let _write_guard = crate::db::write_lock();
     let tx = conn.transaction()?;
     tx.execute_batch(
         "CREATE TEMP TABLE IF NOT EXISTS temp_heartbeat_delete_ids (
@@ -727,6 +734,7 @@ fn delete_orphan_heartbeat_children(pool: &DbPool) -> Result<usize> {
     let conn = pool.get()?;
     let mut total_deleted = 0usize;
     for table in HEARTBEAT_CHILD_TABLES {
+        let _write_guard = crate::db::write_lock();
         let deleted = conn.execute(
             &format!(
                 "DELETE FROM {table}
@@ -757,6 +765,7 @@ fn reconcile_hosts(pool: &DbPool, hostnames: &[String]) -> Result<()> {
     }
 
     let mut conn = pool.get()?;
+    let _write_guard = crate::db::write_lock();
     let tx = conn.transaction()?;
     for hostname in hostnames {
         // One query: count + timestamp bounds in a single pass over the index.
@@ -797,6 +806,7 @@ fn checkpoint_wal_and_incremental_vacuum(pool: &DbPool) -> Result<()> {
     } else {
         tracing::debug!("WAL checkpoint completed");
     }
+    let _write_guard = crate::db::write_lock();
     if let Err(e) = conn.execute_batch("PRAGMA incremental_vacuum(1000);") {
         tracing::warn!(error = %e, "incremental vacuum skipped (non-fatal)");
     } else {

src/db/pool.rs

@@ -9,6 +9,21 @@ use crate::config::StorageConfig;
 
 pub type DbPool = Pool<SqliteConnectionManager>;
 
+/// Process-wide SQLite **write serialization** lock.
+///
+/// SQLite permits only one writer at a time, but cortex runs an r2d2 pool of several
+/// connections with multiple concurrent writer subsystems (syslog/docker ingest,
+/// heartbeat, notifications, AI index, retention maintenance). Without serialization
+/// these race SQLite's single write lock, exceed `busy_timeout`, and surface as
+/// `database is locked` — dropping log batches. Every write transaction acquires this
+/// guard so writers queue in-process instead of colliding at the SQLite layer; reads
+/// stay concurrent on the pool (WAL allows many readers). Reentrant so a write path that
+/// nests guarded helpers on a single thread cannot deadlock.
+pub fn write_lock() -> parking_lot::ReentrantMutexGuard<'static, ()> {
+    static WRITE_LOCK: parking_lot::ReentrantMutex<()> = parking_lot::ReentrantMutex::new(());
+    WRITE_LOCK.lock()
+}
+
 pub const KNOWN_SCHEMA_VERSION: i64 = 20;
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

src/heartbeat.rs

@@ -149,6 +149,7 @@ fn insert_heartbeat(
 ) -> anyhow::Result<HeartbeatIngestResponse> {
     let received_at = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string();
     let mut conn = pool.get()?;
+    let _write_guard = crate::db::write_lock();
     let tx = conn.transaction()?;
     let metadata_json = heartbeat_metadata_json(&request)?;
 

src/notifications/dispatcher.rs

@@ -79,6 +79,7 @@ where
     let exec_start = Instant::now();
     let join_result = tokio::task::spawn_blocking(move || -> Result<()> {
         let mut conn = pool.get()?;
+        let _write_guard = crate::db::write_lock();
         let tx = conn.transaction()?;
         f(&tx)?;
         tx.commit()?;

src/scanner.rs

@@ -757,6 +757,7 @@ fn flush_chunk(
     if batch.is_empty() {
         if let Some(file_metadata) = completion_metadata {
             let mut conn = pool.get()?;
+            let _write_guard = crate::db::write_lock();
             let tx = conn.transaction()?;
             checkpoint::update_source_metadata_in_tx(&tx, source_id, file_metadata)?;
             tx.commit()?;
@@ -776,6 +777,7 @@ fn flush_chunk(
     }
 
     let mut conn = pool.get()?;
+    let _write_guard = crate::db::write_lock();
     let tx = conn.transaction()?;
     let claimed = checkpoint::claim_imports_in_tx(&tx, source_id, imports)?;
     let mut claimed_batch = Vec::with_capacity(batch.len());

src/scanner/checkpoint.rs

@@ -137,6 +137,7 @@ impl<'a> CheckpointStore<'a> {
 
     pub fn reset_source(&self, source_id: i64, canonical_path: &str) -> Result<()> {
         let mut conn = self.pool.get()?;
+        let _write_guard = crate::db::write_lock();
         let tx = conn.transaction()?;
         tx.execute(
             "DELETE FROM transcript_import_records WHERE source_id = ?1",
@@ -304,6 +305,7 @@ impl<'a> CheckpointStore<'a> {
         }
 
         let mut conn = self.pool.get()?;
+        let _write_guard = crate::db::write_lock();
         let tx = conn.transaction()?;
         for checkpoint in &checkpoints {
             let source_id: i64 = tx.query_row(