A bot we cannot reach is paused, and a full disk makes fetching wait (#926)

dcc_fetch.py

@@ -274,6 +274,136 @@ def new_fetch_row(bot, filename, now=None, request_type="file"):
 _seen_absent = set()
 _back_since = {}
 
+# PAUSED BOTS (#926 item 4). A bot we could not connect to this many times in a
+# row is paused - its requests wait, "Paused", until the operator resumes it -
+# the way AutoGet disabled a nick after three "unable to connect" failures: a
+# bot behind a firewall that cannot accept our connection fails every file the
+# same way, and asking on burns its slot and ours. The operator can pause and
+# resume any bot too. Kept in a small file beside the fetch history, so a pause
+# survives a restart; the consecutive-failure count does not need to.
+CONNECT_FAILURES_TO_PAUSE = 3
+_paused = {}               # bot (lowercased) -> {"nick", "reason", "since", "by"}
+_connect_failures = {}     # bot (lowercased) -> consecutive active-connect failures
+
+# A FULL DISK (#926 item 4). No new fetch starts while FETCHED_FILES_DIR has
+# less than this free; a transfer that runs out of space mid-way goes back to
+# pending rather than failing, and everything resumes by itself once space is
+# freed. AutoGet switched itself off when a write failed; waiting is kinder.
+MIN_FREE_BYTES = 200 * 1024 * 1024
+_disk_was_low = [False]
+
+
+def _paused_path():
+    """Beside the fetch history - so wherever that is redirected (tests,
+    FETCH_HISTORY_FILE), this goes with it."""
+    return os.path.join(os.path.dirname(os.path.abspath(db.FETCH_HISTORY_FILE)),
+                        "fetch_paused_bots.json")
+
+
+def load_paused_bots():
+    """Read the paused bots back at startup. A missing or unreadable file is
+    no pauses, not a refusal to start."""
+    import json
+    _paused.clear()
+    try:
+        with open(_paused_path(), "r", encoding="utf-8") as handle:
+            loaded = json.load(handle)
+    except (OSError, ValueError):
+        return
+    if isinstance(loaded, dict):
+        _paused.update({str(key).lower(): value for key, value in loaded.items()
+                        if isinstance(value, dict)})
+
+
+def _save_paused_bots():
+    import json
+    try:
+        with db._disk_lock:
+            db._atomic_write(_paused_path(), json.dumps(_paused, indent=1, sort_keys=True))
+    except Exception as err:
+        print(f"[FETCH] Could not save the paused bots: {err}")
+
+
+def paused_bots():
+    """{bot (lowercased): {"nick", "reason", "since", "by"}}, a copy."""
+    return {key: dict(value) for key, value in _paused.items()}
+
+
+def pause_bot(bot, reason, by="operator"):
+    nick = str(bot or "").strip()
+    if not nick:
+        return False
+    _paused[nick.lower()] = {"nick": nick, "reason": str(reason), "since": time.time(), "by": by}
+    _connect_failures.pop(nick.lower(), None)
+    _save_paused_bots()
+    print(f"[FETCH] Paused fetching from {nick}: {reason}")
+    return True
+
+
+def resume_bot(bot):
+    key = str(bot or "").strip().lower()
+    if _paused.pop(key, None) is None:
+        return False
+    _connect_failures.pop(key, None)
+    _save_paused_bots()
+    print(f"[FETCH] Resumed fetching from {bot}.")
+    return True
+
+
+def _note_connect_failure(bot):
+    """One more time we could not connect to this bot; the third in a row
+    pauses it, and says so where the operator looks."""
+    key = str(bot or "").strip().lower()
+    if not key or key in _paused:
+        return
+    _connect_failures[key] = _connect_failures.get(key, 0) + 1
+    if _connect_failures[key] >= CONNECT_FAILURES_TO_PAUSE:
+        pause_bot(bot, f"could not connect {CONNECT_FAILURES_TO_PAUSE} times in a row", by="auto")
+        try:
+            import announce
+            announce.send_debug(f"Fetching from {bot} is paused: could not connect "
+                                f"{CONNECT_FAILURES_TO_PAUSE} times in a row. Resume it on the "
+                                f"Downloads page when it can take connections.", category="INFO")
+        except Exception:
+            pass
+
+
+def _note_connect_success(bot):
+    _connect_failures.pop(str(bot or "").strip().lower(), None)
+
+
+def _disk_is_low():
+    """Whether FETCHED_FILES_DIR has less than MIN_FREE_BYTES free. Said once
+    when it becomes low and once when it recovers. A disk that cannot be
+    measured is not called low - the write itself still fails safely."""
+    import shutil
+    folder = getattr(config, "FETCHED_FILES_DIR", "") or "."
+    try:
+        free = shutil.disk_usage(platform_compat.long_path(os.path.abspath(folder))).free
+    except OSError:
+        return False
+    low = free < MIN_FREE_BYTES
+    if low != _disk_was_low[0]:
+        _disk_was_low[0] = low
+        message = (f"Fetching is waiting: under {MIN_FREE_BYTES // (1024 * 1024)} MB free where fetched "
+                   f"files go. It carries on by itself once space is freed."
+                   if low else "Fetching carries on: there is space for fetched files again.")
+        print(f"[FETCH] {message}")
+        try:
+            import announce
+            announce.send_debug(message, category="INFO")
+        except Exception:
+            pass
+    return low
+
+
+def _is_disk_full(err):
+    """Whether an error is the disk running out of space (ENOSPC; Windows
+    reports ERROR_DISK_FULL as the same errno)."""
+    import errno
+    return isinstance(err, OSError) and (err.errno == errno.ENOSPC
+                                         or getattr(err, "winerror", None) in (112, 39))
+
 # MAX_UNRESOLVED_FETCHES: the ceiling on how many rows may sit unresolved
 # (pending or in flight) at once, across every requester.
 #
@@ -859,6 +989,10 @@ def check_fetch_queue():
         waiting_bots = {str(row.get("bot", "")).strip().lower()
                         for row in queue.values() if row.get("state") == "pending"}
     readiness = _bot_readiness(waiting_bots, now)
+    for bot in waiting_bots:
+        if bot in _paused:
+            readiness[bot] = "paused"
+    disk_low = bool(waiting_bots) and _disk_is_low()
 
     to_dispatch = []
     with _fetch_lock():
@@ -934,6 +1068,8 @@ def check_fetch_queue():
             row = queue[rid]
             key = str(row.get("bot", "")).strip().lower()
             why = readiness.get(key, "")
+            if not why and disk_low:
+                why = "disk-full"
             if not why and (row.get("retry_at") or 0) > now:
                 why = "retry"
             if not why and max_per_bot > 0 and load.get(key, 0) >= max_per_bot:
@@ -1932,6 +2068,10 @@ def _run_transfer(row, offer, dest_dir, stored_name, sock=None):
         except Exception as connect_err:
             _mark_failed_locked(row, f"connect error: {connect_err}")
             print(f"[FETCH] Could not connect to {offer['ip']}:{offer['port']}: {connect_err}")
+            # Only an ACTIVE connect counts (#926): we dialled them and could
+            # not reach them. A passive offer that nobody connects back to is
+            # about our side, not theirs.
+            _note_connect_failure(row.get("bot"))
             try:
                 sock.close()
             except Exception:
@@ -1952,6 +2092,7 @@ def _run_transfer(row, offer, dest_dir, stored_name, sock=None):
 
     bytes_received = 0
     failure_reason = None
+    disk_full = False
     handle = None
     try:
         # Two different limits, and long_path() only lifts one of them. The
@@ -1990,6 +2131,7 @@ def _run_transfer(row, offer, dest_dir, stored_name, sock=None):
             row["bytes_received"] = bytes_received
     except Exception as recv_err:
         failure_reason = f"transfer error: {recv_err}"
+        disk_full = _is_disk_full(recv_err)
     finally:
         try:
             if handle:
@@ -2004,6 +2146,7 @@ def _run_transfer(row, offer, dest_dir, stored_name, sock=None):
     if failure_reason is None and bytes_received == total_size:
         row["state"] = "complete"
         row["bytes_received"] = bytes_received
+        _note_connect_success(row.get("bot"))
         if row.get("request_type") == "list":
             # The DCC transfer itself succeeded (declared size matched what
             # arrived) - that is what "complete" above means, and is left
@@ -2033,8 +2176,17 @@ def _run_transfer(row, offer, dest_dir, stored_name, sock=None):
     if failure_reason is None:
         failure_reason = f"incomplete transfer ({bytes_received}/{total_size} bytes)"
 
-    _mark_failed_locked(row, failure_reason)
-    print(f"[FETCH] Failed ({failure_reason}): {stored_name}.")
+    if disk_full:
+        # Not this request's fault (#926): it goes back to pending, and the
+        # dispatcher holds everything until there is space again.
+        row.update(state="pending", offered_at=None, bytes_received=0,
+                   reason="the disk filled up - asking again once there is space",
+                   waiting="disk-full")
+        _disk_was_low[0] = False  # so the next check says it
+        print(f"[FETCH] The disk filled up receiving {stored_name}; it will be asked again.")
+    else:
+        _mark_failed_locked(row, failure_reason)
+        print(f"[FETCH] Failed ({failure_reason}): {stored_name}.")
     try:
         if os.path.exists(platform_compat.long_path(dest_path)):
             # Unwrapped, exists() answers False for a >260 path and the

docs/UPDATES-PUBLIC.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- **Added: a bot that keeps failing is paused, and a full disk makes downloads wait.** If DCCore cannot connect to a bot three times in a row, that bot is paused - its downloads stay queued with a **Resume** button on the Downloads page - instead of failing file after file. And when the drive fetched files go to has less than 200 MB free, downloads wait instead of failing, and carry on by themselves once there is space.
 - **Added: the Downloads queue looks after itself.** Files you queue from another bot wait for it if it is offline and are asked for a minute after it comes back; only three are asked of one bot at a time (*Files asked of one bot at once*), the next going when one arrives, so a big selection no longer earns "queue full"; a "busy" answer is asked again later, three times; and unfinished downloads survive a restart. The Downloads page says why each one is waiting.
 - **Changed: searching and downloading keep working while the list rebuilds.** They used to be refused for the whole of `!update` (or a scheduled rebuild) - often a minute or more. The new list is built next to the current one, so the bot now keeps answering from the list people already have and pauses only for the few seconds it takes to swap the new one in; a request that lands in those seconds is asked to try again shortly. *Pause for the whole rebuild* on the Settings page (`PAUSE_FOR_WHOLE_UPDATE`) brings back the old behaviour.
 - **Fixed: fetching from a busy bot now works.** When another bot answered "you're number 12 in my queue", DCCore ignored it, gave up after a minute - and then refused the file when it arrived later. It now understands what file servers answer (OmeNServE, SDFind, SpR, BWI and DCCore itself): a queued request shows its place in their queue on the Downloads page and waits for its turn (up to 12 hours, *Wait for a queued request*), and "I don't have that file" or "queue full" ends the request straight away with the server's own words instead of a minute of silence.

docs/UPDATES.md

@@ -4,6 +4,34 @@ All version changes, optimizations, and bug fixes made over time in the DCCore p
 
 ## 🟨 Unreleased
 
+### ⏸️ A bot we cannot reach is paused, and a full disk makes fetching wait (#926)
+
+Item 4 of #926, stacked on the queue pacing below.
+
+- **Three failed connections pause a bot.** A bot behind a firewall that cannot take our connection fails every
+  file the same way, and asking on burns its slot and ours; AutoGet disabled a nick after three "unable to connect".
+  Now three *active* connect failures in a row (`CONNECT_FAILURES_TO_PAUSE`) pause the bot: its requests stay
+  queued and wait, *Paused - resume it to carry on*, with a **Resume this bot** button on the Downloads page, and
+  the debug feed says so once. A finished transfer resets the count. A passive offer nobody connects back to is not
+  counted - that one is our side, not theirs.
+- **Pause and resume any bot** (`POST /api/fetch/pause` and `/api/fetch/resume`, `GET /api/fetch/paused`). Pauses
+  are saved to `fetch_paused_bots.json` beside the fetch history, so one survives a restart; the failure count does
+  not need to.
+- **A full disk makes fetching wait instead of fail.** With under `MIN_FREE_BYTES` (200 MB) free where fetched files
+  go, no new fetch starts - the rows wait, *Waiting for disk space* - and a transfer that runs out of space goes back
+  to pending rather than failing. Both carry on by themselves once space is freed, and the change is said once each
+  way. A disk that cannot be measured is not called low; the write itself still fails safely. AutoGet switched
+  itself off on a write error; waiting is kinder.
+
+`tests/test_a_failing_bot_is_paused.py` (14): the third failure pausing and saying so, a finished transfer - through
+the real `_run_transfer()` over a socket pair - resetting the count, only the active connect counting; a paused
+bot's requests waiting while others go, resuming, the operator pausing any bot, resuming one not paused, the pause
+surviving a restart (and the resume being saved); a low disk holding fetches and saying so once, carrying on once
+there is space, an unmeasurable disk not blocking, a transfer that fills the disk - the real `_run_transfer()` into
+a file that says no space left - going back to pending, and recognising the error. Mutation-checked: never pausing,
+the real transfer not resetting the count, the pause ignored, the disk ignored, a full disk failing the row, and a
+resume not saved each fail a test.
+
 ### 🗂️ The fetch queue waits for a bot, paces itself per bot, and survives a restart (#926)
 
 Items 2 and 3 of #926, stacked on the reply handling below. The fetch queue could only ask a bot that was in a

oserve.py

@@ -321,6 +321,12 @@ def startup(setup_page=None):
     # in-memory only until now, so a completed download and its Delete
     # button both silently vanished from the dashboard on every restart.
     config.fetch_queue.update(db.load_fetch_history())
+    # And the bots whose fetching is paused (#926), kept beside it.
+    try:
+        import dcc_fetch
+        dcc_fetch.load_paused_bots()
+    except Exception as paused_err:
+        print(f"[FETCH] Could not read the paused bots: {paused_err}")
     # The notices survive a restart, which is the whole point of them: an
     # event worth a badge is by definition one that happened while nobody was
     # looking, and a kick at three in the morning that is gone by nine is a

tests/support.py

@@ -256,6 +256,9 @@ def reset_config(**overrides):
     if fetch_module is not None:
         fetch_module._seen_absent.clear()
         fetch_module._back_since.clear()
+        fetch_module._paused.clear()
+        fetch_module._connect_failures.clear()
+        fetch_module._disk_was_low[0] = False
 
     # What the version check (#572) last found. Read from runtime.py itself,
     # not through config, so reset there: a release "found" by one test would