diff --git a/internal/ebooks/enrichment_queue.go b/internal/ebooks/enrichment_queue.go index 8913a3d0..00f8c431 100644 --- a/internal/ebooks/enrichment_queue.go +++ b/internal/ebooks/enrichment_queue.go @@ -155,42 +155,159 @@ func (q *EnrichmentQueue) Enqueue(ctx context.Context, contentID string, priorit return err } -var reconcileMissingEnrichmentJobsQuery = ` - WITH candidates AS MATERIALIZED ( - SELECT mi.content_id, ` + ebookProtectedFieldsSQL + ` AS protected_fields - FROM media_item_libraries mil - JOIN media_items mi ON mi.content_id = mil.content_id - LEFT JOIN ebook_enrichment_state state ON state.content_id = mi.content_id - WHERE mil.media_folder_id = $1 - AND mi.type = 'ebook' - AND ` + catalog.MangaChapterExclusionWhere("mi") + ` - AND state.content_id IS NULL - ORDER BY mi.content_id - LIMIT $3 +const ensureEnrichmentReconcileCursorQuery = ` + INSERT INTO ebook_enrichment_reconcile_cursors ( + folder_id, after_first_seen_at, after_content_id, updated_at ) - INSERT INTO ebook_enrichment_state ( - content_id, status, priority, next_attempt_at, protected_fields, updated_at - ) - SELECT candidates.content_id, 'pending', $2, now(), candidates.protected_fields, now() - FROM candidates - ON CONFLICT (content_id) DO NOTHING + VALUES ($1, NULL, NULL, now()) + ON CONFLICT (folder_id) DO NOTHING ` -func (q *EnrichmentQueue) ReconcileMissing(ctx context.Context, folderID, priority, limit int) (int, error) { +const lockEnrichmentReconcileCursorQuery = ` + SELECT after_first_seen_at, after_content_id + FROM ebook_enrichment_reconcile_cursors + WHERE folder_id = $1 + FOR UPDATE +` + +var reconcileMissingEnrichmentJobsQuery = ` + WITH membership_candidates AS MATERIALIZED ( + SELECT membership.content_id, membership.first_seen_at + FROM media_item_libraries membership + WHERE membership.media_folder_id = $1 + AND ( + $4::timestamptz IS NULL + OR membership.first_seen_at < $4 + OR ( + membership.first_seen_at = $4 + AND membership.content_id > $5 + ) + ) + ORDER BY membership.first_seen_at DESC, membership.content_id + LIMIT $3 + ), + candidates AS MATERIALIZED ( + SELECT candidate.content_id, ` + ebookProtectedFieldsSQL + ` AS protected_fields + FROM membership_candidates candidate + JOIN media_items mi ON mi.content_id = candidate.content_id + LEFT JOIN ebook_enrichment_state state ON state.content_id = candidate.content_id + WHERE mi.type = 'ebook' + AND ` + catalog.MangaChapterExclusionWhere("mi") + ` + AND state.content_id IS NULL + ), + inserted AS ( + INSERT INTO ebook_enrichment_state ( + content_id, status, priority, next_attempt_at, protected_fields, updated_at + ) + SELECT candidates.content_id, 'pending', $2, now(), candidates.protected_fields, now() + FROM candidates + ON CONFLICT (content_id) DO NOTHING + RETURNING content_id + ), + window_stats AS MATERIALIZED ( + SELECT + COUNT(*)::integer AS inspected, + ( + SELECT first_seen_at + FROM membership_candidates + ORDER BY first_seen_at, content_id DESC + LIMIT 1 + ) AS last_first_seen_at, + ( + SELECT content_id + FROM membership_candidates + ORDER BY first_seen_at, content_id DESC + LIMIT 1 + ) AS last_content_id + FROM membership_candidates + ) + SELECT + (SELECT COUNT(*)::integer FROM inserted) AS reconciled, + window_stats.inspected, + window_stats.last_first_seen_at, + window_stats.last_content_id + FROM window_stats +` + +const updateEnrichmentReconcileCursorQuery = ` + UPDATE ebook_enrichment_reconcile_cursors + SET after_first_seen_at = $2, + after_content_id = $3, + updated_at = now() + WHERE folder_id = $1 +` + +// ReconcileMissing advances a database-persisted keyset cursor while holding +// its folder row lock. Queue repairs and cursor movement commit atomically. +// A server restart resumes from the persisted cursor; completed repairs remain +// durable, and reaching the end resets the cursor so later passes wrap safely. +func (q *EnrichmentQueue) ReconcileMissing( + ctx context.Context, + folderID, priority, limit int, +) (reconciled, inspected int, wrapped bool, err error) { if q == nil || q.pool == nil { - return 0, errors.New("ebook enrichment queue is not configured") + return 0, 0, false, errors.New("ebook enrichment queue is not configured") } if folderID <= 0 { - return 0, errors.New("ebook enrichment folder id is required") + return 0, 0, false, errors.New("ebook enrichment folder id is required") } if limit <= 0 { - return 0, nil + return 0, 0, false, nil } - tag, err := q.pool.Exec(ctx, reconcileMissingEnrichmentJobsQuery, folderID, priority, limit) + + tx, err := q.pool.Begin(ctx) if err != nil { - return 0, err + return 0, 0, false, err } - return int(tag.RowsAffected()), nil + defer func() { + _ = tx.Rollback(ctx) + }() + + if _, err = tx.Exec(ctx, ensureEnrichmentReconcileCursorQuery, folderID); err != nil { + return 0, 0, false, err + } + var afterFirstSeenAt *time.Time + var afterContentID *string + if err = tx.QueryRow( + ctx, + lockEnrichmentReconcileCursorQuery, + folderID, + ).Scan(&afterFirstSeenAt, &afterContentID); err != nil { + return 0, 0, false, err + } + + var lastFirstSeenAt *time.Time + var lastContentID *string + err = tx.QueryRow( + ctx, + reconcileMissingEnrichmentJobsQuery, + folderID, + priority, + limit, + afterFirstSeenAt, + afterContentID, + ).Scan(&reconciled, &inspected, &lastFirstSeenAt, &lastContentID) + if err != nil { + return 0, 0, false, err + } + wrapped = inspected < limit + if wrapped { + lastFirstSeenAt = nil + lastContentID = nil + } + if _, err = tx.Exec( + ctx, + updateEnrichmentReconcileCursorQuery, + folderID, + lastFirstSeenAt, + lastContentID, + ); err != nil { + return 0, 0, false, err + } + if err = tx.Commit(ctx); err != nil { + return 0, 0, false, err + } + return reconciled, inspected, wrapped, nil } // Each indexed window is fixed-size; only their deduplicated union pays for @@ -346,6 +463,8 @@ const hasReadyEnrichmentJobsQueryTemplate = ` WHERE next_attempt_at <= now() AND (status = 'pending' OR (status = 'running' AND lease_until < now())) AND {{lane_predicate}} + ORDER BY next_attempt_at, updated_at, priority DESC + LIMIT 1 ) ` diff --git a/internal/ebooks/enrichment_queue_test.go b/internal/ebooks/enrichment_queue_test.go index 038ff417..fd30ba4f 100644 --- a/internal/ebooks/enrichment_queue_test.go +++ b/internal/ebooks/enrichment_queue_test.go @@ -151,6 +151,8 @@ func TestEnrichmentQueueHasReadyUsesBoundedExistenceQueries(t *testing.T) { "next_attempt_at <= now()", "status = 'pending' OR (status = 'running' AND lease_until < now())", tt.predicate, + "ORDER BY next_attempt_at, updated_at, priority DESC", + "LIMIT 1", } { if !strings.Contains(query, fragment) { t.Fatalf("has-ready query missing %q:\n%s", fragment, tt.query) @@ -166,17 +168,68 @@ func TestEnrichmentQueueHasReadyUsesBoundedExistenceQueries(t *testing.T) { func TestEnrichmentQueueReconcileMissingIsBoundedAndLaneSafe(t *testing.T) { query := strings.Join(strings.Fields(reconcileMissingEnrichmentJobsQuery), " ") for _, fragment := range []string{ - "WHERE mil.media_folder_id = $1", + "membership_candidates AS MATERIALIZED", + "WHERE membership.media_folder_id = $1", + "$4::timestamptz IS NULL", + "membership.first_seen_at < $4", + "membership.content_id > $5", + "ORDER BY membership.first_seen_at DESC, membership.content_id", + "LIMIT $3", + "FROM membership_candidates candidate", "mi.type = 'ebook'", "state.content_id IS NULL", - "LIMIT $3", "SELECT candidates.content_id, 'pending', $2", "ON CONFLICT (content_id) DO NOTHING", + "COUNT(*)::integer AS inspected", + "FROM membership_candidates", + "(SELECT COUNT(*)::integer FROM inserted) AS reconciled", } { if !strings.Contains(query, fragment) { t.Fatalf("reconcile query missing %q:\n%s", fragment, reconcileMissingEnrichmentJobsQuery) } } + windowAt := strings.Index(query, "membership_candidates AS MATERIALIZED") + boundedQuery := query[windowAt:] + limitAt := strings.Index(boundedQuery, "LIMIT $3") + itemJoinAt := strings.Index(boundedQuery, "JOIN media_items mi") + stateJoinAt := strings.Index(boundedQuery, "LEFT JOIN ebook_enrichment_state state") + if windowAt < 0 || limitAt < 0 || itemJoinAt < 0 || stateJoinAt < 0 { + t.Fatalf("reconcile query is missing bounded traversal structure:\n%s", reconcileMissingEnrichmentJobsQuery) + } + if limitAt > itemJoinAt || limitAt > stateJoinAt { + t.Fatalf("reconcile query filters missing jobs before bounding raw membership traversal:\n%s", reconcileMissingEnrichmentJobsQuery) + } + statsAt := strings.Index(query, "COUNT(*)::integer AS inspected") + insertedAt := strings.Index(query, "(SELECT COUNT(*)::integer FROM inserted) AS reconciled") + if statsAt < 0 || insertedAt < 0 || statsAt > insertedAt { + t.Fatalf("cursor advancement is not derived independently from the inspected membership window:\n%s", reconcileMissingEnrichmentJobsQuery) + } + + ensureQuery := strings.Join(strings.Fields(ensureEnrichmentReconcileCursorQuery), " ") + if !strings.Contains(ensureQuery, "ON CONFLICT (folder_id) DO NOTHING") { + t.Fatalf("cursor ensure query is not idempotent: %s", ensureEnrichmentReconcileCursorQuery) + } + lockQuery := strings.Join(strings.Fields(lockEnrichmentReconcileCursorQuery), " ") + for _, fragment := range []string{ + "SELECT after_first_seen_at, after_content_id", + "WHERE folder_id = $1", + "FOR UPDATE", + } { + if !strings.Contains(lockQuery, fragment) { + t.Fatalf("cursor lock query missing %q: %s", fragment, lockEnrichmentReconcileCursorQuery) + } + } + updateQuery := strings.Join(strings.Fields(updateEnrichmentReconcileCursorQuery), " ") + for _, fragment := range []string{ + "UPDATE ebook_enrichment_reconcile_cursors", + "SET after_first_seen_at = $2", + "after_content_id = $3", + "WHERE folder_id = $1", + } { + if !strings.Contains(updateQuery, fragment) { + t.Fatalf("cursor update query missing %q: %s", fragment, updateEnrichmentReconcileCursorQuery) + } + } } func TestEnrichmentScopeValidation(t *testing.T) { diff --git a/internal/scanner/ebook_scan.go b/internal/scanner/ebook_scan.go index 87a70ccf..16976034 100644 --- a/internal/scanner/ebook_scan.go +++ b/internal/scanner/ebook_scan.go @@ -582,7 +582,7 @@ func (s *Scanner) reconcileMissingEbookEnrichment(ctx context.Context, folderID if s == nil || s.ebookEnrichmentQueue == nil || folderID <= 0 { return } - reconciled, err := s.ebookEnrichmentQueue.ReconcileMissing( + reconciled, inspected, wrapped, err := s.ebookEnrichmentQueue.ReconcileMissing( ctx, folderID, ebookEnrichmentPriority, @@ -604,6 +604,13 @@ func (s *Scanner) reconcileMissingEbookEnrichment(ctx context.Context, folderID "reconciled", reconciled, ) } + slog.DebugContext(ctx, "ebook scan: metadata enrichment reconciliation window complete", + "component", "scanner", + "folder_id", folderID, + "inspected", inspected, + "reconciled", reconciled, + "wrapped", wrapped, + ) } func (s *Scanner) autoLinkLiteraryWork(ctx context.Context, contentID string) { diff --git a/internal/scanner/ebook_test.go b/internal/scanner/ebook_test.go index bb0539d6..f9e17c26 100644 --- a/internal/scanner/ebook_test.go +++ b/internal/scanner/ebook_test.go @@ -25,6 +25,8 @@ type fakeEbookEnrichmentQueue struct { reconcileCalls int reconcileLimit int reconcileErr error + inspected int + wrapped bool } func (f *fakeEbookEnrichmentQueue) Enqueue(_ context.Context, contentID string, priority int) error { @@ -33,10 +35,10 @@ func (f *fakeEbookEnrichmentQueue) Enqueue(_ context.Context, contentID string, return f.err } -func (f *fakeEbookEnrichmentQueue) ReconcileMissing(_ context.Context, _ int, _ int, limit int) (int, error) { +func (f *fakeEbookEnrichmentQueue) ReconcileMissing(_ context.Context, _ int, _ int, limit int) (int, int, bool, error) { f.reconcileCalls++ f.reconcileLimit = limit - return 0, f.reconcileErr + return 0, f.inspected, f.wrapped, f.reconcileErr } func TestScannerEbookEnrichmentHookEnqueuesHighPriorityWork(t *testing.T) { diff --git a/internal/scanner/scanner.go b/internal/scanner/scanner.go index ce5b3a00..f0c9cc2b 100644 --- a/internal/scanner/scanner.go +++ b/internal/scanner/scanner.go @@ -204,7 +204,7 @@ type MetadataQueueProducer interface { // provider work in the scanner. type EbookEnrichmentQueue interface { Enqueue(ctx context.Context, contentID string, priority int) error - ReconcileMissing(ctx context.Context, folderID, priority, limit int) (int, error) + ReconcileMissing(ctx context.Context, folderID, priority, limit int) (reconciled, inspected int, wrapped bool, err error) } type LiteraryWorkLinker interface { diff --git a/migrations/ebook_enrichment_reconcile_cursor_test.go b/migrations/ebook_enrichment_reconcile_cursor_test.go new file mode 100644 index 00000000..88f479f4 --- /dev/null +++ b/migrations/ebook_enrichment_reconcile_cursor_test.go @@ -0,0 +1,30 @@ +package migrations + +import ( + "strings" + "testing" +) + +func TestEbookEnrichmentReconcileCursorPersistsAcrossRestartsAndUsesCoveringIndex(t *testing.T) { + migrationBytes, err := FS.ReadFile("sql/20260719173000_ebook_enrichment_reconcile_cursor.sql") + if err != nil { + t.Fatalf("read migration: %v", err) + } + migration := strings.Join(strings.Fields(string(migrationBytes)), " ") + for _, fragment := range []string{ + "-- +goose NO TRANSACTION", + "CREATE TABLE IF NOT EXISTS ebook_enrichment_reconcile_cursors", + "folder_id integer PRIMARY KEY REFERENCES media_folders(id) ON DELETE CASCADE", + "after_first_seen_at timestamptz", + "after_content_id text", + "CHECK ((after_first_seen_at IS NULL) = (after_content_id IS NULL))", + "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_item_libraries_folder_enrichment_cursor", + "ON public.media_item_libraries (media_folder_id, first_seen_at DESC, content_id)", + "DROP INDEX CONCURRENTLY IF EXISTS idx_item_libraries_folder_enrichment_cursor", + "DROP TABLE IF EXISTS ebook_enrichment_reconcile_cursors", + } { + if !strings.Contains(migration, fragment) { + t.Fatalf("cursor migration missing %q", fragment) + } + } +} diff --git a/migrations/sql/20260719173000_ebook_enrichment_reconcile_cursor.sql b/migrations/sql/20260719173000_ebook_enrichment_reconcile_cursor.sql new file mode 100644 index 00000000..bea4b8ca --- /dev/null +++ b/migrations/sql/20260719173000_ebook_enrichment_reconcile_cursor.sql @@ -0,0 +1,37 @@ +-- +goose NO TRANSACTION + +-- +goose Up +CREATE TABLE IF NOT EXISTS ebook_enrichment_reconcile_cursors ( + folder_id integer PRIMARY KEY REFERENCES media_folders(id) ON DELETE CASCADE, + after_first_seen_at timestamptz, + after_content_id text, + updated_at timestamptz NOT NULL DEFAULT now(), + CHECK ((after_first_seen_at IS NULL) = (after_content_id IS NULL)) +); + +-- A failed CREATE INDEX CONCURRENTLY can leave an invalid index that blocks an +-- IF NOT EXISTS retry. Drop only that unusable artifact before rebuilding. +-- +goose StatementBegin +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + JOIN pg_index i ON i.indexrelid = c.oid + WHERE n.nspname = 'public' + AND c.relname = 'idx_item_libraries_folder_enrichment_cursor' + AND NOT i.indisvalid + ) THEN + DROP INDEX public.idx_item_libraries_folder_enrichment_cursor; + END IF; +END; +$$; +-- +goose StatementEnd + +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_item_libraries_folder_enrichment_cursor + ON public.media_item_libraries (media_folder_id, first_seen_at DESC, content_id); + +-- +goose Down +DROP INDEX CONCURRENTLY IF EXISTS idx_item_libraries_folder_enrichment_cursor; +DROP TABLE IF EXISTS ebook_enrichment_reconcile_cursors;