From 0f94388f49d1ec9d146be2f81b50c54595ceded9 Mon Sep 17 00:00:00 2001 From: Cabbagec Date: Fri, 24 Jul 2026 23:27:11 +0000 Subject: [PATCH] fix: advance target steps past peer events --- 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 | 66 ++++++++++++++++---------- tests/test_jobs.py | 66 ++++++++++++++++++++++++++ 7 files changed, 113 insertions(+), 29 deletions(-) diff --git a/deploy/production/lithium/compose.yaml b/deploy/production/lithium/compose.yaml index 3cb219e..83fe98c 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.8 + image: sodium/archive-clients:v0.1.9 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 c2872f4..757d9ea 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.8 + image: sodium/archive-clients:v0.1.9 user: "1001:1001" restart: unless-stopped network_mode: host diff --git a/deploy/production/x2/compose.yaml b/deploy/production/x2/compose.yaml index c66ef03..f48195e 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.8 + image: sodium/archive-clients:v0.1.9 user: "1001:1001" restart: unless-stopped network_mode: host diff --git a/docs/deployment-and-usage.md b/docs/deployment-and-usage.md index 8d4b195..f07fc81 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.8 + image: sodium/archive-clients:v0.1.9 user: "1001:1001" restart: unless-stopped command: ["archive-client", "--config", "/etc/archive-control/client.toml"] diff --git a/pyproject.toml b/pyproject.toml index a463242..0501f4d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "archive-clients" -version = "0.1.8" +version = "0.1.9" 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 d42b5bf..bd40a84 100644 --- a/src/archive_clients/jobs.py +++ b/src/archive_clients/jobs.py @@ -120,7 +120,11 @@ class ClientJobExecutor: replay = self._replay( command.job_id, command.expected_last_event_sequence ) - if replay and replay[-1].type in { + step_replay = [ + event for event in replay + if event.progress.step == command.step + ] + if step_replay and step_replay[-1].type in { control_pb2.JOB_EVENT_TYPE_STEP_SUCCEEDED, control_pb2.JOB_EVENT_TYPE_COMMITTED, control_pb2.JOB_EVENT_TYPE_SUCCEEDED, @@ -132,31 +136,45 @@ class ClientJobExecutor: for event in replay: event_callback(event) return replay - started = replay[0] if replay else self._event( - definition, - sequence=command.expected_last_event_sequence + 1, - revision=command.expected_job_revision + 1, - event_type=control_pb2.JOB_EVENT_TYPE_STEP_STARTED, - state=job_pb2.JOB_STATE_RUNNING, - committed=( - self._committed(command.job_id) - or command.step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP - or ( - command.step == job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK - and self.store.get_job_artifact( - command.job_id, "qb-entry-removed" - ) is not None - ) - ), - step=command.step, - step_state=job_pb2.STEP_STATE_RUNNING, - ) - if not replay: + if step_replay: + started = step_replay[0] + cursor = step_replay[-1] + emitted = list(step_replay) + else: + previous = replay[-1] if replay else None + started = self._event( + definition, + sequence=( + previous.sequence + 1 + if previous is not None + else command.expected_last_event_sequence + 1 + ), + revision=( + previous.job_revision + 1 + if previous is not None + else command.expected_job_revision + 1 + ), + event_type=control_pb2.JOB_EVENT_TYPE_STEP_STARTED, + state=job_pb2.JOB_STATE_RUNNING, + committed=( + self._committed(command.job_id) + or command.step == job_pb2.JOB_STEP_KIND_STAGING_CLEANUP + or ( + command.step == job_pb2.JOB_STEP_KIND_SAFE_FILE_UNLINK + and self.store.get_job_artifact( + command.job_id, "qb-entry-removed" + ) is not None + ) + ), + step=command.step, + step_state=job_pb2.STEP_STATE_RUNNING, + ) self._record(definition, started) + cursor = started + emitted = [started] if event_callback is not None: - event_callback(started) - emitted = [started] - cursor = started + for event in emitted: + event_callback(event) speed_sample = [time.monotonic(), 0] def progress( diff --git a/tests/test_jobs.py b/tests/test_jobs.py index 1b21a74..8320999 100644 --- a/tests/test_jobs.py +++ b/tests/test_jobs.py @@ -111,6 +111,72 @@ class ClientJobHappyPathTests(unittest.TestCase): syncthing=SlowRescanSyncthing(), ) + def test_target_step_advances_past_replayed_source_completion(self): + 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, + resource_display_name="fixture", + 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=1, + )) + source_complete = executor._event( + definition, + sequence=5, + revision=4, + event_type=control_pb2.JOB_EVENT_TYPE_STEP_SUCCEEDED, + state=job_pb2.JOB_STATE_RUNNING, + committed=False, + step=job_pb2.JOB_STEP_KIND_SOURCE_STAGE, + step_state=job_pb2.STEP_STATE_SUCCEEDED, + ) + executor._record(definition, source_complete) + + with patch.object(executor, "_wait_for_syncthing"): + events = executor.execute(control_pb2.ExecuteStepCommand( + job_id=definition.job_id, + expected_job_revision=4, + expected_last_event_sequence=5, + step=job_pb2.JOB_STEP_KIND_SYNCTHING_TRANSFER, + attempt=1, + )) + + self.assertEqual(events[0].sequence, 6) + self.assertEqual( + events[0].progress.step, + job_pb2.JOB_STEP_KIND_SYNCTHING_TRANSFER, + ) + self.assertEqual(events[-1].sequence, 7) + self.assertEqual( + events[-1].type, + control_pb2.JOB_EVENT_TYPE_STEP_SUCCEEDED, + ) + def _run_transfer( self, root: Path,