Tell a stalled source apart from a slow one, and honour a cancel during it

CHANGELOG.md

@@ -8,6 +8,26 @@ project uses [semantic versioning][semver].
 
 ## [Unreleased]
 
+### Added
+
+- **A stall is reported instead of looking like a slow link.** A hung network
+  handle raises nothing — it just stops returning bytes — so nothing could be
+  retried and the job sat at a stale throughput figure. Source reads are now
+  taken in 1 MiB sub-reads and timed: a gap past `--stall-after` (15 s by
+  default) reports a `stalled` stage, the queue shows "no data for 34s" in
+  place of a rate that has stopped being true, and the file is named in the
+  job's warnings afterwards. Timing *bytes* rather than chunks is what makes it
+  trustworthy — an 8 MiB chunk over a degraded link legitimately takes half a
+  minute, and a chunk-granularity timer would call a working copy stalled.
+
+  The duration shown counts from the last byte, not from the warning. The first
+  report only fires once the threshold has already passed, so timing it from
+  there displayed "no data for 0s" on a source that had been silent for fifteen
+  seconds, and stayed that far short for as long as the outage lasted. That
+  figure is what someone reads to decide whether to go and look at the cable,
+  so the engine carries the silence it has already measured and the queue
+  backdates its clock by it.
+
 ### Changed
 
 - **Full verification is the default.** The read-back is the only mode that
@@ -38,6 +58,14 @@ project uses [semantic versioning][semver].
   the drop, which is exactly why the retry is worth making. Recovery costs one
   re-read of the chunk in flight.
 
+- **A cancel is noticed while a read is hung.** The consumer blocked on the
+  chunk queue without a timeout and teardown then joined the reader thread
+  without a deadline, so cancelling a job whose source had stopped responding
+  waited for the operating system's timeout — the very thing the operator was
+  trying to escape. The queue is polled, the checkpoint is checked while
+  waiting, and a reader still blocked after a second is left behind: it is a
+  daemon holding one source handle, which it closes itself.
+
 - **A decoder ffmpeg lacks is probed once per job, not per clip.** Extracting
   thumbnails from BRAW with a stock ffmpeg fails identically for every clip;
   each one still paid four doomed process spawns. The first clip of a suffix

README.md

@@ -105,6 +105,7 @@ offloader verify D:\video\080426\A001
 | `--paranoid` | read each source file twice and compare (offload only) |
 | `--retries N` | attempts per file on a transient read failure (default 3, 1 disables) |
 | `--retry-wait SECONDS` | pause before the first retry, backing off after (default 2) |
+| `--stall-after SECONDS` | report a source that stops delivering bytes without failing, as a hung network mount does (default 15, 0 disables) |
 | `--no-probe` | skip ffprobe metadata and thumbnails |
 | `--quiet` | suppress progress |
 
@@ -245,7 +246,9 @@ bound, and running two at once against the same bus makes both slower and the
 progress readout meaningless. Each row shows live throughput and ETA, and the
 transport controls pause, resume, cancel, reprioritise, and open the reports
 folder. Pause takes effect within one 8 MiB chunk; cancel deletes the partial
-destination file rather than leaving something that looks complete.
+destination file rather than leaving something that looks complete. A source
+that stops delivering bytes reads as `Stalled on …` with the time since the
+last one, rather than a throughput figure that has quietly stopped being true.
 
 Two guards run before anything is queued:
 

ROADMAP.md

@@ -35,7 +35,10 @@ agrees is fine, is reported as the structure-hash mismatch it is. See
 
 A source on a network mount is handled like marginal media, because it fails
 like it: the dropped-session error codes are retried, the handle reopened and
-the read resumed from the last delivered chunk.
+the read resumed from the last delivered chunk. A stall — bytes stopping with
+no error raised at all — is timed at 1 MiB granularity so it is distinguishable
+from a slow link, reported while it happens, and named in the job's warnings
+afterwards.
 
 `--paranoid` reads every source file a second time and compares, which is the
 only thing that catches a read returning wrong bytes without reporting an error.
@@ -64,6 +67,19 @@ has `previousPath` for exactly this and it is not written.
 
 *Where:* `ascmhl.py`, and `verify.py` to read it back.
 
