diff --git a/internal/api/admin_scan.go b/internal/api/admin_scan.go index e1ed856c..70ba6bdf 100644 --- a/internal/api/admin_scan.go +++ b/internal/api/admin_scan.go @@ -74,30 +74,24 @@ type scanTriggerResp struct { } // handleTriggerScan implements POST /api/admin/scan/run. -// Returns 202 on success, 409 if a scan is already in flight. -// The scan runs in a background goroutine detached from the request context. +// Returns 202 on success, 409 if a scan is already in flight (and not stale), +// 500 on DB errors. The lazy reaper inside library.TryStartScan handles +// stale (>1h) in-flight rows automatically — the operator no longer needs +// to manually clear stuck rows after a server crash. func (h *handlers) handleTriggerScan(w http.ResponseWriter, _ *http.Request) { bgCtx := context.Background() - q := dbq.New(h.pool) - - existing, err := q.GetInFlightScanRun(bgCtx) - if err == nil { + started, existing, err := library.TryStartScan( + bgCtx, h.pool, h.scanner, h.coverart, + h.logger.With("source", "manual"), h.scanCfg, + ) + if err != nil { + h.logger.Error("admin: trigger scan failed", "err", err) + writeErr(w, http.StatusInternalServerError, "server_error", "trigger failed") + return + } + if !started { writeJSON(w, http.StatusConflict, scanTriggerResp{ID: uuidToString(existing.ID)}) return } - if !errors.Is(err, pgx.ErrNoRows) { - h.logger.Error("admin: in-flight check failed", "err", err) - writeErr(w, http.StatusInternalServerError, "server_error", "in-flight check failed") - return - } - - // Detached goroutine — must outlive the HTTP request. - go func() { - if _, err := library.RunScan(bgCtx, h.pool, h.scanner, h.coverart, - h.logger.With("component", "scan_run"), h.scanCfg); err != nil { - h.logger.Warn("manual scan run failed", "err", err) - } - }() - writeJSON(w, http.StatusAccepted, scanTriggerResp{}) } diff --git a/internal/library/trigger.go b/internal/library/trigger.go new file mode 100644 index 00000000..7bc9c45b --- /dev/null +++ b/internal/library/trigger.go @@ -0,0 +1,66 @@ +package library + +import ( + "context" + "errors" + "fmt" + "log/slog" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "git.fabledsword.com/bvandeusen/minstrel/internal/coverart" + "git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq" +) + +// StuckScanThreshold is the age past which an in-flight scan_runs row is +// reaped on the next TryStartScan attempt. Real scans on realistic libraries +// finish well within this; the threshold is set far above legitimate scan +// duration so a normally-running scan is never reaped accidentally. +const StuckScanThreshold = 1 * time.Hour + +// TryStartScan attempts to start a new scan, with built-in handling for +// stuck/stale in-flight rows. +// +// Behavior: +// - No in-flight row → starts RunScan in a detached goroutine; returns (true, nil, nil). +// - In-flight row, started_at > StuckScanThreshold ago → reaps via FinishScanRun +// with error_message="reaped (stale)", then starts new scan; returns (true, nil, nil). +// - In-flight row, started_at within threshold → returns (false, &row, nil); caller +// decides what to do (HTTP handler returns 409; scheduler logs skip). +// - DB error during the in-flight check or reap → returns (false, nil, err). +func TryStartScan(ctx context.Context, pool *pgxpool.Pool, + scanner *Scanner, enricher *coverart.Enricher, + logger *slog.Logger, cfg RunScanConfig, +) (started bool, existing *dbq.GetInFlightScanRunRow, err error) { + q := dbq.New(pool) + row, qerr := q.GetInFlightScanRun(ctx) + if qerr == nil { + // Found an in-flight row. Reap if stale, else skip. + age := time.Since(row.StartedAt.Time) + if age > StuckScanThreshold { + logger.Warn("reaping stale in-flight scan", + "id", row.ID, "started_at", row.StartedAt.Time, "age", age) + if rerr := q.FinishScanRun(ctx, dbq.FinishScanRunParams{ + ID: row.ID, + Column2: "reaped (stale)", + }); rerr != nil { + return false, nil, fmt.Errorf("reap stale scan: %w", rerr) + } + // Fall through to start a new scan. + } else { + return false, &row, nil + } + } else if !errors.Is(qerr, pgx.ErrNoRows) { + return false, nil, fmt.Errorf("in-flight check: %w", qerr) + } + + // No in-flight (or just reaped). Start RunScan detached. + go func() { + if _, runErr := RunScan(ctx, pool, scanner, enricher, logger, cfg); runErr != nil { + logger.Warn("scan run failed", "err", runErr) + } + }() + return true, nil, nil +} diff --git a/internal/library/trigger_test.go b/internal/library/trigger_test.go new file mode 100644 index 00000000..fbbec894 --- /dev/null +++ b/internal/library/trigger_test.go @@ -0,0 +1,164 @@ +package library + +import ( + "context" + "errors" + "io" + "log/slog" + "os" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" +) + +// triggerTestPool returns a pool against MINSTREL_TEST_DATABASE_URL or skips. +func triggerTestPool(t *testing.T) *pgxpool.Pool { + t.Helper() + url := os.Getenv("MINSTREL_TEST_DATABASE_URL") + if url == "" { + t.Skip("MINSTREL_TEST_DATABASE_URL not set") + } + pool, err := pgxpool.New(context.Background(), url) + if err != nil { + t.Fatalf("connect: %v", err) + } + t.Cleanup(pool.Close) + // Reset scan_runs so each test starts clean. + if _, err := pool.Exec(context.Background(), `DELETE FROM scan_runs`); err != nil { + t.Fatalf("reset: %v", err) + } + return pool +} + +// quietLogger returns a logger that drops all output. +func quietLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +func TestTryStartScan_NoInFlight_Starts(t *testing.T) { + pool := triggerTestPool(t) + + started, existing, err := TryStartScan( + context.Background(), pool, + nil, nil, // scanner/enricher: nil OK because the goroutine that uses them is detached and we don't await it. + quietLogger(), RunScanConfig{}, + ) + if err != nil { + t.Fatalf("err = %v, want nil", err) + } + if !started { + t.Errorf("started = false, want true") + } + if existing != nil { + t.Errorf("existing = %v, want nil", existing) + } + // Give the detached goroutine a moment. + time.Sleep(50 * time.Millisecond) +} + +func TestTryStartScan_FreshInFlight_Skips(t *testing.T) { + pool := triggerTestPool(t) + + // Seed a fresh in-flight row. + var rowID pgtype.UUID + if err := pool.QueryRow(context.Background(), + `INSERT INTO scan_runs (started_at) VALUES (now()) RETURNING id`, + ).Scan(&rowID); err != nil { + t.Fatalf("seed: %v", err) + } + + started, existing, err := TryStartScan( + context.Background(), pool, + nil, nil, + quietLogger(), RunScanConfig{}, + ) + if err != nil { + t.Fatalf("err = %v, want nil", err) + } + if started { + t.Errorf("started = true, want false") + } + if existing == nil { + t.Fatal("existing = nil, want non-nil") + } + if existing.ID != rowID { + t.Errorf("existing.ID = %v, want %v", existing.ID, rowID) + } +} + +func TestTryStartScan_StaleInFlight_Reaps(t *testing.T) { + pool := triggerTestPool(t) + + // Seed a stale in-flight row (started 2 hours ago). + var staleID pgtype.UUID + if err := pool.QueryRow(context.Background(), + `INSERT INTO scan_runs (started_at) VALUES (now() - interval '2 hours') RETURNING id`, + ).Scan(&staleID); err != nil { + t.Fatalf("seed: %v", err) + } + + started, existing, err := TryStartScan( + context.Background(), pool, + nil, nil, + quietLogger(), RunScanConfig{}, + ) + if err != nil { + t.Fatalf("err = %v, want nil", err) + } + if !started { + t.Errorf("started = false, want true (stale row should reap and proceed)") + } + if existing != nil { + t.Errorf("existing = %v, want nil", existing) + } + + // Verify the stale row was reaped: finished_at non-NULL, error_message contains "reaped". + var finishedAt pgtype.Timestamptz + var errMsg *string + if err := pool.QueryRow(context.Background(), + `SELECT finished_at, error_message FROM scan_runs WHERE id = $1`, staleID, + ).Scan(&finishedAt, &errMsg); err != nil { + t.Fatalf("verify reap: %v", err) + } + if !finishedAt.Valid { + t.Errorf("finished_at = NULL, want non-NULL after reap") + } + if errMsg == nil || *errMsg != "reaped (stale)" { + t.Errorf("error_message = %v, want 'reaped (stale)'", errMsg) + } + + time.Sleep(50 * time.Millisecond) +} + +func TestTryStartScan_DBError_Propagates(t *testing.T) { + url := os.Getenv("MINSTREL_TEST_DATABASE_URL") + if url == "" { + t.Skip("MINSTREL_TEST_DATABASE_URL not set") + } + pool, err := pgxpool.New(context.Background(), url) + if err != nil { + t.Fatalf("connect: %v", err) + } + pool.Close() // close immediately to force errors on subsequent queries + + started, existing, err := TryStartScan( + context.Background(), pool, + nil, nil, + quietLogger(), RunScanConfig{}, + ) + if err == nil { + t.Errorf("err = nil, want non-nil for closed pool") + } + if errors.Is(err, pgx.ErrNoRows) { + t.Errorf("err is ErrNoRows; want a different DB error type") + } + if started { + t.Errorf("started = true, want false on DB error") + } + if existing != nil { + t.Errorf("existing = %v, want nil on DB error", existing) + } +}