fix(worker): cap slskd ETA + rollback on progress-write failure (stop txn-abort job churn)

A near-stalled Soulseek peer drives the EWMA speed toward zero, so
_eta_seconds returned int(remaining/speed) values far past int4 max. Writing
that to Job.downloadEtaSeconds (an integer column) raised "integer out of
range", which aborted the SHARED pipeline connection's transaction. The
on_progress except clause logged but never rolled back, so every subsequent
query cascaded "current transaction is aborted" and the whole job failed and
requeued (9 such failures / 24h observed in prod).

- Cap _eta_seconds at ~100h (359999s), well inside int4.
- Roll back the shared conn in on_progress's except so a failed progress
  write can never poison the pipeline txn, honoring the existing docstring
  promise that "a progress write must never fail the download".
- Regression test for the near-stall cap.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

web/prisma/migrations/20260720200000_add_job_download_eta/migration.sql

@@ -0,0 +1,2 @@
+-- Estimated seconds remaining for the active download (measured throughput). Nullable.
+ALTER TABLE "Job" ADD COLUMN "downloadEtaSeconds" INTEGER;

web/prisma/schema.prisma

@@ -58,6 +58,9 @@ model Job {
   createdAt        DateTime    @default(now())
   updatedAt        DateTime    @updatedAt
   downloadProgress Float       @default(0)
+  // Estimated seconds until the current download finishes, from measured throughput (slskd byte
+  // transfer rate). Null when unknown (no speed sample yet, or a source that doesn't report bytes).
+  downloadEtaSeconds Int?
   paused           Boolean     @default(false)
   candidates       Candidate[]
 }

web/src/app/_ui/job-row.tsx

