fix(clip): stop re-sending undecodable files every pass
A 422 from Apollo's /encode_image (bytes it can't decode — corrupt files, non-images with an image extension) left clip_embedding NULL, so the drain re-sent the file and logged the same "permanent failure (will retry next pass)" on every scan. Permanent rejections now record a clip_failures marker scoped to the row's content_hash; list_clip_unencoded_candidates skips the row while the hash still matches, so a replaced file gets another try with no cleanup. Only 422 (new ClipError::Rejected) marks: other 4xx are deployment problems (auth, routing) and would otherwise blacklist the whole library. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
a605ccd6f8
commit
e75e60786d
9 files changed
+179
-25
No files matched your search
@@ -0,0 +1 @@
|
|||||||
|
DROP TABLE IF EXISTS clip_failures;
|
||||||
@@ -0,0 +1,17 @@
|
|||||||
|
-- "Don't retry" markers for the CLIP drain. A permanent encode failure
|
||||||
|
-- (bytes Apollo can't decode — corrupt or non-image files with an image
|
||||||
|
-- extension) used to leave `clip_embedding` NULL, so the drain re-sent
|
||||||
|
-- the file and logged the same 422 every pass.
|
||||||
|
--
|
||||||
|
-- Keyed on the path, scoped to the bytes that failed: the drain skips a
|
||||||
|
-- row only while its `content_hash` still matches, so replacing the file
|
||||||
|
-- makes it eligible again with no cleanup. To retry everything (e.g.
|
||||||
|
-- after an Apollo decoder upgrade): `DELETE FROM clip_failures;`
|
||||||
|
CREATE TABLE clip_failures (
|
||||||
|
library_id INTEGER NOT NULL REFERENCES libraries(id) ON DELETE CASCADE,
|
||||||
|
rel_path TEXT NOT NULL,
|
||||||
|
content_hash TEXT NOT NULL,
|
||||||
|
error TEXT NOT NULL,
|
||||||
|
failed_at BIGINT NOT NULL,
|
||||||
|
PRIMARY KEY (library_id, rel_path)
|
||||||
|
);
|
||||||
+19
-9
@@ -27,8 +27,10 @@
|
|||||||
//! L2-normalized so the rerank reduces to a plain dot product.
|
//! L2-normalized so the rerank reduces to a plain dot product.
|
||||||
//!
|
//!
|
||||||
//! Error mapping (reflected in [`ClipError`]):
|
//! Error mapping (reflected in [`ClipError`]):
|
||||||
//! - 422 `decode_failed` / `empty_text` → permanent: ImageApi marks the
|
//! - 422 `decode_failed` / `empty_text` → rejected input: ImageApi marks
|
||||||
//! row failed or surfaces the empty-query error to the search caller.
|
//! the row failed or surfaces the empty-query error to the search caller.
|
||||||
|
//! - Any other 4xx → permanent, but not the input's fault (auth, routing),
|
||||||
|
//! so nothing is marked.
|
||||||
//! - 503 `cuda_oom` / `engine_unavailable` → defer-and-retry: no marker.
|
//! - 503 `cuda_oom` / `engine_unavailable` → defer-and-retry: no marker.
|
||||||
//! - Any other 5xx / network error → defer.
|
//! - Any other 5xx / network error → defer.
|
||||||
|
|
||||||
@@ -91,8 +93,12 @@ pub struct ClipHealth {
|
|||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub enum ClipError {
|
pub enum ClipError {
|
||||||
/// Apollo refused for a reason that won't change on retry (decode
|
/// Apollo refused the input itself (422: decode failure on
|
||||||
/// failure on /encode_image, empty text on /encode_text).
|
/// /encode_image, empty text on /encode_text). Retrying the same
|
||||||
|
/// bytes will fail the same way.
|
||||||
|
Rejected(anyhow::Error),
|
||||||
|
/// Any other 4xx: won't fix itself on retry, but it's a deployment
|
||||||
|
/// problem (auth, routing), not this input's.
|
||||||
Permanent(anyhow::Error),
|
Permanent(anyhow::Error),
|
||||||
/// Apollo couldn't process this turn but might next time (CUDA OOM,
|
/// Apollo couldn't process this turn but might next time (CUDA OOM,
|
||||||
/// engine not loaded, network hiccup).
|
/// engine not loaded, network hiccup).
|
||||||
@@ -105,6 +111,7 @@ pub enum ClipError {
|
|||||||
impl std::fmt::Display for ClipError {
|
impl std::fmt::Display for ClipError {
|
||||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
match self {
|
match self {
|
||||||
|
ClipError::Rejected(e) => write!(f, "rejected: {e}"),
|
||||||
ClipError::Permanent(e) => write!(f, "permanent: {e}"),
|
ClipError::Permanent(e) => write!(f, "permanent: {e}"),
|
||||||
ClipError::Transient(e) => write!(f, "transient: {e}"),
|
ClipError::Transient(e) => write!(f, "transient: {e}"),
|
||||||
ClipError::Disabled => write!(f, "clip client disabled"),
|
ClipError::Disabled => write!(f, "clip client disabled"),
|
||||||
@@ -278,7 +285,7 @@ fn classify_error_response(status: u16, body_text: &str) -> ClipError {
|
|||||||
.unwrap_or_default();
|
.unwrap_or_default();
|
||||||
|
|
||||||
if status == 422 {
|
if status == 422 {
|
||||||
return ClipError::Permanent(anyhow::anyhow!(
|
return ClipError::Rejected(anyhow::anyhow!(
|
||||||
"clip {} {}: {}",
|
"clip {} {}: {}",
|
||||||
status,
|
status,
|
||||||
detail_code,
|
detail_code,
|
||||||
@@ -323,6 +330,9 @@ fn classify_error_response(status: u16, body_text: &str) -> ClipError {
|
|||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
|
fn is_rejected(e: &ClipError) -> bool {
|
||||||
|
matches!(e, ClipError::Rejected(_))
|
||||||
|
}
|
||||||
fn is_permanent(e: &ClipError) -> bool {
|
fn is_permanent(e: &ClipError) -> bool {
|
||||||
matches!(e, ClipError::Permanent(_))
|
matches!(e, ClipError::Permanent(_))
|
||||||
}
|
}
|
||||||
@@ -331,16 +341,16 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn classify_422_decode_failed_is_permanent() {
|
fn classify_422_decode_failed_is_rejected() {
|
||||||
assert!(is_permanent(&classify_error_response(
|
assert!(is_rejected(&classify_error_response(
|
||||||
422,
|
422,
|
||||||
r#"{"detail":"decode_failed: bad bytes"}"#
|
r#"{"detail":"decode_failed: bad bytes"}"#
|
||||||
)));
|
)));
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn classify_422_empty_text_is_permanent() {
|
fn classify_422_empty_text_is_rejected() {
|
||||||
assert!(is_permanent(&classify_error_response(
|
assert!(is_rejected(&classify_error_response(
|
||||||
422,
|
422,
|
||||||
r#"{"detail":"empty_text"}"#
|
r#"{"detail":"empty_text"}"#
|
||||||
)));
|
)));
|
||||||
|
|||||||
@@ -227,7 +227,7 @@ async fn main() -> anyhow::Result<()> {
|
|||||||
}
|
}
|
||||||
let _ = call_start;
|
let _ = call_start;
|
||||||
}
|
}
|
||||||
Err(ClipError::Permanent(e)) => {
|
Err(ClipError::Rejected(e) | ClipError::Permanent(e)) => {
|
||||||
perm_fail += 1;
|
perm_fail += 1;
|
||||||
warn!("permanent encode failure for {rel_path}: {e}");
|
warn!("permanent encode failure for {rel_path}: {e}");
|
||||||
}
|
}
|
||||||
|
|||||||
+3
-1
@@ -172,7 +172,9 @@ pub async fn score_photos(
|
|||||||
// 1. Encode the query text. Fast — Apollo's text encoder is ~50ms on CPU.
|
// 1. Encode the query text. Fast — Apollo's text encoder is ~50ms on CPU.
|
||||||
let query_resp = match state.clip_client.encode_text(q_text).await {
|
let query_resp = match state.clip_client.encode_text(q_text).await {
|
||||||
Ok(r) => r,
|
Ok(r) => r,
|
||||||
Err(ClipError::Permanent(e)) => return Err(ScoreError::Rejected(e.to_string())),
|
Err(ClipError::Rejected(e) | ClipError::Permanent(e)) => {
|
||||||
|
return Err(ScoreError::Rejected(e.to_string()));
|
||||||
|
}
|
||||||
Err(ClipError::Transient(e)) => return Err(ScoreError::Unavailable(e.to_string())),
|
Err(ClipError::Transient(e)) => return Err(ScoreError::Unavailable(e.to_string())),
|
||||||
Err(ClipError::Disabled) => return Err(ScoreError::Disabled),
|
Err(ClipError::Disabled) => return Err(ScoreError::Disabled),
|
||||||
};
|
};
|
||||||
|
|||||||
+28
-14
@@ -6,12 +6,12 @@
|
|||||||
//! parallel encode calls to Apollo, and persist the resulting embeddings
|
//! parallel encode calls to Apollo, and persist the resulting embeddings
|
||||||
//! into `image_exif.clip_embedding` / `clip_model_version`.
|
//! into `image_exif.clip_embedding` / `clip_model_version`.
|
||||||
//!
|
//!
|
||||||
//! Unlike the face pipeline, CLIP has no marker rows — a permanent
|
//! A permanent failure (un-decodable bytes) records a `clip_failures`
|
||||||
//! failure (un-decodable bytes) leaves the row's `clip_embedding` NULL
|
//! marker scoped to the row's `content_hash`, so the drain stops
|
||||||
//! and the drain will retry on the next tick. For personal-library
|
//! re-sending the file (and re-logging the same 422) every pass until
|
||||||
//! scale this is fine; the per-tick cap bounds the wasted work, and
|
//! its bytes change. `file_types::is_image_file` filters out videos /
|
||||||
//! `file_types::is_image_file` filters out videos / non-media client-
|
//! non-media client-side, so most permanent failures are corrupt files
|
||||||
//! side so most permanent failures are decoded-but-corrupt files (rare).
|
//! or non-images with an image extension.
|
||||||
//!
|
//!
|
||||||
//! The watcher thread isn't in any pre-existing async context, so we
|
//! The watcher thread isn't in any pre-existing async context, so we
|
||||||
//! build a short-lived tokio runtime per pass and `block_on` the join
|
//! build a short-lived tokio runtime per pass and `block_on` the join
|
||||||
@@ -172,15 +172,29 @@ async fn process_one(
|
|||||||
cand.rel_path, resp.embedding_dim, resp.duration_ms, resp.model_version
|
cand.rel_path, resp.embedding_dim, resp.duration_ms, resp.model_version
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
Err(ClipError::Permanent(e)) => {
|
Err(ClipError::Rejected(e)) => {
|
||||||
// No marker — the row sits with NULL embedding and the drain
|
|
||||||
// retries next pass. For personal-library scale the cost of
|
|
||||||
// re-attempting permanently-broken files is bounded by the
|
|
||||||
// per-tick cap. If this becomes a recurring noise source,
|
|
||||||
// add a `clip_status` column with `failed` semantics like
|
|
||||||
// face_detections has.
|
|
||||||
warn!(
|
warn!(
|
||||||
"clip_watch: permanent failure on {} (will retry next pass): {}",
|
"clip_watch: permanent failure on {}: {} — marking failed",
|
||||||
|
cand.rel_path, e
|
||||||
|
);
|
||||||
|
let mut dao = exif_dao.lock().expect("exif dao");
|
||||||
|
if let Err(me) = dao.mark_clip_failed(
|
||||||
|
library_id,
|
||||||
|
&cand.rel_path,
|
||||||
|
&cand.content_hash,
|
||||||
|
&e.to_string(),
|
||||||
|
) {
|
||||||
|
warn!(
|
||||||
|
"clip_watch: mark failed errored for {}: {:?}",
|
||||||
|
cand.rel_path, me
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(ClipError::Permanent(e)) => {
|
||||||
|
// Not this file's fault (auth / routing) — don't mark it, or
|
||||||
|
// a misconfigured Apollo would blacklist the whole library.
|
||||||
|
warn!(
|
||||||
|
"clip_watch: Apollo refused {} (will retry next pass): {}",
|
||||||
cand.rel_path, e
|
cand.rel_path, e
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -551,6 +551,17 @@ pub trait ExifDao: Sync + Send {
|
|||||||
model_version: &str,
|
model_version: &str,
|
||||||
) -> Result<(), DbError>;
|
) -> Result<(), DbError>;
|
||||||
|
|
||||||
|
/// Record that Apollo permanently rejected these bytes, so
|
||||||
|
/// `list_clip_unencoded_candidates` stops offering the row until its
|
||||||
|
/// `content_hash` changes. Upserts — one marker per path.
|
||||||
|
fn mark_clip_failed(
|
||||||
|
&mut self,
|
||||||
|
library_id: i32,
|
||||||
|
rel_path: &str,
|
||||||
|
content_hash: &str,
|
||||||
|
error: &str,
|
||||||
|
) -> Result<(), DbError>;
|
||||||
|
|
||||||
/// Load every `(content_hash, clip_embedding)` pair from the live
|
/// Load every `(content_hash, clip_embedding)` pair from the live
|
||||||
/// image_exif rows for the given libraries, optionally filtered to a
|
/// image_exif rows for the given libraries, optionally filtered to a
|
||||||
/// single `model_version` (cosine sim across mixed geometries is
|
/// single `model_version` (cosine sim across mixed geometries is
|
||||||
@@ -1519,10 +1530,21 @@ impl ExifDao for SqliteExifDao {
|
|||||||
// (clip_embedding IS NULL AND content_hash IS NOT NULL)
|
// (clip_embedding IS NULL AND content_hash IS NOT NULL)
|
||||||
// filter; the planner hits it directly. ORDER BY id ASC
|
// filter; the planner hits it directly. ORDER BY id ASC
|
||||||
// keeps drain progress monotone across ticks.
|
// keeps drain progress monotone across ticks.
|
||||||
|
// Skip rows whose current bytes already failed permanently
|
||||||
|
// (`mark_clip_failed`); a changed hash makes them eligible.
|
||||||
|
let failed = schema::clip_failures::table
|
||||||
|
.filter(schema::clip_failures::library_id.eq(library_id))
|
||||||
|
.filter(schema::clip_failures::rel_path.eq(rel_path))
|
||||||
|
.filter(
|
||||||
|
schema::clip_failures::content_hash
|
||||||
|
.nullable()
|
||||||
|
.eq(content_hash),
|
||||||
|
);
|
||||||
image_exif
|
image_exif
|
||||||
.filter(library_id.eq(library_id_val))
|
.filter(library_id.eq(library_id_val))
|
||||||
.filter(clip_embedding.is_null())
|
.filter(clip_embedding.is_null())
|
||||||
.filter(content_hash.is_not_null())
|
.filter(content_hash.is_not_null())
|
||||||
|
.filter(diesel::dsl::not(diesel::dsl::exists(failed)))
|
||||||
.select((rel_path, content_hash.assume_not_null()))
|
.select((rel_path, content_hash.assume_not_null()))
|
||||||
.order(id.asc())
|
.order(id.asc())
|
||||||
.limit(limit)
|
.limit(limit)
|
||||||
@@ -1585,6 +1607,41 @@ impl ExifDao for SqliteExifDao {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn mark_clip_failed(
|
||||||
|
&mut self,
|
||||||
|
library_id_val: i32,
|
||||||
|
rel_path_val: &str,
|
||||||
|
content_hash_val: &str,
|
||||||
|
error_val: &str,
|
||||||
|
) -> Result<(), DbError> {
|
||||||
|
trace_db_call("insert", "mark_clip_failed", |_span| {
|
||||||
|
use schema::clip_failures::dsl::*;
|
||||||
|
|
||||||
|
let mut connection = self.connection.lock().expect("Unable to get ExifDao");
|
||||||
|
|
||||||
|
let now = chrono::Utc::now().timestamp();
|
||||||
|
diesel::insert_into(clip_failures)
|
||||||
|
.values((
|
||||||
|
library_id.eq(library_id_val),
|
||||||
|
rel_path.eq(rel_path_val),
|
||||||
|
content_hash.eq(content_hash_val),
|
||||||
|
error.eq(error_val),
|
||||||
|
failed_at.eq(now),
|
||||||
|
))
|
||||||
|
.on_conflict((library_id, rel_path))
|
||||||
|
.do_update()
|
||||||
|
.set((
|
||||||
|
content_hash.eq(content_hash_val),
|
||||||
|
error.eq(error_val),
|
||||||
|
failed_at.eq(now),
|
||||||
|
))
|
||||||
|
.execute(connection.deref_mut())
|
||||||
|
.map(|_| ())
|
||||||
|
.map_err(|e| anyhow::anyhow!("Insert error: {}", e))
|
||||||
|
})
|
||||||
|
.map_err(|e| DbError::log(DbErrorKind::InsertError, e))
|
||||||
|
}
|
||||||
|
|
||||||
fn list_clip_index(
|
fn list_clip_index(
|
||||||
&mut self,
|
&mut self,
|
||||||
library_ids_val: &[i32],
|
library_ids_val: &[i32],
|
||||||
@@ -2559,6 +2616,37 @@ mod exif_dao_tests {
|
|||||||
SqliteExifDao::from_connection(conn)
|
SqliteExifDao::from_connection(conn)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn clip_failure_marker_skips_row_until_its_bytes_change() {
|
||||||
|
let mut dao = setup_two_libraries();
|
||||||
|
insert_row(&mut dao, 1, "bad.jpg", None);
|
||||||
|
insert_row(&mut dao, 1, "good.jpg", None);
|
||||||
|
insert_row(&mut dao, 2, "bad.jpg", None);
|
||||||
|
for (lib, rel) in [(1, "bad.jpg"), (1, "good.jpg"), (2, "bad.jpg")] {
|
||||||
|
dao.backfill_content_hash(lib, rel, "h1", 1).unwrap();
|
||||||
|
}
|
||||||
|
let rels = |dao: &mut SqliteExifDao, lib| -> Vec<String> {
|
||||||
|
dao.list_clip_unencoded_candidates(lib, 10)
|
||||||
|
.unwrap()
|
||||||
|
.into_iter()
|
||||||
|
.map(|(rel, _)| rel)
|
||||||
|
.collect()
|
||||||
|
};
|
||||||
|
|
||||||
|
dao.mark_clip_failed(1, "bad.jpg", "h1", "decode_failed")
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(rels(&mut dao, 1), vec!["good.jpg"]);
|
||||||
|
// Same path in another library is a different file.
|
||||||
|
assert_eq!(rels(&mut dao, 2), vec!["bad.jpg"]);
|
||||||
|
// Re-marking upserts rather than erroring on the primary key.
|
||||||
|
dao.mark_clip_failed(1, "bad.jpg", "h1", "decode_failed")
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
// The file was replaced: new bytes get another try.
|
||||||
|
dao.backfill_content_hash(1, "bad.jpg", "h2", 1).unwrap();
|
||||||
|
assert_eq!(rels(&mut dao, 1), vec!["bad.jpg", "good.jpg"]);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn get_all_with_date_taken_union_returns_all_libraries() {
|
fn get_all_with_date_taken_union_returns_all_libraries() {
|
||||||
let mut dao = setup_two_libraries();
|
let mut dao = setup_two_libraries();
|
||||||
|
|||||||
@@ -18,6 +18,16 @@ diesel::table! {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
diesel::table! {
|
||||||
|
clip_failures (library_id, rel_path) {
|
||||||
|
library_id -> Integer,
|
||||||
|
rel_path -> Text,
|
||||||
|
content_hash -> Text,
|
||||||
|
error -> Text,
|
||||||
|
failed_at -> BigInt,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
diesel::table! {
|
diesel::table! {
|
||||||
daily_conversation_summaries (id) {
|
daily_conversation_summaries (id) {
|
||||||
id -> Integer,
|
id -> Integer,
|
||||||
@@ -348,6 +358,7 @@ diesel::table! {
|
|||||||
diesel::joinable!(entity_facts -> photo_insights (source_insight_id));
|
diesel::joinable!(entity_facts -> photo_insights (source_insight_id));
|
||||||
diesel::joinable!(entity_photo_links -> entities (entity_id));
|
diesel::joinable!(entity_photo_links -> entities (entity_id));
|
||||||
diesel::joinable!(entity_photo_links -> libraries (library_id));
|
diesel::joinable!(entity_photo_links -> libraries (library_id));
|
||||||
|
diesel::joinable!(clip_failures -> libraries (library_id));
|
||||||
diesel::joinable!(face_detections -> libraries (library_id));
|
diesel::joinable!(face_detections -> libraries (library_id));
|
||||||
diesel::joinable!(face_detections -> persons (person_id));
|
diesel::joinable!(face_detections -> persons (person_id));
|
||||||
diesel::joinable!(image_exif -> libraries (library_id));
|
diesel::joinable!(image_exif -> libraries (library_id));
|
||||||
@@ -360,6 +371,7 @@ diesel::joinable!(video_preview_clips -> libraries (library_id));
|
|||||||
|
|
||||||
diesel::allow_tables_to_appear_in_same_query!(
|
diesel::allow_tables_to_appear_in_same_query!(
|
||||||
calendar_events,
|
calendar_events,
|
||||||
|
clip_failures,
|
||||||
daily_conversation_summaries,
|
daily_conversation_summaries,
|
||||||
entities,
|
entities,
|
||||||
entity_facts,
|
entity_facts,
|
||||||
|
|||||||
@@ -1910,6 +1910,16 @@ mod tests {
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn mark_clip_failed(
|
||||||
|
&mut self,
|
||||||
|
_library_id: i32,
|
||||||
|
_rel_path: &str,
|
||||||
|
_content_hash: &str,
|
||||||
|
_error: &str,
|
||||||
|
) -> Result<(), DbError> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
fn list_clip_index(
|
fn list_clip_index(
|
||||||
&mut self,
|
&mut self,
|
||||||
_library_ids: &[i32],
|
_library_ids: &[i32],
|
||||||
|
|||||||
Reference in new issue
Block a user