+### Aborting a hung read, not just reporting it
+
+A stall is now detected and reported, and a cancel is honoured while one is in
+progress. What is still not possible is ending the read itself: recovery waits
+for the operating system to turn the hang into an error, up to `SessionTimeout`
+— 60 seconds on Windows. On Windows `CancelIoEx` against the handle would do
+it, reached through `ctypes` with `msvcrt.get_osfhandle`; POSIX has no portable
+equivalent, which is why this is one platform's fix and not a general one.
+Listed under "What is still not protected" in
+[`docs/data-safety.md`](docs/data-safety.md).
+
+*Where:* `engine.py`, around the `read_ahead` loop.
+
 ### Coordination between instances
 
 One app instance serialises its queue. Two pointed at the same destination do

docs/data-safety.md

@@ -201,6 +201,21 @@ known.
   safety one, and should not be used on a tree whose integrity is in question.
 - **Concurrent instances.** One app instance serialises its queue. Two instances
   pointed at the same destination are not coordinated.
+- **A read that stalls cannot be aborted, only reported.** Retrying needs an
+  error to react to, and a hung network handle raises nothing — it simply stops
+  returning bytes. The job now says so: reads are taken in 1 MiB sub-reads, and
+  a gap longer than `--stall-after` (15 s by default) is reported as a stall
+  rather than left looking like a slow link, with the file named in the job's
+  warnings afterwards. The duration the queue shows counts from the last byte,
+  not from the warning: the first report only fires once the threshold has
+  already passed, so timing it from there would show "no data for 0s" on a
+  source that had been silent for fifteen seconds, and stay that far short for
+  as long as the outage lasted. It is the number someone reads to decide
+  whether to go and look at the cable. A cancel is honoured during it too.
+  What still cannot happen is aborting the read itself, so *recovery* waits on
+  the operating system to turn the hang into one of the codes above — on Windows, the SMB
+  client's `SessionTimeout`, 60 seconds by default. No data is at risk in the
+  meantime; the wait is real.
 
 ## Marginal media and dropped links
 

src/offloader/cli.py

@@ -43,7 +43,9 @@ def __call__(self, event: engine.ProgressEvent) -> None:
 
         pct = (event.job_bytes_done / event.job_bytes_total * 100
                if event.job_bytes_total else 100.0)
