From 7cfaafd360ef5d4c25c985d9b214c2b2fb099183 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 13 May 2026 15:49:07 -0400 Subject: [PATCH] feat(#357): library_changes retention compactor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the last deferred follow-up from #357. The library_changes table is the append-only change log that drives /api/library/sync's delta semantics — every mutation (scanner upsert, like, playlist edit, track delete) writes one row. Without a retention policy the table grows unbounded; the original migration (0025) called out the follow-up explicitly. New goroutine: sync.Compactor runs daily, deletes rows where changed_at < now - 30 days. Logs a row count when non-zero so operators can see compaction activity in the journal. First tick fires on startup so a process that hasn't been compacted in a while catches up immediately. 30-day retention matches the offline-mode spec (docs/superpowers/specs/2026-05-09-flutter-offline-mode-design.md). Clients with a cursor older than that hit the existing 410 fallback path and resync from scratch. Imported as syncpkg in main.go to follow the existing convention (see internal/library/scanner.go). Co-Authored-By: Claude Opus 4.7 (1M context) --- cmd/minstrel/main.go | 8 +++ internal/db/dbq/library_changes.sql.go | 17 +++++ internal/db/queries/library_changes.sql | 6 ++ internal/sync/compactor.go | 85 +++++++++++++++++++++++++ 4 files changed, 116 insertions(+) create mode 100644 internal/sync/compactor.go diff --git a/cmd/minstrel/main.go b/cmd/minstrel/main.go index b5288985..29b9e11c 100644 --- a/cmd/minstrel/main.go +++ b/cmd/minstrel/main.go @@ -25,6 +25,7 @@ import ( "git.fabledsword.com/bvandeusen/minstrel/internal/server" "git.fabledsword.com/bvandeusen/minstrel/internal/similarity" "git.fabledsword.com/bvandeusen/minstrel/internal/subsonic" + syncpkg "git.fabledsword.com/bvandeusen/minstrel/internal/sync" ) func main() { @@ -159,6 +160,13 @@ func run() error { lidarrReconciler := lidarrrequests.NewReconciler(pool, lidarrCfg, logger.With("component", "lidarr"), bus) go lidarrReconciler.Run(ctx) + // library_changes compactor (#357 follow-up). Daily tick; deletes + // rows older than the configured retention so the change-log table + // doesn't grow unbounded. Clients that drop offline longer than + // retention hit the /api/library/sync 410 fallback and resync. + libraryChangesCompactor := syncpkg.NewCompactor(pool, logger.With("component", "library_changes_compactor")) + go libraryChangesCompactor.Run(ctx) + // Per-user system-playlist scheduler (#392 Half B). Fires each // active user's daily build at 03:00 in their stored timezone. // Replaces the 24h-anchored cron loop (removed in the next commit diff --git a/internal/db/dbq/library_changes.sql.go b/internal/db/dbq/library_changes.sql.go index 09256c2a..de326379 100644 --- a/internal/db/dbq/library_changes.sql.go +++ b/internal/db/dbq/library_changes.sql.go @@ -7,8 +7,25 @@ package dbq import ( "context" + + "github.com/jackc/pgx/v5/pgtype" ) +const deleteOldLibraryChanges = `-- name: DeleteOldLibraryChanges :execrows +DELETE FROM library_changes WHERE changed_at < $1 +` + +// 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) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + const getLibraryChangesSince = `-- name: GetLibraryChangesSince :many SELECT id, entity_type, entity_id, op, changed_at FROM library_changes diff --git a/internal/db/queries/library_changes.sql b/internal/db/queries/library_changes.sql index d830c469..463e48e0 100644 --- a/internal/db/queries/library_changes.sql +++ b/internal/db/queries/library_changes.sql @@ -14,3 +14,9 @@ SELECT COALESCE(MAX(id), 0)::BIGINT FROM library_changes; -- name: GetMinLibraryChangeCursor :one SELECT COALESCE(MIN(id), 0)::BIGINT FROM library_changes; + +-- 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; diff --git a/internal/sync/compactor.go b/internal/sync/compactor.go new file mode 100644 index 00000000..e26943f1 --- /dev/null +++ b/internal/sync/compactor.go @@ -0,0 +1,85 @@ +// Periodic compactor for the library_changes log (#357 deferred +// follow-up). 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. +// +// Why daily / 30 days: Flutter clients track a per-app cursor and +// pull deltas from the last seen id. As long as a client checks in +// at least once per retention window, it never sees the cursor go +// stale. Clients that drop offline longer than that hit the /api/ +// library/sync 410 fallback and do a full resync. 30 days is the +// established norm for similar offline-cache delta logs and matches +// the spec at docs/superpowers/specs/2026-05-09-flutter-offline-mode-design.md. + +package sync + +import ( + "context" + "log/slog" + "time" + + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "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. +var CompactorRetention = 30 * 24 * time.Hour + +// CompactorInterval is how often the worker checks. Daily is enough — +// the table grows on the order of a few thousand rows/day for an +// 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. +type Compactor struct { + pool *pgxpool.Pool + logger *slog.Logger + interval time.Duration +} + +// NewCompactor builds a compactor with production defaults. +func NewCompactor(pool *pgxpool.Pool, logger *slog.Logger) *Compactor { + return &Compactor{ + pool: pool, + logger: logger, + interval: CompactorInterval, + } +} + +// Run blocks until ctx is cancelled, ticking every c.interval. +// Fires one tick immediately on startup so a long-running process +// that hasn't been compacted in a while catches up. +func (c *Compactor) Run(ctx context.Context) { + c.tickOnce(ctx) + t := time.NewTicker(c.interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + c.tickOnce(ctx) + } + } +} + +// 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. +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) + } +}