Compare commits
14 Commits
06e7f91897
...
108114ffa3
| Author | SHA1 | Date | |
|---|---|---|---|
| 108114ffa3 | |||
| 6451cf3a3c | |||
| 927266d042 | |||
| c6945bd89e | |||
| 3f624c3395 | |||
| ecec0b6073 | |||
| b78867435a | |||
| 3d30e7dc8b | |||
| 8c95ff735b | |||
| bff590a00f | |||
| 20869bdc13 | |||
| 9c2495825f | |||
| 878eac066e | |||
| 603c62ef87 |
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,179 @@
|
||||
# Press controls + inter-job delay — design
|
||||
|
||||
Date: 2026-07-15
|
||||
|
||||
## Problem
|
||||
|
||||
The monitor can enqueue a large backlog of acquisition jobs at once. The worker drains
|
||||
them serially with **no pacing** (only a 2s idle poll when the queue is empty) and offers
|
||||
**no runtime control** over the queue — the only per-job action on the Floor is Retry (on
|
||||
`needs_attention`). The user wants:
|
||||
|
||||
1. A **configurable delay between downloads** to spread sustained Qobuz load over time.
|
||||
2. **Per-item controls** on the Floor ("the Press"): pause and remove a queued item.
|
||||
3. **Whole-press controls**: pause/resume, clear the queue, and a force-stop of the
|
||||
currently-downloading album.
|
||||
|
||||
## Constraints (from the existing code)
|
||||
|
||||
- The worker is a **single, strictly-serial process** (`worker/lyra_worker/main.py:256-298`).
|
||||
One job runs to completion in a blocking `run_pipeline` call before the loop does anything
|
||||
else.
|
||||
- **Config is read fresh every loop iteration** (`main.py:262`, `get_config`), so any Config
|
||||
change takes effect within ~2s with no restart.
|
||||
- An **in-flight download cannot be cleanly interrupted** — `run_pipeline` and the adapter
|
||||
download are synchronous/blocking with only an `on_progress` callback, no cancel token. The
|
||||
only interruption is process death, after which `reclaim_stuck_jobs` (`claim.py:8-24`)
|
||||
requeues the in-flight job at startup and `clear_staging_root` (`main.py:233`) wipes the
|
||||
partial download.
|
||||
- `claim_next` (`claim.py:27-57`) claims the oldest `state='requested'` job.
|
||||
- Deleting a `Request` cascades to its `Job` and `Candidate` rows (`schema.prisma:51,66`); the
|
||||
linked `MonitoredRelease` is `SetNull`, not cascaded.
|
||||
- `enqueue_due` (`monitor.py:115-147`) re-enqueues a monitored release once it has no live
|
||||
(non-terminal) job and its `lastSearchedAt` is older than `retryIntervalHours` (default 6).
|
||||
So a removed/cleared monitored item reappears later on the normal cadence unless `ignored`.
|
||||
- The web container runs `prisma migrate deploy` on startup (`web/docker-entrypoint.sh`), so a
|
||||
new migration auto-applies on deploy.
|
||||
|
||||
## Decisions (confirmed with the user)
|
||||
|
||||
- **Active item:** add a global **force-kill** ("Stop now") that aborts the in-flight download
|
||||
via worker process-restart. Per-item controls act only on queued items.
|
||||
- **Remove / Clear semantics:** **cancel-for-now**. Manual one-off requests are gone for good;
|
||||
monitored/wanted releases stay wanted and may reappear on the monitor's normal cadence.
|
||||
Permanent skip remains the existing per-release **Ignore**.
|
||||
- **Pause scope:** **downloads only**. Pausing halts claiming/downloading; the monitor keeps
|
||||
finding and enqueueing releases (they pile up in the queue) and Resume drains them.
|
||||
|
||||
## Data model
|
||||
|
||||
- **New column** `Job.paused Boolean @default(false)` — a Prisma migration
|
||||
(`add_job_paused`). Auto-applies on deploy.
|
||||
- **Config keys** (key/value `Config` rows, no migration):
|
||||
- `floor.paused` — `"true"`/`"false"`; global press pause.
|
||||
- `floor.jobDelaySeconds` — integer string, default `"0"`; delay between downloads.
|
||||
- `floor.killRequested` — `"true"`/`"false"`; one-shot force-kill signal, cleared by the
|
||||
worker on consumption.
|
||||
|
||||
## Worker changes
|
||||
|
||||
`worker/lyra_worker/main.py` and `worker/lyra_worker/claim.py`.
|
||||
|
||||
1. **Inter-job delay.** After `finish_job` (`main.py:297`), read `floor.jobDelaySeconds` from
|
||||
the already-fresh `config` and sleep that many seconds before the next loop iteration.
|
||||
Sleep in small (~2s) chunks so the loop stays responsive; the force-kill watcher runs on a
|
||||
separate thread and works regardless. Parse with the existing `_int`-style guard (invalid →
|
||||
0). A value of 0 preserves today's back-to-back behavior.
|
||||
|
||||
2. **Global pause.** Immediately before `claim_next` (`main.py:286`), if `floor.paused` is
|
||||
truthy (`_TRUE` idiom), skip claiming: `time.sleep(IDLE_SLEEP); continue`. Because the loop
|
||||
is serial, any in-flight job has already finished before this check — pause stops the
|
||||
*next* download. Monitor sweep / scan / discovery (above the claim in the loop body)
|
||||
continue to run, so the queue keeps filling while paused.
|
||||
|
||||
3. **Per-item pause.** `claim_next` `WHERE` gains `AND NOT "paused"`. A paused queued job is
|
||||
skipped by the claim but remains `state='requested'`, so it is not "lost" and resumes
|
||||
claiming as soon as `paused` is cleared. `reclaim_stuck_jobs` is unaffected (it only touches
|
||||
in-flight states).
|
||||
|
||||
4. **Force-kill watcher.** A daemon thread started in `run_forever` (modeled on
|
||||
`_liveness_heartbeat`, `main.py:216-226`) polls `floor.killRequested` on its own DB
|
||||
connection every ~1.5s. When set, it clears the flag (commit) and calls `os._exit(0)`. The
|
||||
container restart policy brings the worker back; startup `clear_staging_root` +
|
||||
`reclaim_stuck_jobs` wipe the partial download and requeue the aborted job. The "Stop now"
|
||||
action also sets `floor.paused=true`, so the worker returns **idle** rather than instantly
|
||||
re-downloading the aborted album; the user resumes when ready.
|
||||
|
||||
## Infrastructure (ops repo)
|
||||
|
||||
The prod worker's Docker restart policy is currently **`no`** (`docker inspect` →
|
||||
`RestartPolicy=no`), so process death — from force-kill *or any crash* — leaves it down. The
|
||||
force-kill design requires `restart: unless-stopped` on the `worker` service in the ops repo
|
||||
(`/opt/git/vm-download/docker/lyra/compose.yaml`). The in-repo `docker-compose.yml` already has
|
||||
this (`docker-compose.yml:53`); this change brings prod in line and is good hygiene
|
||||
independent of this feature. This is a one-line change in a separate repo and must be applied
|
||||
before the force-kill feature is usable in prod.
|
||||
|
||||
## API (web) — follows existing `retry` / `library DELETE` patterns
|
||||
|
||||
- **`GET /api/floor`** → `{ paused: boolean, jobDelaySeconds: number }` (reads the Config
|
||||
rows). Used by the Floor to render the paused banner and control states.
|
||||
- **`POST /api/floor`** with `{ action }`:
|
||||
- `"pause"` / `"resume"` → upsert `floor.paused`.
|
||||
- `"clear"` → delete every `Request` whose `Job.state = 'requested'` (queued, not yet
|
||||
claimed). Cascade removes the Jobs and Candidates. The in-flight and terminal jobs are
|
||||
untouched. Returns the number cleared.
|
||||
- `"stop"` → upsert `floor.paused='true'` and `floor.killRequested='true'`.
|
||||
- **`POST /api/requests/[id]/pause`** with `{ paused: boolean }` → set `Job.paused`. Guard:
|
||||
the job must be `state='requested'` (else `400`, matching the existing retry route), since
|
||||
pausing an in-flight job has no effect.
|
||||
- **`DELETE /api/requests/[id]`** → delete the `Request` (cascade Job + Candidates). Guard:
|
||||
only when `Job.state = 'requested'`, so the active download is never touched (that is the
|
||||
job of `Stop now`). Mirrors `DELETE /api/library/[id]`.
|
||||
|
||||
## Web UI
|
||||
|
||||
`web/src/app/queue.tsx` and `web/src/app/_ui/job-row.tsx`; styling via existing `.btn`
|
||||
variants in `design.css`.
|
||||
|
||||
- **Press header** (above the queue list):
|
||||
- `Pause the press` / `Resume` toggle (from `GET /api/floor`).
|
||||
- `Clear queue` — confirm dialog showing the queued count; calls `POST /api/floor
|
||||
{action:"clear"}`.
|
||||
- `Stop now ⚠` — rendered only when a job is actively downloading (derivable from the
|
||||
existing `/api/requests` data: any job in `matching`/`matched`/`downloading`/`tagging`).
|
||||
Confirm dialog ("aborts the current download and pauses the press"); calls `POST /api/floor
|
||||
{action:"stop"}`.
|
||||
- A "Press paused — N queued" banner/chip when `paused`.
|
||||
- **Per-item controls** (queued rows only — `job.state === 'requested'`), placed in the row
|
||||
`note` slot as `.btn sm ghost`:
|
||||
- `Pause` / `Resume` → `POST /api/requests/[id]/pause`.
|
||||
- `Remove` → `DELETE /api/requests/[id]`.
|
||||
- A per-item paused row shows a subtle "paused" chip and the `Resume` button.
|
||||
- Active-download and `needs_attention` rows keep their current rendering (Retry unchanged).
|
||||
- Polling cadence stays 2s; `GET /api/floor` is fetched alongside the existing three calls in
|
||||
`refresh()`.
|
||||
|
||||
## Settings
|
||||
|
||||
Add a numeric field **"Delay between downloads (seconds)"** (default 0, min 0) to the settings
|
||||
form (`web/src/app/settings/settings-form.tsx`), persisted to `floor.jobDelaySeconds` via the
|
||||
existing settings config-write path (the same mechanism used for the other numeric/monitor
|
||||
settings). Include short helptext explaining it paces sustained Qobuz load.
|
||||
|
||||
## Testing
|
||||
|
||||
- **Worker** (pytest, `worker/tests/`):
|
||||
- `claim_next` skips `paused=true` queued jobs and still claims the next unpaused one.
|
||||
- Global pause: the loop skips claiming when `floor.paused` is set (unit-test the claim-gate
|
||||
helper).
|
||||
- Delay: the delay value is read and applied (unit-test the parse/gate helper; keep the
|
||||
sleep injectable/mockable).
|
||||
- Kill watcher: when `floor.killRequested` is set, the watcher clears the flag and triggers
|
||||
exit (mock `os._exit`, assert flag cleared + exit called).
|
||||
- **Web** (route tests, mirroring `requests/[id]/retry/route.test.ts` and
|
||||
`library/bulk/route.test.ts`):
|
||||
- `POST /api/floor` for each action; `clear` deletes only `requested` jobs; `stop` sets both
|
||||
flags.
|
||||
- `POST /api/requests/[id]/pause` toggles `Job.paused`; rejects non-`requested`.
|
||||
- `DELETE /api/requests/[id]` removes a queued Request; refuses an in-flight/terminal job.
|
||||
- `GET /api/floor` returns the Config-derived state.
|
||||
|
||||
## Edge cases
|
||||
|
||||
- **Remove/Clear of monitored items** reappear later by design (cancel-for-now). No extra
|
||||
handling; the per-release Ignore is the permanent path.
|
||||
- **Kill with nothing downloading** still restarts the worker (brief blip). Minimized by only
|
||||
showing `Stop now` while a download is active.
|
||||
- **Delay during pause:** the delay sleeps only *between* jobs; pause is checked before each
|
||||
claim, so a paused press never starts a delayed job.
|
||||
- **Race:** clicking per-item Pause/Remove on a job that was just claimed — the guards
|
||||
(`state='requested'`) reject it; the UI only renders those buttons on queued rows, and the
|
||||
2s refresh reconciles.
|
||||
|
||||
## Out of scope
|
||||
|
||||
- Resuming a partially-downloaded album (no download-resume support; force-killed jobs
|
||||
re-download from scratch).
|
||||
- Reordering the queue / per-item priority.
|
||||
- Any change to the per-release Ignore or the wanted list UI.
|
||||
@@ -0,0 +1,2 @@
|
||||
-- AlterTable
|
||||
ALTER TABLE "Job" ADD COLUMN "paused" BOOLEAN NOT NULL DEFAULT false;
|
||||
@@ -58,6 +58,7 @@ model Job {
|
||||
createdAt DateTime @default(now())
|
||||
updatedAt DateTime @updatedAt
|
||||
downloadProgress Float @default(0)
|
||||
paused Boolean @default(false)
|
||||
candidates Candidate[]
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import { GET, POST } from "./route";
|
||||
import { prisma } from "@/lib/db";
|
||||
|
||||
function post(action: string) {
|
||||
return new Request("http://localhost/api/floor", {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ action }),
|
||||
});
|
||||
}
|
||||
|
||||
async function flag(key: string) {
|
||||
return (await prisma.config.findUnique({ where: { key } }))?.value ?? null;
|
||||
}
|
||||
|
||||
async function queuedRequest(album: string, state = "requested") {
|
||||
return prisma.request.create({
|
||||
data: { artist: "A", album, status: "pending", job: { create: { state: state as never } } },
|
||||
include: { job: true },
|
||||
});
|
||||
}
|
||||
|
||||
describe("floor API", () => {
|
||||
it("GET reports paused=false by default", async () => {
|
||||
const res = await GET();
|
||||
expect(await res.json()).toEqual({ paused: false });
|
||||
});
|
||||
|
||||
it("pause and resume toggle floor.paused", async () => {
|
||||
await POST(post("pause"));
|
||||
expect(await flag("floor.paused")).toBe("true");
|
||||
const paused = await (await GET()).json();
|
||||
expect(paused).toEqual({ paused: true });
|
||||
|
||||
await POST(post("resume"));
|
||||
expect(await flag("floor.paused")).toBe("false");
|
||||
});
|
||||
|
||||
it("stop sets both paused and killRequested", async () => {
|
||||
const res = await POST(post("stop"));
|
||||
expect(res.status).toBe(200);
|
||||
expect(await flag("floor.paused")).toBe("true");
|
||||
expect(await flag("floor.killRequested")).toBe("true");
|
||||
});
|
||||
|
||||
it("clear deletes only queued (requested) requests", async () => {
|
||||
const queued = await queuedRequest("Queued");
|
||||
const active = await queuedRequest("Downloading", "downloading");
|
||||
|
||||
const res = await POST(post("clear"));
|
||||
const body = await res.json();
|
||||
expect(body.cleared).toBe(1);
|
||||
|
||||
expect(await prisma.request.findUnique({ where: { id: queued.id } })).toBeNull();
|
||||
expect(await prisma.request.findUnique({ where: { id: active.id } })).not.toBeNull();
|
||||
});
|
||||
|
||||
it("rejects an unknown action", async () => {
|
||||
const res = await POST(post("nope"));
|
||||
expect(res.status).toBe(400);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,49 @@
|
||||
import { prisma } from "@/lib/db";
|
||||
|
||||
async function getFlag(key: string): Promise<boolean> {
|
||||
const row = await prisma.config.findUnique({ where: { key } });
|
||||
return row?.value === "true";
|
||||
}
|
||||
|
||||
async function setFlag(key: string, value: boolean): Promise<void> {
|
||||
const v = value ? "true" : "false";
|
||||
await prisma.config.upsert({
|
||||
where: { key },
|
||||
create: { key, value: v, secret: false },
|
||||
update: { value: v },
|
||||
});
|
||||
}
|
||||
|
||||
export async function GET() {
|
||||
return Response.json({ paused: await getFlag("floor.paused") });
|
||||
}
|
||||
|
||||
export async function POST(request: Request) {
|
||||
let body: { action?: string };
|
||||
try {
|
||||
body = (await request.json()) as { action?: string };
|
||||
} catch {
|
||||
return Response.json({ error: "invalid JSON" }, { status: 400 });
|
||||
}
|
||||
|
||||
switch (body.action) {
|
||||
case "pause":
|
||||
await setFlag("floor.paused", true);
|
||||
return Response.json({ ok: true, paused: true });
|
||||
case "resume":
|
||||
await setFlag("floor.paused", false);
|
||||
return Response.json({ ok: true, paused: false });
|
||||
case "stop":
|
||||
await setFlag("floor.paused", true);
|
||||
await setFlag("floor.killRequested", true);
|
||||
return Response.json({ ok: true, paused: true, stopped: true });
|
||||
case "clear": {
|
||||
const { count } = await prisma.request.deleteMany({
|
||||
where: { job: { is: { state: "requested" } } },
|
||||
});
|
||||
return Response.json({ ok: true, cleared: count });
|
||||
}
|
||||
default:
|
||||
return Response.json({ error: "unknown action" }, { status: 400 });
|
||||
}
|
||||
}
|
||||
@@ -34,3 +34,23 @@ describe("monitor config API", () => {
|
||||
expect(await prisma.config.count()).toBe(before);
|
||||
});
|
||||
});
|
||||
|
||||
describe("monitor config — floor.jobDelaySeconds", () => {
|
||||
it("defaults jobDelaySeconds to 0 and round-trips a new value", async () => {
|
||||
const before = await (await GET()).json();
|
||||
expect(before["floor.jobDelaySeconds"]).toBe("0");
|
||||
|
||||
await PATCH(
|
||||
new Request("http://localhost/api/monitor/config", {
|
||||
method: "PATCH",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ "floor.jobDelaySeconds": "45" }),
|
||||
}),
|
||||
);
|
||||
|
||||
const stored = await prisma.config.findUnique({ where: { key: "floor.jobDelaySeconds" } });
|
||||
expect(stored?.value).toBe("45");
|
||||
const after = await (await GET()).json();
|
||||
expect(after["floor.jobDelaySeconds"]).toBe("45");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -6,6 +6,7 @@ const DEFAULTS: Record<string, string> = {
|
||||
"monitor.retryIntervalHours": "6",
|
||||
"monitor.qualityCutoff": "2",
|
||||
"monitor.upgradeWindowDays": "14",
|
||||
"floor.jobDelaySeconds": "0",
|
||||
};
|
||||
|
||||
export async function GET() {
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import { POST } from "./route";
|
||||
import { prisma } from "@/lib/db";
|
||||
import type { Prisma } from "@prisma/client";
|
||||
|
||||
function req(id: string, paused: boolean) {
|
||||
return new Request(`http://localhost/api/requests/${id}/pause`, {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json" },
|
||||
body: JSON.stringify({ paused }),
|
||||
});
|
||||
}
|
||||
|
||||
async function seed(jobData: Prisma.JobCreateWithoutRequestInput) {
|
||||
return prisma.request.create({
|
||||
data: { artist: "A", album: "B", status: "pending", job: { create: jobData } },
|
||||
include: { job: true },
|
||||
});
|
||||
}
|
||||
|
||||
describe("per-item pause API", () => {
|
||||
it("pauses a queued job", async () => {
|
||||
const created = await seed({ state: "requested" });
|
||||
const res = await POST(req(created.id, true), { params: Promise.resolve({ id: created.id }) });
|
||||
expect(res.status).toBe(200);
|
||||
const job = await prisma.job.findUnique({ where: { id: created.job!.id } });
|
||||
expect(job?.paused).toBe(true);
|
||||
});
|
||||
|
||||
it("resumes a paused job", async () => {
|
||||
const created = await seed({ state: "requested", paused: true });
|
||||
await POST(req(created.id, false), { params: Promise.resolve({ id: created.id }) });
|
||||
const job = await prisma.job.findUnique({ where: { id: created.job!.id } });
|
||||
expect(job?.paused).toBe(false);
|
||||
});
|
||||
|
||||
it("rejects pausing an in-flight job", async () => {
|
||||
const created = await seed({ state: "downloading", paused: true });
|
||||
const res = await POST(req(created.id, false), { params: Promise.resolve({ id: created.id }) });
|
||||
expect(res.status).toBe(400);
|
||||
const job = await prisma.job.findUnique({ where: { id: created.job!.id } });
|
||||
expect(job?.paused).toBe(true);
|
||||
});
|
||||
|
||||
it("404s for a missing request", async () => {
|
||||
const res = await POST(req("nope", true), { params: Promise.resolve({ id: "nope" }) });
|
||||
expect(res.status).toBe(404);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,26 @@
|
||||
import { prisma } from "@/lib/db";
|
||||
|
||||
export async function POST(request: Request, { params }: { params: Promise<{ id: string }> }) {
|
||||
const { id } = await params;
|
||||
|
||||
let body: { paused?: unknown };
|
||||
try {
|
||||
body = (await request.json()) as { paused?: unknown };
|
||||
} catch {
|
||||
return Response.json({ error: "invalid JSON" }, { status: 400 });
|
||||
}
|
||||
if (typeof body.paused !== "boolean") {
|
||||
return Response.json({ error: "paused must be a boolean" }, { status: 400 });
|
||||
}
|
||||
|
||||
const req = await prisma.request.findUnique({ where: { id }, include: { job: true } });
|
||||
if (!req || !req.job) {
|
||||
return Response.json({ error: "not found" }, { status: 404 });
|
||||
}
|
||||
if (req.job.state !== "requested") {
|
||||
return Response.json({ error: "only queued jobs can be paused" }, { status: 400 });
|
||||
}
|
||||
|
||||
await prisma.job.update({ where: { id: req.job.id }, data: { paused: body.paused } });
|
||||
return Response.json({ ok: true, paused: body.paused });
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import { DELETE } from "./route";
|
||||
import { prisma } from "@/lib/db";
|
||||
import type { Prisma } from "@prisma/client";
|
||||
|
||||
function del(id: string) {
|
||||
return new Request(`http://localhost/api/requests/${id}`, { method: "DELETE" });
|
||||
}
|
||||
|
||||
async function seed(jobData: Prisma.JobCreateWithoutRequestInput) {
|
||||
return prisma.request.create({
|
||||
data: { artist: "A", album: "B", status: "pending", job: { create: jobData } },
|
||||
include: { job: true },
|
||||
});
|
||||
}
|
||||
|
||||
describe("remove request API", () => {
|
||||
it("deletes a queued request and cascades its job", async () => {
|
||||
const created = await seed({ state: "requested" });
|
||||
const res = await DELETE(del(created.id), { params: Promise.resolve({ id: created.id }) });
|
||||
expect(res.status).toBe(200);
|
||||
expect(await prisma.request.findUnique({ where: { id: created.id } })).toBeNull();
|
||||
expect(await prisma.job.findUnique({ where: { id: created.job!.id } })).toBeNull();
|
||||
});
|
||||
|
||||
it("refuses to remove an in-flight job", async () => {
|
||||
const created = await seed({ state: "downloading" });
|
||||
const res = await DELETE(del(created.id), { params: Promise.resolve({ id: created.id }) });
|
||||
expect(res.status).toBe(400);
|
||||
expect(await prisma.request.findUnique({ where: { id: created.id } })).not.toBeNull();
|
||||
});
|
||||
|
||||
it("404s for a missing request", async () => {
|
||||
const res = await DELETE(del("nope"), { params: Promise.resolve({ id: "nope" }) });
|
||||
expect(res.status).toBe(404);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,16 @@
|
||||
import { prisma } from "@/lib/db";
|
||||
|
||||
export async function DELETE(_request: Request, { params }: { params: Promise<{ id: string }> }) {
|
||||
const { id } = await params;
|
||||
|
||||
const req = await prisma.request.findUnique({ where: { id }, include: { job: true } });
|
||||
if (!req) {
|
||||
return Response.json({ error: "not found" }, { status: 404 });
|
||||
}
|
||||
if (req.job && req.job.state !== "requested") {
|
||||
return Response.json({ error: "only queued items can be removed" }, { status: 400 });
|
||||
}
|
||||
|
||||
await prisma.request.delete({ where: { id } }); // cascades Job + Candidates
|
||||
return Response.json({ ok: true });
|
||||
}
|
||||
@@ -64,6 +64,7 @@ export async function GET() {
|
||||
claimedAt: r.job.claimedAt,
|
||||
updatedAt: r.job.updatedAt,
|
||||
downloadProgress: r.job.downloadProgress,
|
||||
paused: r.job.paused,
|
||||
candidateCount: cands.length,
|
||||
chosen: chosen ? { source: chosen.source, format: chosen.format, trackCount: chosen.trackCount } : null,
|
||||
};
|
||||
|
||||
@@ -245,6 +245,9 @@ nav.contents .sep { flex: 1; }
|
||||
.btn.ghost { border-color: var(--rule-2); color: var(--graphite); }
|
||||
.btn.ghost:hover { background: transparent; border-color: var(--ink); color: var(--ink); }
|
||||
|
||||
.press-controls { display: flex; gap: 0.5rem; align-items: center; margin: 0 0 1rem; flex-wrap: wrap; }
|
||||
.item-controls { display: inline-flex; gap: 0.5rem; align-items: center; }
|
||||
|
||||
.toggle {
|
||||
font-family: var(--mono); font-size: 0.64rem; letter-spacing: 0.1em; text-transform: uppercase;
|
||||
color: var(--graphite); display: inline-flex; align-items: center; gap: 7px; cursor: pointer;
|
||||
|
||||
Binary file not shown.
@@ -31,6 +31,7 @@ const MONITOR_FIELDS: [string, string][] = [
|
||||
["monitor.retryIntervalHours", "Retry interval (hours)"],
|
||||
["monitor.qualityCutoff", "Quality cutoff"],
|
||||
["monitor.upgradeWindowDays", "Upgrade window (days)"],
|
||||
["floor.jobDelaySeconds", "Delay between downloads (seconds)"],
|
||||
];
|
||||
|
||||
export function SettingsForm() {
|
||||
|
||||
@@ -13,15 +13,39 @@ def _is_album(g: dict) -> bool:
|
||||
return (g.get("primary-type") or "").casefold() == "album"
|
||||
|
||||
|
||||
def _best_release_group(album: str, groups: list) -> dict | None:
|
||||
"""Choose the best release-group match for `album`. Highest title similarity wins;
|
||||
an Album primary-type breaks ties so a famous album (e.g. Michael Jackson's "Off the
|
||||
Wall") isn't resolved to its same-named single, whose short tracklist would then be
|
||||
mapped positionally onto the full album's files. Returns None below the title threshold.
|
||||
Pure — no network I/O."""
|
||||
def _credited_artist(g: dict) -> str:
|
||||
"""Primary credited artist name of a release-group search result, or '' if absent."""
|
||||
credit = g.get("artist-credit")
|
||||
if isinstance(credit, list) and credit and isinstance(credit[0], dict):
|
||||
return (credit[0].get("artist") or {}).get("name", "") or ""
|
||||
return ""
|
||||
|
||||
|
||||
def _best_release_group(artist: str, album: str, groups: list) -> dict | None:
|
||||
"""Choose the best release-group match for `artist`/`album`.
|
||||
|
||||
Ranked, highest first, on `(artist_ratio, title_ratio, is_album)`:
|
||||
* artist match dominates — MusicBrainz relevance ties same-titled release groups by
|
||||
*different* artists (e.g. "Fun Machine" exists as a Lake Street Dive EP and an
|
||||
unrelated band's Album), and the artist we followed is trustworthy, so a credited
|
||||
artist that matches the request outranks everything else;
|
||||
* title similarity is next;
|
||||
* an Album primary-type only breaks ties *within* the same artist+title, so a famous
|
||||
album (Michael Jackson's "Off the Wall") still isn't resolved to its same-named
|
||||
single whose short tracklist would map positionally onto the album's files.
|
||||
Artist is a ranking signal, never a hard filter — a low match sinks a candidate but
|
||||
never drops the release, so credited-name variations (feat., punctuation) still resolve.
|
||||
Returns None below the title threshold. Pure — no network I/O."""
|
||||
if not groups:
|
||||
return None
|
||||
best = max(groups, key=lambda g: (_ratio(album, g.get("title", "")), _is_album(g)))
|
||||
best = max(
|
||||
groups,
|
||||
key=lambda g: (
|
||||
_ratio(artist, _credited_artist(g)),
|
||||
_ratio(album, g.get("title", "")),
|
||||
_is_album(g),
|
||||
),
|
||||
)
|
||||
if _ratio(album, best.get("title", "")) < _MIN_TITLE_RATIO:
|
||||
return None
|
||||
return best
|
||||
@@ -46,7 +70,7 @@ class MusicBrainzResolver:
|
||||
|
||||
res = musicbrainzngs.search_release_groups(query=album, artist=artist, limit=5)
|
||||
groups = res.get("release-group-list", [])
|
||||
rg = _best_release_group(album, groups)
|
||||
rg = _best_release_group(artist, album, groups)
|
||||
if rg is None:
|
||||
return None
|
||||
|
||||
|
||||
@@ -30,7 +30,7 @@ def claim_next(conn: psycopg.Connection) -> str | None:
|
||||
cur.execute(
|
||||
"""
|
||||
SELECT id FROM "Job"
|
||||
WHERE state = 'requested'
|
||||
WHERE state = 'requested' AND NOT paused
|
||||
ORDER BY "createdAt" ASC
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 1
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
import os
|
||||
|
||||
from lyra_worker.config import get_config
|
||||
|
||||
_TRUE = {"1", "true", "yes", "on"}
|
||||
|
||||
|
||||
def press_paused(config: dict[str, str]) -> bool:
|
||||
"""True when the whole press is paused — the worker stops claiming new jobs."""
|
||||
return str(config.get("floor.paused", "")).strip().lower() in _TRUE
|
||||
|
||||
|
||||
def job_delay_seconds(config: dict[str, str]) -> float:
|
||||
"""Seconds to wait between finishing one download and claiming the next. Invalid or
|
||||
absent -> 0.0 (back-to-back). Negatives are clamped to 0."""
|
||||
try:
|
||||
return max(0.0, float(int(config.get("floor.jobDelaySeconds", "") or 0)))
|
||||
except (TypeError, ValueError):
|
||||
return 0.0
|
||||
|
||||
|
||||
def kill_requested(config: dict[str, str]) -> bool:
|
||||
"""True when a force-stop of the in-flight download has been requested."""
|
||||
return str(config.get("floor.killRequested", "")).strip().lower() in _TRUE
|
||||
|
||||
|
||||
def consume_kill_switch(conn, exit_fn=os._exit) -> bool:
|
||||
"""If a force-stop is pending, clear the flag and terminate the process (default
|
||||
os._exit) so the in-flight download is aborted. The container restart policy brings the
|
||||
worker back; startup clears staging and requeues the aborted job. Returns whether it
|
||||
fired. exit_fn is injectable for tests."""
|
||||
config = get_config(conn)
|
||||
if not kill_requested(config):
|
||||
return False
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
'UPDATE "Config" SET value = \'false\', "updatedAt" = now() WHERE key = %s',
|
||||
("floor.killRequested",),
|
||||
)
|
||||
conn.commit()
|
||||
exit_fn(0)
|
||||
return True
|
||||
@@ -11,6 +11,7 @@ from lyra_worker.config import get_config
|
||||
from lyra_worker.crypto import secret_key_problem
|
||||
from lyra_worker.db import wait_for_db
|
||||
from lyra_worker.discovery import DiscoveryConfig, DiscoveryResult, reset_pending, run_discovery
|
||||
from lyra_worker.floor import press_paused, job_delay_seconds, consume_kill_switch
|
||||
from lyra_worker.library import clear_staging_root
|
||||
from lyra_worker.monitor import MonitorConfig, fulfill_owned_releases, reconcile, sweep
|
||||
from lyra_worker.pipeline import run_pipeline
|
||||
@@ -226,9 +227,30 @@ def _liveness_heartbeat() -> None:
|
||||
time.sleep(HEARTBEAT_SECONDS)
|
||||
|
||||
|
||||
KILL_POLL_SECONDS = 1.5
|
||||
|
||||
|
||||
def _kill_switch_watch() -> None:
|
||||
"""Poll floor.killRequested on a background daemon thread (its own DB connection) so a
|
||||
force-stop aborts the process even while the main loop is blocked in a download. On any
|
||||
DB hiccup, reconnect and keep watching."""
|
||||
conn = wait_for_db()
|
||||
while True:
|
||||
try:
|
||||
consume_kill_switch(conn)
|
||||
except Exception:
|
||||
try:
|
||||
conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
conn = wait_for_db()
|
||||
time.sleep(KILL_POLL_SECONDS)
|
||||
|
||||
|
||||
def run_forever() -> None:
|
||||
_require_secret_key()
|
||||
threading.Thread(target=_liveness_heartbeat, daemon=True).start()
|
||||
threading.Thread(target=_kill_switch_watch, daemon=True).start()
|
||||
conn = wait_for_db()
|
||||
clear_staging_root(STAGING_ROOT) # sweep any staging dirs orphaned by a prior crash
|
||||
reclaimed = reclaim_stuck_jobs(conn) # requeue jobs a prior crash left mid-pipeline
|
||||
@@ -252,6 +274,7 @@ def run_forever() -> None:
|
||||
last_discover_tick = 0.0
|
||||
last_heartbeat = 0.0
|
||||
last_prune = 0.0
|
||||
next_job_allowed = 0.0 # monotonic time before which no new job is claimed (inter-job delay)
|
||||
try:
|
||||
while True:
|
||||
# A transient DB connection loss (e.g. AdminShutdown when Postgres restarts) raises
|
||||
@@ -283,6 +306,12 @@ def run_forever() -> None:
|
||||
|
||||
maybe_run_scan(conn, resolver, probe, browser, config)
|
||||
|
||||
# Press paused (downloads only) or the inter-job delay hasn't elapsed: keep the
|
||||
# loop alive (monitor/scan/discovery above still run) but claim nothing.
|
||||
if press_paused(config) or now < next_job_allowed:
|
||||
time.sleep(IDLE_SLEEP)
|
||||
continue
|
||||
|
||||
job_id = claim_next(conn)
|
||||
if job_id is None:
|
||||
time.sleep(IDLE_SLEEP)
|
||||
@@ -296,6 +325,7 @@ def run_forever() -> None:
|
||||
conn.rollback()
|
||||
finish_job(conn, job_id, mcfg)
|
||||
print(f"worker: finished job {job_id}", flush=True)
|
||||
next_job_allowed = time.monotonic() + job_delay_seconds(config)
|
||||
except psycopg.OperationalError as e:
|
||||
print(f"worker: db connection lost ({e}); reconnecting...", flush=True)
|
||||
try:
|
||||
|
||||
@@ -24,3 +24,17 @@ def test_claim_does_not_repick_claimed_job(conn):
|
||||
insert_request(conn)
|
||||
assert claim_next(conn) is not None
|
||||
assert claim_next(conn) is None
|
||||
|
||||
|
||||
def test_claim_skips_paused_jobs(conn):
|
||||
paused_job = insert_request(conn, album="Paused Album")
|
||||
with conn.cursor() as cur:
|
||||
cur.execute('UPDATE "Job" SET paused = true WHERE id = %s', (paused_job,))
|
||||
conn.commit()
|
||||
unpaused_job = insert_request(conn, album="Live Album")
|
||||
|
||||
claimed = claim_next(conn)
|
||||
assert claimed == unpaused_job # the paused one is skipped
|
||||
|
||||
# With only the paused job left, nothing is claimable.
|
||||
assert claim_next(conn) is None
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
from lyra_worker.floor import (
|
||||
press_paused,
|
||||
job_delay_seconds,
|
||||
kill_requested,
|
||||
consume_kill_switch,
|
||||
)
|
||||
|
||||
|
||||
def test_press_paused_reads_truthy_values():
|
||||
assert press_paused({"floor.paused": "true"}) is True
|
||||
assert press_paused({"floor.paused": "ON"}) is True
|
||||
assert press_paused({"floor.paused": "false"}) is False
|
||||
assert press_paused({}) is False
|
||||
|
||||
|
||||
def test_job_delay_seconds_parses_and_defaults():
|
||||
assert job_delay_seconds({"floor.jobDelaySeconds": "30"}) == 30.0
|
||||
assert job_delay_seconds({}) == 0.0
|
||||
assert job_delay_seconds({"floor.jobDelaySeconds": "-5"}) == 0.0
|
||||
assert job_delay_seconds({"floor.jobDelaySeconds": "junk"}) == 0.0
|
||||
|
||||
|
||||
def test_kill_requested_reads_flag():
|
||||
assert kill_requested({"floor.killRequested": "true"}) is True
|
||||
assert kill_requested({}) is False
|
||||
|
||||
|
||||
def test_consume_kill_switch_clears_flag_and_exits(conn):
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
'INSERT INTO "Config"(key, value, secret, "updatedAt") '
|
||||
"VALUES ('floor.killRequested', 'true', false, now())"
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
calls = []
|
||||
fired = consume_kill_switch(conn, exit_fn=lambda code: calls.append(code))
|
||||
|
||||
assert fired is True
|
||||
assert calls == [0]
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT value FROM \"Config\" WHERE key = 'floor.killRequested'")
|
||||
assert cur.fetchone()[0] == "false"
|
||||
|
||||
|
||||
def test_consume_kill_switch_noop_when_unset(conn):
|
||||
calls = []
|
||||
fired = consume_kill_switch(conn, exit_fn=lambda code: calls.append(code))
|
||||
assert fired is False
|
||||
assert calls == []
|
||||
@@ -4,33 +4,59 @@ from lyra_worker._musicbrainz import _best_release_group
|
||||
# path (network) is covered by test_musicbrainz_live.py.
|
||||
|
||||
|
||||
def _rg(title, primary_type):
|
||||
return {"id": title.lower(), "title": title, "primary-type": primary_type}
|
||||
def _rg(title, primary_type, artist=""):
|
||||
g = {"id": title.lower(), "title": title, "primary-type": primary_type}
|
||||
if artist:
|
||||
g["artist-credit"] = [{"artist": {"name": artist, "id": artist.lower()}}]
|
||||
return g
|
||||
|
||||
|
||||
def test_prefers_album_over_same_titled_single():
|
||||
# Real-world "Off the Wall" bug: MusicBrainz returns the Single release-group
|
||||
# first, tied on title with the Album. We must pick the Album so a 10-track
|
||||
# download isn't tagged from the 2-track single's tracklist.
|
||||
groups = [_rg("Off the Wall", "Single"), _rg("Off the Wall", "Album")]
|
||||
assert _best_release_group("Off the Wall", groups)["primary-type"] == "Album"
|
||||
groups = [
|
||||
_rg("Off the Wall", "Single", "Michael Jackson"),
|
||||
_rg("Off the Wall", "Album", "Michael Jackson"),
|
||||
]
|
||||
assert _best_release_group("Michael Jackson", "Off the Wall", groups)["primary-type"] == "Album"
|
||||
|
||||
|
||||
def test_album_preference_is_order_independent():
|
||||
groups = [_rg("Off the Wall", "Album"), _rg("Off the Wall", "Single")]
|
||||
assert _best_release_group("Off the Wall", groups)["primary-type"] == "Album"
|
||||
groups = [
|
||||
_rg("Off the Wall", "Album", "Michael Jackson"),
|
||||
_rg("Off the Wall", "Single", "Michael Jackson"),
|
||||
]
|
||||
assert _best_release_group("Michael Jackson", "Off the Wall", groups)["primary-type"] == "Album"
|
||||
|
||||
|
||||
def test_closer_title_still_wins_over_album_type():
|
||||
# Title similarity dominates; the album-type preference only breaks ties.
|
||||
groups = [_rg("Continuum", "Single"), _rg("Continuum Deluxe Reissue", "Album")]
|
||||
assert _best_release_group("Continuum", groups)["title"] == "Continuum"
|
||||
# Title similarity dominates the album-type preference (for a single artist).
|
||||
groups = [
|
||||
_rg("Continuum", "Single", "John Mayer"),
|
||||
_rg("Continuum Deluxe Reissue", "Album", "John Mayer"),
|
||||
]
|
||||
assert _best_release_group("John Mayer", "Continuum", groups)["title"] == "Continuum"
|
||||
|
||||
|
||||
def test_below_threshold_returns_none():
|
||||
groups = [_rg("Something Totally Different", "Album")]
|
||||
assert _best_release_group("Off the Wall", groups) is None
|
||||
groups = [_rg("Something Totally Different", "Album", "Michael Jackson")]
|
||||
assert _best_release_group("Michael Jackson", "Off the Wall", groups) is None
|
||||
|
||||
|
||||
def test_empty_groups_returns_none():
|
||||
assert _best_release_group("Off the Wall", []) is None
|
||||
assert _best_release_group("Michael Jackson", "Off the Wall", []) is None
|
||||
|
||||
|
||||
def test_prefers_credited_artist_over_same_titled_other_artist_album():
|
||||
# Real-world "Fun Machine" bug: Lake Street Dive's EP is what we followed, but two
|
||||
# unrelated bands also have a release group literally titled "Fun Machine" as an Album.
|
||||
# All three tie on title (1.0), so the old is_album tiebreak wrongly grabbed a foreign
|
||||
# Album. The followed artist must win over title-tied albums by other artists.
|
||||
groups = [
|
||||
_rg("Fun Machine", "EP", "Lake Street Dive"),
|
||||
_rg("Fun Machine", "Album", "Bastards of Melody"),
|
||||
_rg("Fun Machine", "Album", "Teledildonics 5000"),
|
||||
]
|
||||
best = _best_release_group("Lake Street Dive", "Fun Machine", groups)
|
||||
assert best["artist-credit"][0]["artist"]["name"] == "Lake Street Dive"
|
||||
|
||||
Reference in New Issue
Block a user