feat(server): drift audit batch 7 — periodic GC worker for 5 lifecycle gaps
New `internal/gc` package with a single Worker that runs all five lifecycle / retention sweeps from the 2026-06-02 drift audit on a 1-hour tick. Each sweep is small, idempotent (re-running on already-clean rows is a no-op), and logs its affected-row count. Sweeps (Scribe parent #552): - **#566** GcCloseStalePlayEvents — play_events rows opened > 24h ago that never got a play_ended (client crash, network drop). Synthesizes ended_at from duration_played_ms when known, falls back to now() so the row stops looking "open" to downstream filters (ended_at IS NULL). - **#565** GcClosePlaySessionsWithNoRecentEvents — play_sessions with last_event_at older than 6h get ended_at = last_event_at ("user moved on"); empty sessions older than 1h get closed too (stale handshakes from clients that never recorded a play). The audit caught that the column was added but never populated by any writer — every session row was "open" forever, breaking downstream dedup queries that assume closed semantics. - **#567** GcExpireScrobbleQueueFailedRows — drops scrobble_queue rows in status='failed' older than 14 days. The worker stops retrying after maxAttempts so these otherwise accumulate forever on a persistent ListenBrainz outage / revoked token. - **#574** GcResetStuckSystemPlaylistRuns — flips system_playlist_runs.in_flight back to false on rows whose last_run_at is older than 10 minutes. Catches goroutine-panic wedges where the generator died between SET in_flight=true and SET in_flight=false; the duplicate-prevention check refuses to start a fresh regen while in_flight, so a stuck row would otherwise deadlock all future regens for that user. Records "stuck-row auto-reset by gc" in last_error so the operator can tell auto-reset from a recent real failure. - **#575** GcDeleteExpiredPasswordResets — deletes expired password_resets rows. Unused expired rows go after a 1h grace (gives the operator time to debug an active reset attempt); used rows are kept 7 days for audit. Wiring: - main.go `go gcWorker.Run(ctx)` alongside the other periodic workers (scrobble, similarity, lidarr). - tickOnce fires once at start so a freshly-deployed server does its initial sweep without waiting a full tick, matching the scrobble worker pattern. - Errors per sweep are logged but do NOT abort the remaining ones — a transient pgx error from one query shouldn't prevent the others from running. Tests: - 4 integration tests, one per UPDATE/DELETE sweep, that seed rows-to-sweep + rows-to-leave-alone and assert the right rows changed state. Skip unless MINSTREL_TEST_DATABASE_URL is set (mirrors the api package pattern). - Empty-tables no-op smoke test. - Run() cancellation honoured (no spinning goroutine at test-runner exit). That's all five remaining server-side lifecycle findings from the audit. The Android LOCAL_USER_ID hardcode (#576) is a separate refactor that needs auth-store wiring and stays in the queue.
This commit is contained in:
@@ -15,6 +15,7 @@ import (
|
|||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/coverart"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/coverart"
|
||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/db"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/db"
|
||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/eventbus"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/eventbus"
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/gc"
|
||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/library"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/library"
|
||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/lidarr"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/lidarr"
|
||||||
"git.fabledsword.com/bvandeusen/minstrel/internal/lidarrconfig"
|
"git.fabledsword.com/bvandeusen/minstrel/internal/lidarrconfig"
|
||||||
@@ -161,6 +162,16 @@ func run() error {
|
|||||||
similarityWorker := similarity.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "similarity"))
|
similarityWorker := similarity.NewWorker(pool, listenbrainz.NewClient(), logger.With("component", "similarity"))
|
||||||
go similarityWorker.Run(ctx)
|
go similarityWorker.Run(ctx)
|
||||||
|
|
||||||
|
// Start the GC worker. Runs every 1h and sweeps lifecycle tables
|
||||||
|
// that have no writer-side close path or retention policy:
|
||||||
|
// orphan play_events, stale play_sessions, expired
|
||||||
|
// scrobble_queue failures, stuck system_playlist_runs, expired
|
||||||
|
// password_resets. Each sweep is idempotent — a row that's
|
||||||
|
// already clean is a no-op. Addresses drift audit findings
|
||||||
|
// #565 #566 #567 #574 #575 (Scribe parent #552).
|
||||||
|
gcWorker := gc.NewWorker(pool, logger.With("component", "gc"))
|
||||||
|
go gcWorker.Run(ctx)
|
||||||
|
|
||||||
// Start the Lidarr reconciler worker. Per spec §M5a, polls pending Lidarr
|
// Start the Lidarr reconciler worker. Per spec §M5a, polls pending Lidarr
|
||||||
// import requests and reconciles them against the library. Short-circuits
|
// import requests and reconciles them against the library. Short-circuits
|
||||||
// to no-op when lidarr_config.enabled = false.
|
// to no-op when lidarr_config.enabled = false.
|
||||||
|
|||||||
@@ -0,0 +1,126 @@
|
|||||||
|
// Code generated by sqlc. DO NOT EDIT.
|
||||||
|
// versions:
|
||||||
|
// sqlc v1.31.1
|
||||||
|
// source: gc.sql
|
||||||
|
|
||||||
|
package dbq
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
)
|
||||||
|
|
||||||
|
const gcClosePlaySessionsWithNoRecentEvents = `-- name: GcClosePlaySessionsWithNoRecentEvents :execrows
|
||||||
|
UPDATE play_sessions
|
||||||
|
SET ended_at = COALESCE(last_event_at, started_at)
|
||||||
|
WHERE ended_at IS NULL
|
||||||
|
AND (
|
||||||
|
(track_count > 0 AND last_event_at < now() - INTERVAL '6 hours')
|
||||||
|
OR
|
||||||
|
(track_count = 0 AND started_at < now() - INTERVAL '1 hour')
|
||||||
|
)
|
||||||
|
`
|
||||||
|
|
||||||
|
// #565: play_sessions.ended_at was added but never populated by any
|
||||||
|
// writer. Close sessions whose last_event_at is older than 6h —
|
||||||
|
// treating that as "user moved on" the same way audio_service does
|
||||||
|
// after grace periods. Sessions with NO events (track_count = 0)
|
||||||
|
// older than 1h are also closed (stale handshakes from clients that
|
||||||
|
// never recorded a play). ended_at is set to last_event_at so the
|
||||||
|
// session's duration reads naturally.
|
||||||
|
func (q *Queries) GcClosePlaySessionsWithNoRecentEvents(ctx context.Context) (int64, error) {
|
||||||
|
result, err := q.db.Exec(ctx, gcClosePlaySessionsWithNoRecentEvents)
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
return result.RowsAffected(), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
const gcCloseStalePlayEvents = `-- name: GcCloseStalePlayEvents :execrows
|
||||||
|
|
||||||
|
UPDATE play_events
|
||||||
|
SET ended_at = COALESCE(
|
||||||
|
started_at + (duration_played_ms * INTERVAL '1 millisecond'),
|
||||||
|
now()
|
||||||
|
)
|
||||||
|
WHERE ended_at IS NULL
|
||||||
|
AND started_at < now() - INTERVAL '24 hours'
|
||||||
|
`
|
||||||
|
|
||||||
|
// Background garbage-collector / lifecycle queries. All five address
|
||||||
|
// drift findings from the 2026-06-02 audit (Scribe parent #552):
|
||||||
|
// #565 #566 #567 #574 #575. Sweeper runs on a long tick (default 1h)
|
||||||
|
// so per-query cost is amortised; each is idempotent (re-running on
|
||||||
|
// already-closed/-deleted rows is a no-op).
|
||||||
|
// #566: play_events rows opened more than 24h ago that never got a
|
||||||
|
// play_ended. Client crashed mid-track, network dropped, etc. We
|
||||||
|
// synthesize ended_at = started_at + duration_played_ms when present,
|
||||||
|
// otherwise leave duration_played_ms null and stamp ended_at = now()
|
||||||
|
// so the row stops looking "open" for downstream queries that filter
|
||||||
|
// ended_at IS NULL. Doesn't touch was_skipped (we genuinely don't
|
||||||
|
// know if the user skipped).
|
||||||
|
func (q *Queries) GcCloseStalePlayEvents(ctx context.Context) (int64, error) {
|
||||||
|
result, err := q.db.Exec(ctx, gcCloseStalePlayEvents)
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
return result.RowsAffected(), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
const gcDeleteExpiredPasswordResets = `-- name: GcDeleteExpiredPasswordResets :execrows
|
||||||
|
DELETE FROM password_resets
|
||||||
|
WHERE (used_at IS NOT NULL AND used_at < now() - INTERVAL '7 days')
|
||||||
|
OR (used_at IS NULL AND expires_at < now() - INTERVAL '1 hour')
|
||||||
|
`
|
||||||
|
|
||||||
|
// #575: password_resets accumulates expired + used rows forever. The
|
||||||
|
// validation path already rejects them; this just keeps the table
|
||||||
|
// from growing unbounded. Used rows are kept for 7 days for audit;
|
||||||
|
// unused expired rows go immediately.
|
||||||
|
func (q *Queries) GcDeleteExpiredPasswordResets(ctx context.Context) (int64, error) {
|
||||||
|
result, err := q.db.Exec(ctx, gcDeleteExpiredPasswordResets)
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
return result.RowsAffected(), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
const gcExpireScrobbleQueueFailedRows = `-- name: GcExpireScrobbleQueueFailedRows :execrows
|
||||||
|
DELETE FROM scrobble_queue
|
||||||
|
WHERE status = 'failed'
|
||||||
|
AND enqueued_at < now() - INTERVAL '14 days'
|
||||||
|
`
|
||||||
|
|
||||||
|
// #567: scrobble_queue rows that have been in status='failed' for
|
||||||
|
// more than 14 days. The worker stops retrying after maxAttempts;
|
||||||
|
// failed rows accumulate forever otherwise. CASCADE from play_events
|
||||||
|
// already drops the row when the underlying event is deleted, so this
|
||||||
|
// only handles persistent failures (token revoked, etc.).
|
||||||
|
func (q *Queries) GcExpireScrobbleQueueFailedRows(ctx context.Context) (int64, error) {
|
||||||
|
result, err := q.db.Exec(ctx, gcExpireScrobbleQueueFailedRows)
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
return result.RowsAffected(), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
const gcResetStuckSystemPlaylistRuns = `-- name: GcResetStuckSystemPlaylistRuns :execrows
|
||||||
|
UPDATE system_playlist_runs
|
||||||
|
SET in_flight = false,
|
||||||
|
last_error = COALESCE(last_error, 'stuck-row auto-reset by gc')
|
||||||
|
WHERE in_flight = true
|
||||||
|
AND last_run_at < now() - INTERVAL '10 minutes'
|
||||||
|
`
|
||||||
|
|
||||||
|
// #574: system_playlist_runs.in_flight = true can wedge on a
|
||||||
|
// goroutine panic between SET in_flight=true and SET in_flight=false.
|
||||||
|
// The duplicate-prevention check refuses to start a fresh regen while
|
||||||
|
// in_flight, so a stuck row blocks all future regens for that user.
|
||||||
|
// Reset rows where last_run_at is older than 10 minutes (regens
|
||||||
|
// shouldn't take that long).
|
||||||
|
func (q *Queries) GcResetStuckSystemPlaylistRuns(ctx context.Context) (int64, error) {
|
||||||
|
result, err := q.db.Exec(ctx, gcResetStuckSystemPlaylistRuns)
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
return result.RowsAffected(), nil
|
||||||
|
}
|
||||||
@@ -0,0 +1,70 @@
|
|||||||
|
-- Background garbage-collector / lifecycle queries. All five address
|
||||||
|
-- drift findings from the 2026-06-02 audit (Scribe parent #552):
|
||||||
|
-- #565 #566 #567 #574 #575. Sweeper runs on a long tick (default 1h)
|
||||||
|
-- so per-query cost is amortised; each is idempotent (re-running on
|
||||||
|
-- already-closed/-deleted rows is a no-op).
|
||||||
|
|
||||||
|
-- name: GcCloseStalePlayEvents :execrows
|
||||||
|
-- #566: play_events rows opened more than 24h ago that never got a
|
||||||
|
-- play_ended. Client crashed mid-track, network dropped, etc. We
|
||||||
|
-- synthesize ended_at = started_at + duration_played_ms when present,
|
||||||
|
-- otherwise leave duration_played_ms null and stamp ended_at = now()
|
||||||
|
-- so the row stops looking "open" for downstream queries that filter
|
||||||
|
-- ended_at IS NULL. Doesn't touch was_skipped (we genuinely don't
|
||||||
|
-- know if the user skipped).
|
||||||
|
UPDATE play_events
|
||||||
|
SET ended_at = COALESCE(
|
||||||
|
started_at + (duration_played_ms * INTERVAL '1 millisecond'),
|
||||||
|
now()
|
||||||
|
)
|
||||||
|
WHERE ended_at IS NULL
|
||||||
|
AND started_at < now() - INTERVAL '24 hours';
|
||||||
|
|
||||||
|
-- name: GcClosePlaySessionsWithNoRecentEvents :execrows
|
||||||
|
-- #565: play_sessions.ended_at was added but never populated by any
|
||||||
|
-- writer. Close sessions whose last_event_at is older than 6h —
|
||||||
|
-- treating that as "user moved on" the same way audio_service does
|
||||||
|
-- after grace periods. Sessions with NO events (track_count = 0)
|
||||||
|
-- older than 1h are also closed (stale handshakes from clients that
|
||||||
|
-- never recorded a play). ended_at is set to last_event_at so the
|
||||||
|
-- session's duration reads naturally.
|
||||||
|
UPDATE play_sessions
|
||||||
|
SET ended_at = COALESCE(last_event_at, started_at)
|
||||||
|
WHERE ended_at IS NULL
|
||||||
|
AND (
|
||||||
|
(track_count > 0 AND last_event_at < now() - INTERVAL '6 hours')
|
||||||
|
OR
|
||||||
|
(track_count = 0 AND started_at < now() - INTERVAL '1 hour')
|
||||||
|
);
|
||||||
|
|
||||||
|
-- name: GcExpireScrobbleQueueFailedRows :execrows
|
||||||
|
-- #567: scrobble_queue rows that have been in status='failed' for
|
||||||
|
-- more than 14 days. The worker stops retrying after maxAttempts;
|
||||||
|
-- failed rows accumulate forever otherwise. CASCADE from play_events
|
||||||
|
-- already drops the row when the underlying event is deleted, so this
|
||||||
|
-- only handles persistent failures (token revoked, etc.).
|
||||||
|
DELETE FROM scrobble_queue
|
||||||
|
WHERE status = 'failed'
|
||||||
|
AND enqueued_at < now() - INTERVAL '14 days';
|
||||||
|
|
||||||
|
-- name: GcResetStuckSystemPlaylistRuns :execrows
|
||||||
|
-- #574: system_playlist_runs.in_flight = true can wedge on a
|
||||||
|
-- goroutine panic between SET in_flight=true and SET in_flight=false.
|
||||||
|
-- The duplicate-prevention check refuses to start a fresh regen while
|
||||||
|
-- in_flight, so a stuck row blocks all future regens for that user.
|
||||||
|
-- Reset rows where last_run_at is older than 10 minutes (regens
|
||||||
|
-- shouldn't take that long).
|
||||||
|
UPDATE system_playlist_runs
|
||||||
|
SET in_flight = false,
|
||||||
|
last_error = COALESCE(last_error, 'stuck-row auto-reset by gc')
|
||||||
|
WHERE in_flight = true
|
||||||
|
AND last_run_at < now() - INTERVAL '10 minutes';
|
||||||
|
|
||||||
|
-- name: GcDeleteExpiredPasswordResets :execrows
|
||||||
|
-- #575: password_resets accumulates expired + used rows forever. The
|
||||||
|
-- validation path already rejects them; this just keeps the table
|
||||||
|
-- from growing unbounded. Used rows are kept for 7 days for audit;
|
||||||
|
-- unused expired rows go immediately.
|
||||||
|
DELETE FROM password_resets
|
||||||
|
WHERE (used_at IS NOT NULL AND used_at < now() - INTERVAL '7 days')
|
||||||
|
OR (used_at IS NULL AND expires_at < now() - INTERVAL '1 hour');
|
||||||
@@ -0,0 +1,101 @@
|
|||||||
|
// Package gc runs periodic garbage-collection / lifecycle sweeps
|
||||||
|
// against tables that have NO writer-side close path or NO retention
|
||||||
|
// policy. Each sweep addresses a drift finding from the 2026-06-02
|
||||||
|
// audit (Scribe parent #552) and is idempotent — re-running it on
|
||||||
|
// already-clean rows is a no-op.
|
||||||
|
//
|
||||||
|
// One Worker handles all sweeps so a single long-tick goroutine
|
||||||
|
// amortises the per-tick fixed cost. Each individual sweep is small
|
||||||
|
// (single UPDATE / DELETE with a time-bounded WHERE) and emits a
|
||||||
|
// log line with the affected-row count so the sweep cadence is
|
||||||
|
// visible in the application log without an explicit metrics layer.
|
||||||
|
//
|
||||||
|
// Sweeps:
|
||||||
|
// - GcCloseStalePlayEvents (#566)
|
||||||
|
// - GcClosePlaySessionsWithNoRecentEvents (#565)
|
||||||
|
// - GcExpireScrobbleQueueFailedRows (#567)
|
||||||
|
// - GcResetStuckSystemPlaylistRuns (#574)
|
||||||
|
// - GcDeleteExpiredPasswordResets (#575)
|
||||||
|
package gc
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"log/slog"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
|
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||||
|
)
|
||||||
|
|
||||||
|
// defaultTick is the production sweep cadence. 1 hour is generous
|
||||||
|
// since each sweep's WHERE clause uses a multi-hour staleness
|
||||||
|
// threshold; the worst-case delay between a row becoming sweepable
|
||||||
|
// and the worker noticing is bounded by tick + threshold.
|
||||||
|
const defaultTick = 1 * time.Hour
|
||||||
|
|
||||||
|
// Worker holds the pool + logger + tick interval. Construct with
|
||||||
|
// [NewWorker]; pass the returned Worker to a goroutine that calls
|
||||||
|
// [Worker.Run] with a context that's cancelled on shutdown.
|
||||||
|
type Worker struct {
|
||||||
|
pool *pgxpool.Pool
|
||||||
|
logger *slog.Logger
|
||||||
|
tick time.Duration
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewWorker builds a Worker with the production tick (1h). Tests can
|
||||||
|
// reach into the Worker after construction to override `tick` for
|
||||||
|
// faster iteration.
|
||||||
|
func NewWorker(pool *pgxpool.Pool, logger *slog.Logger) *Worker {
|
||||||
|
return &Worker{pool: pool, logger: logger, tick: defaultTick}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Run blocks until ctx is cancelled, running every sweep on each
|
||||||
|
// tick. Sweeps fire in fixed order; an error in one does NOT abort
|
||||||
|
// the rest (the panic-vs-just-failed distinction matters here — a
|
||||||
|
// pgx transient error from one query shouldn't prevent the others
|
||||||
|
// from running).
|
||||||
|
func (w *Worker) Run(ctx context.Context) {
|
||||||
|
// Fire once at start so a freshly-deployed server doesn't wait a
|
||||||
|
// full tick before doing the initial sweep. Matches the scrobble
|
||||||
|
// + similarity workers' "sweep then tick" pattern.
|
||||||
|
w.tickOnce(ctx)
|
||||||
|
t := time.NewTicker(w.tick)
|
||||||
|
defer t.Stop()
|
||||||
|
for {
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return
|
||||||
|
case <-t.C:
|
||||||
|
w.tickOnce(ctx)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// tickOnce runs each sweep once, logging the affected-row count.
|
||||||
|
// Errors are logged per-sweep but do NOT abort the remaining ones —
|
||||||
|
// each sweep is independent.
|
||||||
|
func (w *Worker) tickOnce(ctx context.Context) {
|
||||||
|
q := dbq.New(w.pool)
|
||||||
|
w.runSweep(ctx, "close_stale_play_events", q.GcCloseStalePlayEvents)
|
||||||
|
w.runSweep(ctx, "close_play_sessions", q.GcClosePlaySessionsWithNoRecentEvents)
|
||||||
|
w.runSweep(ctx, "expire_scrobble_failed", q.GcExpireScrobbleQueueFailedRows)
|
||||||
|
w.runSweep(ctx, "reset_stuck_system_runs", q.GcResetStuckSystemPlaylistRuns)
|
||||||
|
w.runSweep(ctx, "delete_expired_password_resets", q.GcDeleteExpiredPasswordResets)
|
||||||
|
}
|
||||||
|
|
||||||
|
// runSweep is a small adapter so each sweep call site is a one-liner
|
||||||
|
// in tickOnce. Logs at info on rows>0 and debug on rows=0 to keep
|
||||||
|
// the normal-case (nothing-to-do) noise out of operator logs.
|
||||||
|
func (w *Worker) runSweep(ctx context.Context, name string, fn func(context.Context) (int64, error)) {
|
||||||
|
rows, err := fn(ctx)
|
||||||
|
if err != nil {
|
||||||
|
w.logger.Error("gc sweep failed", "sweep", name, "err", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if rows > 0 {
|
||||||
|
w.logger.Info("gc sweep", "sweep", name, "rows_affected", rows)
|
||||||
|
} else {
|
||||||
|
w.logger.Debug("gc sweep", "sweep", name, "rows_affected", 0)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,317 @@
|
|||||||
|
package gc
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"io"
|
||||||
|
"log/slog"
|
||||||
|
"os"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/jackc/pgx/v5/pgtype"
|
||||||
|
"github.com/jackc/pgx/v5/pgxpool"
|
||||||
|
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/db"
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
|
||||||
|
"git.fabledsword.com/bvandeusen/minstrel/internal/dbtest"
|
||||||
|
)
|
||||||
|
|
||||||
|
// testWorker constructs a Worker against MINSTREL_TEST_DATABASE_URL.
|
||||||
|
// Mirrors the api package's testHandlers pattern — skip when not in
|
||||||
|
// integration mode, migrate + reset, return the pool for the caller
|
||||||
|
// to seed.
|
||||||
|
func testWorker(t *testing.T) (*Worker, *pgxpool.Pool) {
|
||||||
|
t.Helper()
|
||||||
|
if testing.Short() {
|
||||||
|
t.Skip("skipping gc integration in -short mode")
|
||||||
|
}
|
||||||
|
dsn := os.Getenv("MINSTREL_TEST_DATABASE_URL")
|
||||||
|
if dsn == "" {
|
||||||
|
t.Skip("MINSTREL_TEST_DATABASE_URL not set")
|
||||||
|
}
|
||||||
|
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
|
||||||
|
if err := db.Migrate(dsn, logger); err != nil {
|
||||||
|
t.Fatalf("migrate: %v", err)
|
||||||
|
}
|
||||||
|
pool, err := pgxpool.New(context.Background(), dsn)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("pool: %v", err)
|
||||||
|
}
|
||||||
|
t.Cleanup(pool.Close)
|
||||||
|
dbtest.ResetDB(t, pool)
|
||||||
|
return NewWorker(pool, logger), pool
|
||||||
|
}
|
||||||
|
|
||||||
|
// seedUser creates a minimal user row for tests that need play_events
|
||||||
|
// / play_sessions / scrobble_queue rows. Username is prefixed so
|
||||||
|
// dbtest.ResetDB cleans up between runs.
|
||||||
|
func seedUser(t *testing.T, pool *pgxpool.Pool, name string) pgtype.UUID {
|
||||||
|
t.Helper()
|
||||||
|
u, err := dbq.New(pool).CreateUser(context.Background(), dbq.CreateUserParams{
|
||||||
|
Username: dbtest.TestUserPrefix + name,
|
||||||
|
PasswordHash: "test-hash",
|
||||||
|
ApiToken: "test-token-" + name,
|
||||||
|
IsAdmin: false,
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("CreateUser: %v", err)
|
||||||
|
}
|
||||||
|
return u.ID
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGcCloseStalePlayEvents_ClosesOnly24hOldRows(t *testing.T) {
|
||||||
|
w, pool := testWorker(t)
|
||||||
|
ctx := context.Background()
|
||||||
|
userID := seedUser(t, pool, "alice")
|
||||||
|
|
||||||
|
// Need a track + session to satisfy FKs on play_events.
|
||||||
|
var trackID pgtype.UUID
|
||||||
|
if err := pool.QueryRow(ctx, `
|
||||||
|
INSERT INTO artists (name) VALUES ('A') RETURNING id
|
||||||
|
`).Scan(new(pgtype.UUID)); err != nil {
|
||||||
|
t.Fatalf("seed artist row: %v", err)
|
||||||
|
}
|
||||||
|
// Get the just-inserted artist id.
|
||||||
|
var artistID pgtype.UUID
|
||||||
|
if err := pool.QueryRow(ctx, `SELECT id FROM artists WHERE name = 'A'`).Scan(&artistID); err != nil {
|
||||||
|
t.Fatalf("read artist: %v", err)
|
||||||
|
}
|
||||||
|
var albumID pgtype.UUID
|
||||||
|
if err := pool.QueryRow(ctx, `
|
||||||
|
INSERT INTO albums (artist_id, title, sort_title) VALUES ($1, 'X', 'X') RETURNING id
|
||||||
|
`, artistID).Scan(&albumID); err != nil {
|
||||||
|
t.Fatalf("seed album: %v", err)
|
||||||
|
}
|
||||||
|
if err := pool.QueryRow(ctx, `
|
||||||
|
INSERT INTO tracks (album_id, artist_id, title, file_path, duration_ms)
|
||||||
|
VALUES ($1, $2, 'T', '/x.mp3', 180000) RETURNING id
|
||||||
|
`, albumID, artistID).Scan(&trackID); err != nil {
|
||||||
|
t.Fatalf("seed track: %v", err)
|
||||||
|
}
|
||||||
|
var sessionID pgtype.UUID
|
||||||
|
if err := pool.QueryRow(ctx, `
|
||||||
|
INSERT INTO play_sessions (user_id, started_at, last_event_at)
|
||||||
|
VALUES ($1, now() - interval '2 hours', now()) RETURNING id
|
||||||
|
`, userID).Scan(&sessionID); err != nil {
|
||||||
|
t.Fatalf("seed session: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Stale row (25h old, no ended_at) — should be closed.
|
||||||
|
if _, err := pool.Exec(ctx, `
|
||||||
|
INSERT INTO play_events (user_id, track_id, session_id, started_at)
|
||||||
|
VALUES ($1, $2, $3, now() - interval '25 hours')
|
||||||
|
`, userID, trackID, sessionID); err != nil {
|
||||||
|
t.Fatalf("seed stale event: %v", err)
|
||||||
|
}
|
||||||
|
// Fresh row (1h old, no ended_at) — should be left alone.
|
||||||
|
if _, err := pool.Exec(ctx, `
|
||||||
|
INSERT INTO play_events (user_id, track_id, session_id, started_at)
|
||||||
|
VALUES ($1, $2, $3, now() - interval '1 hour')
|
||||||
|
`, userID, trackID, sessionID); err != nil {
|
||||||
|
t.Fatalf("seed fresh event: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
w.tickOnce(ctx)
|
||||||
|
|
||||||
|
var closedStale, openFresh bool
|
||||||
|
if err := pool.QueryRow(ctx, `
|
||||||
|
SELECT EXISTS(SELECT 1 FROM play_events
|
||||||
|
WHERE started_at < now() - interval '24 hours'
|
||||||
|
AND ended_at IS NOT NULL)
|
||||||
|
`).Scan(&closedStale); err != nil {
|
||||||
|
t.Fatalf("check stale: %v", err)
|
||||||
|
}
|
||||||
|
if err := pool.QueryRow(ctx, `
|
||||||
|
SELECT EXISTS(SELECT 1 FROM play_events
|
||||||
|
WHERE started_at > now() - interval '2 hours'
|
||||||
|
AND ended_at IS NULL)
|
||||||
|
`).Scan(&openFresh); err != nil {
|
||||||
|
t.Fatalf("check fresh: %v", err)
|
||||||
|
}
|
||||||
|
if !closedStale {
|
||||||
|
t.Errorf("stale play_events row not closed")
|
||||||
|
}
|
||||||
|
if !openFresh {
|
||||||
|
t.Errorf("fresh play_events row was closed (should be left alone)")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGcClosePlaySessions_ClosesIdleAndEmptyStarvedSessions(t *testing.T) {
|
||||||
|
w, pool := testWorker(t)
|
||||||
|
ctx := context.Background()
|
||||||
|
userID := seedUser(t, pool, "bob")
|
||||||
|
|
||||||
|
// Idle session — has events, last_event_at 8h ago.
|
||||||
|
if _, err := pool.Exec(ctx, `
|
||||||
|
INSERT INTO play_sessions (user_id, started_at, last_event_at, track_count)
|
||||||
|
VALUES ($1, now() - interval '8 hours', now() - interval '8 hours', 5)
|
||||||
|
`, userID); err != nil {
|
||||||
|
t.Fatalf("seed idle session: %v", err)
|
||||||
|
}
|
||||||
|
// Empty-starved session — no events, started 2h ago.
|
||||||
|
if _, err := pool.Exec(ctx, `
|
||||||
|
INSERT INTO play_sessions (user_id, started_at, last_event_at, track_count)
|
||||||
|
VALUES ($1, now() - interval '2 hours', now() - interval '2 hours', 0)
|
||||||
|
`, userID); err != nil {
|
||||||
|
t.Fatalf("seed empty session: %v", err)
|
||||||
|
}
|
||||||
|
// Active session — recent last_event_at, has events.
|
||||||
|
if _, err := pool.Exec(ctx, `
|
||||||
|
INSERT INTO play_sessions (user_id, started_at, last_event_at, track_count)
|
||||||
|
VALUES ($1, now() - interval '30 minutes', now() - interval '5 minutes', 3)
|
||||||
|
`, userID); err != nil {
|
||||||
|
t.Fatalf("seed active session: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
w.tickOnce(ctx)
|
||||||
|
|
||||||
|
var closedCount, openCount int
|
||||||
|
if err := pool.QueryRow(ctx, `
|
||||||
|
SELECT count(*) FROM play_sessions
|
||||||
|
WHERE user_id = $1 AND ended_at IS NOT NULL
|
||||||
|
`, userID).Scan(&closedCount); err != nil {
|
||||||
|
t.Fatalf("count closed: %v", err)
|
||||||
|
}
|
||||||
|
if err := pool.QueryRow(ctx, `
|
||||||
|
SELECT count(*) FROM play_sessions
|
||||||
|
WHERE user_id = $1 AND ended_at IS NULL
|
||||||
|
`, userID).Scan(&openCount); err != nil {
|
||||||
|
t.Fatalf("count open: %v", err)
|
||||||
|
}
|
||||||
|
if closedCount != 2 {
|
||||||
|
t.Errorf("closed sessions = %d, want 2 (idle + empty-starved)", closedCount)
|
||||||
|
}
|
||||||
|
if openCount != 1 {
|
||||||
|
t.Errorf("open sessions = %d, want 1 (active)", openCount)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGcResetStuckSystemPlaylistRuns(t *testing.T) {
|
||||||
|
w, pool := testWorker(t)
|
||||||
|
ctx := context.Background()
|
||||||
|
stuckUser := seedUser(t, pool, "stuck")
|
||||||
|
activeUser := seedUser(t, pool, "active")
|
||||||
|
|
||||||
|
// Stuck row: in_flight, last_run_at 30 min ago.
|
||||||
|
if _, err := pool.Exec(ctx, `
|
||||||
|
INSERT INTO system_playlist_runs (user_id, last_run_at, in_flight)
|
||||||
|
VALUES ($1, now() - interval '30 minutes', true)
|
||||||
|
`, stuckUser); err != nil {
|
||||||
|
t.Fatalf("seed stuck run: %v", err)
|
||||||
|
}
|
||||||
|
// Active row: in_flight, last_run_at 2 min ago — still legitimate.
|
||||||
|
if _, err := pool.Exec(ctx, `
|
||||||
|
INSERT INTO system_playlist_runs (user_id, last_run_at, in_flight)
|
||||||
|
VALUES ($1, now() - interval '2 minutes', true)
|
||||||
|
`, activeUser); err != nil {
|
||||||
|
t.Fatalf("seed active run: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
w.tickOnce(ctx)
|
||||||
|
|
||||||
|
var stuckFlight, activeFlight bool
|
||||||
|
if err := pool.QueryRow(ctx, `
|
||||||
|
SELECT in_flight FROM system_playlist_runs WHERE user_id = $1
|
||||||
|
`, stuckUser).Scan(&stuckFlight); err != nil {
|
||||||
|
t.Fatalf("read stuck row: %v", err)
|
||||||
|
}
|
||||||
|
if err := pool.QueryRow(ctx, `
|
||||||
|
SELECT in_flight FROM system_playlist_runs WHERE user_id = $1
|
||||||
|
`, activeUser).Scan(&activeFlight); err != nil {
|
||||||
|
t.Fatalf("read active row: %v", err)
|
||||||
|
}
|
||||||
|
if stuckFlight {
|
||||||
|
t.Errorf("stuck row still in_flight after sweep")
|
||||||
|
}
|
||||||
|
if !activeFlight {
|
||||||
|
t.Errorf("active row was reset (should be left alone — only 2 min old)")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGcDeleteExpiredPasswordResets(t *testing.T) {
|
||||||
|
w, pool := testWorker(t)
|
||||||
|
ctx := context.Background()
|
||||||
|
userID := seedUser(t, pool, "pwd")
|
||||||
|
|
||||||
|
// Expired-unused (> 1h past expires_at) — should delete.
|
||||||
|
if _, err := pool.Exec(ctx, `
|
||||||
|
INSERT INTO password_resets (token, user_id, expires_at)
|
||||||
|
VALUES ('expired-old', $1, now() - interval '2 hours')
|
||||||
|
`, userID); err != nil {
|
||||||
|
t.Fatalf("seed expired: %v", err)
|
||||||
|
}
|
||||||
|
// Used (> 7 days past used_at) — should delete.
|
||||||
|
if _, err := pool.Exec(ctx, `
|
||||||
|
INSERT INTO password_resets (token, user_id, expires_at, used_at)
|
||||||
|
VALUES ('used-old', $1, now() - interval '8 days', now() - interval '8 days')
|
||||||
|
`, userID); err != nil {
|
||||||
|
t.Fatalf("seed used-old: %v", err)
|
||||||
|
}
|
||||||
|
// Active (expires in future) — should survive.
|
||||||
|
if _, err := pool.Exec(ctx, `
|
||||||
|
INSERT INTO password_resets (token, user_id, expires_at)
|
||||||
|
VALUES ('active', $1, now() + interval '1 hour')
|
||||||
|
`, userID); err != nil {
|
||||||
|
t.Fatalf("seed active: %v", err)
|
||||||
|
}
|
||||||
|
// Recently used (< 7 days) — should survive for audit.
|
||||||
|
if _, err := pool.Exec(ctx, `
|
||||||
|
INSERT INTO password_resets (token, user_id, expires_at, used_at)
|
||||||
|
VALUES ('used-recent', $1, now() - interval '1 hour', now() - interval '1 day')
|
||||||
|
`, userID); err != nil {
|
||||||
|
t.Fatalf("seed used-recent: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
w.tickOnce(ctx)
|
||||||
|
|
||||||
|
var remaining []string
|
||||||
|
rows, err := pool.Query(ctx, `SELECT token FROM password_resets WHERE user_id = $1`, userID)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("list remaining: %v", err)
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
for rows.Next() {
|
||||||
|
var tok string
|
||||||
|
if err := rows.Scan(&tok); err != nil {
|
||||||
|
t.Fatalf("scan: %v", err)
|
||||||
|
}
|
||||||
|
remaining = append(remaining, tok)
|
||||||
|
}
|
||||||
|
wantSet := map[string]bool{"active": true, "used-recent": true}
|
||||||
|
if len(remaining) != len(wantSet) {
|
||||||
|
t.Errorf("remaining tokens = %v, want %v", remaining, []string{"active", "used-recent"})
|
||||||
|
}
|
||||||
|
for _, tok := range remaining {
|
||||||
|
if !wantSet[tok] {
|
||||||
|
t.Errorf("unexpected surviving token %q", tok)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Verifies tickOnce doesn't blow up when the tables are completely
|
||||||
|
// empty — sweeps just no-op. Regression guard against an EXEC vs
|
||||||
|
// QUERY-row-count mismatch failing on zero rows.
|
||||||
|
func TestGcTickOnce_NoOpOnEmptyTables(t *testing.T) {
|
||||||
|
w, _ := testWorker(t)
|
||||||
|
w.tickOnce(context.Background())
|
||||||
|
}
|
||||||
|
|
||||||
|
// Sanity check on the Run() loop's cancel behaviour — we don't want
|
||||||
|
// to leave a goroutine spinning at test-runner exit. 10ms tick with
|
||||||
|
// an immediate cancel should return promptly.
|
||||||
|
func TestGcRun_HonoursContextCancel(t *testing.T) {
|
||||||
|
w, _ := testWorker(t)
|
||||||
|
w.tick = 10 * time.Millisecond
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
done := make(chan struct{})
|
||||||
|
go func() {
|
||||||
|
w.Run(ctx)
|
||||||
|
close(done)
|
||||||
|
}()
|
||||||
|
cancel()
|
||||||
|
select {
|
||||||
|
case <-done:
|
||||||
|
case <-time.After(2 * time.Second):
|
||||||
|
t.Fatal("Run did not return after cancel")
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user