From bffd90a3a0ff73684628dd173ba2743d84650934 Mon Sep 17 00:00:00 2001 From: ClaydeCode Date: Wed, 22 Jul 2026 09:19:47 +0000 Subject: [PATCH 1/2] Harden backup pipeline: unblock event loop, pin task refs, build argv as list MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to a4a80bc (issue #127). Four robustness fixes in shard_core/service/backup.py, each with tests in tests/test_backup.py: 1. Offload the Azure marker-blob upload to a worker thread. `_write_marker_blob` called the sync `BlobClient.upload_blob` directly on the event loop; a slow Azure call stalled the whole loop (including the auth path). Now awaits `asyncio.to_thread(...)`. The function becomes async and its caller awaits it. 2. Run `rclone obscure` via `asyncio.create_subprocess_exec` instead of the blocking `subprocess.run`, so it no longer stalls the loop. Also check the exit code and raise BackupFailedError on failure — previously a failed obscure silently returned an empty crypt-password (broken encryption). 3. Hold strong references to fire-and-forget tasks. asyncio holds tasks weakly, so a long backup task (and the unreferenced on_task_done wrapper) could be garbage-collected mid-run. Add a module-level `background_tasks` set with a discard callback, mirroring app_lifecycle.py. 4. Build the rclone argv as a list instead of formatting a string and calling `command.split()`, which would break on any path or SAS URL containing whitespace. Replaces the COMMAND_TEMPLATE / CLEARTEXT_COMMAND_TEMPLATE strings with `_build_backup_command` / `_build_cleartext_backup_command`; the #117 unsupported-flag guard test now runs against both argv builders. Co-Authored-By: Claude Opus 4.8 --- shard_core/service/backup.py | 103 ++++++++++++++++++------- tests/test_backup.py | 145 ++++++++++++++++++++++++++++++++--- 2 files changed, 209 insertions(+), 39 deletions(-) diff --git a/shard_core/service/backup.py b/shard_core/service/backup.py index e623fa5..fa7151d 100644 --- a/shard_core/service/backup.py +++ b/shard_core/service/backup.py @@ -2,7 +2,6 @@ import datetime import json import logging -import subprocess import traceback from pathlib import Path from typing import List @@ -30,23 +29,54 @@ STORE_KEY_BACKUP_PASSPHRASE_LAST_ACCESS = "backup_passphrase_last_access" BACKUP_IN_PROGESS_LOCK = asyncio.Lock() -COMMAND_TEMPLATE = """ -rclone ---azureblob-sas-url {sas_token} ---crypt-password {obscured_password} ---crypt-remote :azureblob:{container_name} -sync {directory} :crypt:{container_name}/{directory} ---create-empty-src-dirs --stats-log-level NOTICE --stats 1000m --use-json-log ---fast-list -""" - -CLEARTEXT_COMMAND_TEMPLATE = """ -rclone ---azureblob-sas-url {sas_token} -sync {directory} :azureblob:{container_name}/{directory} ---create-empty-src-dirs --stats-log-level NOTICE --stats 1000m --use-json-log ---fast-list -""" +# Strong references to fire-and-forget tasks. asyncio only holds tasks weakly, so +# without this a long-running backup could be garbage-collected mid-run. Mirrors +# app_lifecycle.background_tasks. +background_tasks = set() + + +def _build_backup_command( + sas_token: str, obscured_password: str, container_name: str, directory: Path +) -> List[str]: + return [ + "rclone", + "--azureblob-sas-url", + sas_token, + "--crypt-password", + obscured_password, + "--crypt-remote", + f":azureblob:{container_name}", + "sync", + str(directory), + f":crypt:{container_name}/{directory}", + "--create-empty-src-dirs", + "--stats-log-level", + "NOTICE", + "--stats", + "1000m", + "--use-json-log", + "--fast-list", + ] + + +def _build_cleartext_backup_command( + sas_token: str, container_name: str, directory: Path +) -> List[str]: + return [ + "rclone", + "--azureblob-sas-url", + sas_token, + "sync", + str(directory), + f":azureblob:{container_name}/{directory}", + "--create-empty-src-dirs", + "--stats-log-level", + "NOTICE", + "--stats", + "1000m", + "--use-json-log", + "--fast-list", + ] async def start_backup(): @@ -67,6 +97,7 @@ async def start_backup(): directories, sas_url_response.container_name, sas_url_response.sas_url ) ) + _register_background_task(task) async def on_task_done(task: asyncio.Task): if task.exception(): @@ -78,7 +109,14 @@ async def on_task_done(task: asyncio.Task): else: signals.on_backup_update.send() - task.add_done_callback(lambda task: asyncio.create_task(on_task_done(task))) + task.add_done_callback( + lambda task: _register_background_task(asyncio.create_task(on_task_done(task))) + ) + + +def _register_background_task(task: asyncio.Task): + background_tasks.add(task) + task.add_done_callback(background_tasks.discard) async def backup_directories( @@ -130,11 +168,11 @@ async def backup_directories( ) async with db_conn() as conn: await db_backups.insert(conn, report) - _write_marker_blob(container_name, sas_token) + await _write_marker_blob(container_name, sas_token) log.info("Backup done") -def _write_marker_blob(container_name: str, sas_token: str): +async def _write_marker_blob(container_name: str, sas_token: str): """Write a marker blob to the backup container to record the time of the last backup. The blob's Last-Modified timestamp on Azure is updated on every call, giving the @@ -149,7 +187,7 @@ def _write_marker_blob(container_name: str, sas_token: str): blob_url = urlunparse(parsed._replace(path=f"/{container_name}/_last_backup")) blob_client = BlobClient.from_blob_url(blob_url) timestamp = datetime.datetime.now(datetime.timezone.utc).isoformat() - blob_client.upload_blob(timestamp, overwrite=True) + await asyncio.to_thread(blob_client.upload_blob, timestamp, overwrite=True) log.debug("Wrote _last_backup marker blob") except Exception: log.warning("Failed to write _last_backup marker blob", exc_info=True) @@ -162,7 +200,7 @@ def is_backup_in_progress(): async def _backup_directory( container_name, rel_directory, obscured_passphrase, sas_token ): - command = COMMAND_TEMPLATE.format( + command = _build_backup_command( sas_token=sas_token, obscured_password=obscured_passphrase, container_name=container_name, @@ -170,7 +208,7 @@ async def _backup_directory( ) path_root = Path(settings().path_root) process = await asyncio.create_subprocess_exec( - *command.split(), + *command, cwd=path_root, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, @@ -191,10 +229,19 @@ async def _backup_directory( async def _get_obscured_passphrase(): passphrase = await database.get_value(STORE_KEY_BACKUP_PASSPHRASE) - obscured_passphrase = subprocess.run( - ["rclone", "obscure", passphrase], capture_output=True, text=True - ).stdout.strip() - return obscured_passphrase + process = await asyncio.create_subprocess_exec( + "rclone", + "obscure", + passphrase, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await process.communicate() + if process.returncode != 0: + message = stderr.decode().strip() + log.error(f"rclone obscure exited with {process.returncode}: {message}") + raise BackupFailedError(message) + return stdout.decode().strip() def _get_relative_directory(directory: Path) -> Path: diff --git a/tests/test_backup.py b/tests/test_backup.py index d6cc880..d67d7c9 100644 --- a/tests/test_backup.py +++ b/tests/test_backup.py @@ -1,31 +1,65 @@ +import asyncio import json -from unittest.mock import AsyncMock, patch +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch import pytest from shard_core.service import backup from shard_core.service.backup import ( - COMMAND_TEMPLATE, - CLEARTEXT_COMMAND_TEMPLATE, BackupFailedError, _backup_directory, + _build_backup_command, + _build_cleartext_backup_command, + _get_obscured_passphrase, + _write_marker_blob, + start_backup, ) -def test_command_templates_only_use_supported_flags(): +def test_backup_commands_only_use_supported_flags(): """Guard against reintroducing rclone flags missing from the bundled binary. The core image ships rclone 1.60.1 (Debian bookworm), which lacks --azureblob-no-check-container (added in 1.61.0). See issue #117. """ - for template in (COMMAND_TEMPLATE, CLEARTEXT_COMMAND_TEMPLATE): - assert "--azureblob-no-check-container" not in template - assert "--fast-list" in template + argvs = ( + _build_backup_command("sas", "obscured", "container", "dir"), + _build_cleartext_backup_command("sas", "container", "dir"), + ) + for argv in argvs: + assert "--azureblob-no-check-container" not in argv + assert "--fast-list" in argv -def _mock_process(returncode, stderr): +def test_build_backup_command_keeps_whitespace_values_as_single_argv(): + """A SAS URL or path containing whitespace must stay one argv element, + not be split into several tokens (the bug behind command.split()).""" + sas_with_space = "https://acct.blob.core.windows.net/c?sv=a b&sig=x y" + argv = _build_backup_command(sas_with_space, "obscured", "container", "my dir") + + assert sas_with_space in argv + assert "my dir" in argv + assert argv[0] == "rclone" + # the crypt destination for a spaced directory stays intact + assert ":crypt:container/my dir" in argv + + +@pytest.mark.parametrize( + "builder,args", + [ + (_build_backup_command, ("sas", "obscured", "container", "dir")), + (_build_cleartext_backup_command, ("sas", "container", "dir")), + ], +) +def test_backup_command_argv_is_flat_string_list(builder, args): + argv = builder(*args) + assert all(isinstance(token, str) for token in argv) + + +def _mock_process(returncode, stdout=b"", stderr=b""): process = AsyncMock() - process.communicate.return_value = (b"", stderr) + process.communicate.return_value = (stdout, stderr) process.returncode = returncode return process @@ -34,7 +68,7 @@ def _mock_process(returncode, stderr): async def test_backup_directory_returns_stats_on_success(): stats = {"bytes": 42, "errors": 0} stderr = (json.dumps({"stats": stats}) + "\n").encode() - process = _mock_process(0, stderr) + process = _mock_process(0, stderr=stderr) with patch.object(backup.asyncio, "create_subprocess_exec", return_value=process): result = await _backup_directory("container", "dir", "obscured", "sas") assert result == stats @@ -43,8 +77,97 @@ async def test_backup_directory_returns_stats_on_success(): @pytest.mark.asyncio async def test_backup_directory_raises_on_nonzero_exit(): stderr = b"Error: unknown flag: --azureblob-no-check-container\n" - process = _mock_process(1, stderr) + process = _mock_process(1, stderr=stderr) with patch.object(backup.asyncio, "create_subprocess_exec", return_value=process): with pytest.raises(BackupFailedError) as exc_info: await _backup_directory("container", "dir", "obscured", "sas") assert "unknown flag: --azureblob-no-check-container" in str(exc_info.value) + + +@pytest.mark.asyncio +async def test_get_obscured_passphrase_uses_async_subprocess(): + process = _mock_process(0, stdout=b"OBSCURED\n") + create = AsyncMock(return_value=process) + with ( + patch.object(backup.asyncio, "create_subprocess_exec", create), + patch.object(backup.database, "get_value", AsyncMock(return_value="secret")), + ): + result = await _get_obscured_passphrase() + + assert result == "OBSCURED" + create.assert_awaited_once() + assert create.await_args.args == ("rclone", "obscure", "secret") + + +@pytest.mark.asyncio +async def test_get_obscured_passphrase_raises_on_nonzero_exit(): + process = _mock_process(1, stderr=b"boom\n") + with ( + patch.object( + backup.asyncio, "create_subprocess_exec", AsyncMock(return_value=process) + ), + patch.object(backup.database, "get_value", AsyncMock(return_value="secret")), + ): + with pytest.raises(BackupFailedError) as exc_info: + await _get_obscured_passphrase() + assert "boom" in str(exc_info.value) + + +@pytest.mark.asyncio +async def test_write_marker_blob_offloads_upload_to_thread(): + """The blocking Azure upload must run in a worker thread, never directly on + the event loop.""" + blob_client = MagicMock() + to_thread = AsyncMock() + with ( + patch.object(backup.BlobClient, "from_blob_url", return_value=blob_client), + patch.object(backup.asyncio, "to_thread", to_thread), + ): + await _write_marker_blob( + "container", "https://acct.blob.core.windows.net/c?sv=x&sig=y" + ) + + # upload happened via to_thread, not a direct (loop-blocking) call + blob_client.upload_blob.assert_not_called() + to_thread.assert_awaited_once() + assert to_thread.await_args.args[0] == blob_client.upload_blob + + +@pytest.mark.asyncio +async def test_write_marker_blob_swallows_errors(): + with patch.object( + backup.BlobClient, "from_blob_url", side_effect=RuntimeError("nope") + ): + # must not raise - the marker is best-effort + await _write_marker_blob("container", "https://acct/c?sig=y") + + +@pytest.mark.asyncio +async def test_start_backup_keeps_strong_reference_to_task(): + """The backup task must be held by the module-level set so it cannot be + garbage-collected mid-run.""" + release = asyncio.Event() + started = asyncio.Event() + + async def fake_backup(*_args, **_kwargs): + started.set() + await release.wait() + + sas = SimpleNamespace(container_name="c", sas_url="s") + with ( + patch.object(backup, "get_backup_sas_url", AsyncMock(return_value=sas)), + patch.object(backup, "backup_directories", fake_backup), + patch.object(backup.signals, "on_backup_update"), + ): + assert backup.background_tasks == set() + await start_backup() + await asyncio.wait_for(started.wait(), timeout=5) + + assert len(backup.background_tasks) == 1 + + release.set() + for _ in range(100): + await asyncio.sleep(0) + if not backup.background_tasks: + break + assert backup.background_tasks == set() From 63d720054a45527ef3641ae8d03d2542d119490d Mon Sep 17 00:00:00 2001 From: ClaydeCode Date: Wed, 22 Jul 2026 09:26:50 +0000 Subject: [PATCH 2/2] Address review panel: cover both fire-and-forget tasks and the outcome signal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Test-adversary panel found two blocking gaps and cheap advisories: - Strong-ref test only proved the main backup task was registered, not the spawned on_task_done wrapper (reverting its registration stayed green). Now spies on _register_background_task and asserts both tasks are registered. - The backup-outcome signal was entirely unasserted. Add tests that the success path sends on_backup_update() and the failure path sends it with the exception — the path that tells the UI a backup failed. - Assert the built argv list reaches create_subprocess_exec unsplit (the actual call site of the whitespace fix). - Exercise the Path->str conversion (flat-list test now passes a Path) and the marker-blob URL surgery + payload forwarding; cover an error raised inside the upload thread, not just at client setup. - Rename the done-callback lambda param to stop shadowing the outer task. Co-Authored-By: Claude Opus 4.8 --- shard_core/service/backup.py | 4 +- tests/test_backup.py | 117 +++++++++++++++++++++++++++++++---- 2 files changed, 109 insertions(+), 12 deletions(-) diff --git a/shard_core/service/backup.py b/shard_core/service/backup.py index fa7151d..aa05d69 100644 --- a/shard_core/service/backup.py +++ b/shard_core/service/backup.py @@ -110,7 +110,9 @@ async def on_task_done(task: asyncio.Task): signals.on_backup_update.send() task.add_done_callback( - lambda task: _register_background_task(asyncio.create_task(on_task_done(task))) + lambda completed: _register_background_task( + asyncio.create_task(on_task_done(completed)) + ) ) diff --git a/tests/test_backup.py b/tests/test_backup.py index d67d7c9..214ec02 100644 --- a/tests/test_backup.py +++ b/tests/test_backup.py @@ -1,5 +1,6 @@ import asyncio import json +from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch @@ -48,11 +49,13 @@ def test_build_backup_command_keeps_whitespace_values_as_single_argv(): @pytest.mark.parametrize( "builder,args", [ - (_build_backup_command, ("sas", "obscured", "container", "dir")), - (_build_cleartext_backup_command, ("sas", "container", "dir")), + (_build_backup_command, ("sas", "obscured", "container", Path("my dir"))), + (_build_cleartext_backup_command, ("sas", "container", Path("my dir"))), ], ) def test_backup_command_argv_is_flat_string_list(builder, args): + # A Path directory must be stringified into a single str token, never leak + # a Path object into argv (which create_subprocess_exec would reject). argv = builder(*args) assert all(isinstance(token, str) for token in argv) @@ -74,6 +77,25 @@ async def test_backup_directory_returns_stats_on_success(): assert result == stats +@pytest.mark.asyncio +async def test_backup_directory_passes_unsplit_argv_to_subprocess(): + """The whitespace fix lives at the call site: the built list must reach + create_subprocess_exec verbatim, not be re-split.""" + stats = {"bytes": 1, "errors": 0} + stderr = (json.dumps({"stats": stats}) + "\n").encode() + create = AsyncMock(return_value=_mock_process(0, stderr=stderr)) + with patch.object(backup.asyncio, "create_subprocess_exec", create): + await _backup_directory("container", Path("my dir"), "obscured", "sas url") + + expected = _build_backup_command( + sas_token="sas url", + obscured_password="obscured", + container_name="container", + directory=Path("my dir"), + ) + assert create.await_args.args == tuple(expected) + + @pytest.mark.asyncio async def test_backup_directory_raises_on_nonzero_exit(): stderr = b"Error: unknown flag: --azureblob-no-check-container\n" @@ -116,25 +138,33 @@ async def test_get_obscured_passphrase_raises_on_nonzero_exit(): @pytest.mark.asyncio async def test_write_marker_blob_offloads_upload_to_thread(): """The blocking Azure upload must run in a worker thread, never directly on - the event loop.""" + the event loop, and the timestamp payload + overwrite flag must be forwarded.""" blob_client = MagicMock() + ctor = MagicMock(return_value=blob_client) to_thread = AsyncMock() with ( - patch.object(backup.BlobClient, "from_blob_url", return_value=blob_client), + patch.object(backup.BlobClient, "from_blob_url", ctor), patch.object(backup.asyncio, "to_thread", to_thread), ): await _write_marker_blob( "container", "https://acct.blob.core.windows.net/c?sv=x&sig=y" ) + # the blob path is inserted before the query string, replacing the container + ctor.assert_called_once_with( + "https://acct.blob.core.windows.net/container/_last_backup?sv=x&sig=y" + ) # upload happened via to_thread, not a direct (loop-blocking) call blob_client.upload_blob.assert_not_called() to_thread.assert_awaited_once() - assert to_thread.await_args.args[0] == blob_client.upload_blob + func, *rest = to_thread.await_args.args + assert func == blob_client.upload_blob + assert rest and isinstance(rest[0], str) # ISO timestamp payload + assert to_thread.await_args.kwargs == {"overwrite": True} @pytest.mark.asyncio -async def test_write_marker_blob_swallows_errors(): +async def test_write_marker_blob_swallows_setup_errors(): with patch.object( backup.BlobClient, "from_blob_url", side_effect=RuntimeError("nope") ): @@ -143,9 +173,46 @@ async def test_write_marker_blob_swallows_errors(): @pytest.mark.asyncio -async def test_start_backup_keeps_strong_reference_to_task(): - """The backup task must be held by the module-level set so it cannot be - garbage-collected mid-run.""" +async def test_write_marker_blob_swallows_upload_errors(): + with ( + patch.object(backup.BlobClient, "from_blob_url", return_value=MagicMock()), + patch.object( + backup.asyncio, + "to_thread", + AsyncMock(side_effect=RuntimeError("upload failed")), + ), + ): + # an error raised inside the worker thread is also best-effort + await _write_marker_blob("container", "https://acct/c?sig=y") + + +async def _drain_background_tasks(): + # Chained done-callbacks (backup task -> spawned on_task_done task -> discard) + # need several loop turns to settle; yield until the set drains. + for _ in range(1000): + await asyncio.sleep(0) + if not backup.background_tasks: + return + raise AssertionError("background tasks did not drain") + + +async def _start_backup_and_capture_signal(fake_backup): + sas = SimpleNamespace(container_name="c", sas_url="s") + signal = MagicMock() + with ( + patch.object(backup, "get_backup_sas_url", AsyncMock(return_value=sas)), + patch.object(backup, "backup_directories", fake_backup), + patch.object(backup.signals, "on_backup_update", signal), + ): + await start_backup() + await _drain_background_tasks() + return signal + + +@pytest.mark.asyncio +async def test_start_backup_registers_strong_refs_for_both_tasks(): + """Both the backup task and the spawned on_task_done wrapper must be held by + the module-level set (asyncio holds tasks only weakly), then drain to empty.""" release = asyncio.Event() started = asyncio.Event() @@ -154,20 +221,48 @@ async def fake_backup(*_args, **_kwargs): await release.wait() sas = SimpleNamespace(container_name="c", sas_url="s") + register_spy = MagicMock(wraps=backup._register_background_task) with ( patch.object(backup, "get_backup_sas_url", AsyncMock(return_value=sas)), patch.object(backup, "backup_directories", fake_backup), + patch.object(backup, "_register_background_task", register_spy), patch.object(backup.signals, "on_backup_update"), ): assert backup.background_tasks == set() await start_backup() await asyncio.wait_for(started.wait(), timeout=5) + # the still-running backup task is held strongly assert len(backup.background_tasks) == 1 + assert register_spy.call_count == 1 release.set() - for _ in range(100): + for _ in range(1000): await asyncio.sleep(0) - if not backup.background_tasks: + if register_spy.call_count >= 2 and not backup.background_tasks: break + + # the on_task_done wrapper was also registered (not left unreferenced) + assert register_spy.call_count == 2 assert backup.background_tasks == set() + + +@pytest.mark.asyncio +async def test_start_backup_signals_success_on_completion(): + async def fake_backup(*_args, **_kwargs): + return None + + signal = await _start_backup_and_capture_signal(fake_backup) + signal.send.assert_called_once_with() + + +@pytest.mark.asyncio +async def test_start_backup_signals_failure_with_exception(): + boom = RuntimeError("backup blew up") + + async def fake_backup(*_args, **_kwargs): + raise boom + + signal = await _start_backup_and_capture_signal(fake_backup) + signal.send.assert_called_once() + assert signal.send.call_args.args[0] is boom