feat(server/m7-recurring-scans): TryStartScan helper + reaper

Extracts the in-flight check + scan-start logic from handleTriggerScan into a shared library.TryStartScan helper used by both manual and (future) scheduled paths. Folds in the lazy 1-hour stuck-scan reaper: in-flight rows older than StuckScanThreshold get FinishScanRun-ed with error_message="reaped (stale)" and the new scan proceeds. handleTriggerScan becomes a thin wrapper preserving the same external HTTP behavior (202/409/500). Operators no longer need to manually clear stuck rows after a server crash — the next manual trigger reaps it. Tests gated on MINSTREL_TEST_DATABASE_URL: no-in-flight starts; fresh-in-flight skips with row pointer; stale-in-flight reaps (verified via finished_at + error_message); DB error propagates.
2026-05-06 22:15:19 -04:00
parent e27d030e68
commit 8784e938bd
3 changed files with 244 additions and 20 deletions
@@ -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{})
 }
@@ -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
+}
@@ -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)
+	}
+}