Retry insight persistence with exponential backoff to avoid SQLite lock contention
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.
This commit is contained in:
1 parent
4029625dc0
commit
38aaddfd2a
4 files changed
+133
-58
No files matched your search
+62
-46
@@ -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
|
||||
|
||||
+15
-10
@@ -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(_) => {
|
||||
|
||||
Reference in new issue
Block a user