diff --git a/worker/lyra_worker/adapters/_slskd.py b/worker/lyra_worker/adapters/_slskd.py index 85efba2..2548c79 100644 --- a/worker/lyra_worker/adapters/_slskd.py +++ b/worker/lyra_worker/adapters/_slskd.py @@ -29,12 +29,17 @@ 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.""" + 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 int((total_bytes - bytes_now) / speed_bps) + return min(int((total_bytes - bytes_now) / speed_bps), _ETA_CAP_SECONDS) def _norm(s: str) -> str: diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py index e8c6ea1..b9c68aa 100644 --- a/worker/lyra_worker/pipeline.py +++ b/worker/lyra_worker/pipeline.py @@ -191,6 +191,12 @@ def _make_on_progress(conn: psycopg.Connection, job_id: str, reports: "threading try: _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 diff --git a/worker/tests/test_slskd_cancel.py b/worker/tests/test_slskd_cancel.py index fcc804c..29157ed 100644 --- a/worker/tests/test_slskd_cancel.py +++ b/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, _eta_seconds, _parse_search_responses +from lyra_worker.adapters._slskd import ( + _ETA_CAP_SECONDS, + SlskdClient, + _eta_seconds, + _parse_search_responses, +) class _Resp: @@ -157,6 +162,14 @@ def test_eta_seconds_unknown_cases(): 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).