feat: execute durable syncthing route setup

src/archive_clients/daemon.py

@@ -26,8 +26,14 @@ from archive_clients.protocol import (
 from archive_clients.qbittorrent import QBittorrentReader
 from archive_clients.services import ServiceProbe
 from archive_clients.state import ClientStore, CommandConflict
+from archive_clients.syncthing import (
+    RoutePathConflict,
+    RouteSetupError,
+    RouteSetupTimeout,
+    SyncthingRouteManager,
+)
 from archive_control.v1 import (
-    client_pb2, common_pb2, control_pb2, inventory_pb2, job_pb2,
+    client_pb2, common_pb2, control_pb2, inventory_pb2, job_pb2, route_pb2,
 )
 
 
@@ -41,6 +47,7 @@ class ArchiveClientDaemon:
         probes: list[FilesystemProbe],
         service_probes: list[ServiceProbe],
         resource_reader: QBittorrentReader | None = None,
+        route_manager: SyncthingRouteManager | None = None,
     ):
         if len(probes) != 2:
             raise ValueError(
@@ -53,11 +60,25 @@ class ArchiveClientDaemon:
             InventoryService(resource_reader, config.client_id)
             if resource_reader is not None else None
         )
+        healthy_syncthing = any(
+            probe.service == "syncthing"
+            and probe.state == common_pb2.HEALTH_STATE_HEALTHY
+            for probe in service_probes
+        )
+        self.routes = route_manager or (
+            SyncthingRouteManager(
+                config.syncthing,
+                sparse_supported=probes[1].sparse_files,
+            )
+            if healthy_syncthing and probes[1].writable
+            else None
+        )
         self.store = ClientStore(config.state_db)
         self.backups = SQLiteBackupManager(
             config.state_db, config.backup_dir, config.backup
         )
         self._lease = DatabaseLease(config.state_db)
+        self._active_route_commands: set[str] = set()
 
     async def run(self) -> None:
         await asyncio.to_thread(self._lease.acquire)
@@ -148,6 +169,7 @@ class ArchiveClientDaemon:
             writer = asyncio.create_task(self._writer(websocket, outbound))
             command_tasks: set[asyncio.Task[None]] = set()
             try:
+                await self._resume_route_commands(outbound, command_tasks)
                 async for frame in websocket:
                     await self._handle(decode(frame), outbound, command_tasks)
             finally:
@@ -214,6 +236,10 @@ class ArchiveClientDaemon:
                 client_pb2.CLIENT_FEATURE_INVENTORY_CHUNKS,
                 client_pb2.CLIENT_FEATURE_CONTENT_TREE,
             ))
+        if self.routes is not None:
+            request.capabilities.features.append(
+                client_pb2.CLIENT_FEATURE_ROUTE_PROVISIONING
+            )
         for cursor in self.store.list_active_job_cursors():
             active = request.active_jobs.add()
             active.job_id = str(cursor["job_id"])
@@ -344,6 +370,67 @@ class ArchiveClientDaemon:
                     completed, command_tasks
                 )
             )
