Recover transient Syncthing and SQLite job failures

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "archive-clients"
-version = "0.1.14"
+version = "0.1.15"
 requires-python = ">=3.11"
 dependencies = ["protobuf==7.35.1", "websockets==16.0"]
 

src/archive_clients/jobs.py

@@ -671,6 +671,15 @@ class ClientJobExecutor:
                 # Syncthing can expose the per-job directory before the
                 # ready marker and manifest have arrived atomically as a set.
                 pass
+            except RouteSetupError as exc:
+                # A network interruption can temporarily make the local
+                # Syncthing REST API unavailable.  The staged payload is
+                # durable and the job must remain recoverable; keep polling
+                # instead of turning a transient outage into a failed job.
+                logger.warning("syncthing_status_deferred", extra={
+                    "job_id": definition.job_id,
+                    "error_type": type(exc).__name__,
+                })
             time.sleep(self.poll_interval)
 
     def _target_materialize(

src/archive_clients/state.py

@@ -7,6 +7,8 @@ import json
 import os
 import sqlite3
 import stat
+import threading
+from contextlib import contextmanager
 from dataclasses import dataclass
 from pathlib import Path
 
@@ -14,6 +16,12 @@ from pathlib import Path
 SCHEMA_VERSION = 3
 
 
+# A client daemon executes unrelated jobs concurrently.  SQLite still permits
+# only one writer, so serialize this process's short state transactions rather
+# than allowing an otherwise healthy job to fail after its busy timeout.
+_DATABASE_LOCK = threading.RLock()
+
+
 class CommandConflict(RuntimeError):
     pass
 
@@ -557,14 +565,30 @@ class ClientStore:
             ).fetchall()
         return [dict(row) for row in rows]
 
-    def _connect(self) -> sqlite3.Connection:
+    @contextmanager
-        connection = sqlite3.connect(self.database, isolation_level=None, timeout=5)
+    def _connect(self):
-        connection.row_factory = sqlite3.Row
+        """Yield one connection while serializing local SQLite writers.
-        connection.execute("PRAGMA foreign_keys = ON")
+
-        connection.execute("PRAGMA journal_mode = WAL")
+        The longer SQLite timeout also covers a short lock held by a separate
-        connection.execute("PRAGMA synchronous = FULL")
+        maintenance process such as a backup.  The lock is deliberately held
-        connection.execute("PRAGMA busy_timeout = 5000")
+        for the full transaction, including ``BEGIN IMMEDIATE``.
-        return connection
+        """
+        with _DATABASE_LOCK:
+            connection = sqlite3.connect(
+                self.database, isolation_level=None, timeout=30
+            )
+            try:
+                connection.row_factory = sqlite3.Row
+                connection.execute("PRAGMA foreign_keys = ON")
+                connection.execute("PRAGMA journal_mode = WAL")
+                connection.execute("PRAGMA synchronous = FULL")
+                connection.execute("PRAGMA busy_timeout = 30000")
+                # Preserve the original ``with connection`` commit/rollback
+                # behavior used by every store operation.
+                with connection:
+                    yield connection
+            finally:
+                connection.close()
 
 
 def _canonical(value: object) -> str:

tests/test_jobs.py

@@ -12,7 +12,7 @@ from archive_clients.jobs import (
     JobExecutionError,
     _resource_fingerprint,
 )
-from archive_clients.syncthing import RouteSetupError
+from archive_clients.syncthing import RouteSetupError, SyncthingTransferStatus
 from archive_clients.resources import normalize_resource
 from archive_clients.state import ClientStore
 from archive_control.v1 import control_pb2, job_pb2
@@ -39,6 +39,44 @@ class SlowRescanSyncthing(CompleteSyncthing):
 
 
 class ClientJobHappyPathTests(unittest.TestCase):
+    def test_syncthing_api_outage_during_transfer_is_retried(self):
+        """A transient local REST outage must not terminally fail the job."""
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            store = ClientStore(root / "client.db")
+            store.initialize()
+            definition = job_pb2.JobDefinition(
+                job_id=str(uuid4()),
+                transfer={
+                    "source_client_id": "cache-1",
+                    "target_client_id": "archive-1",
+                    "route_id": "route-1",
+                },
+            )
+            observer = Mock()
+            observer.status.side_effect = [
+                RouteSetupError("Syncthing API is unavailable"),
+                SyncthingTransferStatus(1, 42, 42, True, 0),
+            ]
+            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._observer = Mock(return_value=observer)
+            progress = Mock()
+
+            executor._wait_for_syncthing(definition, progress)
+
+            self.assertEqual(observer.status.call_count, 2)
+            progress.assert_called_once_with(1, 42, 42, "0 Syncthing items still needed")
+
     def test_reconnect_replay_never_duplicates_any_transfer_step(self):
         for step in (
             job_pb2.JOB_STEP_KIND_SOURCE_STAGE,

tests/test_state.py

@@ -1,6 +1,7 @@
 import tempfile
 import unittest
 import os
+import threading
 from pathlib import Path
 from uuid import uuid4
 
@@ -13,6 +14,40 @@ from archive_clients.state import (
 
 
 class ClientStoreTests(unittest.TestCase):
+    def test_database_connections_serialize_local_writers(self):
+        """Concurrent jobs share one daemon DB without SQLite lock failures."""
+        with tempfile.TemporaryDirectory() as directory:
+            database = Path(directory) / "state.db"
+            first = ClientStore(database)
+            second = ClientStore(database)
+            first.initialize()
+            barrier = threading.Barrier(2)
+            failures: list[Exception] = []
+
+            def write(store, command_id):
+                try:
+                    barrier.wait()
+                    store.accept_command(command_id, '{"kind":"job"}', '{"ok":true}')
+                except Exception as exc:  # pragma: no cover - assertion below
+                    failures.append(exc)
+
+            left = threading.Thread(target=write, args=(first, "command-1"))
+            right = threading.Thread(target=write, args=(second, "command-2"))
+            left.start()
+            right.start()
+            left.join(1)
+            right.join(1)
+
+            self.assertFalse(left.is_alive())
+            self.assertFalse(right.is_alive())
+            self.assertEqual(failures, [])
+            self.assertEqual(len(first.list_accepted_commands()), 2)
+            with first._connect() as connection:
+                self.assertEqual(
+                    connection.execute("PRAGMA busy_timeout").fetchone()[0],
+                    30000,
+                )
+
     def test_command_acceptance_is_durable_and_content_addressed(self):
         with tempfile.TemporaryDirectory() as directory:
             database = Path(directory) / "state.db"