diff --git a/internal/metadata/artwork_revision_gc.go b/internal/metadata/artwork_revision_gc.go index 536eb29365..88cc5f3875 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,38 @@ 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 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'), + CASE WHEN after_id > 0 THEN cycle_started_at END + FROM artwork_revision_gc_dormant_cursor`, + int64(artworkRevisionDormantRecheck/time.Second)).Scan(&afterID, &cycleDue, &cycleStartedAt) + if errors.Is(err, pgx.ErrNoRows) { + afterID, cycleDue, cycleStartedAt, err = 0, true, nil, 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 < LEAST(NOW() - ($3 * interval '1 second'), $4::timestamptz) + ORDER BY id + 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) } @@ -474,6 +496,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 +507,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..1b676a4495 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,237 @@ 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) + } +} + +// 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 + 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..35e5c1e236 100644 --- a/scripts/ci/db-contracts.txt +++ b/scripts/ci/db-contracts.txt @@ -67,6 +67,9 @@ ./internal/metadata TestArtworkRevisionGCBatchHealRechecksRetrackBetweenSurfaces ./internal/metadata TestArtworkRevisionGCRunRetainsPartialDeletionForHealing ./internal/metadata TestArtworkRevisionGCBatchRechecksOwnershipAndReferences +./internal/metadata TestArtworkRevisionGCDormantSweep +./internal/metadata TestArtworkRevisionGCDormantSweepCycles +./internal/metadata TestArtworkRevisionGCDormantCycleEndsWhileRowsArrive # Persisted notification, request, marker and preview contracts. ./internal/markers TestPopulationSyncResumesAfterQuotaAcrossRuns