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 dao = self.insight_dao.clone();
-            let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao");
+            let stored = retry_with_backoff("store_insight", 3, || {
-            let stored = dao
+                let cx = opentelemetry::Context::new();
-                .store_insight(&cx, new_row)
+                let mut d = dao.lock().expect("Unable to lock InsightDao");
-                .map_err(|e| anyhow!("failed to store amended insight: {:?}", e))?;
+                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 dao = self.insight_dao.clone();
-            let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao");
+            let rows = retry_with_backoff("update_training_messages", 3, || {
-            let rows = dao
+                let cx = opentelemetry::Context::new();
-                .update_training_messages(&cx, req.library_id, &normalized, &json)
+                let mut d = dao.lock().expect("Unable to lock InsightDao");
-                .map_err(|e| anyhow!("failed to persist chat history: {:?}", e))?;
+                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 dao = self.insight_dao.clone();
-        let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao");
+        let rows = retry_with_backoff("update_training_messages", 3, || {
-        let rows = dao
+            let cx = opentelemetry::Context::new();
-            .update_training_messages(&cx, library_id, &normalized, &json)
+            let mut d = dao.lock().expect("Unable to lock InsightDao");
-            .map_err(|e| anyhow!("failed to persist truncated history: {:?}", e))?;
+            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 dao = self.insight_dao.clone();
-            let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao");
+            let stored = retry_with_backoff("store_insight", 3, || {
-            let stored = dao
+                let cx = opentelemetry::Context::new();
-                .store_insight(&cx, new_row)
+                let mut d = dao.lock().expect("Unable to lock InsightDao");
-                .map_err(|e| anyhow!("failed to store amended insight: {:?}", e))?;
+                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 dao = self.insight_dao.clone();
-            let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao");
+            let rows = retry_with_backoff("update_training_messages", 3, || {
-            let rows = dao
+                let cx = opentelemetry::Context::new();
-                .update_training_messages(&cx, req.library_id, &normalized, &json)
+                let mut d = dao.lock().expect("Unable to lock InsightDao");
-                .map_err(|e| anyhow!("failed to persist chat history: {:?}", e))?;
+                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");
+            let mut d = dao.lock().expect("Unable to lock InsightDao");
-            dao.store_insight(&cx, new_row)
+            d.store_insight(&cx, new_row.clone())
-                .map_err(|e| anyhow!("failed to store bootstrap insight: {:?}", e))?
+        })
-        };
+        .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 dao = self.insight_dao.clone();
-            let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao");
+            let stored = retry_with_backoff("store_insight", 3, || {
-            let stored = dao
+                let cx = opentelemetry::Context::new();
-                .store_insight(&cx, new_row)
+                let mut d = dao.lock().expect("Unable to lock InsightDao");
-                .map_err(|e| anyhow!("failed to store amended insight: {:?}", e))?;
+                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 dao = self.insight_dao.clone();
-            let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao");
+            let rows = retry_with_backoff("update_training_messages", 3, || {
-            let rows = dao
+                let cx = opentelemetry::Context::new();
-                .update_training_messages(&cx, req.library_id, &normalized, &json)
+                let mut d = dao.lock().expect("Unable to lock InsightDao");
-                .map_err(|e| anyhow!("failed to persist chat history: {:?}", e))?;
+                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");
+            let mut d = dao.lock().expect("Unable to lock InsightDao");
-            dao.store_insight(&cx, new_row)
+            d.store_insight(&cx, new_row.clone())
-                .map_err(|e| anyhow!("failed to store bootstrap insight: {:?}", e))?
+        })
-        };
+        .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

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 dao = self.insight_dao.clone();
-        let result = dao
+        let result = retry_with_backoff("store_insight", 3, || {
-            .store_insight(&insight_cx, insight)
+            let cx = opentelemetry::Context::new();
-            .map_err(|e| anyhow::anyhow!("Failed to store insight: {:?}", e));
+            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 dao = self.insight_dao.clone();
-            let mut dao = self.insight_dao.lock().expect("Unable to lock InsightDao");
+        let stored = retry_with_backoff("store_insight", 3, || {
-            dao.store_insight(&insight_cx, insight)
+            let cx = opentelemetry::Context::new();
-                .map_err(|e| anyhow::anyhow!("Failed to store agentic insight: {:?}", e))
+            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(_) => {

src/database/models.rs

@@ -192,7 +192,7 @@ pub struct ImageExif {
     pub clip_model_version: Option<String>,
 }
 
-#[derive(Insertable)]
+#[derive(Insertable, Clone)]
 #[diesel(table_name = photo_insights)]
 pub struct InsertPhotoInsight {
     pub library_id: i32,

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<F, T, E>(label: &str, max_retries: u32, mut op: F) -> Result<T, E>
+where
+    F: FnMut() -> Result<T, E>,
+    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