Apps/ImageApi
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge pull request 'Retry insight persistence with exponential backoff to avoid SQLite lock contention' (#108) from feature/db-locked-fix into master

Browse Source 
Reviewed-on: #108
This commit was merged in pull request #108.
This commit is contained in:
cameron
2026-08-07 16:19:41 +00:00
parent 4029625dc0 38aaddfd2a
commit 3732faef71
4 changed files with 133 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ai/insight_chat.rs
+62 -46
View File
@@ -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
src/ai/insight_generator.rs
+15 -10
View File
@@ -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(_) => {
src/database/models.rs
+1 -1
View File
@@ -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
+55 -1
View File
@@ -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