From e75e60786d0e61d727e5a62feb25ae8cc54a4819 Mon Sep 17 00:00:00 2001 From: Cameron Cordes Date: Sat, 17 Oct 2026 00:20:00 -0400 Subject: [PATCH] fix(clip): stop re-sending undecodable files every pass MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../down.sql | 1 + .../up.sql | 17 ++++ src/ai/clip_client.rs | 28 ++++-- src/bin/probe_clip_search.rs | 2 +- src/clip_search.rs | 4 +- src/clip_watch.rs | 42 ++++++--- src/database/mod.rs | 88 +++++++++++++++++++ src/database/schema.rs | 12 +++ src/files.rs | 10 +++ 9 files changed, 179 insertions(+), 25 deletions(-) create mode 100644 migrations/2026-10-18-000000_add_clip_failures/down.sql create mode 100644 migrations/2026-10-18-000000_add_clip_failures/up.sql diff --git a/migrations/2026-10-18-000000_add_clip_failures/down.sql b/migrations/2026-10-18-000000_add_clip_failures/down.sql new file mode 100644 index 0000000..09c84b1 --- /dev/null +++ b/migrations/2026-10-18-000000_add_clip_failures/down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS clip_failures; diff --git a/migrations/2026-10-18-000000_add_clip_failures/up.sql b/migrations/2026-10-18-000000_add_clip_failures/up.sql new file mode 100644 index 0000000..074e3b4 --- /dev/null +++ b/migrations/2026-10-18-000000_add_clip_failures/up.sql @@ -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) +); diff --git a/src/ai/clip_client.rs b/src/ai/clip_client.rs index 3519e8b..46e2615 100644 --- a/src/ai/clip_client.rs +++ b/src/ai/clip_client.rs @@ -27,8 +27,10 @@ //! L2-normalized so the rerank reduces to a plain dot product. //! //! Error mapping (reflected in [`ClipError`]): -//! - 422 `decode_failed` / `empty_text` → permanent: ImageApi marks the -//! row failed or surfaces the empty-query error to the search caller. +//! - 422 `decode_failed` / `empty_text` → rejected input: ImageApi marks +//! 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. //! - Any other 5xx / network error → defer. @@ -91,8 +93,12 @@ pub struct ClipHealth { #[derive(Debug)] pub enum ClipError { - /// Apollo refused for a reason that won't change on retry (decode - /// failure on /encode_image, empty text on /encode_text). + /// Apollo refused the input itself (422: decode failure on + /// /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), /// Apollo couldn't process this turn but might next time (CUDA OOM, /// engine not loaded, network hiccup). @@ -105,6 +111,7 @@ pub enum ClipError { impl std::fmt::Display for ClipError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { + ClipError::Rejected(e) => write!(f, "rejected: {e}"), ClipError::Permanent(e) => write!(f, "permanent: {e}"), ClipError::Transient(e) => write!(f, "transient: {e}"), ClipError::Disabled => write!(f, "clip client disabled"), @@ -278,7 +285,7 @@ fn classify_error_response(status: u16, body_text: &str) -> ClipError { .unwrap_or_default(); if status == 422 { - return ClipError::Permanent(anyhow::anyhow!( + return ClipError::Rejected(anyhow::anyhow!( "clip {} {}: {}", status, detail_code, @@ -323,6 +330,9 @@ fn classify_error_response(status: u16, body_text: &str) -> ClipError { mod tests { use super::*; + fn is_rejected(e: &ClipError) -> bool { + matches!(e, ClipError::Rejected(_)) + } fn is_permanent(e: &ClipError) -> bool { matches!(e, ClipError::Permanent(_)) } @@ -331,16 +341,16 @@ mod tests { } #[test] - fn classify_422_decode_failed_is_permanent() { - assert!(is_permanent(&classify_error_response( + fn classify_422_decode_failed_is_rejected() { + assert!(is_rejected(&classify_error_response( 422, r#"{"detail":"decode_failed: bad bytes"}"# ))); } #[test] - fn classify_422_empty_text_is_permanent() { - assert!(is_permanent(&classify_error_response( + fn classify_422_empty_text_is_rejected() { + assert!(is_rejected(&classify_error_response( 422, r#"{"detail":"empty_text"}"# ))); diff --git a/src/bin/probe_clip_search.rs b/src/bin/probe_clip_search.rs index 9eb71fc..1e54e30 100644 --- a/src/bin/probe_clip_search.rs +++ b/src/bin/probe_clip_search.rs @@ -227,7 +227,7 @@ async fn main() -> anyhow::Result<()> { } let _ = call_start; } - Err(ClipError::Permanent(e)) => { + Err(ClipError::Rejected(e) | ClipError::Permanent(e)) => { perm_fail += 1; warn!("permanent encode failure for {rel_path}: {e}"); } diff --git a/src/clip_search.rs b/src/clip_search.rs index fb8b0a5..40bc75c 100644 --- a/src/clip_search.rs +++ b/src/clip_search.rs @@ -172,7 +172,9 @@ pub async fn score_photos( // 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 { 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::Disabled) => return Err(ScoreError::Disabled), }; diff --git a/src/clip_watch.rs b/src/clip_watch.rs index d0b1802..7574c11 100644 --- a/src/clip_watch.rs +++ b/src/clip_watch.rs @@ -6,12 +6,12 @@ //! parallel encode calls to Apollo, and persist the resulting embeddings //! into `image_exif.clip_embedding` / `clip_model_version`. //! -//! Unlike the face pipeline, CLIP has no marker rows — a permanent -//! failure (un-decodable bytes) leaves the row's `clip_embedding` NULL -//! and the drain will retry on the next tick. For personal-library -//! scale this is fine; the per-tick cap bounds the wasted work, and -//! `file_types::is_image_file` filters out videos / non-media client- -//! side so most permanent failures are decoded-but-corrupt files (rare). +//! A permanent failure (un-decodable bytes) records a `clip_failures` +//! marker scoped to the row's `content_hash`, so the drain stops +//! re-sending the file (and re-logging the same 422) every pass until +//! its bytes change. `file_types::is_image_file` filters out videos / +//! non-media client-side, so most permanent failures are corrupt files +//! or non-images with an image extension. //! //! 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 @@ -172,15 +172,29 @@ async fn process_one( cand.rel_path, resp.embedding_dim, resp.duration_ms, resp.model_version ); } - Err(ClipError::Permanent(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. + Err(ClipError::Rejected(e)) => { 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 ); } diff --git a/src/database/mod.rs b/src/database/mod.rs index 4a26450..c5c87c1 100644 --- a/src/database/mod.rs +++ b/src/database/mod.rs @@ -551,6 +551,17 @@ pub trait ExifDao: Sync + Send { model_version: &str, ) -> 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 /// image_exif rows for the given libraries, optionally filtered to a /// 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) // filter; the planner hits it directly. ORDER BY id ASC // 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 .filter(library_id.eq(library_id_val)) .filter(clip_embedding.is_null()) .filter(content_hash.is_not_null()) + .filter(diesel::dsl::not(diesel::dsl::exists(failed))) .select((rel_path, content_hash.assume_not_null())) .order(id.asc()) .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( &mut self, library_ids_val: &[i32], @@ -2559,6 +2616,37 @@ mod exif_dao_tests { 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 { + 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] fn get_all_with_date_taken_union_returns_all_libraries() { let mut dao = setup_two_libraries(); diff --git a/src/database/schema.rs b/src/database/schema.rs index 051fb20..7924c3c 100644 --- a/src/database/schema.rs +++ b/src/database/schema.rs @@ -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! { daily_conversation_summaries (id) { id -> Integer, @@ -348,6 +358,7 @@ diesel::table! { diesel::joinable!(entity_facts -> photo_insights (source_insight_id)); diesel::joinable!(entity_photo_links -> entities (entity_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 -> persons (person_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!( calendar_events, + clip_failures, daily_conversation_summaries, entities, entity_facts, diff --git a/src/files.rs b/src/files.rs index 7d04779..3044576 100644 --- a/src/files.rs +++ b/src/files.rs @@ -1910,6 +1910,16 @@ mod tests { 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( &mut self, _library_ids: &[i32], -- 2.54.0