fix: a fresh app install syncs the whole library, not the last 30 days of changes (#6778)

The change feed dropped every row older than 30 days, and an app syncing
from cursor 0 builds its library from the feed, so a fresh install, cleared
data or a 410 reset got only what had changed lately.

The compactor now keeps each entity's latest change however old. It drops a
row once a newer one for the same entity exists, and a delete once it is 30
days old, raising a stored mark (library_changes_trimmed) past it; an app
whose cursor is below the mark is sent back to 0, as before. Migration 0101
adds the index the compactor needs, seeds the mark where the old compactor
left off, and gives every artist, album and track whose latest row is not an
upsert a fresh one, so the rows the old compactor dropped come back. The
backup carries the mark with the feed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/api/library_sync.go

@@ -36,7 +36,9 @@ type syncResponse struct {
 // Responses:
 //   - 204 No Content — no changes since cursor
 //   - 200 OK         — JSON syncResponse with upserts, deletes, new cursor
-//   - 410 Gone       — cursor older than oldest log row; client must reset
+//   - 410 Gone       — the client must reset its cursor to 0: the cursor is
+//     older than a delete the log has dropped, or past anything this server
+//     has handed out
 //   - 401 / 500      — standard envelope errors
 func (h *handlers) handleLibrarySync(w http.ResponseWriter, r *http.Request) {
 	user, ok := requireUser(w, r)
@@ -52,16 +54,17 @@ func (h *handlers) handleLibrarySync(w http.ResponseWriter, r *http.Request) {
 
 	q := dbq.New(h.pool)
 
-	// 410 path: if the requested cursor is older than the oldest row,
+	// The log keeps every entity's latest change, so a client syncing from
-	// tell the client to reset. (No compaction job in this slice; the
+	// 0 gets the whole library. It drops deletes older than its retention,
-	// path exists for future compaction support.)
+	// and a cursor below the newest dropped one may have missed it: the
+	// client resets and syncs again from 0.
 	if since > 0 {
-		minCursor, err := q.GetMinLibraryChangeCursor(ctx)
+		trimmed, err := q.GetLibraryChangesTrimmedThrough(ctx)
 		if err != nil {
-			writeErrWithLog(w, h.logger, "library_sync: GetMinLibraryChangeCursor", err)
+			writeErrWithLog(w, h.logger, "library_sync: GetLibraryChangesTrimmedThrough", err)
 			return
 		}
-		if minCursor > 0 && since+1 < minCursor {
+		if since < trimmed {
 			writeErr(w, &apierror.Error{
 				Status:  http.StatusGone,
 				Code:    "cursor_too_old",

internal/api/library_sync_test.go

@@ -250,3 +250,38 @@ func TestLibrarySync_ACursorPastTheServersLastChangeIsReset(t *testing.T) {
 		t.Errorf("the last cursor after compaction: status = %d body = %s, want 204", w.Code, w.Body.String())
 	}
 }
+
+// An app is sent back to 0 only when its cursor is below a delete the log has
+// dropped, which it may have missed. A cursor at that delete has seen it, and
+// one between dropped and kept rows replays what is still there.
+func TestLibrarySync_ACursorBelowADroppedDeleteIsReset(t *testing.T) {
+	h, pool := testHandlers(t)
+	truncateLibrary(t, pool)
+	u := seedUser(t, pool, "alice", "x", false)
+	ctx := context.Background()
+
+	for i := 0; i < 5; i++ {
+		if _, err := pool.Exec(ctx, `INSERT INTO library_changes (entity_type, entity_id, op) VALUES ('artist', $1, 'upsert')`,
+			strconv.Itoa(i)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var dropped int64
+	if err := pool.QueryRow(ctx, `SELECT min(id) + 2 FROM library_changes`).Scan(&dropped); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE library_changes_trimmed SET through = $1`, dropped); err != nil {
+		t.Fatal(err)
+	}
+
+	if w := callLibrarySync(h, u, strconv.FormatInt(dropped-1, 10)); w.Code != http.StatusGone ||
+		!strings.Contains(w.Body.String(), "cursor_too_old") {
+		t.Errorf("a cursor below the dropped delete: status = %d body = %s, want 410 cursor_too_old", w.Code, w.Body.String())
+	}
+	if w := callLibrarySync(h, u, strconv.FormatInt(dropped, 10)); w.Code != http.StatusOK {
+		t.Errorf("a cursor at the dropped delete: status = %d body = %s, want 200", w.Code, w.Body.String())
+	}
+	if w := callLibrarySync(h, u, "0"); w.Code != http.StatusOK {
+		t.Errorf("a sync from 0: status = %d body = %s, want 200", w.Code, w.Body.String())
+	}
+}

internal/backup/tables.go

@@ -111,6 +111,7 @@ var Tables = []Table{
 	// The library change feed is how the apps sync, including a first sync,
 	// so a restored server carries it on from where the backup left it.
 	{Name: "library_changes"},
+	{Name: "library_changes_trimmed"},
 	{Name: "audit_log"},
 	{Name: "playback_errors"},
 }

internal/db/dbq/library_changes.sql.go

@@ -11,15 +11,38 @@ import (
 	"github.com/jackc/pgx/v5/pgtype"
 )
 
-const deleteOldLibraryChanges = `-- name: DeleteOldLibraryChanges :execrows
+const deleteOldLibraryDeletes = `-- name: DeleteOldLibraryDeletes :one
-DELETE FROM library_changes WHERE changed_at < $1
+WITH gone AS (
+    DELETE FROM library_changes WHERE op = 'delete' AND changed_at < $1 RETURNING id
+), mark AS (
+    UPDATE library_changes_trimmed
+       SET through = GREATEST(through, (SELECT COALESCE(max(id), 0) FROM gone))
+)
+SELECT count(*)::BIGINT FROM gone
 `
 
-// Removes library_changes rows older than the given cutoff. Run from
+// Removes deletes older than the cutoff and raises the trimmed mark past the
-// the sync compactor goroutine on a daily tick. Returns the number of
+// highest one, so an app whose cursor is below it starts over. Returns how
-// rows deleted so the worker can log meaningful progress.
+// many went.
-func (q *Queries) DeleteOldLibraryChanges(ctx context.Context, changedAt pgtype.Timestamptz) (int64, error) {
+func (q *Queries) DeleteOldLibraryDeletes(ctx context.Context, changedAt pgtype.Timestamptz) (int64, error) {
-	result, err := q.db.Exec(ctx, deleteOldLibraryChanges, changedAt)
+	row := q.db.QueryRow(ctx, deleteOldLibraryDeletes, changedAt)
+	var column_1 int64
+	err := row.Scan(&column_1)
+	return column_1, err
+}
+
+const deleteSupersededLibraryChanges = `-- name: DeleteSupersededLibraryChanges :execrows
+DELETE FROM library_changes old
+ WHERE EXISTS (SELECT 1 FROM library_changes newer
+                WHERE newer.entity_type = old.entity_type
+                  AND newer.entity_id = old.entity_id
+                  AND newer.id > old.id)
+`
+
+// Removes every row a newer row for the same entity has superseded. An app
+// reading from any cursor still meets the newer row, so it loses nothing.
+func (q *Queries) DeleteSupersededLibraryChanges(ctx context.Context) (int64, error) {
+	result, err := q.db.Exec(ctx, deleteSupersededLibraryChanges)
 	if err != nil {
 		return 0, err
 	}
@@ -79,6 +102,19 @@ func (q *Queries) GetLibraryChangesSince(ctx context.Context, arg GetLibraryChan
 	return items, nil
 }
 
+const getLibraryChangesTrimmedThrough = `-- name: GetLibraryChangesTrimmedThrough :one
+SELECT through FROM library_changes_trimmed
+`
+
+// The highest id of a delete the compactor has dropped. A cursor below it may
+// have missed that delete.
+func (q *Queries) GetLibraryChangesTrimmedThrough(ctx context.Context) (int64, error) {
+	row := q.db.QueryRow(ctx, getLibraryChangesTrimmedThrough)
+	var through int64
+	err := row.Scan(&through)
+	return through, err
+}
+
 const getMaxLibraryChangeCursor = `-- name: GetMaxLibraryChangeCursor :one
 SELECT COALESCE(MAX(id), 0)::BIGINT FROM library_changes
 `
@@ -90,17 +126,6 @@ func (q *Queries) GetMaxLibraryChangeCursor(ctx context.Context) (int64, error)
 	return column_1, err
 }
 
-const getMinLibraryChangeCursor = `-- name: GetMinLibraryChangeCursor :one
-SELECT COALESCE(MIN(id), 0)::BIGINT FROM library_changes
-`
-
-func (q *Queries) GetMinLibraryChangeCursor(ctx context.Context) (int64, error) {
-	row := q.db.QueryRow(ctx, getMinLibraryChangeCursor)
-	var column_1 int64
-	err := row.Scan(&column_1)
-	return column_1, err
-}
-
 const insertLibraryChange = `-- name: InsertLibraryChange :exec
 INSERT INTO library_changes (entity_type, entity_id, op)
 VALUES ($1, $2, $3)

internal/db/dbq/models.go

@@ -544,6 +544,11 @@ type LibraryChange struct {
 	ChangedAt  pgtype.Timestamptz
 }
 
+type LibraryChangesTrimmed struct {
+	ID      bool
+	Through int64
+}
+
 type LibrarySetting struct {
 	ID        bool
 	ScanPaths []string

internal/db/migrations/0101_library_changes_keep_latest.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS library_changes_trimmed;
+DROP INDEX IF EXISTS idx_library_changes_entity;

internal/db/migrations/0101_library_changes_keep_latest.up.sql

@@ -0,0 +1,56 @@
+-- 0101_library_changes_keep_latest.up.sql — the change feed keeps every
+-- entity's latest change, so an app syncing from nothing gets the whole
+-- library.
+--
+-- Until now the compactor dropped every row older than 30 days. An app
+-- syncing from cursor 0 (a fresh install, cleared data, or a reset after a
+-- 410) builds its library from the feed, so it got only what had changed in
+-- the last 30 days. From here the compactor drops a row only once a newer row
+-- for the same entity exists, and drops a delete only once it is 30 days old.
+--
+--   idx_library_changes_entity   finds an entity's rows, newest first, for
+--                                the compactor and the fill below.
+--   library_changes_trimmed      one row: the highest id of a delete the
+--                                compactor has dropped. An app whose cursor
+--                                is below it may have missed that delete, so
+--                                it is told to start over (410). It starts
+--                                where the old compactor left off: everything
+--                                below the oldest remaining row is gone.
+--
+-- Then every artist, album and track whose latest row is not an upsert, or
+-- which has no row at all (the old compactor dropped it), gets one, artists
+-- first so an album never arrives before its artist. Likes and playlists are
+-- left as they are: the apps fetch those from their own endpoints.
+CREATE INDEX idx_library_changes_entity ON library_changes (entity_type, entity_id, id DESC);
+
+CREATE TABLE library_changes_trimmed (
+    id      boolean PRIMARY KEY DEFAULT true CONSTRAINT library_changes_trimmed_one_row CHECK (id),
+    through bigint NOT NULL DEFAULT 0
+);
+
+INSERT INTO library_changes_trimmed (through)
+SELECT COALESCE(min(id), 1) - 1 FROM library_changes;
+
+INSERT INTO library_changes (entity_type, entity_id, op)
+SELECT 'artist', a.id::text, 'upsert'
+  FROM artists a
+ WHERE COALESCE((SELECT c.op FROM library_changes c
+                  WHERE c.entity_type = 'artist' AND c.entity_id = a.id::text
+                  ORDER BY c.id DESC LIMIT 1), 'delete') <> 'upsert'
+ ORDER BY a.id;
+
+INSERT INTO library_changes (entity_type, entity_id, op)
+SELECT 'album', al.id::text, 'upsert'
+  FROM albums al
+ WHERE COALESCE((SELECT c.op FROM library_changes c
+                  WHERE c.entity_type = 'album' AND c.entity_id = al.id::text
+                  ORDER BY c.id DESC LIMIT 1), 'delete') <> 'upsert'
+ ORDER BY al.id;
+
+INSERT INTO library_changes (entity_type, entity_id, op)
+SELECT 'track', t.id::text, 'upsert'
+  FROM tracks t
+ WHERE COALESCE((SELECT c.op FROM library_changes c
+                  WHERE c.entity_type = 'track' AND c.entity_id = t.id::text
+                  ORDER BY c.id DESC LIMIT 1), 'delete') <> 'upsert'
+ ORDER BY t.id;

internal/db/queries/library_changes.sql

@@ -28,11 +28,28 @@ SELECT COALESCE(MAX(id), 0)::BIGINT FROM library_changes;
 -- sequence stands at the backup's last change.
 SELECT COALESCE(pg_sequence_last_value(pg_get_serial_sequence('library_changes', 'id')::regclass), 0)::BIGINT;
 
--- name: GetMinLibraryChangeCursor :one
+-- name: GetLibraryChangesTrimmedThrough :one
-SELECT COALESCE(MIN(id), 0)::BIGINT FROM library_changes;
+-- The highest id of a delete the compactor has dropped. A cursor below it may
+-- have missed that delete.
+SELECT through FROM library_changes_trimmed;
 
--- name: DeleteOldLibraryChanges :execrows
+-- name: DeleteSupersededLibraryChanges :execrows
--- Removes library_changes rows older than the given cutoff. Run from
+-- Removes every row a newer row for the same entity has superseded. An app
--- the sync compactor goroutine on a daily tick. Returns the number of
+-- reading from any cursor still meets the newer row, so it loses nothing.
--- rows deleted so the worker can log meaningful progress.
+DELETE FROM library_changes old
-DELETE FROM library_changes WHERE changed_at < $1;
+ WHERE EXISTS (SELECT 1 FROM library_changes newer
+                WHERE newer.entity_type = old.entity_type
+                  AND newer.entity_id = old.entity_id
+                  AND newer.id > old.id);
+
+-- name: DeleteOldLibraryDeletes :one
+-- Removes deletes older than the cutoff and raises the trimmed mark past the
+-- highest one, so an app whose cursor is below it starts over. Returns how
+-- many went.
+WITH gone AS (
+    DELETE FROM library_changes WHERE op = 'delete' AND changed_at < $1 RETURNING id
+), mark AS (
+    UPDATE library_changes_trimmed
+       SET through = GREATEST(through, (SELECT COALESCE(max(id), 0) FROM gone))
+)
+SELECT count(*)::BIGINT FROM gone;

internal/dbtest/reset.go

@@ -184,6 +184,10 @@ func ResetDB(t *testing.T, pool *pgxpool.Pool) {
 	); err != nil {
 		t.Fatalf("dbtest.ResetDB reset release-type profiles: %v", err)
 	}
+	// The change log's trimmed mark, a singleton: nothing has been trimmed.
+	if _, err := pool.Exec(ctx, "UPDATE library_changes_trimmed SET through = 0"); err != nil {
+		t.Fatalf("dbtest.ResetDB reset library_changes_trimmed: %v", err)
+	}
 	// The server's own backup: no runs, and settings back to off.
 	if _, err := pool.Exec(ctx, `
 		DELETE FROM backup_runs;

internal/sync/compactor.go

@@ -1,17 +1,22 @@
 // Periodic compactor for the library_changes log. Every mutation in the
-// library writes one row, so the table grows unbounded without a retention
+// library writes one row, so the log grows without bound unless trimmed. This
-// policy. This worker trims rows older than CompactorRetention on a daily
+// worker trims it daily, keeping each entity's latest row:
-// tick.
 //
-// Why 30 days: clients keep a cursor and pull deltas from the last seen id.
+//   - A row a newer row for the same entity has superseded goes at once. An
-// A client that checks in at least once per retention window never sees its
+//     app reading from any cursor still meets the newer row.
-// cursor go stale; one offline longer than that gets the /api/library/sync
+//   - A delete goes once it is CompactorRetention old, and raises the trimmed
-// 410 and does a full resync.
+//     mark (library_changes_trimmed) past its id. An app whose cursor is below
+//     the mark may have missed it, so /api/library/sync answers 410 and the
+//     app syncs again from 0.
+//   - The latest upsert of an entity that still exists stays, however old. An
+//     app syncing from 0 (a fresh install, or after a 410) builds its whole
+//     library from the log, so it must hold every artist, album and track.
 
 package sync
 
 import (
 	"context"
+	"fmt"
 	"log/slog"
 	"time"
 
@@ -21,9 +26,9 @@ import (
 	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
 )
 
-// CompactorRetention is the age past which library_changes rows are
+// CompactorRetention is how long a delete stays in the log. An app that
-// dropped. Mutable for tests; production runs should not change it
+// checks in at least once in that window never misses one; one away longer
-// without considering the corresponding client cursor-stale fallback.
+// is sent back to 0. Mutable for tests.
 var CompactorRetention = 30 * 24 * time.Hour
 
 // CompactorInterval is how often the worker checks. Daily is enough —
@@ -31,8 +36,8 @@ var CompactorRetention = 30 * 24 * time.Hour
 // active library, well below a size where intra-day trimming matters.
 var CompactorInterval = 24 * time.Hour
 
-// Compactor is a background worker that periodically deletes
+// Compactor is a background worker that trims the library_changes log down
-// library_changes rows older than CompactorRetention.
+// to each entity's latest change.
 type Compactor struct {
 	pool     *pgxpool.Pool
 	logger   *slog.Logger
@@ -65,18 +70,28 @@ func (c *Compactor) Run(ctx context.Context) {
 	}
 }
 
-// tickOnce deletes rows older than CompactorRetention and logs the
+// tickOnce trims the log once and logs what went. Errors are logged at WARN;
-// row count. Errors are logged at WARN; we never abort the loop on
+// the loop never stops on a transient DB failure.
-// transient DB failures.
 func (c *Compactor) tickOnce(ctx context.Context) {
-	cutoff := pgtype.Timestamptz{Time: time.Now().Add(-CompactorRetention), Valid: true}
+	if err := Compact(ctx, c.pool, time.Now().Add(-CompactorRetention), c.logger); err != nil {
-	rows, err := dbq.New(c.pool).DeleteOldLibraryChanges(ctx, cutoff)
+		c.logger.Warn("library_changes compactor: trim failed", "err", err)
-	if err != nil {
-		c.logger.Warn("library_changes compactor: delete failed", "err", err)
-		return
-	}
-	if rows > 0 {
-		c.logger.Info("library_changes compactor: trimmed rows",
-			"count", rows, "older_than", CompactorRetention)
 	}
 }
+
+// Compact drops every superseded row, then every delete older than cutoff.
+func Compact(ctx context.Context, db dbq.DBTX, cutoff time.Time, logger *slog.Logger) error {
+	q := dbq.New(db)
+	superseded, err := q.DeleteSupersededLibraryChanges(ctx)
+	if err != nil {
+		return fmt.Errorf("drop superseded changes: %w", err)
+	}
+	deletes, err := q.DeleteOldLibraryDeletes(ctx, pgtype.Timestamptz{Time: cutoff, Valid: true})
+	if err != nil {
+		return fmt.Errorf("drop old deletes: %w", err)
+	}
+	if superseded > 0 || deletes > 0 {
+		logger.Info("library_changes compactor: trimmed rows",
+			"superseded", superseded, "old_deletes", deletes, "deletes_older_than", CompactorRetention)
+	}
+	return nil
+}

internal/sync/compactor_test.go

@@ -0,0 +1,59 @@
+package sync_test
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/stretchr/testify/require"
+
+	"git.fabledsword.com/bvandeusen/minstrel/internal/sync"
+)
+
+// The compactor keeps each entity's latest change however old, so an app
+// syncing from 0 still gets an artist nobody has touched in months. It drops
+// what a newer row supersedes, and deletes past their retention, raising the
+// trimmed mark past the newest delete it dropped.
+func TestCompact_KeepsEachEntitysLatestChange_Integration(t *testing.T) {
+	pool := testPool(t)
+	ctx := context.Background()
+	old := time.Now().Add(-90 * 24 * time.Hour)
+	recent := time.Now().Add(-time.Hour)
+
+	add := func(entity, op string, at time.Time) int64 {
+		t.Helper()
+		var id int64
+		require.NoError(t, pool.QueryRow(ctx, `INSERT INTO library_changes (entity_type, entity_id, op, changed_at)
+			VALUES ('artist', $1, $2, $3) RETURNING id`, entity, op, at).Scan(&id))
+		return id
+	}
+	untouched := add("untouched", "upsert", old)             // kept: latest, however old
+	add("edited", "upsert", old)                             // dropped: superseded
+	edited := add("edited", "upsert", recent)                // kept
+	add("removed-long-ago", "upsert", old)                   // dropped: superseded
+	removedLongAgo := add("removed-long-ago", "delete", old) // dropped: past retention
+	add("removed-lately", "upsert", old)                     // dropped: superseded
+	removedLately := add("removed-lately", "delete", recent) // kept: within retention
+
+	require.NoError(t, sync.Compact(ctx, pool, time.Now().Add(-sync.CompactorRetention),
+		slog.New(slog.NewTextHandler(io.Discard, nil))))
+
+	rows, err := pool.Query(ctx, `SELECT id FROM library_changes ORDER BY id`)
+	require.NoError(t, err)
+	left, err := pgx.CollectRows(rows, pgx.RowTo[int64])
+	require.NoError(t, err)
+	require.Equal(t, []int64{untouched, edited, removedLately}, left)
+
+	var through int64
+	require.NoError(t, pool.QueryRow(ctx, `SELECT through FROM library_changes_trimmed`).Scan(&through))
+	require.Equal(t, removedLongAgo, through, "the trimmed mark stands at the newest dropped delete")
+
+	// A second pass finds nothing more, and leaves the mark where it is.
+	require.NoError(t, sync.Compact(ctx, pool, time.Now().Add(-sync.CompactorRetention),
+		slog.New(slog.NewTextHandler(io.Discard, nil))))
+	require.NoError(t, pool.QueryRow(ctx, `SELECT through FROM library_changes_trimmed`).Scan(&through))
+	require.Equal(t, removedLongAgo, through)
+}