From 38aaddfd2ac5219ff6b03938f3ffb2d9c5e571a5 Mon Sep 17 00:00:00 2001 From: Cameron Cordes Date: Wed, 5 Aug 2026 18:22:36 -0400 Subject: [PATCH] Retry insight persistence with exponential backoff to avoid SQLite lock contention MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add retry_with_backoff utility (100ms base, ±25% jitter, 3 attempts) and replace all 11 direct DAO calls in insight_chat.rs and insight_generator.rs. The Mutex guard drops between retries so the maintenance connection can release its SQLite write lock. --- src/ai/insight_chat.rs | 108 +++++++++++++++++++++--------------- src/ai/insight_generator.rs | 25 +++++---- src/database/models.rs | 2 +- src/utils.rs | 56 ++++++++++++++++++- 4 files changed, 133 insertions(+), 58 deletions(-) diff --git a/src/ai/insight_chat.rs b/src/ai/insight_chat.rs index af00731..1a67540 100644 --- a/src/ai/insight_chat.rs +++ b/src/ai/insight_chat.rs @@ -14,7 +14,7 @@ use crate::ai::turn_registry::TurnRegistry; use crate::database::InsightDao; use crate::database::models::InsertPhotoInsight; use crate::otel::global_tracer; -use crate::utils::normalize_path; +use crate::utils::{normalize_path, retry_with_backoff}; use futures::stream::{BoxStream, StreamExt}; use uuid::Uuid; @@ -558,18 +558,22 @@ impl InsightChatService { prompt_eval_count: None, eval_count: None, }; - let cx = opentelemetry::Context::new(); - let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); - let stored = dao - .store_insight(&cx, new_row) - .map_err(|e| anyhow!("failed to store amended insight: {:?}", e))?; + let dao = self.insight_dao.clone(); + let stored = retry_with_backoff("store_insight", 3, || { + let cx = opentelemetry::Context::new(); + let mut d = dao.lock().expect("Unable to lock InsightDao"); + d.store_insight(&cx, new_row.clone()) + }) + .map_err(|e| anyhow!("failed to store amended insight: {:?}", e))?; amended_insight_id = Some(stored.id); } else { - let cx = opentelemetry::Context::new(); - let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); - let rows = dao - .update_training_messages(&cx, req.library_id, &normalized, &json) - .map_err(|e| anyhow!("failed to persist chat history: {:?}", e))?; + let dao = self.insight_dao.clone(); + let rows = retry_with_backoff("update_training_messages", 3, || { + let cx = opentelemetry::Context::new(); + let mut d = dao.lock().expect("Unable to lock InsightDao"); + d.update_training_messages(&cx, req.library_id, &normalized, &json) + }) + .map_err(|e| anyhow!("failed to persist chat history: {:?}", e))?; if rows == 0 { log::warn!( "update_training_messages updated 0 rows for {} (lib {}), \ @@ -642,11 +646,13 @@ impl InsightChatService { let json = serde_json::to_string(truncated) .map_err(|e| anyhow!("failed to serialize truncated history: {}", e))?; - let cx = opentelemetry::Context::new(); - let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); - let rows = dao - .update_training_messages(&cx, library_id, &normalized, &json) - .map_err(|e| anyhow!("failed to persist truncated history: {:?}", e))?; + let dao = self.insight_dao.clone(); + let rows = retry_with_backoff("update_training_messages", 3, || { + let cx = opentelemetry::Context::new(); + let mut d = dao.lock().expect("Unable to lock InsightDao"); + d.update_training_messages(&cx, library_id, &normalized, &json) + }) + .map_err(|e| anyhow!("failed to persist truncated history: {:?}", e))?; if rows == 0 { log::warn!( "update_training_messages (rewind) updated 0 rows for {} (lib {}), \ @@ -955,18 +961,22 @@ impl InsightChatService { prompt_eval_count: None, eval_count: None, }; - let cx = opentelemetry::Context::new(); - let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); - let stored = dao - .store_insight(&cx, new_row) - .map_err(|e| anyhow!("failed to store amended insight: {:?}", e))?; + let dao = self.insight_dao.clone(); + let stored = retry_with_backoff("store_insight", 3, || { + let cx = opentelemetry::Context::new(); + let mut d = dao.lock().expect("Unable to lock InsightDao"); + d.store_insight(&cx, new_row.clone()) + }) + .map_err(|e| anyhow!("failed to store amended insight: {:?}", e))?; amended_insight_id = Some(stored.id); } else { - let cx = opentelemetry::Context::new(); - let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); - let rows = dao - .update_training_messages(&cx, req.library_id, &normalized, &json) - .map_err(|e| anyhow!("failed to persist chat history: {:?}", e))?; + let dao = self.insight_dao.clone(); + let rows = retry_with_backoff("update_training_messages", 3, || { + let cx = opentelemetry::Context::new(); + let mut d = dao.lock().expect("Unable to lock InsightDao"); + d.update_training_messages(&cx, req.library_id, &normalized, &json) + }) + .map_err(|e| anyhow!("failed to persist chat history: {:?}", e))?; if rows == 0 { log::warn!( "update_training_messages (stream) updated 0 rows for {} (lib {}), \ @@ -1140,12 +1150,13 @@ impl InsightChatService { prompt_eval_count: None, eval_count: None, }; - let stored = { + let dao = self.insight_dao.clone(); + let stored = retry_with_backoff("store_insight", 3, || { let cx = opentelemetry::Context::new(); - let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); - dao.store_insight(&cx, new_row) - .map_err(|e| anyhow!("failed to store bootstrap insight: {:?}", e))? - }; + let mut d = dao.lock().expect("Unable to lock InsightDao"); + d.store_insight(&cx, new_row.clone()) + }) + .map_err(|e| anyhow!("failed to store bootstrap insight: {:?}", e))?; let _ = entry .push_event(ChatStreamEvent::Done { @@ -1543,18 +1554,22 @@ impl InsightChatService { prompt_eval_count: None, eval_count: None, }; - let cx = opentelemetry::Context::new(); - let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); - let stored = dao - .store_insight(&cx, new_row) - .map_err(|e| anyhow!("failed to store amended insight: {:?}", e))?; + let dao = self.insight_dao.clone(); + let stored = retry_with_backoff("store_insight", 3, || { + let cx = opentelemetry::Context::new(); + let mut d = dao.lock().expect("Unable to lock InsightDao"); + d.store_insight(&cx, new_row.clone()) + }) + .map_err(|e| anyhow!("failed to store amended insight: {:?}", e))?; amended_insight_id = Some(stored.id); } else { - let cx = opentelemetry::Context::new(); - let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); - let rows = dao - .update_training_messages(&cx, req.library_id, &normalized, &json) - .map_err(|e| anyhow!("failed to persist chat history: {:?}", e))?; + let dao = self.insight_dao.clone(); + let rows = retry_with_backoff("update_training_messages", 3, || { + let cx = opentelemetry::Context::new(); + let mut d = dao.lock().expect("Unable to lock InsightDao"); + d.update_training_messages(&cx, req.library_id, &normalized, &json) + }) + .map_err(|e| anyhow!("failed to persist chat history: {:?}", e))?; if rows == 0 { log::warn!( "update_training_messages (stream) updated 0 rows for {} (lib {}), \ @@ -1749,12 +1764,13 @@ impl InsightChatService { prompt_eval_count: None, eval_count: None, }; - let stored = { + let dao = self.insight_dao.clone(); + let stored = retry_with_backoff("store_insight", 3, || { let cx = opentelemetry::Context::new(); - let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); - dao.store_insight(&cx, new_row) - .map_err(|e| anyhow!("failed to store bootstrap insight: {:?}", e))? - }; + let mut d = dao.lock().expect("Unable to lock InsightDao"); + d.store_insight(&cx, new_row.clone()) + }) + .map_err(|e| anyhow!("failed to store bootstrap insight: {:?}", e))?; // amended_insight_id semantics broaden on bootstrap/regenerate: // populated whenever this turn produced a new insight row, so diff --git a/src/ai/insight_generator.rs b/src/ai/insight_generator.rs index d45fa55..e396ec9 100644 --- a/src/ai/insight_generator.rs +++ b/src/ai/insight_generator.rs @@ -26,7 +26,7 @@ use crate::libraries::Library; use crate::memories::extract_date_from_filename; use crate::otel::global_tracer; use crate::tags::TagDao; -use crate::utils::{earliest_fs_time, normalize_path}; +use crate::utils::{earliest_fs_time, normalize_path, retry_with_backoff}; /// Max location records rendered by `tool_get_location_history`. The DAO /// query is range-bounded, not limited, so the tool caps the rendered list @@ -1539,10 +1539,13 @@ impl InsightGenerator { eval_count: None, }; - let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); - let result = dao - .store_insight(&insight_cx, insight) - .map_err(|e| anyhow::anyhow!("Failed to store insight: {:?}", e)); + let dao = self.insight_dao.clone(); + let result = retry_with_backoff("store_insight", 3, || { + let cx = opentelemetry::Context::new(); + let mut d = dao.lock().expect("Unable to lock InsightDao"); + d.store_insight(&cx, insight.clone()) + }) + .map_err(|e| anyhow::anyhow!("Failed to store insight: {:?}", e)); match &result { Ok(_) => { @@ -4465,11 +4468,13 @@ Return ONLY the summary, nothing else."#, eval_count: last_eval_count, }; - let stored = { - let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao"); - dao.store_insight(&insight_cx, insight) - .map_err(|e| anyhow::anyhow!("Failed to store agentic insight: {:?}", e)) - }; + let dao = self.insight_dao.clone(); + let stored = retry_with_backoff("store_insight", 3, || { + let cx = opentelemetry::Context::new(); + let mut d = dao.lock().expect("Unable to lock InsightDao"); + d.store_insight(&cx, insight.clone()) + }) + .map_err(|e| anyhow::anyhow!("Failed to store agentic insight: {:?}", e)); match &stored { Ok(_) => { diff --git a/src/database/models.rs b/src/database/models.rs index d3d5440..300def1 100644 --- a/src/database/models.rs +++ b/src/database/models.rs @@ -192,7 +192,7 @@ pub struct ImageExif { pub clip_model_version: Option, } -#[derive(Insertable)] +#[derive(Insertable, Clone)] #[diesel(table_name = photo_insights)] pub struct InsertPhotoInsight { pub library_id: i32, diff --git a/src/utils.rs b/src/utils.rs index fdfef9b..d76ad56 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -1,4 +1,58 @@ -use std::time::SystemTime; +use rand::Rng; +use std::time::{Duration, SystemTime}; + +/// Retry a fallible operation with exponential backoff + jitter. +/// +/// Runs the closure immediately, then retries up to `max_retries` times on +/// failure. Each retry sleeps for `base_delay * 2^attempt` plus a uniform +/// jitter of ±25%. Non-final errors are logged at `debug` with the label; +/// the final error is logged at `error`. +/// +/// # Examples +/// ``` +/// use image_api::utils::retry_with_backoff; +/// +/// let result = retry_with_backoff("my-op", 3, || { +/// // something that may fail transiently +/// Ok::<_, anyhow::Error>(42) +/// }); +/// ``` +pub fn retry_with_backoff(label: &str, max_retries: u32, mut op: F) -> Result +where + F: FnMut() -> Result, + E: std::fmt::Debug, +{ + let mut last_err = match op() { + Ok(v) => return Ok(v), + Err(e) => e, + }; + for attempt in 1..=max_retries { + let base = Duration::from_millis(100).saturating_mul(1_u32.pow(attempt - 1)); + let jitter_range = base.as_millis() as f64 * 0.25; + let jitter = rand::thread_rng().gen_range(-jitter_range..=jitter_range) as u128; + let delay = base.as_millis() as i128 + jitter as i128; + std::thread::sleep(Duration::from_millis(delay.max(0) as u64)); + log::debug!( + "{}: attempt {}/{} failed ({:?}), retrying in {}ms", + label, + attempt, + max_retries, + last_err, + delay.max(0) + ); + match op() { + Ok(v) => return Ok(v), + Err(e) => last_err = e, + } + } + log::error!( + "{}: all {} retries exhausted: {:?}", + label, + max_retries, + last_err + ); + Err(last_err) +} /// Normalize a file path to use forward slashes for cross-platform consistency /// This ensures paths stored in the database always use `/` regardless of OS