From 84cd0acb0ca2fb27fdeac288b4f2bbfae805dd02 Mon Sep 17 00:00:00 2001 From: Quick104 <31828688+Quick104@users.noreply.github.com> Date: Thu, 23 Jul 2026 11:10:31 -0400 Subject: [PATCH] fix(ebooks): keep legacy-lane rows in their lane until terminal outcomes Scanner re-enqueues (priority 100) could silently promote pending legacy backlog rows (priority -100) into the incremental lane via the enqueue upsert's GREATEST, and the fail/release requeue branches hardcoded 100 regardless of the row's lane. A mass mtime shift or group-key-version bump would have moved the entire legacy backlog out from under the backfill task's pacing controls into the scheduled sync task. Lane changes now happen only through terminal outcomes (complete or discard); enqueue, fail, and release preserve a negative priority. Co-Authored-By: Claude Fable 5 --- internal/ebooks/enrichment_queue.go | 8 ++++++-- internal/ebooks/enrichment_queue_test.go | 25 +++++++++++++++++++++++- 2 files changed, 30 insertions(+), 3 deletions(-) diff --git a/internal/ebooks/enrichment_queue.go b/internal/ebooks/enrichment_queue.go index 024b7e48..68f7bcf0 100644 --- a/internal/ebooks/enrichment_queue.go +++ b/internal/ebooks/enrichment_queue.go @@ -129,6 +129,9 @@ const mergeEbookProtectedFieldsSQL = ` ) ` +// Legacy-lane rows (priority < 0) may only leave the lane through a terminal +// outcome (Complete/Discard); enqueue, fail, and release must preserve the +// lane so the paced backfill task stays the sole drain for the backlog. var enqueueEnrichmentJobQuery = ` INSERT INTO ebook_enrichment_state ( content_id, status, priority, next_attempt_at, protected_fields, updated_at @@ -145,6 +148,7 @@ var enqueueEnrichmentJobQuery = ` END, priority = CASE WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.priority + WHEN ebook_enrichment_state.priority < 0 THEN ebook_enrichment_state.priority ELSE GREATEST(ebook_enrichment_state.priority, EXCLUDED.priority) END, next_attempt_at = CASE @@ -670,7 +674,7 @@ var failEnrichmentJobQuery = ` WHEN requeue_requested THEN now() ELSE now() + $4::interval END, - priority = CASE WHEN requeue_requested THEN 100 ELSE priority END, + priority = CASE WHEN requeue_requested AND priority >= 0 THEN 100 ELSE priority END, requeue_requested = false, outcome = 'failed', last_error_class = $2, @@ -720,7 +724,7 @@ var releaseEnrichmentJobQuery = ` claim_token = NULL, attempts = GREATEST(attempts - 1, 0), next_attempt_at = CASE WHEN requeue_requested THEN now() ELSE next_attempt_at END, - priority = CASE WHEN requeue_requested THEN 100 ELSE priority END, + priority = CASE WHEN requeue_requested AND priority >= 0 THEN 100 ELSE priority END, requeue_requested = false, updated_at = now() WHERE content_id = $1 diff --git a/internal/ebooks/enrichment_queue_test.go b/internal/ebooks/enrichment_queue_test.go index ede1d664..58e82695 100644 --- a/internal/ebooks/enrichment_queue_test.go +++ b/internal/ebooks/enrichment_queue_test.go @@ -317,6 +317,7 @@ func TestEnrichmentQueueEnqueueMakesPendingRowsDueAndDefersRunningRequeue(t *tes "ON CONFLICT (content_id) DO UPDATE SET", "WHEN ebook_enrichment_state.status = 'running' THEN ebook_enrichment_state.next_attempt_at", "ELSE now()", + "WHEN ebook_enrichment_state.priority < 0 THEN ebook_enrichment_state.priority", "requeue_requested = ebook_enrichment_state.requeue_requested OR ebook_enrichment_state.status = 'running'", } { if !strings.Contains(query, fragment) { @@ -328,6 +329,28 @@ func TestEnrichmentQueueEnqueueMakesPendingRowsDueAndDefersRunningRequeue(t *tes } } +func TestEnrichmentQueueKeepsLegacyLaneRowsUntilTerminalOutcome(t *testing.T) { + enqueue := strings.Join(strings.Fields(enqueueEnrichmentJobQuery), " ") + if !strings.Contains(enqueue, "WHEN ebook_enrichment_state.priority < 0 THEN ebook_enrichment_state.priority") { + t.Fatalf("enqueue must keep legacy-lane rows in the legacy lane:\n%s", enqueueEnrichmentJobQuery) + } + for _, tt := range []struct { + name string + query string + }{ + {name: "fail", query: failEnrichmentJobQuery}, + {name: "release", query: releaseEnrichmentJobQuery}, + } { + query := strings.Join(strings.Fields(tt.query), " ") + if strings.Contains(query, "WHEN requeue_requested THEN 100") { + t.Fatalf("%s requeue must not promote legacy-lane rows without a terminal outcome:\n%s", tt.name, tt.query) + } + if !strings.Contains(query, "WHEN requeue_requested AND priority >= 0 THEN 100 ELSE priority END") { + t.Fatalf("%s requeue must stay within the row's lane:\n%s", tt.name, tt.query) + } + } +} + func TestEnrichmentQueueMigrationIsCrashSafeAndSeedsAllLegacyEbooks(t *testing.T) { body, err := os.ReadFile("../../migrations/sql/20260719090000_ebook_enrichment_jobs.sql") if err != nil { @@ -436,7 +459,7 @@ func TestEnrichmentQueueTransitionsKeepDurableRowsAndReleaseLeases(t *testing.T) failure := strings.Join(strings.Fields(failEnrichmentJobQuery), " ") for _, fragment := range []string{ "WHEN requeue_requested THEN now()", - "WHEN requeue_requested THEN 100 ELSE priority END", + "WHEN requeue_requested AND priority >= 0 THEN 100 ELSE priority END", "AND claim_token = $5", } { if !strings.Contains(failure, fragment) {