Wait for concurrent writes when reserving Slack work

Author: bkrabach

src/store.rs

@@ -377,7 +377,12 @@ impl Store {
     }
     pub fn claim(&mut self, a: &Attention) -> Result<Option<Job>> {
         let now = now_ms();
-        let tx = self.db.transaction()?;
+        // Reserve the write before reading: permission/status connections can
+        // write concurrently, and a deferred read-to-write upgrade can fail
+        // immediately with SQLITE_BUSY instead of honoring the busy timeout.
+        let tx = self
+            .db
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
         let candidate:Option<(String,String,String,String)>=tx.query_row("SELECT id,context,trigger,note FROM jobs j WHERE status='pending' AND available<=?2 AND updated<=?1 AND NOT EXISTS(SELECT 1 FROM jobs x WHERE x.context=j.context AND x.status='running') ORDER BY created LIMIT 1",params![now-a.settle_ms as i64,now],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?))).optional()?;
         let Some((id, context, trigger, note)) = candidate else {
             return Ok(None);
@@ -470,7 +475,12 @@ impl Store {
         Ok(())
     }
     pub fn claim_outgoing(&mut self, c: &Config) -> Result<Option<Outgoing>> {
-        let tx = self.db.transaction()?;
+        // Reserve the write before reading: permission/status connections can
+        // write concurrently, and a deferred read-to-write upgrade can fail
+        // immediately with SQLITE_BUSY instead of honoring the busy timeout.
+        let tx = self
+            .db
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
         let candidate:Option<(Outgoing,String,i64,i64,i64)>=tx.query_row("SELECT o.id,o.channel,o.thread,o.text,j.trigger,j.version,c.human_version,o.media,(SELECT COUNT(*) FROM outbox_parts p WHERE p.outbox_id=o.id AND p.status='sent') FROM outbox o JOIN jobs j ON o.job_id=j.id JOIN contexts c ON c.id=o.context WHERE o.status='ready' AND o.available<=?1 ORDER BY o.created LIMIT 1",[now_ms()],|r| {
             let raw:String=r.get(7)?;
             let media=serde_json::from_str(&raw).map_err(|e|rusqlite::Error::FromSqlConversionFailure(7,rusqlite::types::Type::Text,Box::new(e)))?;

tests/response_delivery.rs

@@ -297,3 +297,43 @@ fn bot_continuation_is_owner_opt_in_and_keeps_hourly_limits() {
     s.db.execute("UPDATE outbox SET available=0", []).unwrap();
     assert!(s.claim_outgoing(&c).unwrap().is_some());
 }
+
+#[test]
+fn reservations_wait_for_other_connections_instead_of_aborting_the_connector() {
+    for reserve_delivery in [false, true] {
+        let dir = tempfile::tempdir().unwrap();
+        let mut s = Store::open(dir.path()).unwrap();
+        let a = attention();
+        s.ingest("work", &message("1.000001"), true).unwrap();
+        if reserve_delivery {
+            let job = s.claim(&a).unwrap().unwrap();
+            s.finish(&job, speak(), &a).unwrap();
+        }
+        // Permissions and status use their own connections. Hold a write open
+        // while the worker reserves a job or reply against the same database.
+        let mut other = rusqlite::Connection::open(dir.path().join("state.sqlite3")).unwrap();
+        let tx = other
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
+            .unwrap();
+        tx.execute(
+            "INSERT INTO settings VALUES('concurrent-check','active')",
+            [],
+        )
+        .unwrap();
+        let (ready, started) = std::sync::mpsc::channel();
+        let worker = std::thread::spawn(move || {
+            ready.send(()).unwrap();
+            if reserve_delivery {
+                s.claim_outgoing(&config()).map(|out| out.map(|v| v.id))
+            } else {
+                s.claim(&a).map(|job| job.map(|v| v.id))
+            }
+        });
+        started.recv().unwrap();
+        std::thread::sleep(std::time::Duration::from_millis(200));
+        tx.commit().unwrap();
+        let result = worker.join().unwrap();
+        assert!(result.is_ok(), "reservation {reserve_delivery}: {result:?}");
+        assert!(result.unwrap().is_some());
+    }
+}