Compare commits

..

2 Commits

Author SHA1 Message Date
Jonathan c49cee9a20 feat: download ETA on the press bar from measured Soulseek throughput
Show a live "~Xm left" countdown on the download progress bar. The slskd download
loop already polls byte transfer every 3s; measure the actual throughput (EWMA-
smoothed bytes/sec), compute seconds-remaining from the album's total bytes, and
surface it. New nullable Job.downloadEtaSeconds (migration; web entrypoint runs
prisma migrate deploy), threaded through the on_progress callback (optional 2nd arg,
so other adapters/callers are unaffected — they report no ETA). API exposes it;
queue.tsx renders etaLabel() after the track count. Null when unknown (no sample yet
or a source that doesn't report bytes). Worker 330 tests, web 214 tests, both green;
verified live-rendered as "42% · 6/14 tracks · ~4m left".

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-20 20:16:58 +02:00
Jonathan db8b43b0e8 feat(web): tabbed On-the-Press view + kebab menu for press actions
The press page interleaved active downloads, queued items, and errors in one long
list — hard to see what's actually pressing. Split the non-imported rows into three
tabs (Active / In queue / Errors) with live counts; the default tab follows the work
(Active if anything's running, else Queue, else Errors) until the user picks one.
Move the press-level actions (Pause/Resume, Clear queue, Stop now) off the toolbar
into a vertical "⋮" kebab menu on the right of the header — Clear queue disables when
the queue is empty, Stop now only shows while downloading.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-20 19:41:11 +02:00
11 changed files with 350 additions and 137 deletions
@@ -0,0 +1,2 @@
-- Estimated seconds remaining for the active download (measured throughput). Nullable.
ALTER TABLE "Job" ADD COLUMN "downloadEtaSeconds" INTEGER;
+3
View File
@@ -58,6 +58,9 @@ model Job {
createdAt DateTime @default(now()) createdAt DateTime @default(now())
updatedAt DateTime @updatedAt updatedAt DateTime @updatedAt
downloadProgress Float @default(0) 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) paused Boolean @default(false)
candidates Candidate[] candidates Candidate[]
} }
+47
View File
@@ -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>
);
}
+16 -1
View File
@@ -1,5 +1,20 @@
import { describe, it, expect } from "vitest"; 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", () => { describe("describeJob", () => {
it("maps pipeline states to literal labels + severity kind", () => { it("maps pipeline states to literal labels + severity kind", () => {
+10
View File
@@ -27,6 +27,16 @@ export function timeAgo(value: string | Date | null | undefined, now: number): s
return `${Math.floor(h / 24)}d`; 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 } { export function describeJob(state: string, stage: string): { label: string; kind: Kind } {
void stage; // stage no longer drives the label; kept for call-site compatibility void stage; // stage no longer drives the label; kept for call-site compatibility
return ( return (
+1
View File
@@ -64,6 +64,7 @@ export async function GET() {
claimedAt: r.job.claimedAt, claimedAt: r.job.claimedAt,
updatedAt: r.job.updatedAt, updatedAt: r.job.updatedAt,
downloadProgress: r.job.downloadProgress, downloadProgress: r.job.downloadProgress,
downloadEtaSeconds: r.job.downloadEtaSeconds,
paused: r.job.paused, paused: r.job.paused,
candidateCount: cands.length, candidateCount: cands.length,
chosen: chosen ? { source: chosen.source, format: chosen.format, trackCount: chosen.trackCount } : null, chosen: chosen ? { source: chosen.source, format: chosen.format, trackCount: chosen.trackCount } : null,
+26
View File
@@ -386,6 +386,32 @@ nav.contents .sep { flex: 1; }
.tab:hover { color: var(--ink); } .tab:hover { color: var(--ink); }
.tab.active { color: var(--ink); border-bottom-color: var(--accent); } .tab.active { color: var(--ink); border-bottom-color: var(--accent); }
.tab .n { color: var(--rule-2); margin-left: 6px; } .tab .n { color: var(--rule-2); margin-left: 6px; }
.tab.has-errors .n { color: var(--alert); }
/* ── 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) ───── */ /* ── Discography row (cover thumb + click-to-modal) ───── */
.disco-open { .disco-open {
+170 -121
View File
@@ -1,14 +1,34 @@
"use client"; "use client";
import { useEffect, useState } from "react"; 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 { StatTiles } from "./_ui/stat-tiles";
import { SectionHeader } from "./_ui/section-header"; import { SectionHeader } from "./_ui/section-header";
import { JobRow } from "./_ui/job-row"; import { JobRow } from "./_ui/job-row";
import { ProgressBar } from "./_ui/progress-bar"; import { ProgressBar } from "./_ui/progress-bar";
import { PressedList } from "./_ui/pressed-list"; import { PressedList } from "./_ui/pressed-list";
import { KebabMenu } from "./_ui/kebab-menu";
import { toast } from "./_ui/toast"; 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 = { type Row = {
id: string; id: string;
artist: string; artist: string;
@@ -26,6 +46,7 @@ type Row = {
candidateCount: number; candidateCount: number;
chosen: { source: string; format: string; trackCount: number } | null; chosen: { source: string; format: string; trackCount: number } | null;
downloadProgress: number; downloadProgress: number;
downloadEtaSeconds: number | null;
paused: boolean; paused: boolean;
} | null; } | null;
}; };
@@ -42,6 +63,7 @@ function jobOf(r: Row) {
candidateCount: 0, candidateCount: 0,
chosen: null, chosen: null,
downloadProgress: 0, downloadProgress: 0,
downloadEtaSeconds: null,
paused: false, paused: false,
} }
); );
@@ -60,6 +82,9 @@ export function Queue() {
const [artist, setArtist] = useState(""); const [artist, setArtist] = useState("");
const [album, setAlbum] = useState(""); const [album, setAlbum] = useState("");
const [paused, setPaused] = useState(false); 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() { async function refresh() {
const [reqRes, wantRes, artRes, floorRes] = await Promise.all([ const [reqRes, wantRes, artRes, floorRes] = await Promise.all([
@@ -159,10 +184,18 @@ export function Queue() {
} }
const active = rows.filter((r) => jobOf(r).state !== "imported"); const active = rows.filter((r) => jobOf(r).state !== "imported");
const downloading = rows.some((r) => { const buckets: Record<Tab, Row[]> = { active: [], queue: [], errors: [] };
const s = jobOf(r).state; for (const r of active) {
return s === "matching" || s === "matched" || s === "downloading" || s === "tagging"; const j = jobOf(r);
}); buckets[categoryOf(j.state, j.currentStage)].push(r);
}
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];
// Dedupe by artist+album — re-requesting/re-downloading the same album leaves several // 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). // completed Requests; show each album once (rows arrive newest-first, so keep the first).
const seenPressed = new Set<string>(); const seenPressed = new Set<string>();
@@ -174,6 +207,95 @@ export function Queue() {
return true; return true;
}); });
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}
meta={meta}
note={note}
bar={bar}
/>
);
}
return ( return (
<div> <div>
<StatTiles <StatTiles
@@ -185,27 +307,29 @@ export function Queue() {
]} ]}
/> />
<SectionHeader title="On the Press" note={`${active.length} running`} /> <div className="dept has-actions">
<h2>On the Press</h2>
<div className="press-controls"> <span className="fill" />
{paused ? ( {paused ? <span className="count">paused</span> : null}
<button className="btn sm accent" onClick={() => floorAction("resume", "Press resumed")}> <KebabMenu label="Press actions">
Resume the press {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>
) : ( {downloading ? (
<button className="btn sm ghost" onClick={() => floorAction("pause", "Press paused")}> <button className="menu-item danger" onClick={stopPress}>
Pause the press Stop now
</button> </button>
)} ) : null}
<button className="btn sm ghost" onClick={clearPress}> </KebabMenu>
Clear queue
</button>
{downloading ? (
<button className="btn sm" onClick={stopPress}>
Stop now
</button>
) : null}
{paused ? <span className="chip">Press paused</span> : null}
</div> </div>
<form className="request-form" onSubmit={submit}> <form className="request-form" onSubmit={submit}>
@@ -225,102 +349,27 @@ export function Queue() {
{active.length === 0 ? ( {active.length === 0 ? (
<p className="empty">The press is quiet. Queue a release above, or send one over from Wanted or Discover.</p> <p className="empty">The press is quiet. Queue a release above, or send one over from Wanted or Discover.</p>
) : ( ) : (
<section className="floor"> <>
{(() => { <div className="tabs" role="tablist">
const now = Date.now(); {(["active", "queue", "errors"] as Tab[]).map((t) => (
return active.map((r) => { <button
const j = jobOf(r); key={t}
const d = describeJob(j.state, j.currentStage); role="tab"
const attention = d.kind === "attention"; aria-selected={activeTab === t}
className={`tab${activeTab === t ? " active" : ""}${t === "errors" && buckets.errors.length ? " has-errors" : ""}`}
let meta: string | undefined; onClick={() => setTab(t)}
if (j.state === "downloading" || j.state === "tagging") { >
const parts: string[] = []; {TAB_LABEL[t]}
if (j.chosen) parts.push(`${j.chosen.source} · ${j.chosen.format}`); <span className="n">{buckets[t].length}</span>
const elapsed = timeAgo(j.claimedAt ?? j.updatedAt, now); </button>
if (elapsed) parts.push(elapsed); ))}
meta = parts.join(" · ") || undefined; </div>
} else if (j.state === "matching" || j.state === "matched") { {shown.length === 0 ? (
meta = <p className="empty">{TAB_EMPTY[activeTab]}</p>
j.candidateCount > 0 ) : (
? `${j.candidateCount} sources${j.chosen ? ` · best: ${j.chosen.source} ${j.chosen.format}` : ""}` <section className="floor">{shown.map(renderRow)}</section>
: "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>
)} )}
{pressed.length > 0 ? ( {pressed.length > 0 ? (
+19 -1
View File
@@ -29,6 +29,14 @@ def _dirname(path: str) -> str:
return "" return ""
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."""
if speed_bps <= 0 or total_bytes <= 0 or bytes_now >= total_bytes:
return None
return int((total_bytes - bytes_now) / speed_bps)
def _norm(s: str) -> str: def _norm(s: str) -> str:
"""Lowercase, strip diacritics, and reduce to alphanumeric tokens, so 'Beyoncé' matches a """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.""" peer's 'Beyonce' folder and punctuation ('I Am… Sasha Fierce') doesn't defeat the match."""
@@ -206,9 +214,13 @@ class SlskdClient:
on_progress(0.0) on_progress(0.0)
total = len(files) total = len(files)
total_bytes = sum(int(f.get("size") or 0) for f in files)
completed = False completed = False
last_bytes = -1 last_bytes = -1
stall = 0 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): for _ in range(_MAX_XFER_POLLS):
time.sleep(_POLL_SECONDS) time.sleep(_POLL_SECONDS)
data = self._get(f"/api/v0/transfers/downloads/{username}") data = self._get(f"/api/v0/transfers/downloads/{username}")
@@ -224,8 +236,14 @@ class SlskdClient:
if states: if states:
if any(bad in s for s in states for bad in ("Errored", "Rejected", "Cancelled")): if any(bad in s for s in states for bad in ("Errored", "Rejected", "Cancelled")):
raise RuntimeError("soulseek transfer failed") 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) 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: if done >= total:
completed = True completed = True
break break
+16 -10
View File
@@ -159,29 +159,35 @@ def _set_state(conn: psycopg.Connection, job_id: str, state: str, stage: str) ->
conn.commit() 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: 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() conn.commit()
def _make_on_progress(conn: psycopg.Connection, job_id: str, reports: "threading.Event"): 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 """Build the on_progress(pct, eta_seconds=None) callback threaded into adapter.download. It
adapter as reporting real byte-level progress (so the file-count poller yields) and 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 throttle-writes Job.downloadProgress (+ the download ETA when the adapter measures one — slskd
pipeline's own thread, so it safely reuses `conn`.""" 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} 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() reports.set()
frac = 0.0 if pct < 0 else 1.0 if pct > 1 else float(pct) frac = 0.0 if pct < 0 else 1.0 if pct > 1 else float(pct)
now = time.monotonic() now = time.monotonic()
# throttle DB writes: on a >=1% move or every 0.5s, not once per received byte # 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 (now - state["t"]) >= 0.5: if frac - state["frac"] >= 0.01 or eta_seconds is not None or (now - state["t"]) >= 0.5:
state["frac"] = frac state["frac"] = frac
state["t"] = now state["t"] = now
try: 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 except Exception as e: # a progress write must never fail the download
print(f"pipeline: on_progress write failed: {e}", flush=True) print(f"pipeline: on_progress write failed: {e}", flush=True)
+40 -4
View File
@@ -6,7 +6,7 @@ import json
import pytest import pytest
import lyra_worker.adapters._slskd as slskd_mod import lyra_worker.adapters._slskd as slskd_mod
from lyra_worker.adapters._slskd import SlskdClient, _parse_search_responses from lyra_worker.adapters._slskd import SlskdClient, _eta_seconds, _parse_search_responses
class _Resp: class _Resp:
@@ -63,7 +63,7 @@ def test_stalled_peer_is_abandoned_and_cancelled(tmp_path, monkeypatch):
fake = slskd_mod.requests fake = slskd_mod.requests
with pytest.raises(TimeoutError): 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) assert any(u.endswith("/api/v0/transfers/downloads/peerA/t1") for u in fake.deleted)
@@ -72,7 +72,7 @@ def test_error_state_is_cancelled(tmp_path, monkeypatch):
monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None) monkeypatch.setattr(slskd_mod.time, "sleep", lambda _s: None)
fake = slskd_mod.requests fake = slskd_mod.requests
with pytest.raises(RuntimeError): 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) assert any(u.endswith("/peerA/t1") for u in fake.deleted)
@@ -92,7 +92,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.time, "sleep", lambda _s: None)
monkeypatch.setattr(slskd_mod, "_STALL_POLLS", 2) # would fire early if progress were ignored 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 out["track_count"] == 1
assert slskd_mod.requests.deleted == [] # completed cleanly → nothing cancelled assert slskd_mod.requests.deleted == [] # completed cleanly → nothing cancelled
@@ -108,3 +108,39 @@ def test_parse_ranks_free_and_fast_peers_first():
] ]
peers = [json.loads(c["source_ref"])["username"] for c in _parse_search_responses(responses)] 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 assert peers == ["fast", "slow", "queued"] # free+fast first; no-slot peer last despite speed
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_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