fix(ebooks): persist bounded reconciliation cursor

2026-07-22 13:43:57 +02:00
parent 9136803130
commit fcaee9bfce
7 changed files with 279 additions and 31 deletions
@@ -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
 	)
 `

@@ -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) {
@@ -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) {
@@ -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) {
@@ -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 {
@@ -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)
+		}
+	}
+}
@@ -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;