feat(tower): instrument push and retry failed

tower/lisael_content.py

@@ -415,18 +415,25 @@ _busy_lock = threading.Lock()
 # ---- transfers journal lock ----
 _history_lock = threading.Lock()
 
-def push_batch(ip, delta):
+def push_batch(ip, delta, subdir="podcasts"):
+    transfers_begin(ip, [(f, subdir) for f in delta])
     ok = 0
-    for f in delta:
-        try:
-            push_file(ip, f); ok += 1
-        except Exception as e:
-            print("push fail", f, e, flush=True)
-    print(f"push done {ip} {ok}/{len(delta)}", flush=True)
+    try:
+        for f in delta:
+            transfers_mark(f, "sending")
+            try:
+                push_file(ip, f, subdir=subdir); ok += 1
+                transfers_mark(f, "done")
+            except Exception as e:
+                transfers_mark(f, "failed", str(e)[:200])
+                print("push fail", f, e, flush=True)
+    finally:
+        transfers_end()
+    print(f"push done {ip} {subdir} {ok}/{len(delta)}", flush=True)
     with _busy_lock:
         _busy.discard(ip)
 
-def try_push(ip, delta):
+def try_push(ip, delta, subdir="podcasts"):
     """Start a background push unless one already runs for ip. Returns started?"""
     if not delta:
         return False
@@ -434,9 +441,39 @@ def try_push(ip, delta):
         if ip in _busy:
             return False
         _busy.add(ip)
-    threading.Thread(target=push_batch, args=(ip, delta), daemon=True).start()
+    threading.Thread(target=push_batch, args=(ip, delta, subdir), daemon=True).start()
     return True
 
+def failed_names(box_ip):
+    """Failed (name, dir) for box_ip: active batch first, else recent history.
+    Skips names already re-sent successfully later in history."""
+    a = transfers_active()
+    pool = (a["files"] if a and a.get("box_ip") == box_ip else []) \
+        + [h for h in transfers_history() if h.get("box_ip") == box_ip]
+    # later success overrides earlier failure: iterate chronological, keep last status
+    status_by = {}
+    for h in pool:
+        status_by[h["name"]] = (h["status"], h.get("dir", "podcasts"))
+    out = []
+    for name, (st, d) in status_by.items():
+        if st == "failed":
+            out.append((name, d))
+    return out
+
+def retry_failed(box_ip):
+    """Re-push failed files for box_ip, grouped by subdir. Returns {started, count}."""
+    fails = failed_names(box_ip)
+    if not fails:
+        return {"started": False, "count": 0}
+    by_dir = {}
+    for name, d in fails:
+        by_dir.setdefault(d, []).append(name)
+    started = False
+    for d, names in by_dir.items():
+        if try_push(box_ip, names, subdir=d):
+            started = True
+    return {"started": started, "count": len(fails)}
+
 # ---- transfers journal ----
 TRANSFERS_JSON = os.environ.get("LISAEL_TRANSFERS", os.path.join(BASE, "transfers.json"))
 TRANSFERS_MAX  = 200

tower/test_transfers.py

@@ -35,6 +35,107 @@ def test_history_truncates_200():
     # the most recent survive
     assert any(h["name"] == "f249.mp3" for h in C.transfers_history())
 