+        elif (
+            accepted is not None
+            and accepted_for_execution
+            and command.WhichOneof("payload") == "ensure_route"
+            and self.routes is not None
+        ):
+            self._schedule_route_command(
+                command,
+                envelope.message_id,
+                outbound,
+                command_tasks,
+            )
+
+    async def _resume_route_commands(
+        self,
+        outbound: asyncio.Queue[str],
+        command_tasks: set[asyncio.Task[None]],
+    ) -> None:
+        for row in await asyncio.to_thread(self.store.list_accepted_commands):
+            acknowledgement = decode_message(
+                str(row["acknowledgement_json"]), control_pb2.CommandAck()
+            )
+            if acknowledgement.status != control_pb2.COMMAND_ACK_STATUS_ACCEPTED:
+                continue
+            command = decode_message(
+                str(row["command_json"]), control_pb2.Command()
+            )
+            if command.WhichOneof("payload") == "ensure_route":
+                self._schedule_route_command(
+                    command, "", outbound, command_tasks
+                )
+
+    def _schedule_route_command(
+        self,
+        command: control_pb2.Command,
+        correlation_id: str,
+        outbound: asyncio.Queue[str],
+        command_tasks: set[asyncio.Task[None]],
+    ) -> None:
+        if command.command_id in self._active_route_commands:
+            return
+        self._active_route_commands.add(command.command_id)
+        task = asyncio.create_task(
+            self._execute_route(command, correlation_id, outbound),
+            name=f"route-{command.ensure_route.route.route_id}",
+        )
+        command_tasks.add(task)
+        task.add_done_callback(
+            lambda completed: self._route_command_finished(
+                command.command_id, completed, command_tasks
+            )
+        )
+
+    def _route_command_finished(
+        self,
+        command_id: str,
+        task: asyncio.Task[None],
+        command_tasks: set[asyncio.Task[None]],
+    ) -> None:
+        self._active_route_commands.discard(command_id)
+        self._command_finished(task, command_tasks)
 
     async def _send_inventory(
         self,
@@ -359,6 +446,207 @@ class ArchiveClientDaemon:
             response.inventory_chunk.CopyFrom(chunk)
             await outbound.put(encode(response))
 
+    async def _execute_route(
+        self,
+        command: Any,
+        correlation_id: str,
+        outbound: asyncio.Queue[str],
+    ) -> None:
+        assert self.routes is not None
+        spec = command.ensure_route.route
+        attempt = await asyncio.to_thread(
+            self.store.begin_route_attempt,
+            command.command_id,
+            spec.route_id,
+            encode_message(spec),
+            str(uuid.uuid4()),
+        )
+        for row in await asyncio.to_thread(
+            self.store.route_update_rows, command.command_id
+        ):
+            response = new_envelope()
+            response.correlation_id = correlation_id
+            response.route_update.CopyFrom(
+                decode_message(str(row["update_json"]), control_pb2.RouteUpdate())
+            )
+            await outbound.put(encode(response))
+        if attempt["state"] in {"ready", "failed"}:
+            return
+
+        deadline = time.monotonic() + spec.setup_timeout_seconds
+        configured = None
+        try:
+            attempt = await asyncio.to_thread(
+                self.store.get_route_attempt, command.command_id
+            )
+            if int(attempt["last_sequence"]) == 0:
+                await self._emit_route_update(
+                    command.command_id,
+                    correlation_id,
+                    outbound,
+                    sequence=1,
+                    state="provisioning",
+                    local_route=self._local_route(
+                        spec, route_pb2.ROUTE_STATE_PROVISIONING
+                    ),
+                    nonce=str(attempt["nonce"]),
+                    detail="route command accepted; configuring Syncthing",
+                )
+            configured = await asyncio.to_thread(
+                self.routes.configure, spec, deadline
+            )
+            await asyncio.to_thread(
+                self.store.record_route_ownership,
+                command.command_id,
+                configured.created_device,
+                configured.created_folder,
+            )
+            attempt = await asyncio.to_thread(
+                self.store.get_route_attempt, command.command_id
+            )
+            configured.local_route.archive_control_created = bool(
+                attempt["created_folder"]
+            )
+            if int(attempt["last_sequence"]) < 2:
+                configured.local_route.state = route_pb2.ROUTE_STATE_VERIFYING
+                configured.local_route.observed_at.GetCurrentTime()
+                await self._emit_route_update(
+                    command.command_id,
+                    correlation_id,
+                    outbound,
+                    sequence=2,
+                    state="verifying",
+                    local_route=configured.local_route,
+                    nonce=str(attempt["nonce"]),
+                    detail="Syncthing configuration verified; exchanging nonces",
+                )
+            local_seen, peer_seen = await asyncio.to_thread(
+                self.routes.verify,
+                configured,
+                spec,
+                self.config.client_id,
+                str(attempt["nonce"]),
+                deadline,
+            )
+            attempt = await asyncio.to_thread(
+                self.store.get_route_attempt, command.command_id
+            )
+            if int(attempt["last_sequence"]) < 3:
+                configured.local_route.state = route_pb2.ROUTE_STATE_READY
+                configured.local_route.detail = "bidirectional nonce verification passed"
+                configured.local_route.observed_at.GetCurrentTime()
+                await self._emit_route_update(
+                    command.command_id,
+                    correlation_id,
+                    outbound,
+                    sequence=3,
+                    state="ready",
+                    local_route=configured.local_route,
+                    nonce=str(attempt["nonce"]),
+                    detail="bidirectional nonce verification passed",
+                    local_nonce_seen_by_peer=local_seen,
+                    peer_nonce_seen_locally=peer_seen,
+                )
+        except Exception as exc:
+            attempt = await asyncio.to_thread(
+                self.store.get_route_attempt, command.command_id
+            )
+            if attempt is None or attempt["state"] == "failed":
+                return
+            code, retryable = _route_error(exc)
+            detail = str(exc) or type(exc).__name__
+            local_route = (
+                configured.local_route
+                if configured is not None
+                else self._local_route(spec, route_pb2.ROUTE_STATE_UNHEALTHY)
+            )
+            local_route.state = route_pb2.ROUTE_STATE_UNHEALTHY
+            local_route.detail = detail
+            local_route.observed_at.GetCurrentTime()
+            await self._emit_route_update(
+                command.command_id,
+                correlation_id,
+                outbound,
+                sequence=int(attempt["last_sequence"]) + 1,
+                state="failed",
+                local_route=local_route,
+                nonce=str(attempt["nonce"]),
+                detail=detail,
+                error_code=code,
+                retryable=retryable,
+            )
+
+    async def _emit_route_update(
+        self,
+        command_id: str,
+        correlation_id: str,
+        outbound: asyncio.Queue[str],
+        *,
+        sequence: int,
+        state: str,
+        local_route: route_pb2.LocalRoute,
+        nonce: str,
+        detail: str,
+        local_nonce_seen_by_peer: bool = False,
+        peer_nonce_seen_locally: bool = False,
+        error_code: int = common_pb2.ERROR_CODE_UNSPECIFIED,
+        retryable: bool = False,
+    ) -> None:
+        update = control_pb2.RouteUpdate(
+            command_id=command_id,
+            update_id=str(uuid.uuid4()),
+            sequence=sequence,
+        )
+        update.route.CopyFrom(local_route)
+        update.verification.route_id = local_route.route_id
+        update.verification.nonce = nonce
+        update.verification.local_nonce_seen_by_peer = local_nonce_seen_by_peer
+        update.verification.peer_nonce_seen_locally = peer_nonce_seen_locally
+        update.verification.state = local_route.state
+        update.verification.detail = detail
+        update.verification.observed_at.GetCurrentTime()
+        if error_code != common_pb2.ERROR_CODE_UNSPECIFIED:
+            update.error.code = error_code
+            update.error.message = detail
+            update.error.retryable = retryable
+        encoded_update = encode_message(update)
+        await asyncio.to_thread(
+            self.store.record_route_update,
+            command_id,
+            sequence,
+            state,
+            encoded_update,
+        )
+        response = new_envelope()
+        response.correlation_id = correlation_id
+        response.route_update.CopyFrom(update)
+        await outbound.put(encode(response))
+
+    def _local_route(
+        self, spec: route_pb2.EnsureRouteSpec, state: int
+    ) -> route_pb2.LocalRoute:
+        device_id = next(
+            (
+                probe.device_id
+                for probe in self.service_probes
+                if probe.service == "syncthing"
+            ),
+            "",
+        )
+        route = route_pb2.LocalRoute(
+            route_id=spec.route_id,
+            local_relative_path=spec.local_relative_path,
+            folder_type=route_pb2.SYNCTHING_FOLDER_TYPE_SEND_RECEIVE,
+            local_syncthing_device_id=device_id,
+            peer_syncthing_device_ids=[spec.peer_syncthing_device_id],
+            state=state,
+            writable=self.probes[1].writable,
+            sparse_supported=self.probes[1].sparse_files,
+            detail="route provisioning",
+        )
+        route.observed_at.GetCurrentTime()
+        return route
+
     @staticmethod
     def _command_finished(
         task: asyncio.Task[None], command_tasks: set[asyncio.Task[None]]
@@ -411,6 +699,24 @@ class ArchiveClientDaemon:
                 acknowledgement.error.message = "inventory scope is unsupported"
             else:
                 acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_ACCEPTED
+        elif command.WhichOneof("payload") == "ensure_route":
+            spec = command.ensure_route.route
+            if self.routes is None:
+                acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED
+                acknowledgement.error.code = common_pb2.ERROR_CODE_UNAVAILABLE
+                acknowledgement.error.message = "Syncthing route adapter is unavailable"
+            elif (
+                not spec.route_id
+                or not spec.peer_client_id
+                or not spec.peer_syncthing_device_id
+                or not spec.local_relative_path
+                or spec.setup_timeout_seconds <= 0
+            ):
+                acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED
+                acknowledgement.error.code = common_pb2.ERROR_CODE_INVALID_ARGUMENT
+                acknowledgement.error.message = "ensure route specification is invalid"
+            else:
+                acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_ACCEPTED
         else:
             acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED
             acknowledgement.error.code = common_pb2.ERROR_CODE_UNSUPPORTED
@@ -418,3 +724,15 @@ class ArchiveClientDaemon:
                 "command is not supported by this client build"
             )
         return acknowledgement
+
+
+def _route_error(exc: Exception) -> tuple[int, bool]:
+    if isinstance(exc, RoutePathConflict):
+        return common_pb2.ERROR_CODE_PATH_CONFLICT, False
+    if isinstance(exc, RouteSetupTimeout):
+        return common_pb2.ERROR_CODE_TIMEOUT, True
+    if isinstance(exc, PermissionError):
+        return common_pb2.ERROR_CODE_PERMISSION_DENIED, False
+    if isinstance(exc, RouteSetupError):
+        return common_pb2.ERROR_CODE_UNAVAILABLE, True
+    return common_pb2.ERROR_CODE_INTERNAL, False