From a12784dd18f2ac5d1a158821665546ccda5a08bc Mon Sep 17 00:00:00 2001 From: owenpkent <20529132+owenpkent@users.noreply.github.com> Date: Mon, 21 Sep 2026 18:09:03 -0400 Subject: [PATCH 1/4] Tell a stalled source apart from a slow one, and honour a cancel during it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A hung network handle raises nothing. It stops returning bytes, so there was no error to retry and no way to distinguish it from a link crawling: the queue went on showing whatever rate it last measured while the job sat frozen, and the operator had only the progress bar to judge by. Time the bytes rather than the chunks. Reads are assembled from 1 MiB sub-reads that stamp a heartbeat, so the liveness signal is independent of the 8 MiB chunk the queue and the hashers work in. That matters: at 350 KB/s a single chunk legitimately takes half a minute, and a chunk-granularity timer would call a working copy stalled. A gap past the threshold reports a "stalled" stage, the queue says "no data for 34s" instead of a rate that has stopped being true, and the file is named in the job's warnings at the end. No watchdog thread was needed: the consumer already runs on a different thread from the read it waits on, which is all a watchdog requires. Polling that wait also fixed a cancel that could not be honoured — the checkpoint is now checked while waiting, and teardown no longer joins a blocked reader without a deadline, which had made cancelling a hung job wait for the very timeout the operator was trying to escape. Reporting only: the read still cannot be aborted, so recovery waits for the operating system to turn the hang into an error. Recorded under what is not protected in data-safety.md, and left on the roadmap with the mechanism named. --- CHANGELOG.md | 20 ++++ README.md | 5 +- ROADMAP.md | 18 +++- docs/data-safety.md | 10 ++ src/offloader/cli.py | 11 +- src/offloader/engine.py | 101 ++++++++++++++++-- src/offloader/gui/queue_view.py | 8 ++ src/offloader/gui/worker.py | 18 ++++ tests/test_stall.py | 176 ++++++++++++++++++++++++++++++++ 9 files changed, 358 insertions(+), 9 deletions(-) create mode 100644 tests/test_stall.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 33aed42..61a5ee5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,18 @@ 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. + ### Changed - **Full verification is the default.** The read-back is the only mode that @@ -38,6 +50,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 diff --git a/README.md b/README.md index 3eaeb7f..34810a3 100644 --- a/README.md +++ b/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: diff --git a/ROADMAP.md b/ROADMAP.md index ae77690..e7d59ac 100644 --- a/ROADMAP.md +++ b/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 diff --git a/docs/data-safety.md b/docs/data-safety.md index 6741dcd..31ab716 100644 --- a/docs/data-safety.md +++ b/docs/data-safety.md @@ -201,6 +201,16 @@ 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. 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 diff --git a/src/offloader/cli.py b/src/offloader/cli.py index dee3e95..7e793d8 100644 --- a/src/offloader/cli.py +++ b/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)), ) diff --git a/src/offloader/engine.py b/src/offloader/engine.py index cfc309c..9542315 100644 --- a/src/offloader/engine.py +++ b/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 @@ -192,6 +213,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 +296,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 +313,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 +337,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 +359,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 +432,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 +468,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 +685,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 +716,21 @@ 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)) + 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 +764,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. diff --git a/src/offloader/gui/queue_view.py b/src/offloader/gui/queue_view.py index 0c9abd9..2a3b30d 100644 --- a/src/offloader/gui/queue_view.py +++ b/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 diff --git a/src/offloader/gui/worker.py b/src/offloader/gui/worker.py index 770b3e4..07f3063 100644 --- a/src/offloader/gui/worker.py +++ b/src/offloader/gui/worker.py @@ -62,6 +62,10 @@ class QueueItem: current_file: str = "" bytes_done: int = 0 bytes_total: int = 0 + #: When the current stall began, or None. Derived from the stage rather + #: than carried in the progress signal, which has no field for it and would + #: have to grow one on every consumer to say the same thing. + stalled_since: float | None = None started_at: float | None = None finished_at: float | None = None job: Job | None = None @@ -108,6 +112,13 @@ def rate_bytes_per_sec(self) -> float: return 0.0 return max(0, self.bytes_done - oldest_bytes) / span + @property + def stalled_for(self) -> float: + """Seconds since the last byte arrived, or 0 when data is moving.""" + if self.stalled_since is None: + return 0.0 + return max(0.0, time.monotonic() - self.stalled_since) + @property def eta_seconds(self) -> float | None: rate = self.rate_bytes_per_sec @@ -343,6 +354,13 @@ def _on_progress(self, identifier: int, fraction: float, stage: str, if item is None: return item.fraction = fraction + # Timed from the first stalled event, so the duration shown is the + # stall's own rather than the age of the last one seen. + if stage == "stalled": + if item.stalled_since is None: + item.stalled_since = time.monotonic() + else: + item.stalled_since = None item.stage = stage item.current_file = filename item.bytes_done = done diff --git a/tests/test_stall.py b/tests/test_stall.py new file mode 100644 index 0000000..f4d23b8 --- /dev/null +++ b/tests/test_stall.py @@ -0,0 +1,176 @@ +"""A source that stops delivering bytes without reporting an error. + +Retrying needs something to react to. A hung network handle gives it nothing — +it simply stops returning, which is why a dropped SMB session over a VPN reads +as a frozen progress bar rather than a failure. The watchdog exists to tell that +apart from a link that is merely slow, and most of these tests are about that +distinction: an 8 MiB chunk at 350 KB/s legitimately takes half a minute, so a +detector timing whole chunks would cry stall on a working copy. +""" + +from __future__ import annotations + +import builtins +import threading +import time +from pathlib import Path + +from offloader import engine, retry +from offloader.models import Profile, VerificationMode + + +def _options(tmp_path: Path, **overrides) -> engine.OffloadOptions: + defaults = dict( + destinations=[tmp_path / "dest"], + algorithm="xxh3-64", + verification=VerificationMode.NONE, + profile=Profile.DATA, + retry=retry.RetryPolicy(attempts=1), + stall_after=0.3, + ) + defaults.update(overrides) + return engine.OffloadOptions(**defaults) + + +class _PacedReader: + """A reader that waits a given time before each successive sub-read. + + Carries the parts of a binary file the engine uses — `read`, `seek`, + `close` — because recovering a chunk reopens the source and seeks back. + """ + + def __init__(self, handle, pauses: list[float], entered: list | None = None): + self._handle = handle + self._pauses = list(pauses) + self._entered = entered + + def read(self, size=-1): + if self._entered is not None: + self._entered.append(time.monotonic()) + if self._pauses: + time.sleep(self._pauses.pop(0)) + return self._handle.read(size) + + def seek(self, offset, whence=0): + return self._handle.seek(offset, whence) + + def close(self): + self._handle.close() + + def __enter__(self): + return self + + def __exit__(self, *args): + self._handle.close() + + +def _pace_source_reads(monkeypatch, card: Path, pauses: list[float], + entered: list | None = None) -> None: + real_open = builtins.open + + def paced_open(path, mode="r", *args, **kwargs): + handle = real_open(path, mode, *args, **kwargs) + try: + inside = Path(path).resolve().is_relative_to(card.resolve()) + except (OSError, ValueError): + inside = False + if inside and "r" in str(mode) and "b" in str(mode): + return _PacedReader(handle, pauses, entered) + return handle + + monkeypatch.setattr(builtins, "open", paced_open) + + +def _run(tmp_path: Path, monkeypatch, payload: bytes, pauses: list[float], + **overrides): + """Offload one file whose reads are paced, collecting the stages emitted.""" + # The consumer only looks between chunks, so the poll is the resolution at + # which a stall is noticed. Shortened here to keep the test quick. + monkeypatch.setattr(engine, "STALL_POLL", 0.02) + card = tmp_path / "card" + card.mkdir() + (card / "A001_C001.mov").write_bytes(payload) + + stages: list[str] = [] + _pace_source_reads(monkeypatch, card, pauses) + job = engine.run(card, _options(tmp_path, **overrides), + progress=lambda event: stages.append(event.stage)) + monkeypatch.undo() + return job, stages + + +def test_a_hung_read_is_reported_as_a_stall(tmp_path: Path, monkeypatch): + """The failure mode that has no error to retry: bytes simply stop.""" + job, stages = _run(tmp_path, monkeypatch, b"x" * 2048, pauses=[1.0]) + + assert "stalled" in stages + assert any("stalled" in w and "no bytes arriving" in w + for w in job.warnings) + + +def test_a_stall_still_produces_a_good_copy(tmp_path: Path, monkeypatch): + """Reporting a stall is all it does. The bytes are not in question — the + link they arrived over is.""" + payload = b"IRREPLACEABLE " * 1000 + job, _ = _run(tmp_path, monkeypatch, payload, pauses=[0.8], + verification=VerificationMode.FULL) + + assert job.final_status == "Verified" + assert (tmp_path / "dest" / "A001_C001.mov").read_bytes() == payload + + +def test_a_slow_link_is_not_a_stall(tmp_path: Path, monkeypatch): + """The false positive the sub-reads exist to prevent. + + Each 1 MiB sub-read lands inside the threshold, but the 8 MiB chunk they + build takes several times longer than it. Timing whole chunks would report + this working copy as stalled; timing bytes does not. + """ + payload = b"y" * (engine.SUBCHUNK_SIZE * 4) + job, stages = _run(tmp_path, monkeypatch, payload, + pauses=[0.12] * 5, stall_after=0.3) + + assert "stalled" not in stages + assert not any("stalled" in w for w in job.warnings) + assert (tmp_path / "dest" / "A001_C001.mov").read_bytes() == payload + + +def test_zero_disables_the_watchdog(tmp_path: Path, monkeypatch): + job, stages = _run(tmp_path, monkeypatch, b"x" * 2048, pauses=[0.8], + stall_after=0.0) + + assert "stalled" not in stages + assert not any("stalled" in w for w in job.warnings) + + +def test_a_cancel_is_noticed_while_a_read_hangs(tmp_path: Path, monkeypatch): + """A hung read never reaches the reader thread's own checkpoint, so without + the consumer's the cancel would wait on the operating system too. + + Cancelled from another thread *while* the read is sleeping — cancelling + before the run would be caught long before any of this, and prove nothing. + """ + monkeypatch.setattr(engine, "STALL_POLL", 0.02) + card = tmp_path / "card" + card.mkdir() + (card / "A001_C001.mov").write_bytes(b"x" * 2048) + + control = engine.JobControl() + entered: list[float] = [] + hang = 5.0 + _pace_source_reads(monkeypatch, card, [hang], entered) + + canceller = threading.Timer(0.2, control.cancel) + canceller.start() + started = time.monotonic() + try: + job = engine.run(card, _options(tmp_path), control=control) + finally: + canceller.cancel() + elapsed = time.monotonic() - started + monkeypatch.undo() + + assert entered, "the read never started, so nothing was hung to cancel" + assert elapsed < hang / 2, ( + f"waited {elapsed:.1f}s for a cancel while a read slept {hang}s") + assert job.cancelled From 200f71116389d7563de119b1ca76dcf64a160ac7 Mon Sep 17 00:00:00 2001 From: owenpkent <20529132+owenpkent@users.noreply.github.com> Date: Mon, 21 Sep 2026 18:22:07 -0400 Subject: [PATCH 2/4] Test the sub-read assembly and what the queue says during a stall MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sub-reads are on the data path, which makes them the part of this change worth attacking. A fencepost in that loop drops or duplicates bytes at a boundary, and because the checksum is computed from whatever the loop produced, a corrupted copy would faithfully match a corrupted source and verify clean at every level. So the sizes either side of a sub-read and a chunk boundary are compared against the bytes on disk and against a hash taken independently, not against another run of the same code. A short read gets its own test. read(n) returning fewer than n bytes does not mean end of file, and a loop that assumed it did would truncate every chunk to the first short read while still hashing consistently, since both sides see the same truncation. The display half was untested and is where the actual lie was: no progress events arrive during a stall, so the throughput column kept showing the last rate it measured. Covered now — no rate and no ETA while stalled, the rate still shown when bytes are moving, the clock started once per stall rather than restamped on every poll, and cleared when data resumes. --- tests/test_gui_stall_display.py | 162 ++++++++++++++++++++++++++++++++ tests/test_stall.py | 112 ++++++++++++++++++++++ 2 files changed, 274 insertions(+) create mode 100644 tests/test_gui_stall_display.py diff --git a/tests/test_gui_stall_display.py b/tests/test_gui_stall_display.py new file mode 100644 index 0000000..3f7d4c9 --- /dev/null +++ b/tests/test_gui_stall_display.py @@ -0,0 +1,162 @@ +"""What the queue says while a source has stopped delivering bytes. + +The engine's side of this is in `test_stall.py`. This is the half the operator +actually sees, and the thing it is fixing is a *lie*: during a stall no +progress events arrive, so the throughput column went on displaying the last +rate it measured. A frozen job that reads "573.8 MB/s" is worse than one that +reads nothing, because it invites waiting. +""" + +from __future__ import annotations + +import os +import time +from pathlib import Path + +import pytest + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") +pytest.importorskip("PySide6", reason="GUI extra not installed") + +from PySide6.QtWidgets import QApplication # noqa: E402 + +from offloader.gui.queue_view import ( # noqa: E402 + STAGE_VERBS, + _active_summary, + _throughput, +) +from offloader.gui.worker import JobState, QueueController, QueueItem # noqa: E402 +from offloader.models import VerificationMode # noqa: E402 +from offloader.presets import Preset # noqa: E402 + +GB = 1024 ** 3 + + +@pytest.fixture(scope="session") +def qapp(): + return QApplication.instance() or QApplication([]) + + +def _preset(tmp_path: Path) -> Preset: + return Preset(name="test", destinations=[tmp_path / "dest"], + algorithm="xxh3-64", verification=VerificationMode.FULL, + thumbnail_count=0, reports=["csv"]) + + +def _running(tmp_path: Path, **overrides) -> QueueItem: + values = dict(identifier=1, source=Path("E:\\"), name="A001", + preset=_preset(tmp_path), state=JobState.RUNNING, + fraction=0.5, stage="copy", + current_file="A001_C007.braw", + bytes_done=int(4 * GB), bytes_total=int(8 * GB), + started_at=time.monotonic() - 60) + values.update(overrides) + return QueueItem(**values) + + +# ------------------------------------------------------------- the item's clock + + +def test_an_item_that_is_not_stalled_has_no_stall_duration(tmp_path): + assert _running(tmp_path).stalled_for == 0.0 + + +def test_the_duration_is_measured_from_when_the_stall_began(tmp_path): + item = _running(tmp_path, stage="stalled") + item.stalled_since = time.monotonic() - 8 + assert item.stalled_for == pytest.approx(8, abs=1) + + +# ------------------------------------------------- the controller's transitions + + +def test_the_first_stalled_event_starts_the_clock(qapp, tmp_path): + controller = QueueController() + try: + item = _running(tmp_path) + controller.items = [item] + controller._on_progress(item.identifier, 0.5, "stalled", + item.current_file, 1, 2) + assert item.stalled_since is not None + finally: + controller.shutdown(2000) + + +def test_later_stalled_events_do_not_restart_the_clock(qapp, tmp_path): + """One event arrives per poll for as long as the stall lasts. Re-stamping + on each would report the age of the last event — about a second, forever — + instead of how long the source has actually been silent.""" + controller = QueueController() + try: + item = _running(tmp_path) + controller.items = [item] + controller._on_progress(item.identifier, 0.5, "stalled", "f", 1, 2) + began = item.stalled_since + time.sleep(0.05) + controller._on_progress(item.identifier, 0.5, "stalled", "f", 1, 2) + assert item.stalled_since == began + finally: + controller.shutdown(2000) + + +def test_any_other_stage_clears_the_clock(qapp, tmp_path): + """Bytes moved again, so the stall is over and the next one must be timed + from its own beginning.""" + controller = QueueController() + try: + item = _running(tmp_path) + controller.items = [item] + controller._on_progress(item.identifier, 0.5, "stalled", "f", 1, 2) + controller._on_progress(item.identifier, 0.6, "copy", "f", 2, 3) + assert item.stalled_since is None + assert item.stalled_for == 0.0 + finally: + controller.shutdown(2000) + + +# ------------------------------------------------------------- what is rendered + + +def test_a_stalled_job_shows_no_rate_at_all(tmp_path): + """The bug this closes: the column is populated from a trailing window of + samples, and during a stall no new ones arrive.""" + item = _running(tmp_path, stage="stalled") + item.stalled_since = time.monotonic() - 34 + item.record_progress(item.bytes_done) + + text = _throughput(item) + assert "MB/s" not in text and "GB/s" not in text + assert "ETA" not in text + assert "34s" in text + + +def test_a_moving_job_still_shows_its_rate(tmp_path): + """The other half — suppressing the rate must be specific to a stall, not + a blanket loss of the figure.""" + item = _running(tmp_path) + item._samples.append((time.monotonic() - 2, + item.bytes_done - 200 * 1024 * 1024)) + + assert "/s" in _throughput(item) + + +def test_the_summary_line_names_the_file_it_is_waiting_on(tmp_path): + item = _running(tmp_path, stage="stalled") + item.stalled_since = time.monotonic() - 12 + + summary = _active_summary(item) + assert "Stalled on" in summary + assert "A001_C007.braw" in summary + + +def test_the_stalled_and_retry_stages_read_as_words(tmp_path): + """Both are stages the engine emits, and neither had a verb — they would + have surfaced as the bare stage name.""" + assert STAGE_VERBS["stalled"] == "Stalled on" + assert STAGE_VERBS["retry"] == "Retrying" + + +def test_an_unmapped_stage_still_says_something(tmp_path): + """`reread` has no verb and should not render as an empty line.""" + item = _running(tmp_path, stage="reread") + assert "Reread" in _active_summary(item) diff --git a/tests/test_stall.py b/tests/test_stall.py index f4d23b8..00c8138 100644 --- a/tests/test_stall.py +++ b/tests/test_stall.py @@ -15,7 +15,10 @@ import time from pathlib import Path +import pytest + from offloader import engine, retry +from offloader.hashers import hash_file from offloader.models import Profile, VerificationMode @@ -143,6 +146,115 @@ def test_zero_disables_the_watchdog(tmp_path: Path, monkeypatch): assert not any("stalled" in w for w in job.warnings) +class _ShortReader: + """A reader that never returns as much as it was asked for. + + Legitimate behaviour for a handle — `read(n)` may return fewer bytes + without being at end of file — and the sub-read loop has to keep asking + rather than treat a short read as the end. + """ + + def __init__(self, handle, cap: int): + self._handle = handle + self._cap = cap + + def read(self, size=-1): + if size is None or size < 0: + return self._handle.read(size) + return self._handle.read(min(size, self._cap)) + + def seek(self, offset, whence=0): + return self._handle.seek(offset, whence) + + def close(self): + self._handle.close() + + def __enter__(self): + return self + + def __exit__(self, *args): + self._handle.close() + + +def _cap_source_reads(monkeypatch, card: Path, cap: int) -> None: + real_open = builtins.open + + def capped_open(path, mode="r", *args, **kwargs): + handle = real_open(path, mode, *args, **kwargs) + try: + inside = Path(path).resolve().is_relative_to(card.resolve()) + except (OSError, ValueError): + inside = False + if inside and "r" in str(mode) and "b" in str(mode): + return _ShortReader(handle, cap) + return handle + + monkeypatch.setattr(builtins, "open", capped_open) + + +@pytest.mark.parametrize("size", [ + 1, # a single short chunk + engine.SUBCHUNK_SIZE - 1, # just under one sub-read + engine.SUBCHUNK_SIZE, # exactly one + engine.SUBCHUNK_SIZE + 1, # spilling into a second + engine.CHUNK_SIZE + engine.SUBCHUNK_SIZE + 7, # past a chunk boundary +]) +def test_sub_reads_reassemble_the_file_exactly(tmp_path: Path, size: int): + """The data path. Reads are assembled from sub-reads now, so a fencepost + in that loop would drop or duplicate bytes at a boundary — and the + checksum is computed from what the loop produced, so a corrupted copy + would faithfully match a corrupted source and verify clean. Compared + against the bytes on disk rather than against another run of the same + code.""" + card = tmp_path / "card" + card.mkdir() + payload = bytes(range(251)) * (size // 251 + 1) + payload = payload[:size] + (card / "A001_C001.mov").write_bytes(payload) + + job = engine.run(card, _options(tmp_path, stall_after=0.0, + verification=VerificationMode.FULL)) + + assert job.final_status == "Verified" + landed = tmp_path / "dest" / "A001_C001.mov" + assert landed.read_bytes() == payload + assert job.files[0].checksum == hash_file(card / "A001_C001.mov", + "xxh3-64") + + +def test_a_short_read_is_not_mistaken_for_the_end_of_the_file( + tmp_path: Path, monkeypatch): + """`read(n)` returning fewer than n bytes does not mean end of file. If + the sub-read loop treated it that way, every chunk would be truncated to + the first short read and the copy would be silently cut off — while still + hashing consistently, because source and destination see the same + truncation.""" + card = tmp_path / "card" + card.mkdir() + payload = b"IRREPLACEABLE " * 900_000 # several chunks + (card / "A001_C001.mov").write_bytes(payload) + + # Well under SUBCHUNK_SIZE, so every single read comes back short. + _cap_source_reads(monkeypatch, card, cap=7919) + job = engine.run(card, _options(tmp_path, stall_after=0.0, + verification=VerificationMode.FULL)) + monkeypatch.undo() + + assert job.final_status == "Verified" + assert (tmp_path / "dest" / "A001_C001.mov").read_bytes() == payload + + +def test_the_stall_warning_reports_the_longest_gap(tmp_path: Path, monkeypatch): + """"Stalled for up to 8s" is the number worth having; the last gap seen + would under-report a job whose worst pause came early.""" + job, _ = _run(tmp_path, monkeypatch, b"x" * 2048, + pauses=[0.9], stall_after=0.2) + + stalled = [w for w in job.warnings if "stalled" in w] + assert len(stalled) == 1, "one warning per file, not one per poll" + assert "no bytes arriving" in stalled[0] + + def test_a_cancel_is_noticed_while_a_read_hangs(tmp_path: Path, monkeypatch): """A hung read never reaches the reader thread's own checkpoint, so without the consumer's the cancel would wait on the operating system too. From 9374c4623e980958b317d13712c0c0024e00e12f Mon Sep 17 00:00:00 2001 From: owenpkent <20529132+owenpkent@users.noreply.github.com> Date: Mon, 21 Sep 2026 21:05:10 -0400 Subject: [PATCH 3/4] Show the silence the engine already measured, not the age of the warning The first stalled event fires only once stall_after has elapsed, and the GUI started its clock when that event arrived. So a source that had supplied no bytes for fifteen seconds first displayed "no data for 0s", and went on understating the outage by the whole threshold for as long as it lasted. That figure is what someone reads to decide whether to go and look at the cable, and understating it by the default threshold is understating it by the only amount that matters at the point they first see it. note_stall already had the real idle duration - it is the number the "stalled for up to Ns" warning is built from - and was throwing it away at the event boundary. ProgressEvent carries it now, the runner passes it through, and the queue backdates its clock by it rather than stamping the arrival. Four regressions across both halves. On the engine side: the first stalled event reports at least the threshold, the reported silence never goes backwards, and no other stage claims a duration, since a leftover value on a copy event would restart a cleared clock. On the GUI side: a controlled duration arriving through _on_progress is what the item displays, and a stall the engine did not time still starts at zero rather than being invented. --- docs/data-safety.md | 11 +++++++--- src/offloader/engine.py | 8 ++++++- src/offloader/gui/worker.py | 17 +++++++++------ tests/test_gui_stall_display.py | 29 +++++++++++++++++++++++++ tests/test_stall.py | 38 +++++++++++++++++++++++++++++++++ 5 files changed, 93 insertions(+), 10 deletions(-) diff --git a/docs/data-safety.md b/docs/data-safety.md index 31ab716..1cb7002 100644 --- a/docs/data-safety.md +++ b/docs/data-safety.md @@ -206,9 +206,14 @@ known. 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. 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 + 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. diff --git a/src/offloader/engine.py b/src/offloader/engine.py index 9542315..ee8a9ec 100644 --- a/src/offloader/engine.py +++ b/src/offloader/engine.py @@ -179,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] @@ -722,7 +727,8 @@ def note_stall(idle: float, _idx=index, _src=source, _st=stat, emit(ProgressEvent(_idx, len(files), _src.name, "stalled", 0, _st.st_size, counters.job_bytes_done, - counters.job_bytes_total)) + counters.job_bytes_total, + stalled_for=idle)) def copy_once(_src=source, _partials=partials, _idx=index, _st=stat) -> _CopyResult: diff --git a/src/offloader/gui/worker.py b/src/offloader/gui/worker.py index 07f3063..35518be 100644 --- a/src/offloader/gui/worker.py +++ b/src/offloader/gui/worker.py @@ -141,7 +141,7 @@ def status_text(self) -> str: class _Runner(QThread): """Runs a single queue item off the UI thread.""" - progressed = Signal(int, float, str, str, int, int) + progressed = Signal(int, float, str, str, int, int, float) completed = Signal(int, object, object, object) # id, Job|None, reports, error def __init__(self, item: QueueItem, parent: QObject | None = None) -> None: @@ -161,7 +161,7 @@ def _on_progress(self, event: engine.ProgressEvent) -> None: if event.job_bytes_total else 0.0) self.progressed.emit( self._item.identifier, fraction, event.stage, event.file_name, - event.job_bytes_done, event.job_bytes_total, + event.job_bytes_done, event.job_bytes_total, event.stalled_for, ) def run(self) -> None: # noqa: D102 - QThread entry point @@ -349,16 +349,21 @@ def shutdown(self, timeout_ms: int = 5000) -> None: # ---------------------------------------------------------------- slots def _on_progress(self, identifier: int, fraction: float, stage: str, - filename: str, done: int, total: int) -> None: + filename: str, done: int, total: int, + stalled_for: float = 0.0) -> None: item = self.find(identifier) if item is None: return item.fraction = fraction - # Timed from the first stalled event, so the duration shown is the - # stall's own rather than the age of the last one seen. + # Backdated by the silence the engine had already measured, not timed + # from the event's arrival. The first stalled event only fires once + # `stall_after` has passed, so starting the clock here showed "no data + # for 0s" on a source that had supplied nothing for fifteen seconds, + # and stayed a whole threshold short for as long as the outage lasted. + # That number is what someone uses to decide whether to pull the cable. if stage == "stalled": if item.stalled_since is None: - item.stalled_since = time.monotonic() + item.stalled_since = time.monotonic() - max(0.0, stalled_for) else: item.stalled_since = None item.stage = stage diff --git a/tests/test_gui_stall_display.py b/tests/test_gui_stall_display.py index 3f7d4c9..bb1cf2d 100644 --- a/tests/test_gui_stall_display.py +++ b/tests/test_gui_stall_display.py @@ -82,6 +82,35 @@ def test_the_first_stalled_event_starts_the_clock(qapp, tmp_path): controller.shutdown(2000) +def test_the_first_displayed_duration_includes_the_threshold(qapp, tmp_path): + """REGRESSION. The engine only reports a stall once `stall_after` has + already passed. Starting the clock when the event arrived showed "no data + for 0s" on a source that had supplied nothing for fifteen seconds, and went + on understating the outage by that much for as long as it lasted. It is the + number someone uses to decide whether to go and pull the cable.""" + controller = QueueController() + try: + item = _running(tmp_path) + controller.items = [item] + controller._on_progress(item.identifier, 0.5, "stalled", + item.current_file, 1, 2, 15.4) + assert item.stalled_for == pytest.approx(15.4, abs=1) + finally: + controller.shutdown(2000) + + +def test_a_stall_the_engine_did_not_time_still_starts_at_zero(qapp, tmp_path): + """The clock is backdated by what was measured, never invented.""" + controller = QueueController() + try: + item = _running(tmp_path) + controller.items = [item] + controller._on_progress(item.identifier, 0.5, "stalled", "f", 1, 2, 0.0) + assert item.stalled_for == pytest.approx(0.0, abs=1) + finally: + controller.shutdown(2000) + + def test_later_stalled_events_do_not_restart_the_clock(qapp, tmp_path): """One event arrives per poll for as long as the stall lasts. Re-stamping on each would report the age of the last event — about a second, forever — diff --git a/tests/test_stall.py b/tests/test_stall.py index 00c8138..38d03af 100644 --- a/tests/test_stall.py +++ b/tests/test_stall.py @@ -255,6 +255,44 @@ def test_the_stall_warning_reports_the_longest_gap(tmp_path: Path, monkeypatch): assert "no bytes arriving" in stalled[0] +def test_the_stalled_event_carries_the_silence_already_observed( + tmp_path: Path, monkeypatch): + """REGRESSION. The first stalled event only fires once `stall_after` has + passed, so a consumer timing from its arrival starts at zero and stays a + whole threshold short of the real outage. The engine has measured that + silence by then; it just was not carrying it.""" + monkeypatch.setattr(engine, "STALL_POLL", 0.02) + card = tmp_path / "card" + card.mkdir() + (card / "A001_C001.mov").write_bytes(b"x" * 2048) + + stalls: list[float] = [] + _pace_source_reads(monkeypatch, card, [0.9]) + engine.run(card, _options(tmp_path, stall_after=0.3), + progress=lambda event: stalls.append(event.stalled_for) + if event.stage == "stalled" else None) + monkeypatch.undo() + + assert stalls, "no stalled event was emitted" + assert stalls[0] >= 0.3, f"the first event reported {stalls[0]:.2f}s" + assert stalls == sorted(stalls), f"the reported silence went backwards: {stalls}" + + +def test_no_other_stage_claims_a_stall_duration(tmp_path: Path, monkeypatch): + """The field is only meaningful on a stalled event, and a copy event + carrying a leftover value would restart a cleared clock.""" + card = tmp_path / "card" + card.mkdir() + (card / "A001_C001.mov").write_bytes(b"x" * 2048) + + seen: list[tuple[str, float]] = [] + engine.run(card, _options(tmp_path, stall_after=0.0), + progress=lambda event: seen.append((event.stage, event.stalled_for))) + + assert seen + assert all(value == 0.0 for stage, value in seen if stage != "stalled") + + def test_a_cancel_is_noticed_while_a_read_hangs(tmp_path: Path, monkeypatch): """A hung read never reaches the reader thread's own checkpoint, so without the consumer's the cancel would wait on the operating system too. From c7976816936bcc1b67037fe69cdd0a53b77a7f41 Mon Sep 17 00:00:00 2001 From: owenpkent <20529132+owenpkent@users.noreply.github.com> Date: Tue, 22 Sep 2026 11:38:31 -0400 Subject: [PATCH 4/4] Record the stall duration correction in the changelog --- CHANGELOG.md | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 61a5ee5..dc73065 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,14 @@ project uses [semantic versioning][semver]. 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