diff --git a/pyproject.toml b/pyproject.toml index 6f2ae30..49f9920 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "archive-clients" -version = "0.1.21" +version = "0.1.22" requires-python = ">=3.11" dependencies = ["protobuf==7.35.1", "websockets==16.0"] diff --git a/src/archive_clients/daemon.py b/src/archive_clients/daemon.py index 565bf62..d561994 100644 --- a/src/archive_clients/daemon.py +++ b/src/archive_clients/daemon.py @@ -427,12 +427,36 @@ class ArchiveClientDaemon: accepted = None accepted_for_execution = False try: - accepted = await asyncio.to_thread( - self.store.accept_command, - command.command_id, - encode_message(command), - encode_message(acknowledgement), - ) + if command.WhichOneof("payload") == "reconcile_job": + reconciliation = command.reconcile_job + accepted = await asyncio.to_thread( + self.store.accept_reconcile_command, + command.command_id, + encode_message(command), + encode_message(acknowledgement), + job_id=reconciliation.authoritative_job.definition.job_id, + definition_json=encode_message( + reconciliation.authoritative_job.definition + ), + state=job_pb2.JobState.Name( + reconciliation.authoritative_job.state + ), + revision=reconciliation.authoritative_job.revision, + last_event_sequence=( + reconciliation.authoritative_last_event_sequence + ), + committed=reconciliation.authoritative_job.committed, + superseded_command_ids=list( + reconciliation.superseded_command_ids + ), + ) + else: + accepted = await asyncio.to_thread( + self.store.accept_command, + command.command_id, + encode_message(command), + encode_message(acknowledgement), + ) if accepted.duplicate: acknowledgement = decode_message( accepted.acknowledgement_json, control_pb2.CommandAck() @@ -526,28 +550,13 @@ class ArchiveClientDaemon: outbound, command_tasks, ) - elif ( - accepted is not None - and accepted_for_execution - and command.WhichOneof("payload") == "reconcile_job" - ): - reconciliation = command.reconcile_job - await asyncio.to_thread( - self.store.reconcile_job, - job_id=reconciliation.authoritative_job.definition.job_id, - definition_json=encode_message(reconciliation.authoritative_job.definition), - state=job_pb2.JobState.Name(reconciliation.authoritative_job.state), - revision=reconciliation.authoritative_job.revision, - last_event_sequence=reconciliation.authoritative_last_event_sequence, - committed=reconciliation.authoritative_job.committed, - superseded_command_ids=list(reconciliation.superseded_command_ids), - ) async def _resume_commands( self, outbound: asyncio.Queue[str], command_tasks: set[asyncio.Task[None]], ) -> None: + latest_route_commands: dict[str, control_pb2.Command] = {} for row in await asyncio.to_thread(self.store.list_accepted_commands): acknowledgement = decode_message( str(row["acknowledgement_json"]), control_pb2.CommandAck() @@ -558,15 +567,21 @@ class ArchiveClientDaemon: str(row["command_json"]), control_pb2.Command() ) if command.WhichOneof("payload") == "ensure_route": - self._schedule_route_command( - command, "", outbound, command_tasks, restore_ready=True - ) + # A newer ensure command for the same route is sufficient to + # restore the process-local route path and replay its own + # updates. Replaying every historical ready command causes + # redundant Syncthing configuration after a restart. + latest_route_commands[command.ensure_route.route.route_id] = command # Job commands are intentionally not replayed here. A command # acknowledgement is durable on both sides; the control daemon # redelivers an unacknowledged command, while registration # reconciliation requests snapshots for active jobs. Replaying # every historical command re-emits overlapping event ranges and # can flood the single connection's bounded outbound queue. + for command in latest_route_commands.values(): + self._schedule_route_command( + command, "", outbound, command_tasks, restore_ready=True + ) async def _resume_route_commands( self, diff --git a/src/archive_clients/state.py b/src/archive_clients/state.py index 242ffbd..445edd5 100644 --- a/src/archive_clients/state.py +++ b/src/archive_clients/state.py @@ -115,6 +115,75 @@ class ClientStore: ) return CommandAcceptance(False, acknowledgement, state) + def accept_reconcile_command( + self, + command_id: str, + command_json: str, + acknowledgement_json: str, + *, + job_id: str, + definition_json: str, + state: str, + revision: int, + last_event_sequence: int, + committed: bool, + superseded_command_ids: list[str], + ) -> CommandAcceptance: + """Atomically durably accept and apply a reconciliation command. + + A reconciliation acknowledgement is meaningful only after its cursor + and retired leases have reached SQLite. Keeping both operations in + one transaction makes a reconnect either redeliver the command or + observe its completed effect; it cannot observe a bare acknowledgement. + """ + payload = _canonical(json.loads(command_json)) + acknowledgement = _canonical(json.loads(acknowledgement_json)) + definition = _canonical(json.loads(definition_json)) + digest = hashlib.sha256(payload.encode()).hexdigest() + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + existing = connection.execute( + """ + SELECT payload_sha256, command_json, acknowledgement_json, state + FROM commands WHERE command_id = ? + """, + (command_id,), + ).fetchone() + if existing: + if existing["payload_sha256"] != digest or existing["command_json"] != payload: + raise CommandConflict("command ID was reused with different content") + return CommandAcceptance( + True, existing["acknowledgement_json"], existing["state"] + ) + status = json.loads(acknowledgement).get("status") + command_state = ( + "rejected" + if status is not None + and status != "COMMAND_ACK_STATUS_ACCEPTED" + else "accepted" + ) + connection.execute( + """ + INSERT INTO commands ( + command_id, payload_sha256, command_json, + acknowledgement_json, state + ) VALUES (?, ?, ?, ?, ?) + """, + (command_id, digest, payload, acknowledgement, command_state), + ) + if command_state == "accepted": + self._reconcile_job_connection( + connection, + job_id=job_id, + definition=definition, + state=state, + revision=revision, + last_event_sequence=last_event_sequence, + committed=committed, + superseded_command_ids=superseded_command_ids, + ) + return CommandAcceptance(False, acknowledgement, command_state) + def list_active_job_cursors(self) -> list[dict[str, object]]: with self._connect() as connection: rows = connection.execute( @@ -256,43 +325,66 @@ class ClientStore: definition = _canonical(json.loads(definition_json)) with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") - existing = connection.execute( - "SELECT definition_json FROM jobs WHERE job_id = ?", (job_id,) - ).fetchone() - if existing is not None and existing["definition_json"] != definition: - raise JobConflict("reconciliation has a different job definition") - connection.execute( - "DELETE FROM events WHERE job_id = ? AND sequence > ?", - (job_id, last_event_sequence), + self._reconcile_job_connection( + connection, + job_id=job_id, + definition=definition, + state=state, + revision=revision, + last_event_sequence=last_event_sequence, + committed=committed, + superseded_command_ids=superseded_command_ids, ) - connection.execute( - """ - INSERT INTO jobs (job_id, definition_json, state, revision, - last_event_sequence, committed) - VALUES (?, ?, ?, ?, ?, ?) - ON CONFLICT(job_id) DO UPDATE SET - state = excluded.state, revision = excluded.revision, - last_event_sequence = excluded.last_event_sequence, - committed = excluded.committed - """, - (job_id, definition, state, revision, last_event_sequence, int(committed)), - ) - if superseded_command_ids: - placeholders = ",".join("?" for _ in superseded_command_ids) - rows = connection.execute( - f"SELECT command_id, command_json FROM commands WHERE command_id IN ({placeholders})", - superseded_command_ids, - ).fetchall() - owned_ids = [ - row["command_id"] for row in rows - if _command_job_id(row["command_json"]) == job_id - ] - if owned_ids: - owned_placeholders = ",".join("?" for _ in owned_ids) - connection.execute( - f"UPDATE commands SET state = 'superseded' WHERE command_id IN ({owned_placeholders})", - owned_ids, - ) + + @staticmethod + def _reconcile_job_connection( + connection: sqlite3.Connection, + *, + job_id: str, + definition: str, + state: str, + revision: int, + last_event_sequence: int, + committed: bool, + superseded_command_ids: list[str], + ) -> None: + existing = connection.execute( + "SELECT definition_json FROM jobs WHERE job_id = ?", (job_id,) + ).fetchone() + if existing is not None and existing["definition_json"] != definition: + raise JobConflict("reconciliation has a different job definition") + connection.execute( + "DELETE FROM events WHERE job_id = ? AND sequence > ?", + (job_id, last_event_sequence), + ) + connection.execute( + """ + INSERT INTO jobs (job_id, definition_json, state, revision, + last_event_sequence, committed) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(job_id) DO UPDATE SET + state = excluded.state, revision = excluded.revision, + last_event_sequence = excluded.last_event_sequence, + committed = excluded.committed + """, + (job_id, definition, state, revision, last_event_sequence, int(committed)), + ) + if superseded_command_ids: + placeholders = ",".join("?" for _ in superseded_command_ids) + rows = connection.execute( + f"SELECT command_id, command_json FROM commands WHERE command_id IN ({placeholders})", + superseded_command_ids, + ).fetchall() + owned_ids = [ + row["command_id"] for row in rows + if _command_job_id(row["command_json"]) == job_id + ] + if owned_ids: + owned_placeholders = ",".join("?" for _ in owned_ids) + connection.execute( + f"UPDATE commands SET state = 'superseded' WHERE command_id IN ({owned_placeholders})", + owned_ids, + ) def begin_file_operation( self, diff --git a/tests/test_state.py b/tests/test_state.py index e34b185..5fbbfb9 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -205,6 +205,39 @@ class ClientStoreTests(unittest.TestCase): self.assertEqual(row["last_event_sequence"], 0) self.assertEqual(row["state"], "JOB_STATE_QUEUED") + def test_reconciliation_acknowledgement_and_state_are_atomic(self): + with tempfile.TemporaryDirectory() as directory: + store = ClientStore(Path(directory) / "state.db") + store.initialize() + store.accept_command( + "stale", '{"executeStep":{"jobId":"job-1"}}', + '{"status":"COMMAND_ACK_STATUS_ACCEPTED"}', + ) + accepted = store.accept_reconcile_command( + "reconcile-1", '{"reconcileJob":{"authoritativeJob":{}}}', + '{"status":"COMMAND_ACK_STATUS_ACCEPTED"}', + job_id="job-1", definition_json='{"jobId":"job-1"}', + state="JOB_STATE_QUEUED", revision=0, + last_event_sequence=0, committed=False, + superseded_command_ids=["stale"], + ) + self.assertFalse(accepted.duplicate) + self.assertEqual(accepted.state, "accepted") + self.assertTrue(store.is_command_active("reconcile-1")) + self.assertFalse(store.is_command_active("stale")) + self.assertEqual( + store.job_snapshot_rows(["job-1"])[0]["last_event_sequence"], 0 + ) + duplicate = store.accept_reconcile_command( + "reconcile-1", '{"reconcileJob":{"authoritativeJob":{}}}', + '{"status":"COMMAND_ACK_STATUS_ACCEPTED"}', + job_id="job-1", definition_json='{"jobId":"job-1"}', + state="JOB_STATE_QUEUED", revision=0, + last_event_sequence=0, committed=False, + superseded_command_ids=["stale"], + ) + self.assertTrue(duplicate.duplicate) + if __name__ == "__main__": unittest.main()