diff --git a/pyproject.toml b/pyproject.toml index b0ecd3f..9d67c08 100644 --- a/pyproject.toml +++ b/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"] diff --git a/src/archive_clients/jobs.py b/src/archive_clients/jobs.py index 8c59779..3391a78 100644 --- a/src/archive_clients/jobs.py +++ b/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( diff --git a/src/archive_clients/state.py b/src/archive_clients/state.py index c9c6671..2504a19 100644 --- a/src/archive_clients/state.py +++ b/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: - connection = sqlite3.connect(self.database, isolation_level=None, timeout=5) - 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 = 5000") - return connection + @contextmanager + def _connect(self): + """Yield one connection while serializing local SQLite writers. + + The longer SQLite timeout also covers a short lock held by a separate + maintenance process such as a backup. The lock is deliberately held + for the full transaction, including ``BEGIN IMMEDIATE``. + """ + 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: diff --git a/tests/test_jobs.py b/tests/test_jobs.py index eef895f..317cb3e 100644 --- a/tests/test_jobs.py +++ b/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, diff --git a/tests/test_state.py b/tests/test_state.py index 8b6c78f..da40aa0 100644 --- a/tests/test_state.py +++ b/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"