From 3e9e46f19098dd5983ffd2bfcb8b53dacdbf2402 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 6 May 2026 21:14:50 -0400 Subject: [PATCH] feat(server/m7-scan-progress): wire stagePublisher into 4 scan stages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each of the 4 stage workers (library walk, MBID backfill, cover enrich, artist art enrich) gains a progressCb parameter that receives a snapshot of the current running state per item processed. The orchestrator stores the snapshot in a local var that the publisher's marshal closure reads at flush time. scanrun.go's 4 stage blocks are rewritten to construct one publisher per stage, pass a Tick-firing closure to the worker as progressCb, and call Flush() after the worker returns (routing any persist error to captureErr — preserves today's diagnostic behavior). The previous post-stage UpdateScanRun* calls are removed; the publisher's Flush handles the final write. Cadence: every 500 items OR every 1s, whichever fires first. Outside-orchestrator callers (scanner_test.go, enricher_test.go) pass nil for progressCb — no-op behavior preserved for tests that don't need progress observability. Co-Authored-By: Claude Sonnet 4.6 --- internal/coverart/artist_enricher.go | 12 ++- internal/coverart/enricher.go | 12 ++- internal/coverart/enricher_test.go | 2 +- internal/library/mbidbackfill.go | 15 +++- internal/library/scanner.go | 11 ++- internal/library/scanner_test.go | 4 +- internal/library/scanrun.go | 130 +++++++++++++++++++-------- 7 files changed, 141 insertions(+), 45 deletions(-) diff --git a/internal/coverart/artist_enricher.go b/internal/coverart/artist_enricher.go index e03778cb..1771793a 100644 --- a/internal/coverart/artist_enricher.go +++ b/internal/coverart/artist_enricher.go @@ -135,7 +135,8 @@ func (e *Enricher) recordArtistArt(ctx context.Context, artistID pgtype.UUID, th // EnrichArtistBatch drains up to limit artists whose eligibility loop // determines they need processing. Iterates serially. Returns // per-batch tallies for the orchestrator's scan-run record. -func (e *Enricher) EnrichArtistBatch(ctx context.Context, limit int, dataDir string) (processed, succeeded, failed int, err error) { +func (e *Enricher) EnrichArtistBatch(ctx context.Context, limit int, dataDir string, + progressCb func(processed, succeeded, failed int)) (processed, succeeded, failed int, err error) { if limit <= 0 { return 0, 0, 0, nil } @@ -156,12 +157,18 @@ func (e *Enricher) EnrichArtistBatch(ctx context.Context, limit int, dataDir str e.logger.Warn("coverart: artist batch entry failed", "artist_id", uuidString(id), "err", eerr) failed++ + if progressCb != nil { + progressCb(processed, succeeded, failed) + } continue } // Re-read to classify. row, rerr := q.GetArtistByID(ctx, id) if rerr != nil { failed++ + if progressCb != nil { + progressCb(processed, succeeded, failed) + } continue } if row.ArtistArtSource != nil && @@ -171,6 +178,9 @@ func (e *Enricher) EnrichArtistBatch(ctx context.Context, limit int, dataDir str } else { failed++ } + if progressCb != nil { + progressCb(processed, succeeded, failed) + } } return processed, succeeded, failed, nil } diff --git a/internal/coverart/enricher.go b/internal/coverart/enricher.go index d33ed266..a3ad2893 100644 --- a/internal/coverart/enricher.go +++ b/internal/coverart/enricher.go @@ -146,7 +146,8 @@ func (e *Enricher) recordAlbumCover(ctx context.Context, albumID pgtype.UUID, pa // EnrichBatch drains up to limit albums whose eligibility loop // determines they need processing. Iterates serially. Returns // per-batch tallies for the orchestrator's scan-run record. -func (e *Enricher) EnrichBatch(ctx context.Context, limit int) (processed, succeeded, failed int, err error) { +func (e *Enricher) EnrichBatch(ctx context.Context, limit int, + progressCb func(processed, succeeded, failed int)) (processed, succeeded, failed int, err error) { if limit <= 0 { return 0, 0, 0, nil } @@ -167,12 +168,18 @@ func (e *Enricher) EnrichBatch(ctx context.Context, limit int) (processed, succe e.logger.Warn("coverart: batch entry failed", "album_id", uuidString(id), "err", eerr) failed++ + if progressCb != nil { + progressCb(processed, succeeded, failed) + } continue } // Re-read to classify. row, rerr := q.GetAlbumWithFirstTrackPath(ctx, id) if rerr != nil { failed++ + if progressCb != nil { + progressCb(processed, succeeded, failed) + } continue } if row.CoverArtSource != nil && @@ -184,6 +191,9 @@ func (e *Enricher) EnrichBatch(ctx context.Context, limit int) (processed, succe } else { failed++ } + if progressCb != nil { + progressCb(processed, succeeded, failed) + } } return processed, succeeded, failed, nil } diff --git a/internal/coverart/enricher_test.go b/internal/coverart/enricher_test.go index 35faed3b..06f73dbf 100644 --- a/internal/coverart/enricher_test.go +++ b/internal/coverart/enricher_test.go @@ -263,7 +263,7 @@ func TestEnrichBatch_DrainsNullSourceOnly(t *testing.T) { } e := newTestEnricher(t, pool) - processed, _, _, err := e.EnrichBatch(context.Background(), 100) + processed, _, _, err := e.EnrichBatch(context.Background(), 100, nil) if err != nil { t.Fatalf("batch: %v", err) } diff --git a/internal/library/mbidbackfill.go b/internal/library/mbidbackfill.go index 804ad39c..b312c1f5 100644 --- a/internal/library/mbidbackfill.go +++ b/internal/library/mbidbackfill.go @@ -46,7 +46,8 @@ type BackfillMBIDsResult struct { // limit caps how many albums are processed in one call. Pass -1 for no // cap. The caller normally batches (e.g. limit=500 per boot) to avoid // stalling startup on huge libraries. -func BackfillMBIDs(ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, limit int) (BackfillMBIDsResult, error) { +func BackfillMBIDs(ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, + limit int, progressCb func(BackfillMBIDsResult)) (BackfillMBIDsResult, error) { q := dbq.New(pool) queryLimit := int32(limit) if limit < 0 { @@ -68,6 +69,9 @@ func BackfillMBIDs(ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, albumMBID, artistMBID := readMBIDsForFile(r.TrackFilePath, logger) if albumMBID == "" { res.Skipped++ + if progressCb != nil { + progressCb(res) + } continue } @@ -85,11 +89,17 @@ func BackfillMBIDs(ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, logger.Info("mbid backfill: duplicate album mbid (canonical row already owns it)", "album_id", r.AlbumID, "mbid", albumMBID) res.Duplicates++ + if progressCb != nil { + progressCb(res) + } continue } logger.Warn("mbid backfill: set album mbid failed", "album_id", r.AlbumID, "err", err) res.Skipped++ + if progressCb != nil { + progressCb(res) + } continue } @@ -124,6 +134,9 @@ func BackfillMBIDs(ctx context.Context, pool *pgxpool.Pool, logger *slog.Logger, "processed", res.Processed, "healed", res.Healed, "skipped", res.Skipped, "duplicates", res.Duplicates) } + if progressCb != nil { + progressCb(res) + } } logger.Info("mbid backfill complete", diff --git a/internal/library/scanner.go b/internal/library/scanner.go index 7381f817..274ec0a3 100644 --- a/internal/library/scanner.go +++ b/internal/library/scanner.go @@ -52,7 +52,10 @@ func New(pool *pgxpool.Pool, logger *slog.Logger, paths []string) *Scanner { // Scan walks every configured root and upserts any audio file whose mtime is // newer than the existing row's updated_at. Walk errors and per-file errors // are logged + counted; the scan keeps going. -func (s *Scanner) Scan(ctx context.Context) (Stats, error) { +// +// progressCb (may be nil) receives the current Stats snapshot after each +// processed file. Used by the orchestrator to drive partial-tally writes. +func (s *Scanner) Scan(ctx context.Context, progressCb func(Stats)) (Stats, error) { var stats Stats q := dbq.New(s.pool) start := time.Now() @@ -65,6 +68,9 @@ func (s *Scanner) Scan(ctx context.Context) (Stats, error) { if err != nil { s.logger.Warn("library scan walk error", "path", path, "err", err) stats.Errored++ + if progressCb != nil { + progressCb(stats) + } return nil } if d.IsDir() { @@ -77,6 +83,9 @@ func (s *Scanner) Scan(ctx context.Context) (Stats, error) { s.logger.Warn("library scan file error", "path", path, "err", err) stats.Errored++ } + if progressCb != nil { + progressCb(stats) + } return nil }); err != nil { return stats, fmt.Errorf("library: walk %q: %w", root, err) diff --git a/internal/library/scanner_test.go b/internal/library/scanner_test.go index a6632dcf..a9dc5250 100644 --- a/internal/library/scanner_test.go +++ b/internal/library/scanner_test.go @@ -82,7 +82,7 @@ func TestScanner_Integration(t *testing.T) { }) scanner := New(pool, logger, []string{root}) - stats, err := scanner.Scan(ctx) + stats, err := scanner.Scan(ctx, nil) if err != nil { t.Fatalf("first scan: %v", err) } @@ -103,7 +103,7 @@ func TestScanner_Integration(t *testing.T) { t.Errorf("artist sort = [%q, %q], want [Artist X, Artist Y]", artists[0].SortName, artists[1].SortName) } - stats2, err := scanner.Scan(ctx) + stats2, err := scanner.Scan(ctx, nil) if err != nil { t.Fatalf("second scan: %v", err) } diff --git a/internal/library/scanrun.go b/internal/library/scanrun.go index 5e2fe5b2..97e13a7f 100644 --- a/internal/library/scanrun.go +++ b/internal/library/scanrun.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "log/slog" + "time" "github.com/jackc/pgx/v5/pgtype" "github.com/jackc/pgx/v5/pgxpool" @@ -93,68 +94,121 @@ func RunScan( // Stage 1: library file-walk scan. if scanner != nil { - stats, serr := scanner.Scan(ctx) + var lastStats Stats + libPub := newStagePublisher(ctx, 500, time.Second, + func() []byte { + blob, _ := json.Marshal(LibraryStageTallies{ + Scanned: lastStats.Scanned, Added: lastStats.Added, + Updated: lastStats.Updated, Skipped: lastStats.Skipped, + Errored: lastStats.Errored, + }) + return blob + }, + func(blob []byte) error { + return q.UpdateScanRunLibrary(ctx, dbq.UpdateScanRunLibraryParams{ + ID: row.ID, Library: blob, + }) + }, + ) + stats, serr := scanner.Scan(ctx, func(s Stats) { + lastStats = s + libPub.Tick() + }) + lastStats = stats // ensure final flush sees post-loop value + if err := libPub.Flush(); err != nil { + captureErr("library_persist", err) + } if serr != nil { captureErr("library", serr) } - blob, _ := json.Marshal(LibraryStageTallies{ - Scanned: stats.Scanned, Added: stats.Added, Updated: stats.Updated, - Skipped: stats.Skipped, Errored: stats.Errored, - }) - if uerr := q.UpdateScanRunLibrary(ctx, dbq.UpdateScanRunLibraryParams{ - ID: row.ID, Library: blob, - }); uerr != nil { - captureErr("library_persist", uerr) - } } // Stage 2: MBID backfill. if cfg.BackfillCap != 0 { - bfRes, bferr := BackfillMBIDs(ctx, pool, logger, cfg.BackfillCap) + var lastRes BackfillMBIDsResult + bfPub := newStagePublisher(ctx, 500, time.Second, + func() []byte { + blob, _ := json.Marshal(MBIDBackfillStageTallies{ + Processed: lastRes.Processed, Healed: lastRes.Healed, + Skipped: lastRes.Skipped, Duplicates: lastRes.Duplicates, + }) + return blob + }, + func(blob []byte) error { + return q.UpdateScanRunMbidBackfill(ctx, dbq.UpdateScanRunMbidBackfillParams{ + ID: row.ID, MbidBackfill: blob, + }) + }, + ) + bfRes, bferr := BackfillMBIDs(ctx, pool, logger, cfg.BackfillCap, func(r BackfillMBIDsResult) { + lastRes = r + bfPub.Tick() + }) + lastRes = bfRes + if err := bfPub.Flush(); err != nil { + captureErr("mbid_backfill_persist", err) + } if bferr != nil { captureErr("mbid_backfill", bferr) } - blob, _ := json.Marshal(MBIDBackfillStageTallies{ - Processed: bfRes.Processed, Healed: bfRes.Healed, - Skipped: bfRes.Skipped, Duplicates: bfRes.Duplicates, - }) - if uerr := q.UpdateScanRunMbidBackfill(ctx, dbq.UpdateScanRunMbidBackfillParams{ - ID: row.ID, MbidBackfill: blob, - }); uerr != nil { - captureErr("mbid_backfill_persist", uerr) - } } // Stage 3: cover enrichment. if enricher != nil && cfg.EnrichCap != 0 { - processed, succeeded, failed, ceerr := enricher.EnrichBatch(ctx, cfg.EnrichCap) + var lastP, lastS, lastF int + coverPub := newStagePublisher(ctx, 500, time.Second, + func() []byte { + blob, _ := json.Marshal(CoverEnrichStageTallies{ + Processed: lastP, Succeeded: lastS, Failed: lastF, + }) + return blob + }, + func(blob []byte) error { + return q.UpdateScanRunCoverEnrich(ctx, dbq.UpdateScanRunCoverEnrichParams{ + ID: row.ID, CoverEnrich: blob, + }) + }, + ) + processed, succeeded, failed, ceerr := enricher.EnrichBatch(ctx, cfg.EnrichCap, func(p, s, f int) { + lastP, lastS, lastF = p, s, f + coverPub.Tick() + }) + lastP, lastS, lastF = processed, succeeded, failed + if err := coverPub.Flush(); err != nil { + captureErr("cover_enrich_persist", err) + } if ceerr != nil { captureErr("cover_enrich", ceerr) } - blob, _ := json.Marshal(CoverEnrichStageTallies{ - Processed: processed, Succeeded: succeeded, Failed: failed, - }) - if uerr := q.UpdateScanRunCoverEnrich(ctx, dbq.UpdateScanRunCoverEnrichParams{ - ID: row.ID, CoverEnrich: blob, - }); uerr != nil { - captureErr("cover_enrich_persist", uerr) - } } // Stage 4: artist art enrichment. if enricher != nil && cfg.ArtistEnrichCap != 0 && cfg.DataDir != "" { - processed, succeeded, failed, ceerr := enricher.EnrichArtistBatch(ctx, cfg.ArtistEnrichCap, cfg.DataDir) + var lastP, lastS, lastF int + aaPub := newStagePublisher(ctx, 500, time.Second, + func() []byte { + blob, _ := json.Marshal(ArtistArtEnrichStageTallies{ + Processed: lastP, Succeeded: lastS, Failed: lastF, + }) + return blob + }, + func(blob []byte) error { + return q.UpdateScanRunArtistArtEnrich(ctx, dbq.UpdateScanRunArtistArtEnrichParams{ + ID: row.ID, ArtistArtEnrich: blob, + }) + }, + ) + processed, succeeded, failed, ceerr := enricher.EnrichArtistBatch(ctx, cfg.ArtistEnrichCap, cfg.DataDir, func(p, s, f int) { + lastP, lastS, lastF = p, s, f + aaPub.Tick() + }) + lastP, lastS, lastF = processed, succeeded, failed + if err := aaPub.Flush(); err != nil { + captureErr("artist_art_enrich_persist", err) + } if ceerr != nil { captureErr("artist_art_enrich", ceerr) } - blob, _ := json.Marshal(ArtistArtEnrichStageTallies{ - Processed: processed, Succeeded: succeeded, Failed: failed, - }) - if uerr := q.UpdateScanRunArtistArtEnrich(ctx, dbq.UpdateScanRunArtistArtEnrichParams{ - ID: row.ID, ArtistArtEnrich: blob, - }); uerr != nil { - captureErr("artist_art_enrich_persist", uerr) - } } // Finalize: set finished_at + error message (or empty string if all stages succeeded).