Prevent memory writer starvation and release 1.14.1

.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

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"
 

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"
 

crates/synapsecore/src/brain/ingest.rs

@@ -66,6 +66,7 @@ impl Brain {
         includeflagged: bool,
     ) -> Result<ImportReport> {
         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 = ?",

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<usize> {
+        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) \

crates/synapsecore/src/brain/store.rs

@@ -22,6 +22,7 @@ pub struct Brain {
     pub(super) pool: SqlitePool,
     path: PathBuf,
     _lock: Arc<File>,
+    writer: Arc<tokio::sync::Mutex<()>>,
 }
 
 impl Brain {
@@ -40,6 +41,7 @@ impl Brain {
 
     fn from(opened: crate::database::Opened, path: PathBuf) -> Result<Self> {
         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<i64> {
         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<Option<Memory>> {
         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<Option<Memory>> {
         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<u64> {
+        let _writer = self.writing().await;
         let mut transaction = self.pool.begin_with("BEGIN IMMEDIATE").await?;
         sqlx::query("DELETE FROM memoryorigin")
             .execute(&mut *transaction)

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;

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<Vec<(PathBuf, Weak<Queue<()>>)>> = 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<Arc<Queue<()>>> {
+    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)
+}

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 \

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)

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 (?,?,?,?,?) \

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 \

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 (?, ?, ?, ?, ?, ?, ?)",

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
+    );
+}