@@ -14,6 +14,7 @@ export function JobRow({
   album,
   kind,
   label,
+  marker,
   meta,
   note,
   bar,
@@ -22,6 +23,7 @@ export function JobRow({
   album: string;
   kind: Kind;
   label: string;
+  marker?: string;
   meta?: ReactNode;
   note?: ReactNode;
   bar?: ReactNode;
@@ -31,6 +33,7 @@ export function JobRow({
       <div className="stripe" />
       <div>
         <h3 className="title">
+          {marker ? <span className="nextup">{marker}</span> : null}
           {album} <span className="artist">· {artist}</span>
         </h3>
         {meta ? <div className="meta">{meta}</div> : null}

web/src/app/_ui/kebab-menu.tsx

@@ -0,0 +1,47 @@
+"use client";
+
+import { useEffect, useRef, useState, type ReactNode } from "react";
+
+/** A vertical "⋮" trigger that opens a small popup menu. Closes on outside click, Escape, or any
+ *  click inside the popup (so menu items don't each need to manage it). Children are the items —
+ *  use <button className="menu-item">…</button>. */
+export function KebabMenu({ label = "Actions", children }: { label?: string; children: ReactNode }) {
+  const [open, setOpen] = useState(false);
+  const ref = useRef<HTMLDivElement>(null);
+
+  useEffect(() => {
+    if (!open) return;
+    function onDoc(e: MouseEvent) {
+      if (ref.current && !ref.current.contains(e.target as Node)) setOpen(false);
+    }
+    function onKey(e: KeyboardEvent) {
+      if (e.key === "Escape") setOpen(false);
+    }
+    document.addEventListener("mousedown", onDoc);
+    document.addEventListener("keydown", onKey);
+    return () => {
+      document.removeEventListener("mousedown", onDoc);
+      document.removeEventListener("keydown", onKey);
+    };
+  }, [open]);
+
+  return (
+    <div className="menu" ref={ref}>
+      <button
+        type="button"
+        className="menu-trigger"
+        aria-haspopup="menu"
+        aria-expanded={open}
+        aria-label={label}
+        onClick={() => setOpen((o) => !o)}
+      >
+        ⋮
+      </button>
+      {open ? (
+        <div className="menu-pop" role="menu" onClick={() => setOpen(false)}>
+          {children}
+        </div>
+      ) : null}
+    </div>
+  );
+}

web/src/app/_ui/status.test.ts

@@ -1,5 +1,20 @@
 import { describe, it, expect } from "vitest";
-import { describeJob, timeAgo } from "./status";
+import { describeJob, etaLabel, timeAgo } from "./status";
+
+describe("etaLabel", () => {
+  it("formats seconds, minutes, and hours", () => {
+    expect(etaLabel(45)).toBe("~45s left");
+    expect(etaLabel(240)).toBe("~4m left");
+    expect(etaLabel(3900)).toBe("~1h 5m left");
+  });
+
+  it("returns null for missing or non-positive input", () => {
+    expect(etaLabel(null)).toBeNull();
+    expect(etaLabel(undefined)).toBeNull();
+    expect(etaLabel(0)).toBeNull();
+    expect(etaLabel(-5)).toBeNull();
+  });
+});
 
 describe("describeJob", () => {
   it("maps pipeline states to literal labels + severity kind", () => {

web/src/app/_ui/status.ts

@@ -27,6 +27,16 @@ export function timeAgo(value: string | Date | null | undefined, now: number): s
   return `${Math.floor(h / 24)}d`;
 }
 
+/** Compact download ETA like "~4m left", "~45s left", "~1h 5m left". Null/≤0 → no label. */
+export function etaLabel(seconds: number | null | undefined): string | null {
+  if (seconds == null || seconds <= 0) return null;
+  if (seconds < 60) return `~${seconds}s left`;
+  const m = Math.round(seconds / 60);
+  if (m < 60) return `~${m}m left`;
+  const h = Math.floor(m / 60);
+  return `~${h}h ${m % 60}m left`;
+}
+
 export function describeJob(state: string, stage: string): { label: string; kind: Kind } {
   void stage; // stage no longer drives the label; kept for call-site compatibility
   return (

web/src/app/api/requests/route.ts

@@ -64,6 +64,7 @@ export async function GET() {
               claimedAt: r.job.claimedAt,
               updatedAt: r.job.updatedAt,
               downloadProgress: r.job.downloadProgress,
+              downloadEtaSeconds: r.job.downloadEtaSeconds,
               paused: r.job.paused,
               candidateCount: cands.length,
               chosen: chosen ? { source: chosen.source, format: chosen.format, trackCount: chosen.trackCount } : null,

web/src/app/design.css

@@ -386,6 +386,39 @@ nav.contents .sep { flex: 1; }
 .tab:hover { color: var(--ink); }
 .tab.active { color: var(--ink); border-bottom-color: var(--accent); }
 .tab .n { color: var(--rule-2); margin-left: 6px; }
+.tab.has-errors .n { color: var(--alert); }
+
+/* "Next up" badge on the first queued row (the item the worker claims next) */
+.nextup {
+  display: inline-block; vertical-align: middle; margin-right: 9px; transform: translateY(-2px);
+  font-family: var(--mono); font-size: 0.56rem; letter-spacing: 0.13em; text-transform: uppercase;
+  color: var(--accent-ink); background: var(--accent); padding: 2px 7px; border-radius: 4px;
+}
+
+/* ── Kebab menu (⋮ popup for press actions) ───────────── */
+.dept.has-actions { align-items: center; }
+.dept.has-actions .fill { transform: none; }
+.menu { position: relative; display: inline-flex; }
+.menu-trigger {
+  font-size: 1.15rem; line-height: 1; color: var(--graphite); background: transparent;
+  border: 1px solid var(--rule); border-radius: 6px; width: 32px; height: 30px; cursor: pointer;
+  display: inline-flex; align-items: center; justify-content: center;
+}
+.menu-trigger:hover { color: var(--ink); border-color: var(--rule-2); }
+.menu-pop {
+  position: absolute; top: calc(100% + 6px); right: 0; z-index: 30; min-width: 190px;
+  background: var(--paper-2); border: 1px solid var(--rule-2); border-radius: 9px;
+  box-shadow: 0 10px 30px rgba(0, 0, 0, 0.22); padding: 5px;
+  display: flex; flex-direction: column; gap: 1px;
+}
+.menu-item {
+  font-family: var(--mono); font-size: 0.68rem; letter-spacing: 0.1em; text-transform: uppercase;
+  text-align: left; background: transparent; border: 0; color: var(--ink);
+  padding: 10px 11px; border-radius: 6px; cursor: pointer; white-space: nowrap;
+}
+.menu-item:hover { background: var(--rule); }
+.menu-item:disabled { color: var(--rule-2); cursor: default; background: transparent; }
+.menu-item.danger { color: var(--alert); }
 
 /* ── Discography row (cover thumb + click-to-modal) ───── */
 .disco-open {

web/src/app/queue.tsx

@@ -1,14 +1,34 @@
 "use client";
 
 import { useEffect, useState } from "react";
-import { describeJob, timeAgo } from "./_ui/status";
+import { describeJob, etaLabel, timeAgo } from "./_ui/status";
 import { StatTiles } from "./_ui/stat-tiles";
 import { SectionHeader } from "./_ui/section-header";
 import { JobRow } from "./_ui/job-row";
 import { ProgressBar } from "./_ui/progress-bar";
 import { PressedList } from "./_ui/pressed-list";
+import { KebabMenu } from "./_ui/kebab-menu";
 import { toast } from "./_ui/toast";
 
+type Tab = "active" | "queue" | "errors";
+
+// Split the on-the-press rows into three honest buckets so active presses aren't buried between
+// queued items and errors. Active = being worked (searching/downloading/finishing), Queue =
+// waiting to be claimed, Errors = needs attention.
+function categoryOf(state: string, stage: string): Tab {
+  const kind = describeJob(state, stage).kind;
+  if (kind === "attention") return "errors";
+  if (state === "requested") return "queue";
+  return "active";
+}
+
+const TAB_LABEL: Record<Tab, string> = { active: "Active", queue: "In queue", errors: "Errors" };
+const TAB_EMPTY: Record<Tab, string> = {
+  active: "Nothing pressing right now.",
+  queue: "The queue is empty.",
+  errors: "No errors — everything's flowing.",
+};
+
 type Row = {
   id: string;
   artist: string;
@@ -26,6 +46,7 @@ type Row = {
     candidateCount: number;
     chosen: { source: string; format: string; trackCount: number } | null;
     downloadProgress: number;
+    downloadEtaSeconds: number | null;
     paused: boolean;
   } | null;
 };
@@ -42,14 +63,16 @@ function jobOf(r: Row) {
       candidateCount: 0,
       chosen: null,
       downloadProgress: 0,
+      downloadEtaSeconds: null,
       paused: false,
     }
   );
 }
 
 function pressedMark(r: Row): string {
-  if (!r.createdAt) return "Lyra";
-  const d = new Date(r.createdAt);
+  const when = r.job?.updatedAt ?? r.createdAt; // when it was pressed, not when it was requested
+  if (!when) return "Lyra";
+  const d = new Date(when);
   return `Lyra · ${d.toLocaleDateString("en-GB", { day: "2-digit", month: "short" }).toUpperCase()}`;
 }
 
@@ -60,6 +83,9 @@ export function Queue() {
   const [artist, setArtist] = useState("");
   const [album, setAlbum] = useState("");
   const [paused, setPaused] = useState(false);
+  // null = follow the smart default (Active if anything's active, else Queue, else Errors); once
+  // the user picks a tab, their choice sticks.
+  const [tab, setTab] = useState<Tab | null>(null);
 
   async function refresh() {
     const [reqRes, wantRes, artRes, floorRes] = await Promise.all([
@@ -159,53 +185,166 @@ export function Queue() {
   }
 
   const active = rows.filter((r) => jobOf(r).state !== "imported");
-  const downloading = rows.some((r) => {
-    const s = jobOf(r).state;
-    return s === "matching" || s === "matched" || s === "downloading" || s === "tagging";
-  });
-  // Dedupe by artist+album — re-requesting/re-downloading the same album leaves several
-  // completed Requests; show each album once (rows arrive newest-first, so keep the first).
+  const buckets: Record<Tab, Row[]> = { active: [], queue: [], errors: [] };
+  for (const r of active) {
+    const j = jobOf(r);
+    buckets[categoryOf(j.state, j.currentStage)].push(r);
+  }
+  // The API returns rows newest-first, but the worker claims the OLDEST queued job next
+  // (claim_next: ORDER BY createdAt ASC). Order the queue tab to match, so the item that runs
+  // next sits at the top instead of the bottom.
+  const ts = (r: Row) => (r.createdAt ? new Date(r.createdAt).getTime() : 0);
+  buckets.queue.sort((a, b) => ts(a) - ts(b));
+  // "Next up" = the first queued item the worker will actually claim (oldest, not individually
+  // paused) — mark it so the ordering reads clearly.
+  const nextUpId = buckets.queue.find((r) => !jobOf(r).paused)?.id ?? null;
+  const queuedCount = buckets.queue.length;
+  const downloading = buckets.active.length > 0;
+  // Smart default: land on the tab that has something worth looking at.
+  const defaultTab: Tab = buckets.active.length ? "active" : queuedCount ? "queue" : "errors";
+  const activeTab: Tab = tab ?? defaultTab;
+  const shown = buckets[activeTab];
+
+  // "Recently Pressed": imported albums ordered by when they were PRESSED (job.updatedAt ≈ import
+  // time), newest first — an album requested long ago but imported just now is recently pressed
+  // even though its Request is old (rows arrive by request createdAt, which buried fresh imports).
+  // Dedupe by artist+album (re-downloads leave several completed Requests; keep the most recent).
   const seenPressed = new Set<string>();
-  const pressed = rows.filter((r) => {
-    if (jobOf(r).state !== "imported") return false;
-    const key = `${r.artist} ${r.album}`.toLowerCase();
-    if (seenPressed.has(key)) return false;
-    seenPressed.add(key);
-    return true;
-  });
+  const pressedAll = rows
+    .filter((r) => jobOf(r).state === "imported")
+    .sort((a, b) => new Date(jobOf(b).updatedAt ?? 0).getTime() - new Date(jobOf(a).updatedAt ?? 0).getTime())
+    .filter((r) => {
+      const key = `${r.artist} ${r.album}`.toLowerCase();
+      if (seenPressed.has(key)) return false;
+      seenPressed.add(key);
+      return true;
+    });
+  const pressed = pressedAll.slice(0, 10); // show only the 10 most recently pressed
+
+  const now = Date.now();
+  function renderRow(r: Row) {
+    const j = jobOf(r);
+    const d = describeJob(j.state, j.currentStage);
+    const attention = d.kind === "attention";
+
+    let meta: string | undefined;
+    if (j.state === "downloading" || j.state === "tagging") {
+      const parts: string[] = [];
+      if (j.chosen) parts.push(`${j.chosen.source} · ${j.chosen.format}`);
+      const elapsed = timeAgo(j.claimedAt ?? j.updatedAt, now);
+      if (elapsed) parts.push(elapsed);
+      meta = parts.join(" · ") || undefined;
+    } else if (j.state === "matching" || j.state === "matched") {
+      meta =
+        j.candidateCount > 0
+          ? `${j.candidateCount} sources${j.chosen ? ` · best: ${j.chosen.source} ${j.chosen.format}` : ""}`
+          : "Searching sources…";
+    } else if (j.state === "requested") {
+      const elapsed = timeAgo(r.createdAt, now);
+      meta = elapsed ? `In the queue · ${elapsed}` : "In the queue";
+    } else {
+      meta = undefined;
+    }
+
+    const note = attention ? (
+      <>
+        {j.error?.trim() || "No source met the quality cutoff, or the pipeline stalled."}{" "}
+        <button className="btn sm" onClick={() => retry(r.id)}>
+          Retry
+        </button>
+      </>
+    ) : j.state === "requested" ? (
+      <span className="item-controls">
+        {j.paused ? (
+          <>
+            <span className="chip">Paused</span>{" "}
+            <button className="btn sm ghost" onClick={() => pauseItem(r.id, false)}>
+              Resume
+            </button>
+          </>
+        ) : (
+          <button className="btn sm ghost" onClick={() => pauseItem(r.id, true)}>
+            Pause
+          </button>
+        )}{" "}
+        <button className="btn sm ghost" onClick={() => removeItem(r.id)}>
+          Remove
+        </button>
+      </span>
+    ) : undefined;
+
+    let bar: React.ReactNode;
+    if (j.state === "matching" || j.state === "matched") {
+      bar = <ProgressBar kind="working" indeterminate caption="searching" />;
+    } else if (j.state === "downloading") {
+      const pct = Math.round((j.downloadProgress || 0) * 100);
+      if (pct <= 0) {
+        // Bytes haven't started flowing yet — Qobuz login/metadata/artwork, a Soulseek peer
+        // queue, or a YouTube info-fetch. Show an indeterminate bar instead of a stuck-looking
+        // 0% (it flips to the % bar on the first byte).
+        bar = <ProgressBar kind="working" indeterminate caption="preparing…" />;
+      } else {
+        const total = j.chosen?.trackCount ?? 0;
+        const done = total ? Math.round((pct / 100) * total) : 0;
+        const base = total ? `${pct}% · ${done}/${total} tracks` : `${pct}%`;
+        const eta = etaLabel(j.downloadEtaSeconds);
+        bar = <ProgressBar kind="working" percent={pct} caption={eta ? `${base} · ${eta}` : base} />;
+      }
+    } else if (j.state === "tagging") {
+      bar = <ProgressBar kind="verify" indeterminate caption="finishing" />;
+    } else {
+      bar = undefined;
+    }
+
+    return (
+      <JobRow
+        key={r.id}
+        artist={r.artist}
+        album={r.album}
+        kind={d.kind}
+        label={d.label}
+        marker={r.id === nextUpId ? "Next up" : undefined}
+        meta={meta}
+        note={note}
+        bar={bar}
+      />
+    );
+  }
 
   return (
     <div>
       <StatTiles
         items={[
           { k: "On the press", v: active.length },
-          { k: "Pressed", v: pressed.length, accent: true },
+          { k: "Pressed", v: pressedAll.length, accent: true },
           { k: "Wanted", v: wanted },
           { k: "Watching", v: watching, unit: "artists" },
         ]}
       />
 
-      <SectionHeader title="On the Press" note={`${active.length} running`} />
-
-      <div className="press-controls">
-        {paused ? (
-          <button className="btn sm accent" onClick={() => floorAction("resume", "Press resumed")}>
-            Resume the press
+      <div className="dept has-actions">
+        <h2>On the Press</h2>
+        <span className="fill" />
+        {paused ? <span className="count">paused</span> : null}
+        <KebabMenu label="Press actions">
+          {paused ? (
+            <button className="menu-item" onClick={() => floorAction("resume", "Press resumed")}>
+              Resume the press
+            </button>
+          ) : (
+            <button className="menu-item" onClick={() => floorAction("pause", "Press paused")}>
+              Pause the press
+            </button>
+          )}
+          <button className="menu-item" onClick={clearPress} disabled={queuedCount === 0}>
+            Clear queue
           </button>
-        ) : (
-          <button className="btn sm ghost" onClick={() => floorAction("pause", "Press paused")}>
-            Pause the press
-          </button>
-        )}
-        <button className="btn sm ghost" onClick={clearPress}>
-          Clear queue
-        </button>
-        {downloading ? (
-          <button className="btn sm" onClick={stopPress}>
-            Stop now
-          </button>
-        ) : null}
-        {paused ? <span className="chip">Press paused</span> : null}
+          {downloading ? (
+            <button className="menu-item danger" onClick={stopPress}>
+              Stop now
+            </button>
+          ) : null}
+        </KebabMenu>
       </div>
 
       <form className="request-form" onSubmit={submit}>
@@ -225,107 +364,32 @@ export function Queue() {
       {active.length === 0 ? (
         <p className="empty">The press is quiet. Queue a release above, or send one over from Wanted or Discover.</p>
       ) : (
-        <section className="floor">
-          {(() => {
-            const now = Date.now();
-            return active.map((r) => {
-              const j = jobOf(r);
-              const d = describeJob(j.state, j.currentStage);
-              const attention = d.kind === "attention";
-
-              let meta: string | undefined;
-              if (j.state === "downloading" || j.state === "tagging") {
-                const parts: string[] = [];
-                if (j.chosen) parts.push(`${j.chosen.source} · ${j.chosen.format}`);
-                const elapsed = timeAgo(j.claimedAt ?? j.updatedAt, now);
-                if (elapsed) parts.push(elapsed);
-                meta = parts.join(" · ") || undefined;
-              } else if (j.state === "matching" || j.state === "matched") {
-                meta =
-                  j.candidateCount > 0
-                    ? `${j.candidateCount} sources${j.chosen ? ` · best: ${j.chosen.source} ${j.chosen.format}` : ""}`
-                    : "Searching sources…";
-              } else if (j.state === "requested") {
-                const elapsed = timeAgo(r.createdAt, now);
-                meta = elapsed ? `In the queue · ${elapsed}` : "In the queue";
-              } else {
-                meta = undefined;
-              }
-
-              const note = attention ? (
-                <>
-                  {j.error?.trim() || "No source met the quality cutoff, or the pipeline stalled."}{" "}
-                  <button className="btn sm" onClick={() => retry(r.id)}>
-                    Retry
-                  </button>
-                </>
-              ) : j.state === "requested" ? (
-                <span className="item-controls">
-                  {j.paused ? (
-                    <>
-                      <span className="chip">Paused</span>{" "}
-                      <button className="btn sm ghost" onClick={() => pauseItem(r.id, false)}>
-                        Resume
-                      </button>
-                    </>
-                  ) : (
-                    <button className="btn sm ghost" onClick={() => pauseItem(r.id, true)}>
-                      Pause
-                    </button>
-                  )}{" "}
-                  <button className="btn sm ghost" onClick={() => removeItem(r.id)}>
-                    Remove
-                  </button>
-                </span>
-              ) : undefined;
-
-              let bar: React.ReactNode;
-              if (j.state === "matching" || j.state === "matched") {
-                bar = <ProgressBar kind="working" indeterminate caption="searching" />;
-              } else if (j.state === "downloading") {
-                const pct = Math.round((j.downloadProgress || 0) * 100);
-                if (pct <= 0) {
-                  // Bytes haven't started flowing yet — Qobuz login/metadata/artwork, a
-                  // Soulseek peer queue, or a YouTube info-fetch. Show an indeterminate bar
-                  // instead of a stuck-looking 0% (it flips to the % bar on the first byte).
-                  bar = <ProgressBar kind="working" indeterminate caption="preparing…" />;
-                } else {
-                  const total = j.chosen?.trackCount ?? 0;
-                  const done = total ? Math.round((pct / 100) * total) : 0;
-                  bar = (
-                    <ProgressBar
-                      kind="working"
-                      percent={pct}
-                      caption={total ? `${pct}% · ${done}/${total} tracks` : `${pct}%`}
-                    />
-                  );
-                }
-              } else if (j.state === "tagging") {
-                bar = <ProgressBar kind="verify" indeterminate caption="finishing" />;
-              } else {
-                bar = undefined;
-              }
-
-              return (
-                <JobRow
-                  key={r.id}
-                  artist={r.artist}
-                  album={r.album}
-                  kind={d.kind}
-                  label={d.label}
-                  meta={meta}
-                  note={note}
-                  bar={bar}
-                />
-              );
-            });
-          })()}
-        </section>
+        <>
+          <div className="tabs" role="tablist">
+            {(["active", "queue", "errors"] as Tab[]).map((t) => (
+              <button
+                key={t}
+                role="tab"
+                aria-selected={activeTab === t}
+                className={`tab${activeTab === t ? " active" : ""}${t === "errors" && buckets.errors.length ? " has-errors" : ""}`}
+                onClick={() => setTab(t)}
+              >
+                {TAB_LABEL[t]}
+                <span className="n">{buckets[t].length}</span>
+              </button>
+            ))}
+          </div>
+          {shown.length === 0 ? (
+            <p className="empty">{TAB_EMPTY[activeTab]}</p>
+          ) : (
+            <section className="floor">{shown.map(renderRow)}</section>
+          )}
+        </>
       )}
 
       {pressed.length > 0 ? (
         <>
-          <SectionHeader title="Recently Pressed" note={`${pressed.length} in library`} />
+          <SectionHeader title="Recently Pressed" note={`last ${pressed.length} · ${pressedAll.length} in library`} />
           <PressedList
             items={pressed.map((r) => ({ id: r.id, artist: r.artist, album: r.album, matrix: pressedMark(r), rgMbid: r.rgMbid ?? null }))}
           />

worker/lyra_worker/_musicbrainz.py

@@ -37,6 +37,12 @@ def _is_album(g: dict) -> bool:
     return (g.get("primary-type") or "").casefold() == "album"
 
 
+def _rg_year(g: dict) -> int:
+    """First-release year of a release-group, or 9999 when unknown (so undated groups lose ties)."""
+    frd = (g.get("first-release-date") or "")[:4]
+    return int(frd) if frd.isdigit() else 9999
+
+
 def _credited_artist(g: dict) -> str:
     """Primary credited artist name of a release-group search result, or '' if absent."""
     credit = g.get("artist-credit")
@@ -56,7 +62,12 @@ def _best_release_group(artist: str, album: str, groups: list) -> dict | None:
       * 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.
+        single whose short tracklist would map positionally onto the album's files;
+      * finally, the EARLIEST release-group wins — MusicBrainz lists the original album and its
+        later reissues/anniversary editions as separate same-titled groups (Linkin Park "Hybrid
+        Theory": a 2000 original and a 2023 13-track reissue), which tie on everything above, so
+        MB's arbitrary relevance order used to pick a reissue whose longer tracklist/runtime then
+        failed completeness checks against the standard album a source delivers. Prefer the original.
     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."""
@@ -68,6 +79,7 @@ def _best_release_group(artist: str, album: str, groups: list) -> dict | None:
             _ratio(artist, _credited_artist(g)),
             _ratio(album, g.get("title", "")),
             _is_album(g),
+            -_rg_year(g),  # tie-break: the original (earliest) group over later reissues
         ),
     )
     if _ratio(album, best.get("title", "")) < _MIN_TITLE_RATIO:
@@ -75,6 +87,23 @@ def _best_release_group(artist: str, album: str, groups: list) -> dict | None:
     return best
 
 
+def _pick_release(releases: list) -> dict | None:
+    """Choose which release of a release-group to read the tracklist from. MB returns them in an
+    arbitrary order and ``releases[0]`` is often a later reissue/deluxe/anniversary edition with
+    more tracks and a longer runtime than the standard album a source actually delivers — which
+    then makes a *complete* download fail the pipeline's track-count/duration completeness checks
+    and re-download forever (Linkin Park "Hybrid Theory": MB's arbitrary pick was a 2023 13-track
+    reissue at ~52 min vs the 2000 original's 12 tracks / ~38 min). Prefer the ORIGINAL standard
+    edition: an Official release, then the earliest date, then a dated one. Pure — unit-tested."""
+    if not releases:
+        return None
+    def _key(r: dict) -> tuple:
+        official = 0 if (r.get("status") == "Official") else 1  # official editions first
+        date = r.get("date") or ""
+        return (official, 0 if date else 1, date)  # then dated, then earliest date
+    return min(releases, key=_key)
+
+
 class MusicBrainzResolver:
     """Real MbResolver using the MusicBrainz webservice via musicbrainzngs.
 
@@ -119,9 +148,10 @@ class MusicBrainzResolver:
         if rgid:
             rgfull = _with_retry(lambda: musicbrainzngs.get_release_group_by_id(rgid, includes=["releases"]))
             releases = rgfull.get("release-group", {}).get("release-list", [])
-            if releases:
+            chosen = _pick_release(releases)
+            if chosen:
                 rel = _with_retry(
-                    lambda: musicbrainzngs.get_release_by_id(releases[0]["id"], includes=["recordings"])
+                    lambda: musicbrainzngs.get_release_by_id(chosen["id"], includes=["recordings"])
                 ).get("release", {})
                 tracks = []
                 for medium in rel.get("medium-list", []):

worker/lyra_worker/adapters/_slskd.py

@@ -29,6 +29,19 @@ def _dirname(path: str) -> str:
     return ""
 
 
+_ETA_CAP_SECONDS = 359_999  # ~100h; keeps the value well inside Job.downloadEtaSeconds (int4)
+
+
+def _eta_seconds(total_bytes: int, bytes_now: int, speed_bps: float) -> int | None:
+    """Seconds until the transfer finishes at the current measured rate, or None when it can't be
+    estimated (no throughput sample yet, unknown total, or already complete). Pure — unit-tested.
+    A near-stalled peer drives speed_bps toward zero, so the estimate is capped: an uncapped value
+    can exceed int4 and overflow the downloadEtaSeconds column write (poisoning the pipeline txn)."""
+    if speed_bps <= 0 or total_bytes <= 0 or bytes_now >= total_bytes:
+        return None
+    return min(int((total_bytes - bytes_now) / speed_bps), _ETA_CAP_SECONDS)
+
+
 def _norm(s: str) -> str:
     """Lowercase, strip diacritics, and reduce to alphanumeric tokens, so 'Beyoncé' matches a
     peer's 'Beyonce' folder and punctuation ('I Am… Sasha Fierce') doesn't defeat the match."""
@@ -62,16 +75,20 @@ def _keep_matching(responses: list, needle: str) -> list:
     return out
 
 
-def _parse_search_responses(responses: list) -> list[dict]:
+def _parse_search_responses(responses: list, artist: str = "") -> list[dict]:
     """Turn slskd search responses into album candidates (one per peer+directory), ordered so the
-    peers most likely to deliver quickly come first: a free upload slot, then higher upload speed,
-    then a shorter queue. The pipeline tries candidates in this order, so ranking fast/free peers
-    first (over slow or queued ones) is what makes the Soulseek fall-through actually converge.
-    Pure — no I/O — so it is unit-tested offline."""
+    peers that will FINISH first come first: a free upload slot, then the shortest estimated
+    transfer time (album bytes ÷ the peer's advertised speed), then a shorter queue. Ranking on
+    estimated time — not raw speed — matters because Soulseek FLAC rips of the same album vary
+    ~2x in size (a 24-bit/bloated rip vs a standard CD rip), and Lyra scores them the SAME quality
+    class (the search exposes no bit-depth), so a peer advertising high speed but serving huge
+    files can lose a race — and blow past the download backstop — against a peer with a smaller
+    standard rip. The pipeline tries candidates in this order, so this is what makes the Soulseek
+    fall-through converge on a copy that actually completes. Pure — no I/O — unit-tested offline."""
     scored: list[tuple] = []
     for resp in responses:
         username = resp.get("username", "")
-        has_slot = 1 if resp.get("hasFreeUploadSlot") else 0
+        has_slot = 0 if resp.get("hasFreeUploadSlot") else 1  # 0 sorts first (free slot = starts now)
         speed = int(resp.get("uploadSpeed") or 0)
         queue = int(resp.get("queueLength") or 0)
         by_dir: dict[str, list] = {}
@@ -86,9 +103,20 @@ def _parse_search_responses(responses: list) -> list[dict]:
         for directory, dfiles in by_dir.items():
             if not dfiles:
                 continue
+            # Estimated seconds to pull this album from this peer at its advertised rate. Unknown/
+            # zero speed → a low nominal (1 B/s) so total size still orders those peers last.
+            total_bytes = sum(int(f["size"] or 0) for f in dfiles)
+            est_seconds = total_bytes / (speed if speed > 0 else 1)
             lossless = all(f["ext"] in _LOSSLESS_EXT for f in dfiles)
             dirname = _basename(directory)
-            if " - " in dirname:
+            # Trust the REQUESTED artist when it appears anywhere in the path — many rips put the
+            # artist in a parent folder ("Rihanna\(2009) Rated R\…") or bury it ("2009 - Rihanna -
+            # Rated R"), which the "Artist - Album" folder parse gets wrong; that tanked the
+            # confidence score and lost real FLACs to clean-named MP3s. album stays the folder name
+            # (fuzzy-matched downstream); artist is what the parse gets wrong.
+            if artist and _path_matches(artist, dfiles[0]["filename"]):
+                guess_artist, guess_album = artist, dirname
+            elif " - " in dirname:
                 guess_artist, guess_album = (p.strip() for p in dirname.split(" - ", 1))
             else:
                 guess_artist, guess_album = "", dirname
@@ -103,8 +131,9 @@ def _parse_search_responses(responses: list) -> list[dict]:
                 "format": "FLAC" if lossless else "MP3",
                 "bitrate": None,
             }
-            scored.append((has_slot, speed, -queue, candidate))
-    scored.sort(key=lambda t: (t[0], t[1], t[2]), reverse=True)  # free + fast + short-queue first
+            scored.append((has_slot, est_seconds, queue, candidate))
+    # ascending: free slot first, then shortest estimated transfer, then shortest queue
+    scored.sort(key=lambda t: (t[0], t[1], t[2]))
     return [c for *_rest, c in scored]
 
 
@@ -177,7 +206,7 @@ class SlskdClient:
                 if alt:
                     responses = alt
                     break
-        return _parse_search_responses(responses)
+        return _parse_search_responses(responses, artist)
 
     def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict:
         ref = json.loads(source_ref)
@@ -206,9 +235,13 @@ class SlskdClient:
         on_progress(0.0)
 
         total = len(files)
+        total_bytes = sum(int(f.get("size") or 0) for f in files)
         completed = False
         last_bytes = -1
         stall = 0
+        speed = 0.0  # bytes/sec, EWMA-smoothed so the ETA doesn't jitter poll to poll
+        prev_bytes = 0
+        prev_t = time.monotonic()
         for _ in range(_MAX_XFER_POLLS):
             time.sleep(_POLL_SECONDS)
             data = self._get(f"/api/v0/transfers/downloads/{username}")
@@ -224,8 +257,14 @@ class SlskdClient:
             if states:
                 if any(bad in s for s in states for bad in ("Errored", "Rejected", "Cancelled")):
                     raise RuntimeError("soulseek transfer failed")
+                now = time.monotonic()
+                dt = now - prev_t
+                if dt > 0 and bytes_now >= prev_bytes:
+                    inst = (bytes_now - prev_bytes) / dt
+                    speed = inst if speed == 0.0 else 0.6 * speed + 0.4 * inst
+                prev_bytes, prev_t = bytes_now, now
                 done = sum(1 for s in states if "Completed" in s)
-                on_progress(min(1.0, done / max(total, 1)))
+                on_progress(min(1.0, done / max(total, 1)), _eta_seconds(total_bytes, bytes_now, speed))
                 if done >= total:
                     completed = True
                     break

worker/lyra_worker/pipeline.py

@@ -22,8 +22,10 @@ _AUDIO_EXT = {".flac", ".mp3", ".m4a", ".opus", ".ogg", ".aac", ".wav"}
 # Reject a download whose total playtime falls below this fraction of MusicBrainz's expected
 # total — catches truncated files / preview clips substituted for real tracks that keep the
 # track COUNT right. Generous, so edition/encoding differences don't false-positive; bonus
-# tracks (over-long) are always fine.
-_DURATION_MIN_RATIO = 0.85
+# tracks (over-long) are always fine. Kept well below 1.0 because a source legitimately delivers a
+# different edition than MB's resolved release (shorter radio edits, no bonus/live tracks), and only
+# a genuinely truncated/preview-filled download (playtime a fraction of expected) should be caught.
+_DURATION_MIN_RATIO = 0.65
 # Cap how many ranked candidates a single job will actually download+verify before giving up.
 # Without this, the incomplete-download fall-through can grind through dozens of sources (a popular
 # album returns ~200 Soulseek candidates), each a slow attempt that also leaves an abandoned slskd
@@ -159,30 +161,42 @@ def _set_state(conn: psycopg.Connection, job_id: str, state: str, stage: str) ->
     conn.commit()
 
 
-def _set_download_progress(conn: psycopg.Connection, job_id: str, frac: float) -> None:
+def _set_download_progress(
+    conn: psycopg.Connection, job_id: str, frac: float, eta_seconds: int | None = None
+) -> None:
     with conn.cursor() as cur:
-        cur.execute('UPDATE "Job" SET "downloadProgress" = %s WHERE id = %s', (frac, job_id))
+        cur.execute(
+            'UPDATE "Job" SET "downloadProgress" = %s, "downloadEtaSeconds" = %s WHERE id = %s',
+            (frac, eta_seconds, job_id),
+        )
     conn.commit()
 
 
 def _make_on_progress(conn: psycopg.Connection, job_id: str, reports: "threading.Event"):
-    """Build the on_progress(pct) callback threaded into adapter.download. It marks the
-    adapter as reporting real byte-level progress (so the file-count poller yields) and
-    throttle-writes Job.downloadProgress. Called synchronously from adapter.download on the
-    pipeline's own thread, so it safely reuses `conn`."""
+    """Build the on_progress(pct, eta_seconds=None) callback threaded into adapter.download. It
+    marks the adapter as reporting real byte-level progress (so the file-count poller yields) and
+    throttle-writes Job.downloadProgress (+ the download ETA when the adapter measures one — slskd
+    does; others pass None). Called synchronously from adapter.download on the pipeline's own
+    thread, so it safely reuses `conn`."""
     state = {"frac": -1.0, "t": 0.0}
 
-    def on_progress(pct: float) -> None:
+    def on_progress(pct: float, eta_seconds: int | None = None) -> None:
         reports.set()
         frac = 0.0 if pct < 0 else 1.0 if pct > 1 else float(pct)
         now = time.monotonic()
-        # throttle DB writes: on a >=1% move or every 0.5s, not once per received byte
-        if frac - state["frac"] >= 0.01 or (now - state["t"]) >= 0.5:
+        # throttle DB writes: on a >=1% move, a new ETA, or every 0.5s — not once per received byte
+        if frac - state["frac"] >= 0.01 or eta_seconds is not None or (now - state["t"]) >= 0.5:
             state["frac"] = frac
             state["t"] = now
             try:
-                _set_download_progress(conn, job_id, frac)
+                _set_download_progress(conn, job_id, frac, eta_seconds)
             except Exception as e:  # a progress write must never fail the download
+                # Roll back so a failed write can't leave the shared pipeline conn in an aborted
+                # txn (which would cascade "current transaction is aborted" into every later query).
+                try:
+                    conn.rollback()
+                except Exception:
+                    pass
                 print(f"pipeline: on_progress write failed: {e}", flush=True)
 
     return on_progress

worker/tests/test_musicbrainz.py

@@ -1,7 +1,7 @@
 import musicbrainzngs
 import pytest
 
-from lyra_worker._musicbrainz import _best_release_group, _with_retry
+from lyra_worker._musicbrainz import _best_release_group, _pick_release, _with_retry
 
 # Offline unit tests for the pure release-group selection logic. The live resolve()
 # path (network) is covered by test_musicbrainz_live.py.
@@ -51,6 +51,21 @@ def test_empty_groups_returns_none():
     assert _best_release_group("Michael Jackson", "Off the Wall", []) is None
 
 
+def test_prefers_original_group_over_later_reissue():
+    # Real "Hybrid Theory" bug: MB lists a 2000 original and a 2023 reissue as separate same-titled
+    # Album groups (tie on artist/title/type). The reissue's longer tracklist fails completeness
+    # checks, so the original (earliest) must win regardless of MB's relevance order.
+    groups = [
+        {"id": "reissue", "title": "Hybrid Theory", "primary-type": "Album",
+         "first-release-date": "2023-07-21",
+         "artist-credit": [{"artist": {"name": "Linkin Park"}}]},
+        {"id": "original", "title": "Hybrid Theory", "primary-type": "Album",
+         "first-release-date": "2000-10-24",
+         "artist-credit": [{"artist": {"name": "Linkin Park"}}]},
+    ]
+    assert _best_release_group("Linkin Park", "Hybrid Theory", groups)["id"] == "original"
+
+
 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.
@@ -68,6 +83,30 @@ def test_prefers_credited_artist_over_same_titled_other_artist_album():
 # --- _with_retry: transient MusicBrainz failures (SSL EOF, 503 rate-limit) ---
 
 
+def test_pick_release_prefers_original_official_over_later_reissue():
+    # Real "Hybrid Theory" bug: MB's arbitrary releases[0] was a 2023 13-track reissue; the 2000
+    # original is what sources deliver, so completeness checks must reference it.
+    releases = [
+        {"id": "reissue", "status": "Official", "date": "2023-06-01"},
+        {"id": "original", "status": "Official", "date": "2000-10-24"},
+        {"id": "deluxe", "status": "Official", "date": "2020-10-09"},
+    ]
+    assert _pick_release(releases)["id"] == "original"
+
+
+def test_pick_release_skips_non_official_and_undated():
+    releases = [
+        {"id": "promo", "status": "Promotion", "date": "1999-01-01"},  # earliest but not official
+        {"id": "undated", "status": "Official", "date": ""},
+        {"id": "official", "status": "Official", "date": "2001-03-03"},
+    ]
+    assert _pick_release(releases)["id"] == "official"
+
+
+def test_pick_release_empty_is_none():
+    assert _pick_release([]) is None
+
+
 def test_retry_returns_value_without_backoff_when_first_call_succeeds():
     sleeps: list[float] = []
     calls = {"n": 0}

worker/tests/test_slskd_cancel.py

@@ -6,7 +6,12 @@ import json
 import pytest
 
 import lyra_worker.adapters._slskd as slskd_mod
-from lyra_worker.adapters._slskd import SlskdClient, _parse_search_responses
+from lyra_worker.adapters._slskd import (
+    _ETA_CAP_SECONDS,
+    SlskdClient,
+    _eta_seconds,
+    _parse_search_responses,
+)
 
 
 class _Resp:
@@ -63,7 +68,7 @@ def test_stalled_peer_is_abandoned_and_cancelled(tmp_path, monkeypatch):
     fake = slskd_mod.requests
 
     with pytest.raises(TimeoutError):
-        _client(tmp_path).download(_REF, str(tmp_path / "dest"), lambda _p: None)
+        _client(tmp_path).download(_REF, str(tmp_path / "dest"), lambda *_a: None)
     assert any(u.endswith("/api/v0/transfers/downloads/peerA/t1") for u in fake.deleted)
 
 
@@ -72,7 +77,7 @@ def test_error_state_is_cancelled(tmp_path, monkeypatch):
     monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None)
     fake = slskd_mod.requests
     with pytest.raises(RuntimeError):
-        _client(tmp_path).download(_REF, str(tmp_path / "dest"), lambda _p: None)
+        _client(tmp_path).download(_REF, str(tmp_path / "dest"), lambda *_a: None)
     assert any(u.endswith("/peerA/t1") for u in fake.deleted)
 
 
@@ -92,7 +97,7 @@ def test_slow_but_progressing_peer_is_not_abandoned(tmp_path, monkeypatch):
     monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None)
     monkeypatch.setattr(slskd_mod, "_STALL_POLLS", 2)  # would fire early if progress were ignored
 
-    out = _client(tmp_path).download(ref, str(tmp_path / "dest"), lambda _p: None)
+    out = _client(tmp_path).download(ref, str(tmp_path / "dest"), lambda *_a: None)
     assert out["track_count"] == 1
     assert slskd_mod.requests.deleted == []  # completed cleanly → nothing cancelled
 
@@ -108,3 +113,83 @@ def test_parse_ranks_free_and_fast_peers_first():
     ]
     peers = [json.loads(c["source_ref"])["username"] for c in _parse_search_responses(responses)]
     assert peers == ["fast", "slow", "queued"]  # free+fast first; no-slot peer last despite speed
+
+
+def test_parse_uses_requested_artist_when_present_in_path():
+    # Real "Rated R" MP3-over-FLAC bug: the artist is in a PARENT folder ("Rihanna\(2009) Rated
+    # R\…"), so the "Artist - Album" folder parse yields no artist and confidence tanks. When the
+    # requested artist is in the path, trust it so the candidate scores as Rihanna.
+    responses = [
+        {"username": "p", "hasFreeUploadSlot": True, "uploadSpeed": 1000, "queueLength": 0,
+         "files": [{"filename": r"@@x\Music\Rihanna\(2009) Rated R\01. Mad House.flac", "size": 5}]},
+    ]
+    c = _parse_search_responses(responses, artist="Rihanna")[0]
+    assert c["artist"] == "Rihanna"
+
+
+def test_parse_without_artist_falls_back_to_folder_parse():
+    responses = [
+        {"username": "p", "hasFreeUploadSlot": True, "uploadSpeed": 1000, "queueLength": 0,
+         "files": [{"filename": r"x\Some Band - Some Album\01.flac", "size": 5}]},
+    ]
+    c = _parse_search_responses(responses)[0]  # no artist arg → folder parse
+    assert c["artist"] == "Some Band"
+
+
+def test_parse_prefers_shorter_transfer_over_raw_speed():
+    # The Hybrid Theory case: a peer advertising higher speed but serving a ~2x-larger rip (24-bit
+    # / bloated FLAC) should lose to a peer with lower speed but a smaller standard rip that
+    # finishes sooner — both are the same quality class to Lyra, so faster-to-finish wins.
+    responses = [
+        {"username": "bloated", "hasFreeUploadSlot": True, "uploadSpeed": 20000, "queueLength": 0,
+         "files": [{"filename": rf"x\LP - Album\{i:02}.flac", "size": 40_000_000} for i in range(12)]},
+        {"username": "lean", "hasFreeUploadSlot": True, "uploadSpeed": 15000, "queueLength": 0,
+         "files": [{"filename": rf"y\LP - Album\{i:02}.flac", "size": 24_000_000} for i in range(12)]},
+    ]
+    # bloated est = 480MB/20000 = 24000s; lean est = 288MB/15000 = 19200s → lean finishes first
+    peers = [json.loads(c["source_ref"])["username"] for c in _parse_search_responses(responses)]
+    assert peers[0] == "lean"
+
+
+def test_eta_seconds_from_measured_rate():
+    assert _eta_seconds(1000, 250, 250.0) == 3  # 750 bytes left / 250 B/s
+    assert _eta_seconds(1000, 0, 100.0) == 10
+
+
+def test_eta_seconds_unknown_cases():
+    assert _eta_seconds(1000, 250, 0.0) is None      # no speed sample yet
+    assert _eta_seconds(0, 0, 100.0) is None          # unknown total size
+    assert _eta_seconds(1000, 1000, 100.0) is None    # already complete
+
+
+def test_eta_seconds_capped_when_peer_nearly_stalls():
+    # A near-stalled peer (tiny speed) would yield an ETA far past int4, overflowing the
+    # downloadEtaSeconds column write. The estimate must be clamped to a safe ceiling.
+    eta = _eta_seconds(30_000_000, 0, 0.001)  # 30MB / 0.001 B/s ~= 3e10s uncapped
+    assert eta == _ETA_CAP_SECONDS
+    assert eta < 2_147_483_647  # fits in int4
+
+
+def test_download_reports_shrinking_eta(tmp_path, monkeypatch):
+    # A 900-byte file arriving 300 bytes per 3s poll → measured speed 100 B/s → ETA counts down
+    # (6s, 3s) then None on completion. on_progress gets (fraction, eta_seconds).
+    (tmp_path / "01.flac").write_bytes(b"x" * 900)  # so _retrieve finds the finished file
+    ref = json.dumps({"username": "peerA", "files": [{"filename": r"d\Album\01.flac", "size": 900}]})
+    polls = [
+        {"directories": [{"files": [{"filename": r"d\Album\01.flac", "size": 900,
+                                     "state": "InProgress", "bytesTransferred": b, "id": "t1"}]}]}
+        for b in (300, 600)
+    ] + [
+        {"directories": [{"files": [{"filename": r"d\Album\01.flac", "size": 900,
+                                     "state": "Completed, Succeeded", "bytesTransferred": 900, "id": "t1"}]}]}
+    ]
+    monkeypatch.setattr(slskd_mod, "requests", _SeqRequests(polls))
+    monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None)
+    clock = iter(float(i) * 3 for i in range(100))  # 0,3,6,… deterministic monotonic time
+    monkeypatch.setattr(slskd_mod.time, "monotonic", lambda: next(clock))
+
+    calls: list = []
+    _client(tmp_path).download(ref, str(tmp_path / "dest"), lambda *a: calls.append(a))
+
+    etas = [a[1] for a in calls if len(a) > 1]
+    assert etas == [6, 3, None]  # counts down at 100 B/s, then None once complete