diff --git a/internal/api/library_sync.go b/internal/api/library_sync.go index ecd45de2..e95e6448 100644 --- a/internal/api/library_sync.go +++ b/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, - // tell the client to reset. (No compaction job in this slice; the - // path exists for future compaction support.) + // The log keeps every entity's latest change, so a client syncing from + // 0 gets the whole library. It drops deletes older than its retention, + // 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", diff --git a/internal/api/library_sync_test.go b/internal/api/library_sync_test.go index 84aaf277..18983d54 100644 --- a/internal/api/library_sync_test.go +++ b/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()) + } +} diff --git a/internal/backup/tables.go b/internal/backup/tables.go index 95c599d8..5ac351a8 100644 --- a/internal/backup/tables.go +++ b/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"}, } diff --git a/internal/db/dbq/library_changes.sql.go b/internal/db/dbq/library_changes.sql.go index bed9d3a2..b78b26d0 100644 --- a/internal/db/dbq/library_changes.sql.go +++ b/internal/db/dbq/library_changes.sql.go @@ -11,15 +11,38 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) -const deleteOldLibraryChanges = `-- name: DeleteOldLibraryChanges :execrows -DELETE FROM library_changes WHERE changed_at < $1 +const deleteOldLibraryDeletes = `-- name: DeleteOldLibraryDeletes :one +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 -// the sync compactor goroutine on a daily tick. Returns the number of -// rows deleted so the worker can log meaningful progress. -func (q *Queries) DeleteOldLibraryChanges(ctx context.Context, changedAt pgtype.Timestamptz) (int64, error) { - result, err := q.db.Exec(ctx, deleteOldLibraryChanges, changedAt) +// 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. +func (q *Queries) DeleteOldLibraryDeletes(ctx context.Context, changedAt pgtype.Timestamptz) (int64, error) { + 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) diff --git a/internal/db/dbq/models.go b/internal/db/dbq/models.go index f3a06cb1..580029fd 100644 --- a/internal/db/dbq/models.go +++ b/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 diff --git a/internal/db/migrations/0101_library_changes_keep_latest.down.sql b/internal/db/migrations/0101_library_changes_keep_latest.down.sql new file mode 100644 index 00000000..ffee2bf1 --- /dev/null +++ b/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; diff --git a/internal/db/migrations/0101_library_changes_keep_latest.up.sql b/internal/db/migrations/0101_library_changes_keep_latest.up.sql new file mode 100644 index 00000000..4d9933cd --- /dev/null +++ b/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; diff --git a/internal/db/queries/library_changes.sql b/internal/db/queries/library_changes.sql index 3b9f7733..712850cc 100644 --- a/internal/db/queries/library_changes.sql +++ b/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 -SELECT COALESCE(MIN(id), 0)::BIGINT FROM library_changes; +-- name: GetLibraryChangesTrimmedThrough :one +-- 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 --- Removes library_changes rows older than the given cutoff. Run from --- the sync compactor goroutine on a daily tick. Returns the number of --- rows deleted so the worker can log meaningful progress. -DELETE FROM library_changes WHERE changed_at < $1; +-- name: DeleteSupersededLibraryChanges :execrows +-- 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. +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); + +-- 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; diff --git a/internal/dbtest/reset.go b/internal/dbtest/reset.go index ac250d5e..e8b5807d 100644 --- a/internal/dbtest/reset.go +++ b/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; diff --git a/internal/sync/compactor.go b/internal/sync/compactor.go index 93829b2b..b975540f 100644 --- a/internal/sync/compactor.go +++ b/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 -// policy. This worker trims rows older than CompactorRetention on a daily -// tick. +// library writes one row, so the log grows without bound unless trimmed. This +// worker trims it daily, keeping each entity's latest row: // -// Why 30 days: clients keep a cursor and pull deltas from the last seen id. -// A client that checks in at least once per retention window never sees its -// cursor go stale; one offline longer than that gets the /api/library/sync -// 410 and does a full resync. +// - A row a newer row for the same entity has superseded goes at once. An +// app reading from any cursor still meets the newer row. +// - A delete goes once it is CompactorRetention old, and raises the trimmed +// 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 -// dropped. Mutable for tests; production runs should not change it -// without considering the corresponding client cursor-stale fallback. +// CompactorRetention is how long a delete stays in the log. An app that +// checks in at least once in that window never misses one; one away longer +// 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 -// library_changes rows older than CompactorRetention. +// Compactor is a background worker that trims the library_changes log down +// 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 -// row count. Errors are logged at WARN; we never abort the loop on -// transient DB failures. +// tickOnce trims the log once and logs what went. Errors are logged at WARN; +// the loop never stops on a transient DB failure. func (c *Compactor) tickOnce(ctx context.Context) { - cutoff := pgtype.Timestamptz{Time: time.Now().Add(-CompactorRetention), Valid: true} - rows, err := dbq.New(c.pool).DeleteOldLibraryChanges(ctx, cutoff) - 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) + if err := Compact(ctx, c.pool, time.Now().Add(-CompactorRetention), c.logger); err != nil { + c.logger.Warn("library_changes compactor: trim failed", "err", err) } } + +// 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 +} diff --git a/internal/sync/compactor_test.go b/internal/sync/compactor_test.go new file mode 100644 index 00000000..49bb13c9 --- /dev/null +++ b/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) +}