From 794e2d786a987f419e3729eedfcf2cac0e2de8a4 Mon Sep 17 00:00:00 2001 From: rxwatcher Date: Sun, 19 Jul 2026 17:26:52 +0200 Subject: [PATCH] fix(ebooks): use planner-safe queue cursors --- internal/ebooks/enrichment_queue.go | 18 +++++++----------- internal/ebooks/enrichment_queue_test.go | 14 +++++++++----- .../ebook_enrichment_reconcile_cursor_test.go | 2 +- ...73000_ebook_enrichment_reconcile_cursor.sql | 2 +- 4 files changed, 18 insertions(+), 18 deletions(-) diff --git a/internal/ebooks/enrichment_queue.go b/internal/ebooks/enrichment_queue.go index 00f8c431..17f243e2 100644 --- a/internal/ebooks/enrichment_queue.go +++ b/internal/ebooks/enrichment_queue.go @@ -177,13 +177,9 @@ var reconcileMissingEnrichmentJobsQuery = ` 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 - ) + OR (membership.first_seen_at, membership.content_id) < ($4, $5) ) - ORDER BY membership.first_seen_at DESC, membership.content_id + ORDER BY membership.first_seen_at DESC, membership.content_id DESC LIMIT $3 ), candidates AS MATERIALIZED ( @@ -210,13 +206,13 @@ var reconcileMissingEnrichmentJobsQuery = ` ( SELECT first_seen_at FROM membership_candidates - ORDER BY first_seen_at, content_id DESC + ORDER BY first_seen_at, content_id LIMIT 1 ) AS last_first_seen_at, ( SELECT content_id FROM membership_candidates - ORDER BY first_seen_at, content_id DESC + ORDER BY first_seen_at, content_id LIMIT 1 ) AS last_content_id FROM membership_candidates @@ -457,15 +453,15 @@ var ( ) const hasReadyEnrichmentJobsQueryTemplate = ` - SELECT EXISTS ( - SELECT 1 + SELECT COALESCE(( + SELECT true FROM ebook_enrichment_state 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 - ) + ), false) ` var ( diff --git a/internal/ebooks/enrichment_queue_test.go b/internal/ebooks/enrichment_queue_test.go index fd30ba4f..332a4e57 100644 --- a/internal/ebooks/enrichment_queue_test.go +++ b/internal/ebooks/enrichment_queue_test.go @@ -135,7 +135,7 @@ func TestEnrichmentQueueReadyCountUsesTheSameLiteralLaneAsClaims(t *testing.T) { } } -func TestEnrichmentQueueHasReadyUsesBoundedExistenceQueries(t *testing.T) { +func TestEnrichmentQueueHasReadyUsesBoundedOrderedScalarQueries(t *testing.T) { for _, tt := range []struct { name string query string @@ -147,17 +147,22 @@ func TestEnrichmentQueueHasReadyUsesBoundedExistenceQueries(t *testing.T) { t.Run(tt.name, func(t *testing.T) { query := strings.Join(strings.Fields(tt.query), " ") for _, fragment := range []string{ - "SELECT EXISTS", + "SELECT COALESCE((", + "SELECT true", "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", + "), false)", } { if !strings.Contains(query, fragment) { t.Fatalf("has-ready query missing %q:\n%s", fragment, tt.query) } } + if strings.Contains(query, "EXISTS") { + t.Fatalf("has-ready query uses EXISTS, which lets PostgreSQL discard ordering:\n%s", tt.query) + } if strings.Contains(query, "COUNT(") { t.Fatalf("has-ready query performs an exact count:\n%s", tt.query) } @@ -171,9 +176,8 @@ func TestEnrichmentQueueReconcileMissingIsBoundedAndLaneSafe(t *testing.T) { "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", + "(membership.first_seen_at, membership.content_id) < ($4, $5)", + "ORDER BY membership.first_seen_at DESC, membership.content_id DESC", "LIMIT $3", "FROM membership_candidates candidate", "mi.type = 'ebook'", diff --git a/migrations/ebook_enrichment_reconcile_cursor_test.go b/migrations/ebook_enrichment_reconcile_cursor_test.go index 88f479f4..e7e9aec1 100644 --- a/migrations/ebook_enrichment_reconcile_cursor_test.go +++ b/migrations/ebook_enrichment_reconcile_cursor_test.go @@ -19,7 +19,7 @@ func TestEbookEnrichmentReconcileCursorPersistsAcrossRestartsAndUsesCoveringInde "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)", + "ON public.media_item_libraries (media_folder_id, first_seen_at DESC, content_id DESC)", "DROP INDEX CONCURRENTLY IF EXISTS idx_item_libraries_folder_enrichment_cursor", "DROP TABLE IF EXISTS ebook_enrichment_reconcile_cursors", } { diff --git a/migrations/sql/20260719173000_ebook_enrichment_reconcile_cursor.sql b/migrations/sql/20260719173000_ebook_enrichment_reconcile_cursor.sql index bea4b8ca..00414708 100644 --- a/migrations/sql/20260719173000_ebook_enrichment_reconcile_cursor.sql +++ b/migrations/sql/20260719173000_ebook_enrichment_reconcile_cursor.sql @@ -30,7 +30,7 @@ $$; -- +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); + ON public.media_item_libraries (media_folder_id, first_seen_at DESC, content_id DESC); -- +goose Down DROP INDEX CONCURRENTLY IF EXISTS idx_item_libraries_folder_enrichment_cursor;