diff --git a/CHANGELOG.md b/CHANGELOG.md index d83d0787..2bb4409e 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/Cargo.lock b/Cargo.lock index 15130ba1..65601563 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -418,7 +418,7 @@ checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" [[package]] name = "cortex" -version = "1.1.2" +version = "1.1.3" dependencies = [ "anyhow", "axum", @@ -435,6 +435,7 @@ dependencies = [ "lru", "notify", "opentelemetry-proto", + "parking_lot", "prost", "r2d2", "r2d2_sqlite", diff --git a/Cargo.toml b/Cargo.toml index 8e9c5967..f61012df 100644 --- a/Cargo.toml +++ b/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" diff --git a/mcpb/manifest.json b/mcpb/manifest.json index fef0a7a1..9d723e65 100644 --- a/mcpb/manifest.json +++ b/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": { diff --git a/server.json b/server.json index efe9ce0c..99fda00e 100644 --- a/server.json +++ b/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", diff --git a/src/app/error_detection/scanner.rs b/src/app/error_detection/scanner.rs index ca945a7b..8dd13ef1 100644 --- a/src/app/error_detection/scanner.rs +++ b/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 { diff --git a/src/app/service.rs b/src/app/service.rs index ac25a4d5..dc4201e1 100644 --- a/src/app/service.rs +++ b/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, diff --git a/src/db.rs b/src/db.rs index 1a0c8067..e982a324 100644 --- a/src/db.rs +++ b/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, diff --git a/src/db/ingest.rs b/src/db/ingest.rs index c46e9a13..c1deb764 100644 --- a/src/db/ingest.rs +++ b/src/db/ingest.rs @@ -33,6 +33,7 @@ pub fn insert_logs_batch(pool: &DbPool, entries: &[LogBatchEntry]) -> Result Result { 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()?; diff --git a/src/db/maintenance.rs b/src/db/maintenance.rs index c55a9086..a9590169 100644 --- a/src/db/maintenance.rs +++ b/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 Result { 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 { 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 { diff --git a/src/db/pool.rs b/src/db/pool.rs index dff0bd46..c1c24fae 100644 --- a/src/db/pool.rs +++ b/src/db/pool.rs @@ -9,6 +9,21 @@ use crate::config::StorageConfig; pub type DbPool = Pool; +/// 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)] diff --git a/src/heartbeat.rs b/src/heartbeat.rs index 0b2bf5a4..4c9eb0a5 100644 --- a/src/heartbeat.rs +++ b/src/heartbeat.rs @@ -149,6 +149,7 @@ fn insert_heartbeat( ) -> anyhow::Result { 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)?; diff --git a/src/notifications/dispatcher.rs b/src/notifications/dispatcher.rs index b9aaed7f..acde76a6 100644 --- a/src/notifications/dispatcher.rs +++ b/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()?; diff --git a/src/scanner.rs b/src/scanner.rs index f3575335..61a295ff 100644 --- a/src/scanner.rs +++ b/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()); diff --git a/src/scanner/checkpoint.rs b/src/scanner/checkpoint.rs index b1d838a7..31353688 100644 --- a/src/scanner/checkpoint.rs +++ b/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(