From 32ac79f80792467d5613b785d6dea8d970227024 Mon Sep 17 00:00:00 2001 From: Wess Cope Date: Tue, 20 Oct 2026 11:24:09 -0400 Subject: [PATCH] Queue competing memory writes fairly Windows sustained contention exposed writer starvation after the snapshot upgrade fix. Share a FIFO gate across handles of the same database, acquire it before a connection, and use it for memory, learning, and passive usage transactions. Keep reads available and retain weak references only while handles are alive. Exercise independent handles and read availability under a held writer, repeat contention tests on Windows CI, and advance desktop and core to 1.14.1. The unpublished 1.14.0 draft remains withheld. --- .github/workflows/ci.yml | 7 +++ Cargo.lock | 4 +- crates/synapse/Cargo.toml | 2 +- crates/synapsecore/Cargo.lock | 2 +- crates/synapsecore/Cargo.toml | 2 +- crates/synapsecore/src/brain/ingest.rs | 2 + crates/synapsecore/src/brain/rekey.rs | 1 + crates/synapsecore/src/brain/store.rs | 11 +++++ crates/synapsecore/src/database.rs | 2 + crates/synapsecore/src/database/writer.rs | 24 ++++++++++ crates/synapsecore/src/learning/approve.rs | 3 +- crates/synapsecore/src/learning/feedback.rs | 5 +- crates/synapsecore/src/learning/promotion.rs | 3 +- crates/synapsecore/src/learning/resolve.rs | 3 +- crates/synapsecore/src/usage/record.rs | 3 +- .../synapsecore/tests/learning/concurrency.rs | 47 +++++++++++++++++-- 16 files changed, 107 insertions(+), 14 deletions(-) create mode 100644 crates/synapsecore/src/database/writer.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 50689ee..fd18a9b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -60,6 +60,13 @@ jobs: run: cargo test --locked - name: Core tests run: cargo test --locked --manifest-path crates/synapsecore/Cargo.toml + - name: Memory write contention + shell: pwsh + run: | + 1..3 | ForEach-Object { + cargo test --locked --manifest-path crates/synapsecore/Cargo.toml --test learning concurrency + if ($LASTEXITCODE -ne 0) { throw 'Memory write contention failed' } + } - uses: oven-sh/setup-bun@v2 - name: Windows pi bridge run: bun test pi/tests diff --git a/Cargo.lock b/Cargo.lock index 65b718d..4e1610b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5782,7 +5782,7 @@ dependencies = [ [[package]] name = "synapse" -version = "1.14.0" +version = "1.14.1" dependencies = [ "anyhow", "async-channel 2.5.0", @@ -5797,7 +5797,7 @@ dependencies = [ [[package]] name = "synapsecore" -version = "1.14.0" +version = "1.14.1" dependencies = [ "anyhow", "chacha20poly1305", diff --git a/crates/synapse/Cargo.toml b/crates/synapse/Cargo.toml index 6fc4cb6..82a7525 100644 --- a/crates/synapse/Cargo.toml +++ b/crates/synapse/Cargo.toml @@ -2,7 +2,7 @@ name = "synapse" # Bumping this on main is the release trigger. Keep it in step with # crates/synapsecore, which is where `synapse version` reads its number from. -version = "1.14.0" +version = "1.14.1" edition.workspace = true description = "Local memory and scoped credentials for developer tools" diff --git a/crates/synapsecore/Cargo.lock b/crates/synapsecore/Cargo.lock index 623d859..cfc5044 100644 --- a/crates/synapsecore/Cargo.lock +++ b/crates/synapsecore/Cargo.lock @@ -1789,7 +1789,7 @@ dependencies = [ [[package]] name = "synapsecore" -version = "1.14.0" +version = "1.14.1" dependencies = [ "anyhow", "chacha20poly1305", diff --git a/crates/synapsecore/Cargo.toml b/crates/synapsecore/Cargo.toml index b2c6a7f..382aa7b 100644 --- a/crates/synapsecore/Cargo.toml +++ b/crates/synapsecore/Cargo.toml @@ -20,7 +20,7 @@ name = "synapsecore" # Keep this in step with crates/synapse: `synapse version` reads the version of # the crate that holds the command, and the two ship as one product. Only the # desktop crate's version triggers the release workflow. -version = "1.14.0" +version = "1.14.1" edition = "2024" description = "Memory, credentials, and the command surface for Synapse, without a window" diff --git a/crates/synapsecore/src/brain/ingest.rs b/crates/synapsecore/src/brain/ingest.rs index 446e4fd..58d644d 100644 --- a/crates/synapsecore/src/brain/ingest.rs +++ b/crates/synapsecore/src/brain/ingest.rs @@ -66,6 +66,7 @@ impl Brain { includeflagged: bool, ) -> Result { let created = now()?; + let _writer = self.writing().await; let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?; let batchid = sqlx::query("INSERT INTO importbatch(provider, created) VALUES (?, ?)") .bind(preview.provider.value()) @@ -207,6 +208,7 @@ impl Brain { .await? .context("import batch not found")?; anyhow::ensure!(!batch.undone, "import batch {id} was already undone"); + let _writer = self.writing().await; let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?; let memories = sqlx::query_scalar::<_, i64>( "SELECT DISTINCT memoryid FROM memoryorigin WHERE batchid = ?", diff --git a/crates/synapsecore/src/brain/rekey.rs b/crates/synapsecore/src/brain/rekey.rs index ecbe16e..0bea6cd 100644 --- a/crates/synapsecore/src/brain/rekey.rs +++ b/crates/synapsecore/src/brain/rekey.rs @@ -54,6 +54,7 @@ impl Brain { /// Where two checkouts of one repository each hold an identical memory, the /// second keeps its old identity rather than colliding with the first. async fn move_project(&self, from: &str, to: &str) -> Result { + let _writer = self.writing().await; let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?; let rows = sqlx::query_as::<_, (i64, String, String, i64)>( "SELECT meta.memoryid, memory.body, memory.source, CAST(memory.created AS INTEGER) \ diff --git a/crates/synapsecore/src/brain/store.rs b/crates/synapsecore/src/brain/store.rs index aafb52d..5b6a38d 100644 --- a/crates/synapsecore/src/brain/store.rs +++ b/crates/synapsecore/src/brain/store.rs @@ -22,6 +22,7 @@ pub struct Brain { pub(super) pool: SqlitePool, path: PathBuf, _lock: Arc, + writer: Arc>, } impl Brain { @@ -40,6 +41,7 @@ impl Brain { fn from(opened: crate::database::Opened, path: PathBuf) -> Result { Ok(Self { + writer: crate::database::writequeue(&path)?, pool: opened.pool, path, _lock: opened.lock, @@ -85,6 +87,7 @@ impl Brain { .duration_since(UNIX_EPOCH) .context("system clock is before the Unix epoch")? .as_secs() as i64; + let _writer = self.writing().await; let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?; let result = sqlx::query("INSERT INTO memory(body, source, created) VALUES (?, ?, ?)") .bind(body) @@ -169,6 +172,7 @@ impl Brain { ) -> Result { let body = body.trim(); anyhow::ensure!(!body.is_empty(), "memory content cannot be empty"); + let _writer = self.writing().await; let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?; let result = sqlx::query("INSERT INTO memory(body, source, created) VALUES (?, ?, ?)") .bind(body) @@ -206,6 +210,10 @@ impl Brain { } } + pub(crate) async fn writing(&self) -> tokio::sync::MutexGuard<'_, ()> { + self.writer.lock().await + } + pub(crate) fn databasepath(&self) -> &Path { &self.path } @@ -527,6 +535,7 @@ impl Brain { ) -> Result> { let body = stored(body)?; let body = body.as_str(); + let _writer = self.writing().await; let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?; crate::brain::gram::unindex(&mut transaction, id).await?; let result = @@ -554,6 +563,7 @@ impl Brain { pub async fn deletememory(&self, id: i64) -> Result> { let memory = self.memory(id).await?; if memory.is_some() { + let _writer = self.writing().await; let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?; // Anything this memory replaced comes back. Leaving those rows // pointing at an id that no longer resolves would hide them for @@ -583,6 +593,7 @@ impl Brain { } pub async fn wipememories(&self) -> Result { + let _writer = self.writing().await; let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?; sqlx::query("DELETE FROM memoryorigin") .execute(&mut *transaction) diff --git a/crates/synapsecore/src/database.rs b/crates/synapsecore/src/database.rs index 9ab4fa8..d042bf9 100644 --- a/crates/synapsecore/src/database.rs +++ b/crates/synapsecore/src/database.rs @@ -4,10 +4,12 @@ mod migration; mod permission; #[cfg(test)] mod tests; +mod writer; pub use backup::{folder as backupfolder, snapshots}; pub use lifecycle::{check, export, restore}; pub use permission::{securedirectory, securefile}; +pub(crate) use writer::writequeue; use anyhow::{Context, Result}; use sqlx::SqlitePool; diff --git a/crates/synapsecore/src/database/writer.rs b/crates/synapsecore/src/database/writer.rs new file mode 100644 index 0000000..cb492a6 --- /dev/null +++ b/crates/synapsecore/src/database/writer.rs @@ -0,0 +1,24 @@ +use anyhow::Result; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex, Weak}; +use tokio::sync::Mutex as Queue; + +static WRITERS: Mutex>)>> = Mutex::new(Vec::new()); + +/// SQLite still arbitrates between processes. Within one process, queue long +/// writes fairly before taking a connection so competing busy handlers cannot +/// starve a writer or occupy the entire read pool. +pub(crate) fn writequeue(path: &Path) -> Result>> { + let path = std::fs::canonicalize(path)?; + let mut writers = WRITERS.lock().unwrap_or_else(|error| error.into_inner()); + writers.retain(|(_, queue)| queue.strong_count() > 0); + if let Some(queue) = writers + .iter() + .find_map(|(known, queue)| (known == &path).then(|| queue.upgrade()).flatten()) + { + return Ok(queue); + } + let queue = Arc::new(Queue::new(())); + writers.push((path, Arc::downgrade(&queue))); + Ok(queue) +} diff --git a/crates/synapsecore/src/learning/approve.rs b/crates/synapsecore/src/learning/approve.rs index 29fe5d7..a8cf1e0 100644 --- a/crates/synapsecore/src/learning/approve.rs +++ b/crates/synapsecore/src/learning/approve.rs @@ -81,7 +81,8 @@ impl Brain { ) .await?; let now = now()?; - let mut transaction = self.pool().begin().await?; + let _writer = self.writing().await; + let mut transaction = self.pool().begin_with("BEGIN IMMEDIATE").await?; let id: i64 = sqlx::query_scalar( "INSERT INTO learningpromotion(project,lesson,feedback,skill,description,instructions,origin,created) \ VALUES (?,?,?,?,?,?,'revision',?) ON CONFLICT(project,lesson) DO UPDATE SET \ diff --git a/crates/synapsecore/src/learning/feedback.rs b/crates/synapsecore/src/learning/feedback.rs index 1dfe76f..24e9116 100644 --- a/crates/synapsecore/src/learning/feedback.rs +++ b/crates/synapsecore/src/learning/feedback.rs @@ -27,7 +27,8 @@ impl Brain { let lesson = digest(&(&target, &problem, &correction, &query, &expected))?; let session = digest(&session)?; let now = now()?; - let mut transaction = self.pool().begin().await?; + let _writer = self.writing().await; + let mut transaction = self.pool().begin_with("BEGIN IMMEDIATE").await?; let id: i64 = sqlx::query_scalar( "INSERT INTO learningfeedback(project, target, digest, lesson, session, outcome, \ problem, correction, evidence, query, expected, created, updated) \ @@ -114,6 +115,7 @@ impl Brain { "the target changed; collect fresh evidence" ); let now = now()?; + let _writer = self.writing().await; let mut transaction = self.pool().begin_with("BEGIN IMMEDIATE").await?; let mut current: Record = sqlx::query_as("SELECT * FROM learningfeedback WHERE id = ? AND project = ?") @@ -174,6 +176,7 @@ impl Brain { .bind(current.checked).bind(current.reviewed).bind(current.unstable).bind(now).bind(id) .execute(&mut *transaction).await?; transaction.commit().await?; + drop(_writer); let record = self.feedbackrecord(id, folder).await?; self.collectlearning(&record, folder).await?; Ok(record) diff --git a/crates/synapsecore/src/learning/promotion.rs b/crates/synapsecore/src/learning/promotion.rs index b1472d3..9d7f02e 100644 --- a/crates/synapsecore/src/learning/promotion.rs +++ b/crates/synapsecore/src/learning/promotion.rs @@ -59,7 +59,8 @@ impl Brain { ); let now = now()?; let assessment = self.assessevidence(record).await?; - let mut transaction = self.pool().begin().await?; + let _writer = self.writing().await; + let mut transaction = self.pool().begin_with("BEGIN IMMEDIATE").await?; if !record.query.is_empty() { sqlx::query( "INSERT INTO learningcase(project,feedback,query,expected,created) VALUES (?,?,?,?,?) \ diff --git a/crates/synapsecore/src/learning/resolve.rs b/crates/synapsecore/src/learning/resolve.rs index 7f56995..4c25185 100644 --- a/crates/synapsecore/src/learning/resolve.rs +++ b/crates/synapsecore/src/learning/resolve.rs @@ -119,7 +119,8 @@ impl Brain { let checkout = crate::brain::projectroot(folder)?; let fingerprint = fingerprint(finding, &hashes, checkout.as_deref())?; let now = now()?; - let mut transaction = self.pool().begin().await?; + let _writer = self.writing().await; + let mut transaction = self.pool().begin_with("BEGIN IMMEDIATE").await?; let id: i64 = sqlx::query_scalar( "INSERT INTO driftresolution(project, fingerprint, memoryid, disposition, note, created) \ VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(project, fingerprint) DO UPDATE SET \ diff --git a/crates/synapsecore/src/usage/record.rs b/crates/synapsecore/src/usage/record.rs index 7a4d3af..c8f43d7 100644 --- a/crates/synapsecore/src/usage/record.rs +++ b/crates/synapsecore/src/usage/record.rs @@ -103,7 +103,8 @@ pub async fn record(brain: &Brain, entry: Entry) -> Result<()> { .duration_since(UNIX_EPOCH) .map(|elapsed| elapsed.as_secs() as i64) .unwrap_or_default(); - let mut transaction = brain.pool().begin().await?; + let _writer = brain.writing().await; + let mut transaction = brain.pool().begin_with("BEGIN IMMEDIATE").await?; sqlx::query( "INSERT INTO usage(kind, memories, reach, matched, sent, overhead, created) \ VALUES (?, ?, ?, ?, ?, ?, ?)", diff --git a/crates/synapsecore/tests/learning/concurrency.rs b/crates/synapsecore/tests/learning/concurrency.rs index f90018b..1fe742c 100644 --- a/crates/synapsecore/tests/learning/concurrency.rs +++ b/crates/synapsecore/tests/learning/concurrency.rs @@ -1,15 +1,23 @@ use super::Fixture; +use anyhow::Context; use std::sync::Arc; -use synapsecore::brain::MemoryScope; +use synapsecore::brain::{Brain, MemoryScope}; use synapsecore::usage::{self, Entry}; #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn memory_writes_and_passive_bookkeeping_can_arrive_together() { let f = Fixture::new().await; + let peer = Brain::glance(f._directory.path().join("brain.db")) + .await + .unwrap(); let ready = Arc::new(tokio::sync::Barrier::new(16)); let mut tasks = Vec::new(); for worker in 0..16 { - let brain = f.brain.clone(); + let brain = if worker % 4 < 2 { + f.brain.clone() + } else { + peer.clone() + }; let folder = f.root.clone(); let ready = ready.clone(); tasks.push(tokio::spawn(async move { @@ -23,9 +31,12 @@ async fn memory_writes_and_passive_bookkeeping_can_arrive_together() { MemoryScope::Project, Some(&folder), ) - .await?; + .await + .with_context(|| format!("memory worker {worker}, write {n}"))?; } else { - usage::record(&brain, Entry::session(1_000, 100)).await?; + usage::record(&brain, Entry::session(1_000, 100)) + .await + .with_context(|| format!("bookkeeping worker {worker}, write {n}"))?; } } anyhow::Ok(()) @@ -36,3 +47,31 @@ async fn memory_writes_and_passive_bookkeeping_can_arrive_together() { } assert_eq!(f.brain.stats().await.unwrap().entries, 401); } + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn queued_writers_leave_connections_available_for_reads() { + let f = Fixture::new().await; + let held = f.brain.pool().begin_with("BEGIN IMMEDIATE").await.unwrap(); + let mut writers = Vec::new(); + for _ in 0..16 { + let brain = f.brain.clone(); + writers.push(tokio::spawn(async move { + usage::record(&brain, Entry::session(1_000, 100)) + .await + .unwrap(); + })); + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + let stats = tokio::time::timeout(std::time::Duration::from_secs(2), f.brain.stats()).await; + held.commit().await.unwrap(); + for writer in writers { + writer.await.unwrap(); + } + assert_eq!( + stats + .expect("waiting writers occupied every read connection") + .unwrap() + .entries, + 1 + ); +}