feat(dates): backfill drain converts mixed-convention rows to true instants (#118)
Per-tick drain (backfill_utc_dates) at the existing date-drain seam, inside the per-library online gate. Selects date_utc IS NULL AND date_taken IS NOT NULL (new partial index idx_image_exif_utc_backfill), re-runs resolve_dates_batch with empty priors (one exiftool subprocess per batch, same call shape as the #117 audit), and writes only the new-contract triple; the legacy pair stays frozen. Capped by DATE_BACKFILL_MAX_PER_TICK; missing-file rows are paged past without spending the cap. Legacy manual overrides convert via manual_date_resolution rather than being re-resolved from the file. Day-boundary invalidation deletes the day-span precomputed reels covering a re-dated row's old/new display-zone month-day so the nightly pre-gen regenerates them. Progress: imageserver_date_backfill_remaining gauge + date_backfill_remaining on GET /libraries. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
ab16bc7da6
commit
98db2a48db
11 files changed
+1185
-4
No files matched your search
@@ -43,6 +43,17 @@ pub struct DuplicateRow {
|
||||
pub duplicate_decided_at: Option<i64>,
|
||||
}
|
||||
|
||||
/// One row the true-instant backfill drain (ADR-0001, ticket #118) still
|
||||
/// has to convert: on the legacy convention (`date_utc IS NULL`) with a
|
||||
/// legacy `date_taken` to compare the converted day against.
|
||||
#[derive(Debug, Clone, Queryable)]
|
||||
pub struct UtcBackfillCandidate {
|
||||
pub id: i32,
|
||||
pub rel_path: String,
|
||||
pub date_taken: i64,
|
||||
pub date_taken_source: Option<String>,
|
||||
}
|
||||
|
||||
pub mod calendar_dao;
|
||||
pub mod daily_summary_dao;
|
||||
pub mod insight_generation_job_dao;
|
||||
@@ -513,6 +524,46 @@ pub trait ExifDao: Sync + Send {
|
||||
date_taken_tz_source: Option<&str>,
|
||||
) -> Result<(), DbError>;
|
||||
|
||||
/// Rows the true-instant drain (ADR-0001, ticket #118) still has to
|
||||
/// convert: `date_utc IS NULL AND date_taken IS NOT NULL` in
|
||||
/// `library_id`, `id > after_id`, ordered by id ASC. Rows with both
|
||||
/// NULL belong to `get_rows_needing_date_backfill` (whose writer fills
|
||||
/// the new triple too). The `after_id` cursor lets the drain page past
|
||||
/// rows it can't convert this tick (missing files) without spending
|
||||
/// its per-tick cap on them. Backed by `idx_image_exif_utc_backfill`.
|
||||
fn get_rows_needing_utc_backfill(
|
||||
&mut self,
|
||||
context: &opentelemetry::Context,
|
||||
library_id: i32,
|
||||
after_id: i32,
|
||||
limit: i64,
|
||||
) -> Result<Vec<UtcBackfillCandidate>, DbError>;
|
||||
|
||||
/// Persist a converted row's new-contract triple (`date_utc`,
|
||||
/// `camera_offset_minutes`, `date_taken_tz_source`) by primary key.
|
||||
/// Never touches the frozen legacy `date_taken` / `date_taken_source`
|
||||
/// pair. Guarded on `date_utc IS NULL` so a manual edit or ingest
|
||||
/// that raced the drain wins. Returns whether a row was written.
|
||||
fn backfill_date_utc(
|
||||
&mut self,
|
||||
context: &opentelemetry::Context,
|
||||
id: i32,
|
||||
date_utc: i64,
|
||||
camera_offset_minutes: Option<i32>,
|
||||
date_taken_tz_source: &str,
|
||||
) -> Result<bool, DbError>;
|
||||
|
||||
/// Progress surface for the true-instant drain: rows in `library_id`
|
||||
/// not yet on the new contract (`date_utc IS NULL`). Counts every such
|
||||
/// row — including NULL-date rows the legacy drain resolves and rows
|
||||
/// whose file is missing (until the missing-file scan retires them) —
|
||||
/// because cutover (#120) waits on exactly this set reaching zero.
|
||||
fn count_rows_needing_utc_backfill(
|
||||
&mut self,
|
||||
context: &opentelemetry::Context,
|
||||
library_id: i32,
|
||||
) -> Result<i64, DbError>;
|
||||
|
||||
/// Find image_exif rows needing a CLIP embedding for semantic search:
|
||||
/// `clip_embedding IS NULL AND content_hash IS NOT NULL`, ordered by id
|
||||
/// ASC, limited. Hash-less rows wait for `backfill_unhashed_backlog` to
|
||||
@@ -1501,6 +1552,95 @@ impl ExifDao for SqliteExifDao {
|
||||
})
|
||||
}
|
||||
|
||||
fn get_rows_needing_utc_backfill(
|
||||
&mut self,
|
||||
context: &opentelemetry::Context,
|
||||
library_id_val: i32,
|
||||
after_id: i32,
|
||||
limit: i64,
|
||||
) -> Result<Vec<UtcBackfillCandidate>, DbError> {
|
||||
trace_db_call(context, "query", "get_rows_needing_utc_backfill", |_span| {
|
||||
use schema::image_exif::dsl::*;
|
||||
|
||||
let mut connection = self.connection.lock().expect("Unable to get ExifDao");
|
||||
|
||||
// Partial index `(library_id, id) WHERE date_utc IS NULL`
|
||||
// serves the filter + order directly.
|
||||
image_exif
|
||||
.filter(library_id.eq(library_id_val))
|
||||
.filter(date_utc.is_null())
|
||||
.filter(date_taken.is_not_null())
|
||||
.filter(id.gt(after_id))
|
||||
.select((
|
||||
id,
|
||||
rel_path,
|
||||
date_taken.assume_not_null(),
|
||||
date_taken_source,
|
||||
))
|
||||
.order(id.asc())
|
||||
.limit(limit)
|
||||
.load::<UtcBackfillCandidate>(connection.deref_mut())
|
||||
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
|
||||
})
|
||||
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
|
||||
}
|
||||
|
||||
fn backfill_date_utc(
|
||||
&mut self,
|
||||
context: &opentelemetry::Context,
|
||||
id_val: i32,
|
||||
date_utc_val: i64,
|
||||
camera_offset_minutes_val: Option<i32>,
|
||||
date_taken_tz_source_val: &str,
|
||||
) -> Result<bool, DbError> {
|
||||
trace_db_call(context, "update", "backfill_date_utc", |_span| {
|
||||
use schema::image_exif::dsl::*;
|
||||
|
||||
let mut connection = self.connection.lock().expect("Unable to get ExifDao");
|
||||
|
||||
// New-contract triple only — the legacy pair is frozen
|
||||
// (ADR-0001).
|
||||
diesel::update(image_exif.find(id_val).filter(date_utc.is_null()))
|
||||
.set((
|
||||
date_utc.eq(Some(date_utc_val)),
|
||||
camera_offset_minutes.eq(camera_offset_minutes_val),
|
||||
date_taken_tz_source.eq(Some(date_taken_tz_source_val)),
|
||||
))
|
||||
.execute(connection.deref_mut())
|
||||
.map(|rows| rows > 0)
|
||||
.map_err(|e| anyhow::anyhow!("diesel update failed (id={}): {}", id_val, e))
|
||||
})
|
||||
.map_err(|e| {
|
||||
log::warn!("backfill_date_utc: {}", e);
|
||||
DbError::new(DbErrorKind::UpdateError)
|
||||
})
|
||||
}
|
||||
|
||||
fn count_rows_needing_utc_backfill(
|
||||
&mut self,
|
||||
context: &opentelemetry::Context,
|
||||
library_id_val: i32,
|
||||
) -> Result<i64, DbError> {
|
||||
trace_db_call(
|
||||
context,
|
||||
"query",
|
||||
"count_rows_needing_utc_backfill",
|
||||
|_span| {
|
||||
use schema::image_exif::dsl::*;
|
||||
|
||||
let mut connection = self.connection.lock().expect("Unable to get ExifDao");
|
||||
|
||||
image_exif
|
||||
.filter(library_id.eq(library_id_val))
|
||||
.filter(date_utc.is_null())
|
||||
.count()
|
||||
.get_result::<i64>(connection.deref_mut())
|
||||
.map_err(|e| anyhow::anyhow!("Query error: {}", e))
|
||||
},
|
||||
)
|
||||
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
|
||||
}
|
||||
|
||||
fn list_clip_unencoded_candidates(
|
||||
&mut self,
|
||||
context: &opentelemetry::Context,
|
||||
|
||||
@@ -58,6 +58,24 @@ pub trait PrecomputedReelDao: Sync + Send {
|
||||
/// to protect files a ledger row still points at.
|
||||
#[allow(dead_code)]
|
||||
fn all_cache_keys(&mut self, context: &opentelemetry::Context) -> Result<Vec<String>, DbError>;
|
||||
|
||||
/// Every reel row for `span` whose `library_key` is one of
|
||||
/// `library_keys`. Used by the true-instant date drain (ticket #118)
|
||||
/// to find the day reels a re-dated photo invalidates.
|
||||
fn list_for_span(
|
||||
&mut self,
|
||||
context: &opentelemetry::Context,
|
||||
span: &str,
|
||||
library_keys: &[String],
|
||||
) -> Result<Vec<PrecomputedReel>, DbError>;
|
||||
|
||||
/// Delete the given reel rows (stale-marking: the nightly pre-gen
|
||||
/// finds no current row and regenerates). Returns rows deleted.
|
||||
fn delete_ids(
|
||||
&mut self,
|
||||
context: &opentelemetry::Context,
|
||||
ids: &[i32],
|
||||
) -> Result<usize, DbError>;
|
||||
}
|
||||
|
||||
pub struct SqlitePrecomputedReelDao {
|
||||
@@ -219,6 +237,49 @@ impl PrecomputedReelDao for SqlitePrecomputedReelDao {
|
||||
})
|
||||
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
|
||||
}
|
||||
|
||||
fn list_for_span(
|
||||
&mut self,
|
||||
context: &opentelemetry::Context,
|
||||
span: &str,
|
||||
library_keys: &[String],
|
||||
) -> Result<Vec<PrecomputedReel>, DbError> {
|
||||
trace_db_call(context, "query", "list_for_span", |_span| {
|
||||
use schema::precomputed_reels::dsl;
|
||||
|
||||
let mut connection = self
|
||||
.connection
|
||||
.lock()
|
||||
.expect("Unable to lock PrecomputedReelDao");
|
||||
|
||||
dsl::precomputed_reels
|
||||
.filter(dsl::span.eq(span))
|
||||
.filter(dsl::library_key.eq_any(library_keys))
|
||||
.load::<PrecomputedReel>(connection.deref_mut())
|
||||
.map_err(|e| anyhow::anyhow!("Failed to list reels for span: {}", e))
|
||||
})
|
||||
.map_err(|e| DbError::log(DbErrorKind::QueryError, e))
|
||||
}
|
||||
|
||||
fn delete_ids(
|
||||
&mut self,
|
||||
context: &opentelemetry::Context,
|
||||
ids: &[i32],
|
||||
) -> Result<usize, DbError> {
|
||||
trace_db_call(context, "delete", "delete_ids", |_span| {
|
||||
use schema::precomputed_reels::dsl;
|
||||
|
||||
let mut connection = self
|
||||
.connection
|
||||
.lock()
|
||||
.expect("Unable to lock PrecomputedReelDao");
|
||||
|
||||
diesel::delete(dsl::precomputed_reels.filter(dsl::id.eq_any(ids)))
|
||||
.execute(connection.deref_mut())
|
||||
.map_err(|e| anyhow::anyhow!("Failed to delete reels: {}", e))
|
||||
})
|
||||
.map_err(|e| DbError::log(DbErrorKind::UpdateError, e))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
Reference in new issue
Block a user