fix: atomically apply reconciliation leases

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"]
 

src/archive_clients/daemon.py

@@ -427,12 +427,36 @@ class ArchiveClientDaemon:
         accepted = None
         accepted_for_execution = False
         try:
-            accepted = await asyncio.to_thread(
+            if command.WhichOneof("payload") == "reconcile_job":
-                self.store.accept_command,
+                reconciliation = command.reconcile_job
-                command.command_id,
+                accepted = await asyncio.to_thread(
-                encode_message(command),
+                    self.store.accept_reconcile_command,
-                encode_message(acknowledgement),
+                    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(
+                # A newer ensure command for the same route is sufficient to
-                    command, "", outbound, command_tasks, restore_ready=True
+                # 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,

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(
+            self._reconcile_job_connection(
-                "SELECT definition_json FROM jobs WHERE job_id = ?", (job_id,)
+                connection,
-            ).fetchone()
+                job_id=job_id,
-            if existing is not None and existing["definition_json"] != definition:
+                definition=definition,
-                raise JobConflict("reconciliation has a different job definition")
+                state=state,
-            connection.execute(
+                revision=revision,
-                "DELETE FROM events WHERE job_id = ? AND sequence > ?",
+                last_event_sequence=last_event_sequence,
-                (job_id, last_event_sequence),
+                committed=committed,
+                superseded_command_ids=superseded_command_ids,
             )
-            connection.execute(
+
-                """
+    @staticmethod
-                INSERT INTO jobs (job_id, definition_json, state, revision,
+    def _reconcile_job_connection(
-                                  last_event_sequence, committed)
+        connection: sqlite3.Connection,
-                VALUES (?, ?, ?, ?, ?, ?)
+        *,
-                ON CONFLICT(job_id) DO UPDATE SET
+        job_id: str,
-                    state = excluded.state, revision = excluded.revision,
+        definition: str,
-                    last_event_sequence = excluded.last_event_sequence,
+        state: str,
-                    committed = excluded.committed
+        revision: int,
-                """,
+        last_event_sequence: int,
-                (job_id, definition, state, revision, last_event_sequence, int(committed)),
+        committed: bool,
-            )
+        superseded_command_ids: list[str],
-            if superseded_command_ids:
+    ) -> None:
-                placeholders = ",".join("?" for _ in superseded_command_ids)
+        existing = connection.execute(
-                rows = connection.execute(
+            "SELECT definition_json FROM jobs WHERE job_id = ?", (job_id,)
-                    f"SELECT command_id, command_json FROM commands WHERE command_id IN ({placeholders})",
+        ).fetchone()
-                    superseded_command_ids,
+        if existing is not None and existing["definition_json"] != definition:
-                ).fetchall()
+            raise JobConflict("reconciliation has a different job definition")
-                owned_ids = [
+        connection.execute(
-                    row["command_id"] for row in rows
+            "DELETE FROM events WHERE job_id = ? AND sequence > ?",
-                    if _command_job_id(row["command_json"]) == job_id
+            (job_id, last_event_sequence),
-                ]
+        )
-                if owned_ids:
+        connection.execute(
-                    owned_placeholders = ",".join("?" for _ in owned_ids)
+            """
-                    connection.execute(
+            INSERT INTO jobs (job_id, definition_json, state, revision,
-                        f"UPDATE commands SET state = 'superseded' WHERE command_id IN ({owned_placeholders})",
+                              last_event_sequence, committed)
-                        owned_ids,
+            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,

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()