-        line = (f"  [{pct:5.1f}%] {event.stage:<6} "
+        # 7 wide: the longest stage name is "stalled", and a field that a stage
+        # overflows shifts every column after it.
+        line = (f"  [{pct:5.1f}%] {event.stage:<7} "
                 f"{event.file_index + 1}/{event.file_total}  {event.file_name}")
         line = line[:110]
         pad = max(0, self._width - len(line))
@@ -176,6 +178,12 @@ def _common_options(parser: argparse.ArgumentParser) -> None:
     parser.add_argument("--retry-wait", type=float, default=2.0, metavar="SECONDS",
                         help="pause before the first retry, backing off after "
                              "(default: %(default)s)")
+    parser.add_argument("--stall-after", type=float, default=15.0,
+                        metavar="SECONDS",
+                        help="report a source that has gone this long without "
+                             "delivering a byte, which a hung network mount "
+                             "does without raising any error "
+                             "(default: %(default)s, 0 disables)")
     parser.add_argument("--profile", choices=[p.value for p in Profile],
                         default=Profile.MEDIA.value,
                         help="'media' (default) offloads camera cards with "
@@ -261,6 +269,7 @@ def _options_from(args: argparse.Namespace, destinations: list[Path]) -> engine.
         retry=retry.RetryPolicy(attempts=max(1, args.retries),
                                 delay=max(0.0, args.retry_wait)),
         paranoid=getattr(args, "paranoid", False),
+        stall_after=max(0.0, getattr(args, "stall_after", 15.0)),
     )
 
 

src/offloader/engine.py

@@ -13,6 +13,7 @@
 import queue
 import shutil
 import threading
+import time
 from collections.abc import Callable, Iterable, Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -49,6 +50,26 @@
 #: side runs.
 READ_AHEAD = 3
 
+#: Bytes the reader asks for in one call. Smaller than `CHUNK_SIZE` purely so
+#: the stall watchdog hears from a slow link *between* chunks: an 8 MiB chunk
+#: over a degraded network mount can legitimately take half a minute, and a
+#: detector that cannot tell that from a hung handle is worse than none. At
+#: 1 MiB the liveness signal is independent of the chunk size the queue and the
+#: hashers work in, which stays tuned for throughput.
+SUBCHUNK_SIZE = 1 << 20  # 1 MiB
+
+#: How often the consumer wakes to ask whether anything has arrived. Also the
+#: resolution at which a cancel is noticed while a read is hung, which is the
+#: other thing this poll buys.
+STALL_POLL = 1.0
+
+#: How long teardown waits for the reader thread before leaving it behind. A
+#: read that has hung is not going to return on request, and an unbounded join
+#: here would hand the cancel back to whatever the operator was trying to
+#: escape. The thread is a daemon holding one source handle, which it closes
+#: itself on the way out, so abandoning it costs nothing.
+ABANDON_READER_AFTER = 1.0
+
 
 #: Extension worn by a copy that is still in flight. A destination file only
 #: takes its real name once it is complete — and, under full verification, once
@@ -158,6 +179,11 @@ class ProgressEvent:
     bytes_total: int = 0
     job_bytes_done: int = 0
     job_bytes_total: int = 0
+    #: On a "stalled" event, how long the source has actually supplied nothing.
+    #: Carried because the first such event only fires once the threshold has
+    #: already passed: a consumer timing from its arrival starts at zero and
+    #: stays a whole threshold short of the outage for as long as it lasts.
+    stalled_for: float = 0.0
 
 
 ProgressCallback = Callable[[ProgressEvent], None]
@@ -192,6 +218,12 @@ class OffloadOptions:
     #: pass over the card, and is the only thing that catches a read which
     #: returned wrong bytes without the operating system noticing.
     paranoid: bool = False
+    #: Seconds without a single byte arriving before the job says so. A hung
+    #: network handle raises nothing — it stops returning bytes — so this is
+    #: the only way a stall is distinguishable from a slow link. Reporting
+    #: only: the read still cannot be aborted, so recovery waits on the
+    #: operating system to turn the hang into an error. 0 disables.
+    stall_after: float = 15.0
 
     def __post_init__(self) -> None:
         # The data profile is defined by the absence of media work, so enforce
@@ -269,7 +301,9 @@ class _CopyResult:
 def _copy_fanout(source: Path, targets: Sequence[Path], algorithm: str,
                  on_chunk: Callable[[int], None],
                  control: JobControl | None = None,
-                 retry: retry_mod.RetryPolicy = retry_mod.NO_RETRY) -> _CopyResult:
+                 retry: retry_mod.RetryPolicy = retry_mod.NO_RETRY,
+                 on_stall: Callable[[float], None] | None = None,
+                 stall_after: float = 0.0) -> _CopyResult:
     """Stream `source` into every target at once.
 
     `targets` are the *in-flight* paths — the caller renames them into place
@@ -284,6 +318,11 @@ def _copy_fanout(source: Path, targets: Sequence[Path], algorithm: str,
     the caller's whole-file retry: a write that fails part-way leaves the
     destination at a length nothing here knows, whereas a failed read has
     produced nothing at all.
+
+    `on_stall` is called with the seconds since the last byte arrived, once per
+    `STALL_POLL` for as long as nothing is arriving. It needs no thread of its
+    own: the consumer below already runs on a different thread from the read it
+    is waiting on, which is the only thing a watchdog requires.
     """
     source = Path(source)
     src_hasher = new_hasher(algorithm)
