docs: worker robustness design (scan chunking + DB reconnect)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,110 @@
|
||||
# Worker robustness: scan chunking + DB reconnect
|
||||
|
||||
**Date:** 2026-07-12
|
||||
**Status:** approved, implementing
|
||||
|
||||
## Problem
|
||||
|
||||
Two worker-loop robustness gaps surfaced after the discovery work shipped:
|
||||
|
||||
1. **The library scan blocks the worker loop for its entire duration.** `scan_library`
|
||||
walks the whole `/music` tree in one synchronous call, MB-resolving each album at
|
||||
MusicBrainz's ~1 req/s. On a large library that is minutes-to-hours during which
|
||||
`claim_next` never runs — no downloads, no monitor sweep, no discovery. It is also
|
||||
not resumable: a crash mid-scan restarts from the top.
|
||||
|
||||
2. **A transient DB connection loss crashes the whole worker.** The loop uses one
|
||||
long-lived `conn` created once by `wait_for_db()`. Any DB op on a dropped connection
|
||||
(e.g. `AdminShutdown` when Postgres restarts) raises out of `run_forever` and the
|
||||
process exits. `restart: unless-stopped` brings it back, but ungracefully: a scary
|
||||
traceback, loss of in-memory adapter/source state, and a crash mid-pipeline can leave
|
||||
a job dirty. Observed live during the 2026-07-12 deploy when `docker compose` recreated
|
||||
the `db` container.
|
||||
|
||||
**Not in scope — discovery.** `run_discovery` is already bounded: it processes at most
|
||||
`discover.maxSeeds` (default 50) due seeds per run and advances a `lastDiscoveredAt`
|
||||
cursor, so it is chunked and resumable across runs. Left as-is by conscious decision.
|
||||
|
||||
## Design
|
||||
|
||||
### A. Scan chunking
|
||||
|
||||
Process a bounded chunk of albums per loop iteration, persisting progress in `Config`,
|
||||
so the loop returns to `claim_next` / monitor between chunks. Deterministic ordering
|
||||
(the existing `sorted(os.listdir)` walk) makes a path cursor stable; all inserts are
|
||||
already `ON CONFLICT`, so overlap/re-processing is idempotent.
|
||||
|
||||
**Config state:**
|
||||
- `scan.requested` (existing) — request a fresh scan.
|
||||
- `scan.cursor` (new) — last fully-processed `"<artist>/<album>"` key. Non-empty ⇒ a
|
||||
scan is in progress.
|
||||
- `scan.progress` (new) — running totals, e.g. `"imported=12 skipped=3"`.
|
||||
- `scan.result` (existing) — final summary, written once on completion.
|
||||
|
||||
**New function** in `scan.py`:
|
||||
```
|
||||
scan_chunk(conn, resolver, probe, browser, dest_root, cursor, limit)
|
||||
-> (new_cursor: str, imported_delta: int, skipped_delta: int, done: bool)
|
||||
```
|
||||
Enumerates the same nested sorted `(artist_folder, album_folder)` walk, skips every pair
|
||||
whose `"artist/album"` key is `<= cursor`, processes up to `limit` albums via the existing
|
||||
`_ensure_artist` / `_populate_discography` / `_record_owned` helpers (unchanged), and
|
||||
returns the last-processed key as `new_cursor` plus `done=True` when it reached the end
|
||||
without hitting `limit`. `scan_library` is kept (or reimplemented as: run `scan_chunk` with
|
||||
an unbounded limit) so its direct tests stay valid.
|
||||
|
||||
**Loop integration** (replaces the current `if scan.requested:` block in `main.py`):
|
||||
- `scan.requested` truthy AND no scan in progress → **start**: clear `scan.cursor` and
|
||||
`scan.progress`, clear `scan.requested`.
|
||||
- scan in progress (`scan.cursor` present, or just-started) → run **exactly one**
|
||||
`scan_chunk`, persist the advanced cursor and accumulated totals. On `done`: write
|
||||
`scan.result`, clear `scan.cursor`/`scan.progress`.
|
||||
- One chunk per iteration ⇒ job-claim and the monitor tick run between chunks.
|
||||
|
||||
**Config:** `scan.chunkSize`, default **25** (~25–50 s/chunk at MB ~1 req/s).
|
||||
|
||||
**Accepted trade-off:** an artist whose albums straddle a chunk boundary has its
|
||||
discography browsed twice. Idempotent (`ON CONFLICT DO UPDATE`) and rare (albums are
|
||||
contiguous per artist in sorted order), so we do not persist a cross-chunk `browsed` set.
|
||||
|
||||
**Library-change semantics during a scan:** additions after the cursor are picked up;
|
||||
additions/removals before the cursor wait for the next scan. Acceptable.
|
||||
|
||||
### B. Worker DB reconnect
|
||||
|
||||
Wrap the loop body in `run_forever` with a connection-error boundary:
|
||||
```
|
||||
while True:
|
||||
try:
|
||||
... existing iteration ...
|
||||
except psycopg.OperationalError as e:
|
||||
print(f"worker: db connection lost ({e}); reconnecting...", flush=True)
|
||||
try: conn.close()
|
||||
except Exception: pass
|
||||
conn = wait_for_db()
|
||||
continue
|
||||
```
|
||||
`AdminShutdown` subclasses `OperationalError`, so this catches the observed case. The
|
||||
inner `conn.rollback()` calls that would themselves fail on a dead connection re-raise
|
||||
`OperationalError` and compose up to this boundary. Reuses `wait_for_db()`'s existing
|
||||
retry/backoff. Net effect: a transient blip reconnects in-process instead of crashing.
|
||||
|
||||
## Testing
|
||||
|
||||
- **scan_chunk** (integration, `lyra_test` + a temp `/music` tree via tmp_path):
|
||||
- respects `limit` and returns the correct `new_cursor`;
|
||||
- resumes from a given cursor (processes only albums after it);
|
||||
- `done=False` mid-walk, `done=True` at the end;
|
||||
- two sequential chunks cover the same albums as one unbounded `scan_library` run
|
||||
(same imported/skipped, same rows) — idempotent equivalence;
|
||||
- cursor past the end is a no-op with `done=True`.
|
||||
- **loop orchestration** (extract a testable `maybe_run_scan(...)` mirroring
|
||||
`maybe_run_discovery`): start clears cursor+requested; one chunk per call; completion
|
||||
writes `scan.result` and clears the cursor.
|
||||
- **DB reconnect**: unit-test that a helper reconnects a closed connection (drive
|
||||
`wait_for_db` semantics / a small boundary helper); full-loop reconnect verified by
|
||||
reasoning since the infinite loop isn't directly unit-testable.
|
||||
|
||||
## Out of scope
|
||||
Discovery per-run blocking (already bounded); the `_upsert_artist` score-accumulate
|
||||
tweak; the Last.fm source.
|
||||
Reference in New Issue
Block a user