+def test_failed_names_from_history():
+    # seed a finished batch with one failed file for a box
+    C.transfers_begin("9.9.9.9", [("ok.mp3", "routines", 10), ("bad.mp3", "routines", 10)])
+    C.transfers_mark("ok.mp3", "sending"); C.transfers_mark("ok.mp3", "done")
+    C.transfers_mark("bad.mp3", "sending"); C.transfers_mark("bad.mp3", "failed", "net")
+    C.transfers_end()
+    fn = C.failed_names("9.9.9.9")
+    assert ("bad.mp3", "routines") in fn
+    assert ("ok.mp3", "routines") not in fn
+    # unknown box -> nothing
+    assert C.failed_names("0.0.0.0") == []
+
+def test_failed_names_from_active():
+    # active batch (not yet ended) also visible
+    C.transfers_begin("10.0.0.1", [("x.mp3", "podcasts", 20), ("y.mp3", "podcasts", 20)])
+    C.transfers_mark("x.mp3", "sending"); C.transfers_mark("x.mp3", "failed", "timeout")
+    C.transfers_mark("y.mp3", "sending"); C.transfers_mark("y.mp3", "done")
+    fn = C.failed_names("10.0.0.1")
+    assert ("x.mp3", "podcasts") in fn
+    assert ("y.mp3", "podcasts") not in fn
+    # clean up active so next test starts fresh
+    C.transfers_end()
+
+def test_push_batch_instrumented():
+    """push_batch calls begin/mark/end in the right sequence without real network."""
+    journal = []
+
+    def fake_push(ip, name, subdir="podcasts"):
+        journal.append(("push", name, subdir))
+        if name == "fail.mp3":
+            raise OSError("net error")
+
+    orig_push = C.push_file
+    C.push_file = fake_push
+    try:
+        C.push_batch("1.1.1.1", ["ok.mp3", "fail.mp3"], subdir="podcasts")
+    finally:
+        C.push_file = orig_push
+
+    # journal: begin fired (check active was set, indirectly via history after end)
+    hist = C.transfers_history()
+    by_name = {h["name"]: h for h in hist}
+    assert by_name["ok.mp3"]["status"] == "done"
+    assert by_name["fail.mp3"]["status"] == "failed"
+    assert "net error" in (by_name["fail.mp3"]["error"] or "")
+    # push calls happened in order
+    assert journal == [("push", "ok.mp3", "podcasts"), ("push", "fail.mp3", "podcasts")]
+    # _busy must be released after push_batch
+    import threading
+    assert "1.1.1.1" not in C._busy
+
+def test_push_batch_subdir_threaded():
+    """push_batch passes subdir to push_file and releases _busy."""
+    calls = []
+
+    def fake_push(ip, name, subdir="podcasts"):
+        calls.append(subdir)
+
+    orig = C.push_file
+    C.push_file = fake_push
+    try:
+        C.push_batch("2.2.2.2", ["r.mp3"], subdir="routines")
+    finally:
+        C.push_file = orig
+
+    assert calls == ["routines"]
+    assert "2.2.2.2" not in C._busy
+
+def test_retry_failed():
+    """retry_failed re-pushes only failed files, grouped by dir."""
+    pushed = []
+
+    def fake_push(ip, name, subdir="podcasts"):
+        pushed.append((name, subdir))
+
+    # seed history: one failed podcast, one failed routine, one done
+    C.transfers_begin("3.3.3.3", [
+        ("a.mp3", "podcasts", 10),
+        ("b.mp3", "routines", 10),
+        ("c.mp3", "podcasts", 10),
+    ])
+    C.transfers_mark("a.mp3", "sending"); C.transfers_mark("a.mp3", "failed", "err")
+    C.transfers_mark("b.mp3", "sending"); C.transfers_mark("b.mp3", "failed", "err")
+    C.transfers_mark("c.mp3", "sending"); C.transfers_mark("c.mp3", "done")
+    C.transfers_end()
+
+    orig = C.push_file
+    C.push_file = fake_push
+    try:
+        result = C.retry_failed("3.3.3.3")
+    finally:
+        C.push_file = orig
+        # drain any background thread
+        import time; time.sleep(0.1)
+
+    assert result["count"] == 2
+    assert result["started"] is True
+    # unknown box returns 0
+    r2 = C.retry_failed("0.0.0.0")
+    assert r2 == {"started": False, "count": 0}
+
 if __name__ == "__main__":
     for name, fn in sorted(globals().items()):
         if name.startswith("test_"):