From 9df2e6991ea13c927a114879923a6ba7a6d761f0 Mon Sep 17 00:00:00 2001 From: Cabbagec Date: Sat, 25 Jul 2026 02:50:22 +0000 Subject: [PATCH] fix: publish initial sparse transfer progress --- deploy/production/lithium/compose.yaml | 2 +- deploy/production/x1/compose.yaml | 2 +- deploy/production/x2/compose.yaml | 2 +- docs/deployment-and-usage.md | 2 +- pyproject.toml | 2 +- src/archive_clients/jobs.py | 15 +++++++-- tests/test_jobs.py | 43 ++++++++++++++++++++++++++ 7 files changed, 60 insertions(+), 8 deletions(-) diff --git a/deploy/production/lithium/compose.yaml b/deploy/production/lithium/compose.yaml index 06cd807..1746922 100644 --- a/deploy/production/lithium/compose.yaml +++ b/deploy/production/lithium/compose.yaml @@ -2,7 +2,7 @@ name: archive-control-archive services: archive-client: - image: sodium/archive-clients:v0.1.10 + image: sodium/archive-clients:v0.1.11 user: "1000:1000" restart: unless-stopped command: ["--config", "/etc/archive-control/client.toml"] diff --git a/deploy/production/x1/compose.yaml b/deploy/production/x1/compose.yaml index 5233777..8ebc07c 100644 --- a/deploy/production/x1/compose.yaml +++ b/deploy/production/x1/compose.yaml @@ -2,7 +2,7 @@ name: archive-control-cache services: archive-client: - image: sodium/archive-clients:v0.1.10 + image: sodium/archive-clients:v0.1.11 user: "1001:1001" restart: unless-stopped network_mode: host diff --git a/deploy/production/x2/compose.yaml b/deploy/production/x2/compose.yaml index e4ae4b9..e65d26f 100644 --- a/deploy/production/x2/compose.yaml +++ b/deploy/production/x2/compose.yaml @@ -2,7 +2,7 @@ name: archive-control-cache services: archive-client: - image: sodium/archive-clients:v0.1.10 + image: sodium/archive-clients:v0.1.11 user: "1001:1001" restart: unless-stopped network_mode: host diff --git a/docs/deployment-and-usage.md b/docs/deployment-and-usage.md index ea68f3a..0a16319 100644 --- a/docs/deployment-and-usage.md +++ b/docs/deployment-and-usage.md @@ -218,7 +218,7 @@ cache/archive routes according to policy. ```yaml services: archive-client: - image: sodium/archive-clients:v0.1.10 + image: sodium/archive-clients:v0.1.11 user: "1001:1001" restart: unless-stopped command: ["archive-client", "--config", "/etc/archive-control/client.toml"] diff --git a/pyproject.toml b/pyproject.toml index 71b5d53..c1cf818 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "archive-clients" -version = "0.1.10" +version = "0.1.11" requires-python = ">=3.11" dependencies = ["protobuf==7.35.1", "websockets==16.0"] diff --git a/src/archive_clients/jobs.py b/src/archive_clients/jobs.py index bd40a84..82ef3a6 100644 --- a/src/archive_clients/jobs.py +++ b/src/archive_clients/jobs.py @@ -175,7 +175,12 @@ class ClientJobExecutor: if event_callback is not None: for event in emitted: event_callback(event) - speed_sample = [time.monotonic(), 0] + # A newly-started step must publish its first observation. In + # particular, a sparse Syncthing temporary file may retain the same + # allocated-block count for a long time while data is still needed; + # suppressing that first observation made a healthy transfer look + # permanently stalled to the control daemon. + speed_sample: list[float | int | None] = [None, 0] def progress( fraction: float, @@ -185,8 +190,12 @@ class ClientJobExecutor: ) -> None: nonlocal cursor now = time.monotonic() - elapsed = now - float(speed_sample[0]) - if fraction < 1 and elapsed < 1: + previous_time = speed_sample[0] + elapsed = ( + now - float(previous_time) + if previous_time is not None else 0.0 + ) + if previous_time is not None and fraction < 1 and elapsed < 1: return speed = ( max(0, bytes_complete - int(speed_sample[1])) / elapsed diff --git a/tests/test_jobs.py b/tests/test_jobs.py index 8320999..19f2722 100644 --- a/tests/test_jobs.py +++ b/tests/test_jobs.py @@ -177,6 +177,49 @@ class ClientJobHappyPathTests(unittest.TestCase): control_pb2.JOB_EVENT_TYPE_STEP_SUCCEEDED, ) + def test_first_partial_progress_is_durable(self): + """A sparse transfer must not lose its only initial observation.""" + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + store = ClientStore(root / "client.db") + store.initialize() + definition = job_pb2.JobDefinition( + job_id=str(uuid4()), + idempotency_key=str(uuid4()), + operation=job_pb2.JOB_OPERATION_ARCHIVE, + transfer={ + "source_client_id": "cache-1", + "target_client_id": "archive-1", + "route_id": "route-1", + }, + ) + definition.resource_id.info_hash_v1_hex = "a" * 40 + definition.created_at.GetCurrentTime() + executor = ClientJobExecutor( + client_id="archive-1", qbittorrent=Mock(), store=store, + qb_root=root, qb_api_root=Path("/downloads"), + route_path=lambda _: root, syncthing_transport=Mock(), + sparse_supported=True, poll_interval=0, + ) + executor.assign(control_pb2.AssignJobCommand( + job=definition, expected_job_revision=1, + expected_last_event_sequence=0, + )) + + def partial_progress(_definition, _step, progress): + progress(0.001, 10, 10_000, "still receiving") + + with patch.object(executor, "_execute_step", partial_progress): + events = executor.execute(control_pb2.ExecuteStepCommand( + job_id=definition.job_id, expected_job_revision=1, + expected_last_event_sequence=1, + step=job_pb2.JOB_STEP_KIND_SYNCTHING_TRANSFER, attempt=1, + )) + + self.assertEqual(len(events), 3) + self.assertEqual(events[1].type, control_pb2.JOB_EVENT_TYPE_PROGRESS) + self.assertEqual(events[1].progress.bytes_complete, 10) + def _run_transfer( self, root: Path,