From 10430037fe2cdab8f2200319d69a91cce0f6ea52 Mon Sep 17 00:00:00 2001
From: Jonathan <claude.cs@jger.nl>
Date: Tue, 21 Jul 2026 20:42:13 +0200
Subject: [PATCH] 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>
---
 worker/lyra_worker/adapters/_slskd.py |  9 +++++++--
 worker/lyra_worker/pipeline.py        |  6 ++++++
 worker/tests/test_slskd_cancel.py     | 15 ++++++++++++++-
 3 files changed, 27 insertions(+), 3 deletions(-)

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).