test: harden claim race test

crates/adapters/sqlite/src/event_queue/tests.rs

@@ -10,11 +10,15 @@ use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
 use super::{DbEventQueueConfig, claim_batch};
 
 async fn pool_on(path: &std::path::Path) -> SqlitePool {
+    pool_with_timeout(path, 5).await
+}
+
+async fn pool_with_timeout(path: &std::path::Path, busy_secs: u64) -> SqlitePool {
     let opts = SqliteConnectOptions::from_str(&format!("sqlite://{}", path.display()))
         .unwrap()
         .create_if_missing(true)
         .journal_mode(SqliteJournalMode::Wal)
-        .busy_timeout(std::time::Duration::from_secs(5));
+        .busy_timeout(std::time::Duration::from_secs(busy_secs));
     SqlitePoolOptions::new()
         .max_connections(4)
         .connect_with(opts)
@@ -46,7 +50,7 @@ async fn claiming_while_another_connection_writes_does_not_fail() {
     let path = dir.path().join("q.db");
     let claimer = pool_on(&path).await;
     crate::migrate(&claimer).await.unwrap();
-    let writer = pool_on(&path).await;
+    let writer = pool_with_timeout(&path, 30).await;
     insert(&claimer, 50).await;
 
     let writer_task = {
@@ -65,10 +69,14 @@ async fn claiming_while_another_connection_writes_does_not_fail() {
     let cfg = config(5);
     let mut claimed = 0;
     while !writer_task.is_finished() {
-        claimed += claim_batch(&claimer, &cfg)
+        let batch = claim_batch(&claimer, &cfg)
             .await
-            .expect("claim must not fail while another connection writes")
-            .len();
+            .expect("claim must not fail while another connection writes");
+        if batch.is_empty() {
+            // an empty UPDATE still takes the write lock; don't starve the writer
+            tokio::task::yield_now().await;
+        }
+        claimed += batch.len();
     }
     writer_task.await.unwrap();
     assert!(claimed > 0);