Compare commits
2
Commits
fcaf8487c2
...
b7b59c1ba2
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b7b59c1ba2 | ||
|
|
1fad08dea3 |
No files matched your search
@@ -437,6 +437,28 @@ a stronger tool comes online (exiftool install, new filename regex),
|
||||
re-resolve out-of-band rather than re-introducing the steady-state
|
||||
eligibility.
|
||||
|
||||
**True-instant drain (`backfill_utc_dates`, ticket #118).** Runs right
|
||||
after `backfill_missing_date_taken` inside the per-library online gate,
|
||||
sharing its `DATE_BACKFILL_MAX_PER_TICK` cap. Converts legacy
|
||||
mixed-convention rows (`date_utc IS NULL AND date_taken IS NOT NULL`,
|
||||
partial index `idx_image_exif_utc_backfill`) by re-running
|
||||
`resolve_dates_batch` with empty priors (one exiftool subprocess per
|
||||
batch) and writing **only** the new-contract triple — the legacy pair
|
||||
stays frozen. Legacy `manual` rows convert via
|
||||
`camera_tz::manual_date_resolution` instead of being re-resolved from
|
||||
the file. Missing-file rows are paged past without spending the cap, and
|
||||
a per-library resume cursor (carried across ticks by the watcher, wraps
|
||||
at end of table) keeps unconvertible rows from stalling the rows behind
|
||||
them; EXIF/exiftool-legacy rows wait when exiftool is absent (they'd
|
||||
otherwise fall to fs_time). When a converted row's display-zone day differs from
|
||||
the day its legacy value rendered on, the `day`-span `precomputed_reels`
|
||||
rows covering either month-day (this library + `all`) are deleted and
|
||||
unlinked; the nightly pre-gen regenerates. Daily conversation summaries
|
||||
are SMS-derived and unaffected. Progress: Prometheus
|
||||
`imageserver_date_backfill_remaining` (per tick) and
|
||||
`date_backfill_remaining` per library on `GET /libraries` (live count of
|
||||
`date_utc IS NULL`).
|
||||
|
||||
`/memories` is a single SQL query against this column
|
||||
(`get_memories_in_window` in `src/database/mod.rs`), using
|
||||
`strftime('%m-%d' | '%W' | '%m', date_taken, 'unixepoch', tz)` for
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
DROP INDEX IF EXISTS idx_image_exif_utc_backfill;
|
||||
@@ -0,0 +1,9 @@
|
||||
-- Partial index backing the true-instant backfill drain (ADR-0001, ticket
|
||||
-- #118). The drain selects rows still on the legacy convention
|
||||
-- (`date_utc IS NULL`) per library in id order, and the progress surface
|
||||
-- (Prometheus `imageserver_date_backfill_remaining` + `/libraries`) counts
|
||||
-- the same set. Mirrors `idx_image_exif_date_backfill`: the index shrinks
|
||||
-- to nothing as the drain converges, so it costs ~0 after cutover.
|
||||
CREATE INDEX idx_image_exif_utc_backfill
|
||||
ON image_exif (library_id, id)
|
||||
WHERE date_utc IS NULL;
|
||||
+968
-7
File diff suppressed because it is too large.
Load diff
@@ -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)]
|
||||
|
||||
@@ -351,7 +351,7 @@ const EXIFTOOL_EVIDENCE_TAGS: &[&str] = &[
|
||||
/// Cache the "exiftool exists on PATH" check across the process lifetime so
|
||||
/// the per-tick backfill doesn't fork a doomed subprocess every iteration on
|
||||
/// deploys without exiftool installed.
|
||||
fn exiftool_available() -> bool {
|
||||
pub fn exiftool_available() -> bool {
|
||||
static AVAIL: OnceLock<bool> = OnceLock::new();
|
||||
*AVAIL.get_or_init(|| {
|
||||
let ok = Command::new("exiftool")
|
||||
|
||||
@@ -1756,6 +1756,35 @@ mod tests {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn get_rows_needing_utc_backfill(
|
||||
&mut self,
|
||||
_context: &opentelemetry::Context,
|
||||
_library_id: i32,
|
||||
_after_id: i32,
|
||||
_limit: i64,
|
||||
) -> Result<Vec<crate::database::UtcBackfillCandidate>, DbError> {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
|
||||
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> {
|
||||
Ok(false)
|
||||
}
|
||||
|
||||
fn count_rows_needing_utc_backfill(
|
||||
&mut self,
|
||||
_context: &opentelemetry::Context,
|
||||
_library_id: i32,
|
||||
) -> Result<i64, DbError> {
|
||||
Ok(0)
|
||||
}
|
||||
|
||||
fn set_manual_date_taken(
|
||||
&mut self,
|
||||
_context: &opentelemetry::Context,
|
||||
|
||||
@@ -460,6 +460,10 @@ pub struct LibraryStatus {
|
||||
#[serde(flatten)]
|
||||
pub library: Library,
|
||||
pub health: LibraryHealth,
|
||||
/// Rows in this library not yet converted to the true-instant date
|
||||
/// contract (`date_utc IS NULL`, ADR-0001 / ticket #118). Drains to 0
|
||||
/// over watcher ticks; `None` when the count query failed.
|
||||
pub date_backfill_remaining: Option<i64>,
|
||||
}
|
||||
|
||||
#[derive(serde::Serialize)]
|
||||
@@ -488,6 +492,12 @@ pub async fn list_libraries(_claims: Claims, app_state: Data<AppState>) -> impl
|
||||
.library_health
|
||||
.read()
|
||||
.unwrap_or_else(|e| e.into_inner());
|
||||
let context = opentelemetry::Context::new();
|
||||
let mut exif_dao = app_state
|
||||
.insight_generator
|
||||
.exif_dao()
|
||||
.lock()
|
||||
.expect("Unable to lock ExifDao");
|
||||
let libraries = live_guard
|
||||
.iter()
|
||||
.map(|lib| LibraryStatus {
|
||||
@@ -496,6 +506,11 @@ pub async fn list_libraries(_claims: Claims, app_state: Data<AppState>) -> impl
|
||||
.get(&lib.id)
|
||||
.cloned()
|
||||
.unwrap_or(LibraryHealth::Online),
|
||||
// Live count off the partial index — cheap, and exact even for
|
||||
// a stale library the watcher isn't refreshing.
|
||||
date_backfill_remaining: exif_dao
|
||||
.count_rows_needing_utc_backfill(&context, lib.id)
|
||||
.ok(),
|
||||
})
|
||||
.collect();
|
||||
HttpResponse::Ok().json(LibrariesResponse {
|
||||
|
||||
@@ -151,6 +151,12 @@ fn main() -> std::io::Result<()> {
|
||||
.registry
|
||||
.register(Box::new(thumbnails::VIDEO_GAUGE.clone()))
|
||||
.unwrap();
|
||||
// True-instant date drain progress (#118). Refreshed per watcher
|
||||
// tick; see `backfill::refresh_utc_backfill_progress`.
|
||||
prometheus
|
||||
.registry
|
||||
.register(Box::new(backfill::DATE_BACKFILL_REMAINING.clone()))
|
||||
.unwrap();
|
||||
// HLS readiness gauges. Updated by the watcher every full-scan
|
||||
// tick and on every `/hls/stats` request. See `hls_stats`.
|
||||
prometheus
|
||||
|
||||
+48
-1
@@ -26,7 +26,10 @@ use log::{debug, error, info, warn};
|
||||
use crate::backfill;
|
||||
use crate::content_hash;
|
||||
use crate::database::models::InsertImageExif;
|
||||
use crate::database::{ExifDao, PreviewDao, SqliteExifDao, SqlitePreviewDao};
|
||||
use crate::database::{
|
||||
ExifDao, PrecomputedReelDao, PreviewDao, SqliteExifDao, SqlitePrecomputedReelDao,
|
||||
SqlitePreviewDao, SqliteUserAiPrefsDao, UserAiPrefsDao,
|
||||
};
|
||||
use crate::date_resolver;
|
||||
use crate::exif;
|
||||
use crate::face_watch;
|
||||
@@ -336,6 +339,20 @@ pub fn watch_files(
|
||||
Box::new(SqliteTagDao::default()) as Box<dyn tags::TagDao>
|
||||
));
|
||||
|
||||
// True-instant date drain (#118): day-reel invalidation + the
|
||||
// display zone the day comparison runs in.
|
||||
let precomputed_reel_dao = Arc::new(Mutex::new(
|
||||
Box::new(SqlitePrecomputedReelDao::new()) as Box<dyn PrecomputedReelDao>
|
||||
));
|
||||
let user_ai_prefs_dao: Mutex<Box<dyn UserAiPrefsDao>> =
|
||||
Mutex::new(Box::new(SqliteUserAiPrefsDao::new()));
|
||||
// Last-known per-library remaining count behind the
|
||||
// `imageserver_date_backfill_remaining` gauge.
|
||||
let mut utc_backfill_remaining: HashMap<i32, i64> = HashMap::new();
|
||||
// Per-library resume cursor for the true-instant drain; in-memory,
|
||||
// so a restart resumes from 0 (harmless — the drain is idempotent).
|
||||
let mut utc_backfill_cursors: HashMap<i32, i32> = HashMap::new();
|
||||
|
||||
let mut last_quick_scan = SystemTime::now();
|
||||
// Initialize to UNIX_EPOCH so the *first* tick is treated as a
|
||||
// full scan. That replaces the legacy startup ScanDirectoryMessage
|
||||
@@ -494,6 +511,36 @@ pub fn watch_files(
|
||||
{
|
||||
let context = opentelemetry::Context::new();
|
||||
backfill::backfill_missing_date_taken(&context, lib, &exif_dao);
|
||||
|
||||
// True-instant drain (ADR-0001, #118): convert
|
||||
// legacy mixed-convention rows, invalidating day
|
||||
// reels whose day moved. Inside the online gate, so a
|
||||
// stale library is skipped whole and resumes next
|
||||
// tick. The display zone is re-read per tick so a
|
||||
// prefs change applies without restart.
|
||||
let display_tz = {
|
||||
let mut prefs = user_ai_prefs_dao.lock().expect("prefs dao poisoned");
|
||||
match prefs.get_prefs(&context) {
|
||||
Ok(Some(p)) => {
|
||||
crate::camera_tz::display_zone_for(p.display_tz.as_deref())
|
||||
}
|
||||
_ => crate::camera_tz::display_zone_for(None),
|
||||
}
|
||||
};
|
||||
backfill::backfill_utc_dates(
|
||||
&context,
|
||||
lib,
|
||||
&exif_dao,
|
||||
&precomputed_reel_dao,
|
||||
&display_tz,
|
||||
utc_backfill_cursors.entry(lib.id).or_insert(0),
|
||||
);
|
||||
backfill::refresh_utc_backfill_progress(
|
||||
&context,
|
||||
lib,
|
||||
&exif_dao,
|
||||
&mut utc_backfill_remaining,
|
||||
);
|
||||
}
|
||||
|
||||
if is_full_scan {
|
||||
|
||||
Reference in new issue
Block a user