@@ -303,6 +342,10 @@ def _copy_fanout(source: Path, targets: Sequence[Path], algorithm: str,
     stop = threading.Event()
     failure: list[BaseException] = []
     recovered: list[tuple[int, int]] = []
+    #: When the reader last had bytes in its hand. A list because it is written
+    #: on the reader thread and read on the consumer's, and a bare float would
+    #: rebind rather than mutate.
+    last_byte = [time.monotonic()]
 
     def read_ahead() -> None:
         """Keep the queue fed so the next read overlaps the current write.
@@ -321,7 +364,19 @@ def read_ahead() -> None:
             reader = longpath.open_binary(source, "rb")
 
             def read_one() -> bytes:
-                return reader.read(CHUNK_SIZE)
+                # Assembled from sub-reads so a link that is crawling rather
+                # than hung still marks itself alive on the way. A short read
+                # means end of file; a failure part-way discards the lot,
+                # because `recover` below seeks back to the start of the chunk
+                # and nothing partial has been hashed or delivered.
+                buffer = bytearray()
+                while len(buffer) < CHUNK_SIZE:
+                    part = reader.read(min(SUBCHUNK_SIZE, CHUNK_SIZE - len(buffer)))
+                    if not part:
+                        break
+                    buffer += part
+                    last_byte[0] = time.monotonic()
+                return bytes(buffer)
 
             def recover() -> None:
                 # Reopen rather than seek alone: a reader that dropped off the
@@ -382,7 +437,22 @@ def recover() -> None:
         started = True
 
         while True:
-            chunk = chunks.get()
+            try:
+                chunk = chunks.get(timeout=STALL_POLL)
+            except queue.Empty:
+                # Deliberately no check on whether the reader is still alive:
+                # it delivers its sentinel from a `finally`, and up to
+                # READ_AHEAD chunks can still be queued behind a thread that
+                # has already exited. Leaving on liveness would drop them.
+                if control is not None:
+                    # A hung read never reaches the reader's own checkpoint, so
+                    # without this a cancel waits on the operating system too.
+                    control.checkpoint()
+                if stall_after and on_stall is not None:
+                    idle = time.monotonic() - last_byte[0]
+                    if idle >= stall_after:
+                        on_stall(idle)
+                continue
             if chunk is None:
                 break
             src_hasher.update(chunk)
@@ -403,8 +473,12 @@ def recover() -> None:
     finally:
         stop.set()
         if started:
-            # Drain so a reader parked on a full queue can observe `stop`.
-            while thread.is_alive():
+            # Drain so a reader parked on a full queue can observe `stop`,
+            # which it does within one put timeout. Bounded well above that,
+            # because the other reason the thread may not be finishing is a
+            # read that has hung — see ABANDON_READER_AFTER.
+            deadline = time.monotonic() + ABANDON_READER_AFTER
+            while thread.is_alive() and time.monotonic() < deadline:
                 try:
                     chunks.get_nowait()
                 except queue.Empty:
@@ -616,6 +690,9 @@ def emit(event: ProgressEvent) -> None:
         partials = [t.with_name(t.name + PARTIAL_SUFFIX) for t in targets]
 
         bytes_at_start = counters.job_bytes_done
+        #: Longest gap between bytes on this file, if it ever stalled. A list
+        #: rather than a nonlocal because it belongs to this iteration.
+        stalls: list[float] = []
 
         try:
             # These close over the loop variables and are all invoked inside
@@ -644,11 +721,22 @@ def note_retry(attempt: int, exc: BaseException, pause: float,
                     f"{_src.name}: read failed ({exc}); "
                     f"attempt {attempt} of {options.retry.attempts}")
 
+            def note_stall(idle: float, _idx=index, _src=source, _st=stat,
+                           _stalls=stalls) -> None:
+                _stalls.append(idle)
+                emit(ProgressEvent(_idx, len(files), _src.name, "stalled",
+                                   0, _st.st_size,
+                                   counters.job_bytes_done,
+                                   counters.job_bytes_total,
+                                   stalled_for=idle))
+
             def copy_once(_src=source, _partials=partials, _idx=index,
                           _st=stat) -> _CopyResult:
                 nonlocal reread_noted
                 result = _copy_fanout(_src, _partials, options.algorithm,
-                                      on_chunk, control, options.retry)
+                                      on_chunk, control, options.retry,
+                                      on_stall=note_stall,
+                                      stall_after=options.stall_after)
                 if not options.paranoid:
                     return result
                 emit(ProgressEvent(_idx, len(files), _src.name, "reread",
@@ -682,6 +770,13 @@ def copy_once(_src=source, _partials=partials, _idx=index,
                 job.warnings.append(
                     f"{source.name} copied on attempt {used} of "
                     f"{options.retry.attempts} — the source may be failing")
+            if stalls:
+                # The copy is as good as any other; the link it came over is
+                # not, and a job that took an hour for this reason should say
+                # which files it waited on rather than look merely slow.
+                job.warnings.append(
+                    f"{source.name} stalled for up to {max(stalls):.0f}s with "
+                    f"no bytes arriving — a link that dropped, not slow media")
             if result.recovered_reads:
                 # Recovered without restarting the file, which is why the copy
                 # succeeded at all — but the sectors that needed it are real.

src/offloader/gui/queue_view.py

@@ -34,10 +34,18 @@
     "verify": "Verifying",
     "probe": "Reading metadata",
     "thumbs": "Extracting thumbnails",
+    "retry": "Retrying",
+    "stalled": "Stalled on",
 }
 
+#: Stages during which no bytes are moving, so a rate and an ETA computed from
+#: the last few seconds describe a past that has stopped being true.
+_NO_RATE_STAGES = frozenset({"stalled"})
+
 
 def _throughput(item: QueueItem) -> str:
+    if item.state is JobState.RUNNING and item.stage in _NO_RATE_STAGES:
+        return f"no data for {item.stalled_for:.0f}s"
     if item.state is JobState.RUNNING:
         rate = item.rate_bytes_per_sec
         eta = item.eta_seconds