haeaeae/silo-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ebooks): use planner-safe queue cursors

Browse Source
This commit is contained in:
rxwatcher
2026-07-22 13:43:57 +02:00
parent fcaee9bfce
commit 794e2d786a
4 changed files with 18 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/ebooks/enrichment_queue.go
+7 -11
View File
@@ -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 (
internal/ebooks/enrichment_queue_test.go
+9 -5
View File
@@ -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'",
migrations/ebook_enrichment_reconcile_cursor_test.go
+1 -1
View File
@@ -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",
} {
migrations/sql/20260719173000_ebook_enrichment_reconcile_cursor.sql
+1 -1
View File
@@ -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;