diff --git a/web/prisma/migrations/20260720200000_add_job_download_eta/migration.sql b/web/prisma/migrations/20260720200000_add_job_download_eta/migration.sql
new file mode 100644
index 0000000..874478b
--- /dev/null
+++ b/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;
diff --git a/web/prisma/schema.prisma b/web/prisma/schema.prisma
index f26ef53..8ae186d 100644
--- a/web/prisma/schema.prisma
+++ b/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[]
}
diff --git a/web/src/app/_ui/status.test.ts b/web/src/app/_ui/status.test.ts
index 5dc72c0..7f5c17a 100644
--- a/web/src/app/_ui/status.test.ts
+++ b/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", () => {
diff --git a/web/src/app/_ui/status.ts b/web/src/app/_ui/status.ts
index 51d357b..0489eee 100644
--- a/web/src/app/_ui/status.ts
+++ b/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 (
diff --git a/web/src/app/api/requests/route.ts b/web/src/app/api/requests/route.ts
index 67de383..30f5f42 100644
--- a/web/src/app/api/requests/route.ts
+++ b/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,
diff --git a/web/src/app/queue.tsx b/web/src/app/queue.tsx
index d4de511..3c060c0 100644
--- a/web/src/app/queue.tsx
+++ b/web/src/app/queue.tsx
@@ -1,7 +1,7 @@
"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";
@@ -46,6 +46,7 @@ type Row = {
candidateCount: number;
chosen: { source: string; format: string; trackCount: number } | null;
downloadProgress: number;
+ downloadEtaSeconds: number | null;
paused: boolean;
} | null;
};
@@ -62,6 +63,7 @@ function jobOf(r: Row) {
candidateCount: 0,
chosen: null,
downloadProgress: 0,
+ downloadEtaSeconds: null,
paused: false,
}
);
@@ -270,13 +272,9 @@ export function Queue() {
} else {
const total = j.chosen?.trackCount ?? 0;
const done = total ? Math.round((pct / 100) * total) : 0;
- bar = (
-
- );
+ const base = total ? `${pct}% · ${done}/${total} tracks` : `${pct}%`;
+ const eta = etaLabel(j.downloadEtaSeconds);
+ bar = ;
}
} else if (j.state === "tagging") {
bar = ;
diff --git a/worker/lyra_worker/adapters/_slskd.py b/worker/lyra_worker/adapters/_slskd.py
index aa2376d..e0f3f1e 100644
--- a/worker/lyra_worker/adapters/_slskd.py
+++ b/worker/lyra_worker/adapters/_slskd.py
@@ -29,6 +29,14 @@ def _dirname(path: str) -> str:
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:
"""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."""
@@ -206,9 +214,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 +236,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
diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py
index 4f5b30f..3e42c72 100644
--- a/worker/lyra_worker/pipeline.py
+++ b/worker/lyra_worker/pipeline.py
@@ -159,29 +159,35 @@ 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
print(f"pipeline: on_progress write failed: {e}", flush=True)
diff --git a/worker/tests/test_slskd_cancel.py b/worker/tests/test_slskd_cancel.py
index f6c61ee..967fa59 100644
--- a/worker/tests/test_slskd_cancel.py
+++ b/worker/tests/test_slskd_cancel.py
@@ -6,7 +6,7 @@ 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 SlskdClient, _eta_seconds, _parse_search_responses
class _Resp:
@@ -63,7 +63,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 +72,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 +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, "_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 +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)]
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