From 585eb7612c57128e00f7c834f7b2c319d3aa525d Mon Sep 17 00:00:00 2001 From: blurbery <271863866+blurbery@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:16:28 +1100 Subject: [PATCH 1/2] perf(metadata): stop the artwork GC sweep rewriting referenced revisions The artwork revision GC's dormant sweep re-verifies parked revisions so a reference that disappears through a surface without a displacement trigger still becomes collectible. It picked rows by updated_at and re-stamped updated_at on every row still referenced, so each parked row was rewritten once a day. updated_at is the key of the sweep's own index, so none of those updates could be HOT and each added entries to all seven indexes. On a server with about 110k parked revisions that was 258k row rewrites and 753 MB of WAL in three days, for rows that didn't change. The sweep now walks parked rows in id order from a cursor persisted in artwork_revision_gc_dormant_cursor and writes only the rows it requeues. It still skips rows changed within the recheck interval, and starts a new pass at most once per interval. The cursor only advances from the position a sweep read, so concurrent sweeps repeat work at worst. The (updated_at, id) index existed only for the old ordering and is dropped. --- internal/metadata/artwork_revision_gc.go | 105 ++++++---- internal/metadata/artwork_revision_gc_test.go | 183 ++++++++++++++++++ ...artwork_revision_gc_dormant_cursor_test.go | 31 +++ ...147_artwork_revision_gc_dormant_cursor.sql | 35 ++++ scripts/ci/db-contracts.txt | 2 + 5 files changed, 317 insertions(+), 39 deletions(-) create mode 100644 migrations/artwork_revision_gc_dormant_cursor_test.go create mode 100644 migrations/sql/20261008041147_artwork_revision_gc_dormant_cursor.sql diff --git a/internal/metadata/artwork_revision_gc.go b/internal/metadata/artwork_revision_gc.go index 536eb29365..62ef25cda3 100644 --- a/internal/metadata/artwork_revision_gc.go +++ b/internal/metadata/artwork_revision_gc.go @@ -20,10 +20,10 @@ import ( const ( artworkRevisionGCBatchSize = 10000 artworkRevisionGCLease = 15 * time.Minute - // artworkRevisionDormantRecheck bounds how stale a parked (referenced) - // revision may get before the sweep re-verifies it. Displacement triggers - // are the fast path; the sweep guarantees a reference that disappears - // through an untriggered surface still becomes collectible eventually. + // artworkRevisionDormantRecheck is how often the sweep starts another pass + // over the parked (referenced) revisions. Displacement triggers are the fast + // path; the sweep guarantees a reference that disappears through an + // untriggered surface still becomes collectible eventually. artworkRevisionDormantRecheck = 24 * time.Hour ) @@ -457,16 +457,35 @@ func referencedArtworkPaths(ctx context.Context, q interface { } // sweepDormant re-verifies a bounded batch of parked revisions whose last -// check is older than the recheck interval, re-arming any that lost every -// reference through a surface without a displacement trigger. +// change is older than the recheck interval, re-arming any that lost every +// reference through a surface without a displacement trigger. A persisted +// keyset cursor walks the parked rows in id order, so a still referenced row +// is read once per cycle but never rewritten. A new cycle starts at most once +// per recheck interval. func (g *ArtworkRevisionGarbageCollector) sweepDormant(ctx context.Context, limit int) (checked, requeued int, err error) { + var afterID int64 + var cycleDue bool + err = g.pool.QueryRow(ctx, ` + SELECT after_id, cycle_started_at <= NOW() - ($1 * interval '1 second') + FROM artwork_revision_gc_dormant_cursor`, + int64(artworkRevisionDormantRecheck/time.Second)).Scan(&afterID, &cycleDue) + if errors.Is(err, pgx.ErrNoRows) { + afterID, cycleDue, err = 0, true, nil + } + if err != nil { + return 0, 0, fmt.Errorf("artwork revision GC: read dormant cursor: %w", err) + } + if afterID == 0 && !cycleDue { + return 0, 0, nil + } + rows, err := g.pool.Query(ctx, ` SELECT id, original_path FROM artwork_revision_gc_candidates - WHERE next_attempt_at IS NULL - AND updated_at < NOW() - ($2 * interval '1 second') - ORDER BY updated_at, id - LIMIT $1`, limit, int64(artworkRevisionDormantRecheck/time.Second)) + WHERE next_attempt_at IS NULL AND id > $1 + AND updated_at < NOW() - ($3 * interval '1 second') + ORDER BY id + LIMIT $2`, afterID, limit, int64(artworkRevisionDormantRecheck/time.Second)) if err != nil { return 0, 0, fmt.Errorf("artwork revision GC: list dormant revisions: %w", err) } @@ -474,6 +493,7 @@ func (g *ArtworkRevisionGarbageCollector) sweepDormant(ctx context.Context, limi ids := make(map[string][]int64) var paths []string + lastID := afterID for rows.Next() { var id int64 var path string @@ -484,45 +504,52 @@ func (g *ArtworkRevisionGarbageCollector) sweepDormant(ctx context.Context, limi paths = append(paths, path) } ids[path] = append(ids[path], id) + lastID = id checked++ } if err := rows.Err(); err != nil { return 0, 0, fmt.Errorf("artwork revision GC: dormant revisions: %w", err) } - if checked == 0 { - return 0, 0, nil - } - referenced, err := g.referencedPaths(ctx, paths) - if err != nil { - return checked, 0, err - } - var touch, requeue []int64 - for path, pathIDs := range ids { - if _, ok := referenced[path]; ok { - touch = append(touch, pathIDs...) - continue + if checked > 0 { + referenced, err := g.referencedPaths(ctx, paths) + if err != nil { + return checked, 0, err } - requeue = append(requeue, pathIDs...) - } - if len(requeue) > 0 { - if _, err := g.pool.Exec(ctx, ` - UPDATE artwork_revision_gc_candidates - SET next_attempt_at = GREATEST(not_before, NOW()), - updated_at = NOW() - WHERE id = ANY($1) AND next_attempt_at IS NULL`, requeue); err != nil { - return checked, 0, fmt.Errorf("artwork revision GC: requeue dormant revisions: %w", err) + var requeue []int64 + for path, pathIDs := range ids { + if _, ok := referenced[path]; !ok { + requeue = append(requeue, pathIDs...) + } } - requeued = len(requeue) - } - if len(touch) > 0 { - if _, err := g.pool.Exec(ctx, ` - UPDATE artwork_revision_gc_candidates - SET updated_at = NOW() - WHERE id = ANY($1) AND next_attempt_at IS NULL`, touch); err != nil { - return checked, requeued, fmt.Errorf("artwork revision GC: touch dormant revisions: %w", err) + if len(requeue) > 0 { + if _, err := g.pool.Exec(ctx, ` + UPDATE artwork_revision_gc_candidates + SET next_attempt_at = GREATEST(not_before, NOW()), + updated_at = NOW() + WHERE id = ANY($1) AND next_attempt_at IS NULL`, requeue); err != nil { + return checked, 0, fmt.Errorf("artwork revision GC: requeue dormant revisions: %w", err) + } + requeued = len(requeue) } } + + // A short batch ends the cycle. The cursor only moves from the position + // this sweep read, so a concurrent sweep of the same batch is harmless. + nextID := lastID + if checked < limit { + nextID = 0 + } + if _, err := g.pool.Exec(ctx, ` + INSERT INTO artwork_revision_gc_dormant_cursor AS c (singleton, after_id, cycle_started_at, updated_at) + VALUES (true, $2, NOW(), NOW()) + ON CONFLICT (singleton) DO UPDATE SET + after_id = EXCLUDED.after_id, + cycle_started_at = CASE WHEN c.after_id = 0 THEN NOW() ELSE c.cycle_started_at END, + updated_at = NOW() + WHERE c.after_id = $1`, afterID, nextID); err != nil { + return checked, requeued, fmt.Errorf("artwork revision GC: advance dormant cursor: %w", err) + } return checked, requeued, nil } diff --git a/internal/metadata/artwork_revision_gc_test.go b/internal/metadata/artwork_revision_gc_test.go index 9cd5a7510c..e571c8b1c9 100644 --- a/internal/metadata/artwork_revision_gc_test.go +++ b/internal/metadata/artwork_revision_gc_test.go @@ -867,6 +867,9 @@ func TestArtworkRevisionGCDormantSweep(t *testing.T) { _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = $1`, referencedContentID) _, _ = pool.Exec(ctx, `DELETE FROM artwork_revision_gc_candidates WHERE original_path = ANY($1)`, []string{referencedPath, orphanPath}) }) + // Start a sweep cycle just before both rows. + placeDormantCursor(t, pool, dormantCandidateID(t, pool, referencedPath)-1) + referencedVersion := dormantCandidateVersion(t, pool, referencedPath) deleter := &blockingArtworkRevisionDeleter{started: make(chan struct{})} collector := NewArtworkRevisionGarbageCollector(pool, deleter) @@ -896,6 +899,186 @@ func TestArtworkRevisionGCDormantSweep(t *testing.T) { if referencedNext != nil { t.Fatalf("referenced dormant row was re-armed: %v", *referencedNext) } + if got := dormantCandidateVersion(t, pool, referencedPath); got != referencedVersion { + t.Fatalf("referenced dormant row was rewritten: row version %s, want %s", got, referencedVersion) + } +} + +// TestArtworkRevisionGCDormantSweepCycles covers the persisted cursor: each +// sweep continues where the last one stopped, a row behind the cursor waits +// for the next cycle, and a new cycle starts only once the recheck interval +// has passed since the last one started. +func TestArtworkRevisionGCDormantSweepCycles(t *testing.T) { + pool := artworkRevisionGCTestPool(t) + ctx := context.Background() + suffix := time.Now().UnixNano() + referencedContentID := fmt.Sprintf("gc-dormant-cycle-ref-%d", suffix) + orphanPath := fmt.Sprintf("tmdb/movies/%d/poster/original.cycle-orphan.webp", suffix) + referencedPath := fmt.Sprintf("tmdb/movies/%d/poster/original.cycle-ref.webp", suffix) + if _, err := pool.Exec(ctx, ` + INSERT INTO media_items (content_id, type, title, status, genres, poster_path) + VALUES ($1, 'movie', 'GC Dormant Cycle Ref', 'matched', '{}'::text[], $2)`, referencedContentID, referencedPath); err != nil { + t.Fatalf("seed referenced item: %v", err) + } + // The orphan gets the lower id, so a cursor can stand between the two. + for _, path := range []string{orphanPath, referencedPath} { + if _, err := pool.Exec(ctx, ` + INSERT INTO artwork_revision_gc_candidates ( + original_path, image_type, object_keys, not_before, next_attempt_at + ) VALUES ($1, 'poster', '{}', NOW() - interval '2 days', NULL)`, path); err != nil { + t.Fatalf("seed dormant candidate: %v", err) + } + } + t.Cleanup(func() { + _, _ = pool.Exec(ctx, `DELETE FROM media_items WHERE content_id = $1`, referencedContentID) + _, _ = pool.Exec(ctx, `DELETE FROM artwork_revision_gc_candidates WHERE original_path = ANY($1)`, []string{orphanPath, referencedPath}) + }) + if _, err := pool.Exec(ctx, ` + UPDATE artwork_revision_gc_candidates + SET updated_at = NOW() - interval '2 days' + WHERE original_path = ANY($1)`, []string{orphanPath, referencedPath}); err != nil { + t.Fatalf("age dormant candidates: %v", err) + } + orphanID := dormantCandidateID(t, pool, orphanPath) + referencedID := dormantCandidateID(t, pool, referencedPath) + if referencedID <= orphanID { + t.Fatalf("referenced id %d is not above orphan id %d", referencedID, orphanID) + } + collector := NewArtworkRevisionGarbageCollector(pool, &blockingArtworkRevisionDeleter{started: make(chan struct{})}) + orphanParked := func() bool { + t.Helper() + var next *time.Time + if err := pool.QueryRow(ctx, `SELECT next_attempt_at FROM artwork_revision_gc_candidates WHERE id = $1`, orphanID).Scan(&next); err != nil { + t.Fatalf("load orphan candidate: %v", err) + } + return next == nil + } + + // A bounded sweep stops at its last row and the next one resumes there. + placeDormantCursor(t, pool, referencedID-1) + if checked, _, err := collector.sweepDormant(ctx, 1); err != nil || checked != 1 { + t.Fatalf("bounded sweep checked %d rows, err %v; want 1", checked, err) + } + if afterID, _ := dormantCursor(t, pool); afterID != referencedID { + t.Fatalf("cursor after bounded sweep = %d, want %d", afterID, referencedID) + } + + // The orphan sits behind the cursor, so this cycle passes it by and ends. + placeDormantCursor(t, pool, orphanID) + if _, _, err := collector.sweepDormant(ctx, artworkRevisionGCBatchSize); err != nil { + t.Fatalf("sweep rest of cycle: %v", err) + } + afterID, cycleAge := dormantCursor(t, pool) + if afterID != 0 || cycleAge > time.Minute { + t.Fatalf("cursor after a short batch = %d (cycle started %s ago), want a fresh cycle at 0", afterID, cycleAge) + } + if !orphanParked() { + t.Fatal("sweep re-armed a row behind its cursor") + } + + // The next cycle waits for the recheck interval. + if checked, _, err := collector.sweepDormant(ctx, artworkRevisionGCBatchSize); err != nil || checked != 0 { + t.Fatalf("sweep before the next cycle is due checked %d rows, err %v; want 0", checked, err) + } + if !orphanParked() { + t.Fatal("orphan re-armed before the next cycle") + } + if _, err := pool.Exec(ctx, ` + UPDATE artwork_revision_gc_dormant_cursor + SET cycle_started_at = NOW() - $1 * interval '1 second' - interval '1 minute'`, + int64(artworkRevisionDormantRecheck/time.Second)); err != nil { + t.Fatalf("age dormant cycle: %v", err) + } + if _, _, err := collector.sweepDormant(ctx, artworkRevisionGCBatchSize); err != nil { + t.Fatalf("sweep next cycle: %v", err) + } + if _, cycleAge := dormantCursor(t, pool); cycleAge > time.Minute { + t.Fatalf("new cycle started %s ago, want it recorded as starting now", cycleAge) + } + // Finish the cycle, however many parked rows sort before the orphan. + for range 100 { + if afterID, _ := dormantCursor(t, pool); afterID == 0 || !orphanParked() { + break + } + if _, _, err := collector.sweepDormant(ctx, artworkRevisionGCBatchSize); err != nil { + t.Fatalf("sweep next cycle: %v", err) + } + } + if orphanParked() { + t.Fatal("next cycle did not re-arm the orphan") + } + + // A missing cursor row reads as a new cycle and is recreated. + if _, err := pool.Exec(ctx, `DELETE FROM artwork_revision_gc_dormant_cursor`); err != nil { + t.Fatalf("delete dormant cursor: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `INSERT INTO artwork_revision_gc_dormant_cursor DEFAULT VALUES ON CONFLICT (singleton) DO NOTHING`) + }) + if _, _, err := collector.sweepDormant(ctx, 1); err != nil { + t.Fatalf("sweep without a cursor row: %v", err) + } + if afterID, cycleAge := dormantCursor(t, pool); cycleAge > time.Minute { + t.Fatalf("recreated cursor = %d (cycle started %s ago), want a cycle that started now", afterID, cycleAge) + } +} + +func dormantCandidateID(t *testing.T, pool *pgxpool.Pool, path string) int64 { + t.Helper() + var id int64 + if err := pool.QueryRow(context.Background(), ` + SELECT id FROM artwork_revision_gc_candidates WHERE original_path = $1`, path).Scan(&id); err != nil { + t.Fatalf("load candidate id: %v", err) + } + return id +} + +// dormantCandidateVersion identifies the row version: any UPDATE writes a new +// one with a new ctid and xmin. +func dormantCandidateVersion(t *testing.T, pool *pgxpool.Pool, path string) string { + t.Helper() + var version string + if err := pool.QueryRow(context.Background(), ` + SELECT ctid::text || '/' || xmin::text FROM artwork_revision_gc_candidates WHERE original_path = $1`, path).Scan(&version); err != nil { + t.Fatalf("load candidate version: %v", err) + } + return version +} + +// placeDormantCursor points the dormant sweep's cursor at afterID and restores +// the shared cursor row when the test ends. Position 0 has a new cycle due; any +// other position is part of a cycle that started now. +func placeDormantCursor(t *testing.T, pool *pgxpool.Pool, afterID int64) { + t.Helper() + ctx := context.Background() + var savedAfterID int64 + var savedStarted string + if err := pool.QueryRow(ctx, ` + SELECT after_id, cycle_started_at::text FROM artwork_revision_gc_dormant_cursor`).Scan(&savedAfterID, &savedStarted); err != nil { + t.Fatalf("read dormant cursor: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), ` + UPDATE artwork_revision_gc_dormant_cursor SET after_id = $1, cycle_started_at = $2::timestamptz`, savedAfterID, savedStarted) + }) + if _, err := pool.Exec(ctx, ` + UPDATE artwork_revision_gc_dormant_cursor + SET after_id = $1::bigint, cycle_started_at = CASE WHEN $1::bigint = 0 THEN '-infinity' ELSE NOW() END`, afterID); err != nil { + t.Fatalf("place dormant cursor: %v", err) + } +} + +// dormantCursor returns the cursor position and how long ago its cycle began. +func dormantCursor(t *testing.T, pool *pgxpool.Pool) (int64, time.Duration) { + t.Helper() + var afterID int64 + var ageSeconds float64 + if err := pool.QueryRow(context.Background(), ` + SELECT after_id, LEAST(EXTRACT(EPOCH FROM NOW() - cycle_started_at), 1e9)::float8 + FROM artwork_revision_gc_dormant_cursor`).Scan(&afterID, &ageSeconds); err != nil { + t.Fatalf("read dormant cursor: %v", err) + } + return afterID, time.Duration(ageSeconds * float64(time.Second)) } type callbackArtworkRevisionDeleter struct { diff --git a/migrations/artwork_revision_gc_dormant_cursor_test.go b/migrations/artwork_revision_gc_dormant_cursor_test.go new file mode 100644 index 0000000000..5158bf985a --- /dev/null +++ b/migrations/artwork_revision_gc_dormant_cursor_test.go @@ -0,0 +1,31 @@ +package migrations + +import ( + "strings" + "testing" +) + +func TestArtworkRevisionGCDormantCursorMigrationChangesIndexConcurrently(t *testing.T) { + migrationBytes, err := FS.ReadFile("sql/20261008041147_artwork_revision_gc_dormant_cursor.sql") + if err != nil { + t.Fatalf("read migration: %v", err) + } + migration := string(migrationBytes) + if !strings.HasPrefix(migration, "-- +goose NO TRANSACTION") { + t.Fatal("concurrent index changes require NO TRANSACTION") + } + up, down, found := strings.Cut(migration, "-- +goose Down") + if !found { + t.Fatal("migration has no down marker") + } + up, down = strings.Join(strings.Fields(up), " "), strings.Join(strings.Fields(down), " ") + const drop = "DROP INDEX CONCURRENTLY IF EXISTS public.artwork_revision_gc_dormant_idx;" + if !strings.Contains(up, drop) { + t.Fatalf("up must drop the old dormant index concurrently: %q", drop) + } + create := "CREATE INDEX CONCURRENTLY artwork_revision_gc_dormant_idx ON public.artwork_revision_gc_candidates (updated_at, id) WHERE next_attempt_at IS NULL;" + dropAt, createAt := strings.Index(down, drop), strings.Index(down, create) + if dropAt < 0 || createAt < 0 || dropAt > createAt { + t.Fatalf("down must clean up a failed build concurrently before %q", create) + } +} diff --git a/migrations/sql/20261008041147_artwork_revision_gc_dormant_cursor.sql b/migrations/sql/20261008041147_artwork_revision_gc_dormant_cursor.sql new file mode 100644 index 0000000000..01dcd0c7ba --- /dev/null +++ b/migrations/sql/20261008041147_artwork_revision_gc_dormant_cursor.sql @@ -0,0 +1,35 @@ +-- +goose NO TRANSACTION + +-- +goose Up +-- The artwork revision GC's dormant sweep re-verifies parked revisions. It +-- used to re-stamp every still-referenced row's updated_at once a day and +-- picked rows by an (updated_at, id) index, so each check rewrote the row and +-- every index entry. It now walks parked rows in id order from a persisted +-- cursor and writes only the rows it requeues. after_id = 0 means no cycle is +-- in progress; a new cycle starts once cycle_started_at is a recheck interval +-- old. +CREATE TABLE IF NOT EXISTS public.artwork_revision_gc_dormant_cursor ( + singleton boolean PRIMARY KEY DEFAULT true, + after_id bigint NOT NULL DEFAULT 0, + cycle_started_at timestamptz NOT NULL DEFAULT '-infinity', + updated_at timestamptz NOT NULL DEFAULT NOW(), + CONSTRAINT artwork_revision_gc_dormant_cursor_singleton CHECK (singleton), + CONSTRAINT artwork_revision_gc_dormant_cursor_after_id_check CHECK (after_id >= 0) +); + +INSERT INTO public.artwork_revision_gc_dormant_cursor DEFAULT VALUES +ON CONFLICT (singleton) DO NOTHING; + +-- Only the old sweep ordered by updated_at. +DROP INDEX CONCURRENTLY IF EXISTS public.artwork_revision_gc_dormant_idx; + +-- +goose Down +-- A failed concurrent build can leave an invalid index. Keep retry cleanup +-- outside a transaction so it cannot take a blocking ordinary index-drop lock. +DROP INDEX CONCURRENTLY IF EXISTS public.artwork_revision_gc_dormant_idx; + +CREATE INDEX CONCURRENTLY artwork_revision_gc_dormant_idx + ON public.artwork_revision_gc_candidates (updated_at, id) + WHERE next_attempt_at IS NULL; + +DROP TABLE IF EXISTS public.artwork_revision_gc_dormant_cursor; diff --git a/scripts/ci/db-contracts.txt b/scripts/ci/db-contracts.txt index c6a94e2e66..1f0980c2ba 100644 --- a/scripts/ci/db-contracts.txt +++ b/scripts/ci/db-contracts.txt @@ -67,6 +67,8 @@ ./internal/metadata TestArtworkRevisionGCBatchHealRechecksRetrackBetweenSurfaces ./internal/metadata TestArtworkRevisionGCRunRetainsPartialDeletionForHealing ./internal/metadata TestArtworkRevisionGCBatchRechecksOwnershipAndReferences +./internal/metadata TestArtworkRevisionGCDormantSweep +./internal/metadata TestArtworkRevisionGCDormantSweepCycles # Persisted notification, request, marker and preview contracts. ./internal/markers TestPopulationSyncResumesAfterQuotaAcrossRuns From 96505625b3e44d353cb0bb436be395e980a3b053 Mon Sep 17 00:00:00 2001 From: Quick104 <31828688+Quick104@users.noreply.github.com> Date: Sat, 10 Oct 2026 00:47:40 -0400 Subject: [PATCH 2/2] fix(metadata): end each dormant sweep cycle at rows that predate it A cycle ended only on a short batch, and the batch query accepted any parked row past the recheck interval above the cursor, including rows parked after the cycle started. With a full batch of such rows maturing every run, the cursor never wrapped and rows behind it were never rechecked. Limit each cycle to rows that last changed before it began. Co-Authored-By: Claude Opus 5.5 --- internal/metadata/artwork_revision_gc.go | 15 +++--- internal/metadata/artwork_revision_gc_test.go | 51 +++++++++++++++++++ scripts/ci/db-contracts.txt | 1 + 3 files changed, 61 insertions(+), 6 deletions(-) diff --git a/internal/metadata/artwork_revision_gc.go b/internal/metadata/artwork_revision_gc.go index 62ef25cda3..88cc5f3875 100644 --- a/internal/metadata/artwork_revision_gc.go +++ b/internal/metadata/artwork_revision_gc.go @@ -461,16 +461,19 @@ func referencedArtworkPaths(ctx context.Context, q interface { // reference through a surface without a displacement trigger. A persisted // keyset cursor walks the parked rows in id order, so a still referenced row // is read once per cycle but never rewritten. A new cycle starts at most once -// per recheck interval. +// per recheck interval and covers only rows that last changed before it +// started, so rows that keep maturing above the cursor cannot hold it open. func (g *ArtworkRevisionGarbageCollector) sweepDormant(ctx context.Context, limit int) (checked, requeued int, err error) { var afterID int64 var cycleDue bool + var cycleStartedAt *time.Time err = g.pool.QueryRow(ctx, ` - SELECT after_id, cycle_started_at <= NOW() - ($1 * interval '1 second') + SELECT after_id, cycle_started_at <= NOW() - ($1 * interval '1 second'), + CASE WHEN after_id > 0 THEN cycle_started_at END FROM artwork_revision_gc_dormant_cursor`, - int64(artworkRevisionDormantRecheck/time.Second)).Scan(&afterID, &cycleDue) + int64(artworkRevisionDormantRecheck/time.Second)).Scan(&afterID, &cycleDue, &cycleStartedAt) if errors.Is(err, pgx.ErrNoRows) { - afterID, cycleDue, err = 0, true, nil + afterID, cycleDue, cycleStartedAt, err = 0, true, nil, nil } if err != nil { return 0, 0, fmt.Errorf("artwork revision GC: read dormant cursor: %w", err) @@ -483,9 +486,9 @@ func (g *ArtworkRevisionGarbageCollector) sweepDormant(ctx context.Context, limi SELECT id, original_path FROM artwork_revision_gc_candidates WHERE next_attempt_at IS NULL AND id > $1 - AND updated_at < NOW() - ($3 * interval '1 second') + AND updated_at < LEAST(NOW() - ($3 * interval '1 second'), $4::timestamptz) ORDER BY id - LIMIT $2`, afterID, limit, int64(artworkRevisionDormantRecheck/time.Second)) + LIMIT $2`, afterID, limit, int64(artworkRevisionDormantRecheck/time.Second), cycleStartedAt) if err != nil { return 0, 0, fmt.Errorf("artwork revision GC: list dormant revisions: %w", err) } diff --git a/internal/metadata/artwork_revision_gc_test.go b/internal/metadata/artwork_revision_gc_test.go index e571c8b1c9..1b676a4495 100644 --- a/internal/metadata/artwork_revision_gc_test.go +++ b/internal/metadata/artwork_revision_gc_test.go @@ -1023,6 +1023,57 @@ func TestArtworkRevisionGCDormantSweepCycles(t *testing.T) { } } +// TestArtworkRevisionGCDormantCycleEndsWhileRowsArrive keeps a full batch of +// parked rows maturing above the cursor during a long cycle. The cycle must +// still end, or the rows behind the cursor are never checked again. +func TestArtworkRevisionGCDormantCycleEndsWhileRowsArrive(t *testing.T) { + pool := artworkRevisionGCTestPool(t) + ctx := context.Background() + prefix := fmt.Sprintf("tmdb/movies/%d/poster/original.arrival-", time.Now().UnixNano()) + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM artwork_revision_gc_candidates WHERE original_path LIKE $1`, prefix+"%") + }) + // seed parks a row that last changed age ago. + seed := func(n int, age string) int64 { + t.Helper() + path := fmt.Sprintf("%s%d.webp", prefix, n) + if _, err := pool.Exec(ctx, ` + INSERT INTO artwork_revision_gc_candidates ( + original_path, image_type, object_keys, not_before, next_attempt_at + ) VALUES ($1, 'poster', '{}', NOW() - interval '5 days', NULL)`, path); err != nil { + t.Fatalf("seed dormant candidate: %v", err) + } + if _, err := pool.Exec(ctx, ` + UPDATE artwork_revision_gc_candidates SET updated_at = NOW() - $2::interval + WHERE original_path = $1`, path, age); err != nil { + t.Fatalf("age dormant candidate: %v", err) + } + return dormantCandidateID(t, pool, path) + } + // Two rows parked before a cycle that started three days ago. + first := seed(0, "4 days") + seed(1, "4 days") + placeDormantCursor(t, pool, first-1) + if _, err := pool.Exec(ctx, ` + UPDATE artwork_revision_gc_dormant_cursor SET cycle_started_at = NOW() - interval '3 days'`); err != nil { + t.Fatalf("age dormant cycle: %v", err) + } + collector := NewArtworkRevisionGarbageCollector(pool, &blockingArtworkRevisionDeleter{started: make(chan struct{})}) + + // Each sweep reads one row while another row, parked during the cycle and + // now past the recheck interval, lands above the cursor. + for n := 2; n < 12; n++ { + if _, _, err := collector.sweepDormant(ctx, 1); err != nil { + t.Fatalf("sweep dormant: %v", err) + } + if afterID, _ := dormantCursor(t, pool); afterID == 0 { + return + } + seed(n, "2 days") + } + t.Fatal("dormant cycle never ended while parked rows kept arriving above its cursor") +} + func dormantCandidateID(t *testing.T, pool *pgxpool.Pool, path string) int64 { t.Helper() var id int64 diff --git a/scripts/ci/db-contracts.txt b/scripts/ci/db-contracts.txt index 1f0980c2ba..35e5c1e236 100644 --- a/scripts/ci/db-contracts.txt +++ b/scripts/ci/db-contracts.txt @@ -69,6 +69,7 @@ ./internal/metadata TestArtworkRevisionGCBatchRechecksOwnershipAndReferences ./internal/metadata TestArtworkRevisionGCDormantSweep ./internal/metadata TestArtworkRevisionGCDormantSweepCycles +./internal/metadata TestArtworkRevisionGCDormantCycleEndsWhileRowsArrive # Persisted notification, request, marker and preview contracts. ./internal/markers TestPopulationSyncResumesAfterQuotaAcrossRuns