From 63da21f07dbbb9b9ac78a7a6bfbaffac677eec6c Mon Sep 17 00:00:00 2001 From: Cabbagec Date: Thu, 23 Jul 2026 10:11:18 +0000 Subject: [PATCH] test: expand phase 8 fault matrix --- e2e/README.md | 16 + e2e/scenarios/complex_jobs.py | 582 ++++++++++++++++++++++++++++++++++ e2e/scripts/complex-matrix.sh | 382 ++++++++++++++++++++++ e2e/scripts/up.sh | 3 + src/archive_clients/daemon.py | 15 + tests/test_daemon.py | 10 + 6 files changed, 1008 insertions(+) create mode 100644 e2e/scenarios/complex_jobs.py create mode 100755 e2e/scripts/complex-matrix.sh diff --git a/e2e/README.md b/e2e/README.md index ac03dc9..7dda515 100644 --- a/e2e/README.md +++ b/e2e/README.md @@ -34,6 +34,22 @@ jobs, waits for the durable five/three/five-step flows, and verifies target qBittorrent selections, archive retention, safe cache removal, and target file digests. +Set `E2E_RUN_COMPLEX=1` to run the Phase 8 selective and recovery matrix. It +uses a four-file torrent with complementary cache selections to verify +selective archive placement, multi-job archive/cache merging, union coverage +across two archive nodes, uncovered/no-op/stale-preview rejection, and +retention of shared and unknown files during eviction. It then pauses the +test scheduler at durable boundaries to verify record-only queued +cancellation, rollback after target materialization, control restart after +source staging, and command replay after a target client disconnect. The final +hostile cases change the source selection and make an absent target appear +after confirmation; each must fail before commit with its exact precondition +reason. qBittorrent is also checked for zero downloaded bytes at the replayed +target. + +The scheduler pause/resume endpoints exist only on the loopback test adapter; +normal control and Telegram orchestration always use the automatic scheduler. + The Syncthing 2.1.2 and LinuxServer qBittorrent multi-platform image indexes are digest-pinned. Runtime secrets are generated with mode 0600 and ignored by Git. The qBittorrent test config limits its authentication bypass to the diff --git a/e2e/scenarios/complex_jobs.py b/e2e/scenarios/complex_jobs.py new file mode 100644 index 0000000..f848950 --- /dev/null +++ b/e2e/scenarios/complex_jobs.py @@ -0,0 +1,582 @@ +#!/usr/bin/env python3 +"""Drive selective, negative, cancellation, and recovery E2E jobs.""" + +from __future__ import annotations + +import argparse +import json +import sys +import time +import urllib.error +import urllib.parse +import urllib.request + + +CONTROL = "http://127.0.0.1:18081/test/v1" + + +class HttpError(RuntimeError): + def __init__(self, status: int, body: object): + super().__init__(f"HTTP {status}: {body}") + self.status = status + self.body = body + + +def get_json(url: str): + try: + with urllib.request.urlopen(url, timeout=35) as response: + return json.load(response) + except urllib.error.HTTPError as exc: + raise HttpError(exc.code, _error_body(exc)) from exc + + +def post_json(url: str, value: object): + request = urllib.request.Request( + url, + data=json.dumps(value, separators=(",", ":")).encode(), + method="POST", + headers={"Content-Type": "application/json"}, + ) + try: + with urllib.request.urlopen(request, timeout=35) as response: + return json.load(response) + except urllib.error.HTTPError as exc: + raise HttpError(exc.code, _error_body(exc)) from exc + + +def _error_body(error: urllib.error.HTTPError): + try: + return json.load(error) + except (ValueError, OSError): + return error.read().decode("utf-8", "replace") + + +def indices(value: str) -> list[int]: + if not value: + return [] + result = [int(item) for item in value.split(",")] + if result != sorted(set(result)) or any(item < 0 for item in result): + raise ValueError("indices must be unique, sorted, non-negative integers") + return result + + +def selection_indices(value: dict[str, object]) -> list[int]: + result = [] + for item in value.get("ranges", []): + first = int(item.get("first", 0)) + last = int(item.get("last", 0)) + result.extend(range(first, last + 1)) + return result + + +def resource_id(info_hash: str) -> dict[str, str]: + key = "info_hash_v1_hex" if len(info_hash) == 40 else "info_hash_v2_hex" + return {key: info_hash} + + +def qb_json(endpoint: str, path: str, **parameters: str): + query = urllib.parse.urlencode(parameters) + suffix = f"?{query}" if query else "" + return get_json(f"{endpoint}{path}{suffix}") + + +def transfer_preview(args: argparse.Namespace): + request = { + "operation": args.operation, + "source_client_id": args.source_client, + "target_client_id": args.target_client, + "resource_id": resource_id(args.info_hash), + } + selected = indices(args.selected) + if selected: + request["selected_file_indices"] = selected + return post_json(f"{CONTROL}/jobs/preview", request) + + +def create_previewed(preview: dict[str, object]) -> str: + definition = preview["definition"] + job_id = definition["job_id"] + created = post_json(f"{CONTROL}/jobs", { + "preview_revision": preview["preview_revision"], + "definition": definition, + }) + if created["job_id"] != job_id: + raise RuntimeError("control returned the wrong job") + return job_id + + +def wait_job(job_id: str, expected: str, timeout: float = 180): + deadline = time.monotonic() + timeout + last = None + while time.monotonic() < deadline: + last = get_json(f"{CONTROL}/jobs/{job_id}") + state = last["state"] + if state == expected: + return last + if state in { + "JOB_STATE_FAILED", + "JOB_STATE_CANCELLED", + "JOB_STATE_SUCCEEDED", + }: + raise RuntimeError( + f"job reached {state} while waiting for {expected}: {last}" + ) + time.sleep(0.25) + raise RuntimeError( + f"timed out waiting for {job_id} to reach {expected}: {last}" + ) + + +def wait_latest( + job_id: str, + *, + event_type: str, + step: str | None = None, + minimum_sequence: int = 0, + timeout: float = 90, +): + deadline = time.monotonic() + timeout + last = None + while time.monotonic() < deadline: + last = get_json(f"{CONTROL}/jobs/{job_id}") + if last["state"] == "JOB_STATE_FAILED": + raise RuntimeError(f"job failed while waiting for event: {last}") + event = last.get("latest_event") or {} + progress = event.get("progress") or {} + if ( + event.get("type") == event_type + and int(event.get("sequence", 0)) >= minimum_sequence + and (step is None or progress.get("step") == step) + ): + return last + time.sleep(0.1) + raise RuntimeError(f"timed out waiting for job event: {last}") + + +def advance_for(job_id: str, action: str): + outcomes = post_json(f"{CONTROL}/scheduler/advance", {}) + matching = [ + item for item in outcomes + if item.get("job_id") == job_id and item.get("action") == action + ] + if len(matching) != 1: + raise RuntimeError( + f"expected scheduler action {action} for {job_id}: {outcomes}" + ) + return matching[0] + + +def assert_transfer_preview( + preview: dict[str, object], expected: list[int], delta: list[int] +) -> None: + transfer = preview["definition"]["transfer"] + actual_requested = selection_indices(transfer["requested_files"]) + actual_delta = selection_indices(transfer["transfer_delta_files"]) + if actual_requested != expected or actual_delta != delta: + raise RuntimeError( + "unexpected transfer selection: " + f"requested={actual_requested} delta={actual_delta}" + ) + + +def command_transfer(args: argparse.Namespace) -> None: + preview = transfer_preview(args) + expected = indices(args.expected_selection or args.selected) + delta = indices(args.expected_delta or args.expected_selection or args.selected) + if expected: + assert_transfer_preview(preview, expected, delta) + job_id = create_previewed(preview) + job = wait_job(job_id, "JOB_STATE_SUCCEEDED") + if not job["committed"]: + raise RuntimeError("successful transfer was not committed") + print(json.dumps({ + "job_id": job_id, + "operation": args.operation, + "selection": expected, + "delta": delta, + "state": job["state"], + }, sort_keys=True)) + + +def command_preview_error(args: argparse.Namespace) -> None: + try: + if args.operation == "evict_cache": + post_json(f"{CONTROL}/jobs/preview", { + "operation": args.operation, + "cache_client_id": args.cache_client, + "resource_id": resource_id(args.info_hash), + }) + else: + transfer_preview(args) + except HttpError as exc: + rendered = json.dumps(exc.body, sort_keys=True) + if exc.status != args.status or args.contains not in rendered: + raise RuntimeError( + f"unexpected preview failure: status={exc.status} body={rendered}" + ) from exc + print(json.dumps({ + "expected_status": exc.status, + "matched": args.contains, + }, sort_keys=True)) + return + raise RuntimeError("preview unexpectedly succeeded") + + +def command_stale_preview(args: argparse.Namespace) -> None: + preview = transfer_preview(args) + definition = preview["definition"] + definition["resource_display_name"] += "-tampered" + try: + post_json(f"{CONTROL}/jobs", { + "preview_revision": preview["preview_revision"], + "definition": definition, + }) + except HttpError as exc: + if exc.status != 409: + raise + print(json.dumps({"stale_preview_rejected": True}, sort_keys=True)) + return + raise RuntimeError("tampered preview revision was accepted") + + +def command_evict(args: argparse.Namespace) -> None: + preview = post_json(f"{CONTROL}/jobs/preview", { + "operation": "evict_cache", + "cache_client_id": args.cache_client, + "resource_id": resource_id(args.info_hash), + }) + coverage = preview["definition"]["eviction"]["archive_coverage"] + if len(coverage) < args.minimum_coverage_proofs: + raise RuntimeError( + f"expected at least {args.minimum_coverage_proofs} coverage proofs" + ) + job_id = create_previewed(preview) + job = wait_job(job_id, "JOB_STATE_SUCCEEDED") + if not job["committed"]: + raise RuntimeError("successful eviction was not committed") + print(json.dumps({ + "job_id": job_id, + "coverage_proofs": len(coverage), + "state": job["state"], + }, sort_keys=True)) + + +_STEPS = [ + ( + "source_stage", + "source_stage", + "JOB_STEP_KIND_SOURCE_STAGE", + ), + ( + "syncthing_transfer", + "syncthing_transfer", + "JOB_STEP_KIND_SYNCTHING_TRANSFER", + ), + ( + "target_materialize", + "target_materialize", + "JOB_STEP_KIND_TARGET_MATERIALIZE", + ), +] + + +def command_drive(args: argparse.Namespace) -> None: + post_json(f"{CONTROL}/scheduler/pause", {}) + preview = transfer_preview(args) + expected = indices(args.expected_selection or args.selected) + delta = indices(args.expected_delta or args.expected_selection or args.selected) + if expected: + assert_transfer_preview(preview, expected, delta) + job_id = create_previewed(preview) + + advance_for(job_id, "assign_source") + wait_latest( + job_id, event_type="JOB_EVENT_TYPE_ASSIGNED", minimum_sequence=1 + ) + advance_for(job_id, "assign_target") + wait_latest( + job_id, event_type="JOB_EVENT_TYPE_ASSIGNED", minimum_sequence=2 + ) + if args.through == "assigned": + print(job_id) + return + for boundary, action, step in _STEPS: + advance_for(job_id, action) + wait_latest( + job_id, + event_type="JOB_EVENT_TYPE_STEP_SUCCEEDED", + step=step, + ) + if args.through == boundary: + print(job_id) + return + raise RuntimeError(f"unsupported drive boundary: {args.through}") + + +def command_create_paused(args: argparse.Namespace) -> None: + post_json(f"{CONTROL}/scheduler/pause", {}) + preview = transfer_preview(args) + expected = indices(args.expected_selection or args.selected) + delta = indices(args.expected_delta or args.expected_selection or args.selected) + if expected: + assert_transfer_preview(preview, expected, delta) + print(create_previewed(preview)) + + +def command_queued_cancel(args: argparse.Namespace) -> None: + post_json(f"{CONTROL}/scheduler/pause", {}) + job_id = create_previewed(transfer_preview(args)) + cancelled = post_json(f"{CONTROL}/jobs/{job_id}/cancel", {}) + if cancelled.get("disposition") != "removed": + raise RuntimeError(f"queued cancellation was not record-only: {cancelled}") + try: + get_json(f"{CONTROL}/jobs/{job_id}") + except HttpError as exc: + if exc.status == 404: + print(json.dumps({ + "job_id": job_id, + "record_removed": True, + }, sort_keys=True)) + return + raise + raise RuntimeError("queued cancelled job record still exists") + + +def command_cancel(args: argparse.Namespace) -> None: + post_json(f"{CONTROL}/jobs/{args.job_id}/cancel", {}) + deadline = time.monotonic() + 90 + last = None + while time.monotonic() < deadline: + last = get_json(f"{CONTROL}/jobs/{args.job_id}") + if last["state"] == "JOB_STATE_CANCELLED": + if last["committed"]: + raise RuntimeError("cancelled precommit job became committed") + print(json.dumps({ + "job_id": args.job_id, + "state": last["state"], + "committed": last["committed"], + }, sort_keys=True)) + return + if last["state"] == "JOB_STATE_FAILED": + raise RuntimeError(f"cancellation failed: {last}") + post_json(f"{CONTROL}/scheduler/advance", {}) + time.sleep(0.2) + raise RuntimeError(f"timed out waiting for cancellation: {last}") + + +def command_advance(args: argparse.Namespace) -> None: + outcome = advance_for(args.job_id, args.action) + print(json.dumps(outcome, sort_keys=True)) + + +def command_resume(args: argparse.Namespace) -> None: + post_json(f"{CONTROL}/scheduler/resume", {}) + job = wait_job(args.job_id, "JOB_STATE_SUCCEEDED") + if not job["committed"]: + raise RuntimeError("resumed job was not committed") + print(json.dumps({ + "job_id": args.job_id, + "state": job["state"], + "committed": job["committed"], + }, sort_keys=True)) + + +def command_expect_failure(args: argparse.Namespace) -> None: + post_json(f"{CONTROL}/scheduler/resume", {}) + deadline = time.monotonic() + 120 + last = None + while time.monotonic() < deadline: + last = get_json(f"{CONTROL}/jobs/{args.job_id}") + if last["state"] == "JOB_STATE_FAILED": + message = ( + (last.get("latest_event") or {}) + .get("error", {}) + .get("message", "") + ) + if args.contains not in message: + raise RuntimeError( + f"failure reason did not contain {args.contains!r}: {last}" + ) + if last["committed"]: + raise RuntimeError("hostile precondition failure committed") + print(json.dumps({ + "job_id": args.job_id, + "state": last["state"], + "matched": args.contains, + }, sort_keys=True)) + return + if last["state"] in { + "JOB_STATE_SUCCEEDED", + "JOB_STATE_CANCELLED", + }: + raise RuntimeError(f"hostile job reached wrong terminal state: {last}") + time.sleep(0.25) + raise RuntimeError(f"timed out waiting for hostile failure: {last}") + + +def command_assert_qb(args: argparse.Namespace) -> None: + expected = indices(args.selected) + deadline = time.monotonic() + args.timeout + last = None + while time.monotonic() < deadline: + records = qb_json( + args.endpoint, "/torrents/info", hashes=args.info_hash + ) + if args.absent: + if records == []: + print(json.dumps({"absent": True}, sort_keys=True)) + return + elif len(records) == 1: + runtime_state = str(records[0].get("state", "")) + files = qb_json( + args.endpoint, "/torrents/files", hash=args.info_hash + ) + selected = [ + int(item["index"]) for item in files + if int(item.get("priority", 0)) > 0 + ] + selected_complete = [ + int(item["index"]) for item in files + if ( + int(item.get("priority", 0)) > 0 + and float(item.get("progress", 0)) >= 1 + ) + ] + last = { + "runtime_state": runtime_state, + "selected": selected, + "selected_complete": selected_complete, + } + no_downloaded = True + if args.no_downloaded: + properties = qb_json( + args.endpoint, + "/torrents/properties", + hash=args.info_hash, + ) + downloaded = int(properties.get("total_downloaded", -1)) + last["total_downloaded"] = downloaded + no_downloaded = downloaded == 0 + if ( + selected == expected + and selected_complete == expected + and runtime_state + not in {"checkingUP", "checkingDL", "checkingResumeData"} + and no_downloaded + ): + print(json.dumps(last, sort_keys=True)) + return + else: + last = records + time.sleep(0.25) + raise RuntimeError(f"qBittorrent assertion timed out: {last}") + + +def add_transfer_arguments(parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "--operation", choices=("archive", "unarchive"), required=True + ) + parser.add_argument("--source-client", required=True) + parser.add_argument("--target-client", required=True) + parser.add_argument("--info-hash", required=True) + parser.add_argument("--selected", default="") + parser.add_argument("--expected-selection", default="") + parser.add_argument("--expected-delta", default="") + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser() + commands = parser.add_subparsers(dest="command", required=True) + + transfer = commands.add_parser("transfer") + add_transfer_arguments(transfer) + transfer.set_defaults(run=command_transfer) + + preview_error = commands.add_parser("preview-error") + preview_error.add_argument( + "--operation", + choices=("archive", "unarchive", "evict_cache"), + required=True, + ) + preview_error.add_argument("--source-client", default="") + preview_error.add_argument("--target-client", default="") + preview_error.add_argument("--cache-client", default="") + preview_error.add_argument("--info-hash", required=True) + preview_error.add_argument("--selected", default="") + preview_error.add_argument("--status", type=int, default=409) + preview_error.add_argument("--contains", required=True) + preview_error.set_defaults(run=command_preview_error) + + stale = commands.add_parser("stale-preview") + add_transfer_arguments(stale) + stale.set_defaults(run=command_stale_preview) + + evict = commands.add_parser("evict") + evict.add_argument("--cache-client", required=True) + evict.add_argument("--info-hash", required=True) + evict.add_argument("--minimum-coverage-proofs", type=int, default=1) + evict.set_defaults(run=command_evict) + + drive = commands.add_parser("drive") + add_transfer_arguments(drive) + drive.add_argument( + "--through", + choices=( + "assigned", + "source_stage", + "syncthing_transfer", + "target_materialize", + ), + required=True, + ) + drive.set_defaults(run=command_drive) + + create_paused = commands.add_parser("create-paused") + add_transfer_arguments(create_paused) + create_paused.set_defaults(run=command_create_paused) + + queued_cancel = commands.add_parser("queued-cancel") + add_transfer_arguments(queued_cancel) + queued_cancel.set_defaults(run=command_queued_cancel) + + cancel = commands.add_parser("cancel") + cancel.add_argument("--job-id", required=True) + cancel.set_defaults(run=command_cancel) + + advance = commands.add_parser("advance") + advance.add_argument("--job-id", required=True) + advance.add_argument("--action", required=True) + advance.set_defaults(run=command_advance) + + resume = commands.add_parser("resume") + resume.add_argument("--job-id", required=True) + resume.set_defaults(run=command_resume) + + failure = commands.add_parser("expect-failure") + failure.add_argument("--job-id", required=True) + failure.add_argument("--contains", required=True) + failure.set_defaults(run=command_expect_failure) + + qb = commands.add_parser("assert-qb") + qb.add_argument("--endpoint", required=True) + qb.add_argument("--info-hash", required=True) + qb.add_argument("--selected", default="") + qb.add_argument("--absent", action="store_true") + qb.add_argument("--no-downloaded", action="store_true") + qb.add_argument("--timeout", type=float, default=60) + qb.set_defaults(run=command_assert_qb) + return parser + + +def main() -> int: + args = build_parser().parse_args() + args.run(args) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: + print(f"complex E2E scenario failed: {exc}", file=sys.stderr) + raise diff --git a/e2e/scripts/complex-matrix.sh b/e2e/scripts/complex-matrix.sh new file mode 100755 index 0000000..7d99861 --- /dev/null +++ b/e2e/scripts/complex-matrix.sh @@ -0,0 +1,382 @@ +#!/usr/bin/env bash +set -euo pipefail + +source "$(dirname "$0")/lib.sh" + +bundle_hash=547cec172aab2f4d418d73d4f5fc0d2794e971b7 +bundle_torrent=ZDQ6aW5mb2Q1OmZpbGVzbGQ2Omxlbmd0aGkxNjM4NGU0OnBhdGhsOTphbHBoYS5iaW5lZWQ2Omxlbmd0aGkxNjM4NGU0OnBhdGhsNjpuZXN0ZWQ5OmJyYXZvLmJpbmVlZDY6bGVuZ3RoaTE2Mzg0ZTQ6cGF0aGw2Om5lc3RlZDExOmNoYXJsaWUuYmluZWVkNjpsZW5ndGhpMTYzODRlNDpwYXRobDk6ZGVsdGEuYmluZWVlNDpuYW1lNjpidW5kbGUxMjpwaWVjZSBsZW5ndGhpMTYzODRlNjpwaWVjZXM4MDp9tnus7uyzpCC/N6e+ykpFGF+PPHmZb79ZCq2dp84v4EgIIzMZE7bwEMQCvl+7qybWEDECinYwiKcGKn82sWFjoNungr001vTmtGeLhKBGd2Vl +legacy_bundle_hash=305af159425d76140eea48d20cfcfbb3f8dccf46 +shared_hash=03fbcf663bd3fe80c298ca54a55780a6b1d4b43b +shared_torrent=ZDQ6aW5mb2Q1OmZpbGVzbGQ2Omxlbmd0aGkxNjM4NGU0OnBhdGhsNjpuZXN0ZWQ5OmJyYXZvLmJpbmVlZTQ6bmFtZTY6YnVuZGxlMTI6cGllY2UgbGVuZ3RoaTE2Mzg0ZTY6cGllY2VzMjA6eZlvv1kKrZ2nzi/gSAgjMxkTtvBlZQ== +cancel_hash=330be0cb7c2201135a2de63b28e77993745ff688 +cancel_torrent=ZDQ6aW5mb2Q2Omxlbmd0aGkzMWU0Om5hbWUxMTpmaXh0dXJlLmJpbjEyOnBpZWNlIGxlbmd0aGkxNjM4NGU2OnBpZWNlczIwOlFlB05MtIUU4oem2MNz5LJW/GI6ZWU= + +control_scenario() { + compose_control exec -T control \ + python /e2e/scenarios/complex_jobs.py "$@" +} + +qb_endpoint() { + case "$1" in + cache-1) printf '%s' http://qb-cache-1:8080/api/v2 ;; + cache-2) printf '%s' http://qb-cache-2:8080/api/v2 ;; + archive-1) printf '%s' http://qb-archive-1:8080/api/v2 ;; + archive-2) printf '%s' http://qb-archive-2:8080/api/v2 ;; + *) printf 'unknown E2E node: %s\n' "$1" >&2; return 2 ;; + esac +} + +wait_qb_present() { + local node=$1 + local info_hash=$2 + local deadline=$((SECONDS + 30)) + while (( SECONDS < deadline )); do + local records + records=$(compose_node "$node" exec -T qbittorrent curl -fsS \ + "http://127.0.0.1:8080/api/v2/torrents/info?hashes=$info_hash") + if [[ "$records" != "[]" ]]; then + return + fi + sleep 1 + done + printf 'timed out waiting for torrent %s on %s\n' \ + "$info_hash" "$node" >&2 + return 1 +} + +delete_torrent() { + local node=$1 + local info_hash=$2 + compose_node "$node" exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/delete \ + --data-urlencode "hashes=$info_hash" \ + --data-urlencode "deleteFiles=true" >/dev/null || true + control_scenario assert-qb \ + --endpoint "$(qb_endpoint "$node")" \ + --info-hash "$info_hash" \ + --absent >/dev/null +} + +install_torrent() { + local node=$1 + local info_hash=$2 + local encoded=$3 + local save_path=$4 + local selected=$5 + local all_indices=$6 + compose_node "$node" exec -T qbittorrent /bin/sh -c \ + "printf '%s' '$encoded' | base64 -d > /tmp/e2e.torrent" + compose_node "$node" exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/add \ + -F torrents=@/tmp/e2e.torrent \ + -F "savepath=$save_path" \ + -F stopped=true >/dev/null + wait_qb_present "$node" "$info_hash" + compose_node "$node" exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/filePrio \ + --data-urlencode "hash=$info_hash" \ + --data-urlencode "id=$all_indices" \ + --data-urlencode "priority=0" >/dev/null + compose_node "$node" exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/filePrio \ + --data-urlencode "hash=$info_hash" \ + --data-urlencode "id=$selected" \ + --data-urlencode "priority=1" >/dev/null + compose_node "$node" exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/recheck \ + --data-urlencode "hashes=$info_hash" >/dev/null +} + +assert_qb() { + local node=$1 + local info_hash=$2 + local selected=$3 + control_scenario assert-qb \ + --endpoint "$(qb_endpoint "$node")" \ + --info-hash "$info_hash" \ + --selected "$selected" >/dev/null +} + +for node in cache-1 cache-2 archive-1 archive-2; do + delete_torrent "$node" "$legacy_bundle_hash" + delete_torrent "$node" "$bundle_hash" + delete_torrent "$node" "$shared_hash" + delete_torrent "$node" "$cancel_hash" + compose_node "$node" exec -T qbittorrent /bin/sh -c \ + 'rm -f /downloads/bundle/alpha.bin \ + /downloads/bundle/nested/bravo.bin \ + /downloads/bundle/nested/charlie.bin \ + /downloads/bundle/delta.bin \ + /downloads/bundle/unknown.keep \ + /downloads/fixture.bin; \ + rmdir /downloads/bundle/nested /downloads/bundle 2>/dev/null || true' + compose_node "$node" exec -T qbittorrent /bin/sh -c \ + 'chown -R 1001:1001 /downloads/bundle 2>/dev/null || true' +done + +compose_node cache-1 exec -T --user 1001:1001 qbittorrent /bin/sh -c \ + 'mkdir -p /downloads/bundle/nested; \ + dd if=/dev/zero bs=16384 count=1 2>/dev/null | tr "\000" A \ + > /downloads/bundle/alpha.bin; \ + dd if=/dev/zero bs=16384 count=1 2>/dev/null | tr "\000" C \ + > /downloads/bundle/nested/charlie.bin' +install_torrent cache-1 "$bundle_hash" "$bundle_torrent" \ + /downloads '0|2' '0|1|2|3' +assert_qb cache-1 "$bundle_hash" '0,2' + +compose_node cache-2 exec -T --user 1001:1001 qbittorrent /bin/sh -c \ + 'mkdir -p /downloads/bundle/nested; \ + dd if=/dev/zero bs=16384 count=1 2>/dev/null | tr "\000" B \ + > /downloads/bundle/nested/bravo.bin; \ + dd if=/dev/zero bs=16384 count=1 2>/dev/null | tr "\000" D \ + > /downloads/bundle/delta.bin' +install_torrent cache-2 "$bundle_hash" "$bundle_torrent" \ + /downloads '1|3' '0|1|2|3' +assert_qb cache-2 "$bundle_hash" '1,3' + +control_scenario stale-preview \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-1 \ + --info-hash "$bundle_hash" \ + --selected '0' >/dev/null + +control_scenario transfer \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-1 \ + --info-hash "$bundle_hash" \ + --selected '0,2' \ + --expected-selection '0,2' \ + --expected-delta '0,2' +assert_qb archive-1 "$bundle_hash" '0,2' + +control_scenario preview-error \ + --operation evict_cache \ + --cache-client cache-2 \ + --info-hash "$bundle_hash" \ + --status 400 \ + --contains 'missing file indices' + +control_scenario transfer \ + --operation archive \ + --source-client cache-2 \ + --target-client archive-2 \ + --info-hash "$bundle_hash" \ + --selected '1,3' \ + --expected-selection '1,3' \ + --expected-delta '1,3' +assert_qb archive-2 "$bundle_hash" '1,3' + +compose_node cache-2 exec -T --user 1001:1001 qbittorrent /bin/sh -c \ + 'dd if=/dev/zero bs=16384 count=1 2>/dev/null | tr "\000" A \ + > /downloads/bundle/alpha.bin; \ + dd if=/dev/zero bs=16384 count=1 2>/dev/null | tr "\000" C \ + > /downloads/bundle/nested/charlie.bin' +compose_node cache-2 exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/filePrio \ + --data-urlencode "hash=$bundle_hash" \ + --data-urlencode 'id=0|1|2|3' \ + --data-urlencode priority=1 >/dev/null +compose_node cache-2 exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/recheck \ + --data-urlencode "hashes=$bundle_hash" >/dev/null +assert_qb cache-2 "$bundle_hash" '0,1,2,3' + +control_scenario evict \ + --cache-client cache-2 \ + --info-hash "$bundle_hash" \ + --minimum-coverage-proofs 2 +control_scenario assert-qb \ + --endpoint "$(qb_endpoint cache-2)" \ + --info-hash "$bundle_hash" \ + --absent >/dev/null + +control_scenario transfer \ + --operation unarchive \ + --source-client archive-1 \ + --target-client cache-2 \ + --info-hash "$bundle_hash" \ + --selected '0,2' \ + --expected-selection '0,2' \ + --expected-delta '0,2' +assert_qb cache-2 "$bundle_hash" '0,2' + +control_scenario transfer \ + --operation unarchive \ + --source-client archive-2 \ + --target-client cache-2 \ + --info-hash "$bundle_hash" \ + --selected '1,3' \ + --expected-selection '1,3' \ + --expected-delta '1,3' +assert_qb cache-2 "$bundle_hash" '0,1,2,3' + +control_scenario transfer \ + --operation archive \ + --source-client cache-2 \ + --target-client archive-1 \ + --info-hash "$bundle_hash" \ + --selected '1,3' \ + --expected-selection '1,3' \ + --expected-delta '1,3' +assert_qb archive-1 "$bundle_hash" '0,1,2,3' + +control_scenario transfer \ + --operation archive \ + --source-client cache-2 \ + --target-client archive-2 \ + --info-hash "$bundle_hash" \ + --selected '0,2' \ + --expected-selection '0,2' \ + --expected-delta '0,2' +assert_qb archive-2 "$bundle_hash" '0,1,2,3' + +control_scenario preview-error \ + --operation archive \ + --source-client cache-2 \ + --target-client archive-1 \ + --info-hash "$bundle_hash" \ + --selected '0' \ + --status 400 \ + --contains 'already contains' + +control_scenario evict \ + --cache-client cache-1 \ + --info-hash "$bundle_hash" +control_scenario transfer \ + --operation unarchive \ + --source-client archive-1 \ + --target-client cache-1 \ + --info-hash "$bundle_hash" \ + --selected '1' \ + --expected-selection '1' \ + --expected-delta '1' +assert_qb cache-1 "$bundle_hash" '1' + +compose_node cache-1 exec -T --user 1001:1001 qbittorrent /bin/sh -c \ + 'printf "preserve-unknown\n" > /downloads/bundle/unknown.keep' +install_torrent cache-1 "$shared_hash" "$shared_torrent" \ + /downloads '0' '0' +assert_qb cache-1 "$shared_hash" '0' +control_scenario evict \ + --cache-client cache-1 \ + --info-hash "$bundle_hash" +control_scenario assert-qb \ + --endpoint "$(qb_endpoint cache-1)" \ + --info-hash "$bundle_hash" \ + --absent >/dev/null +assert_qb cache-1 "$shared_hash" '0' +compose_node cache-1 exec -T qbittorrent test -f \ + /downloads/bundle/nested/bravo.bin +compose_node cache-1 exec -T qbittorrent test -f \ + /downloads/bundle/unknown.keep + +compose_node cache-1 exec -T --user 1001:1001 qbittorrent /bin/sh -c \ + 'printf "archive-control-e2e-happy-path\n" > /downloads/fixture.bin' +install_torrent cache-1 "$cancel_hash" "$cancel_torrent" \ + /downloads '0' '0' +assert_qb cache-1 "$cancel_hash" '0' + +control_scenario queued-cancel \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-1 \ + --info-hash "$cancel_hash" +compose_node cache-1 exec -T qbittorrent test -f /downloads/fixture.bin + +cancel_job=$(control_scenario drive \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-2 \ + --info-hash "$cancel_hash" \ + --through target_materialize) +control_scenario cancel --job-id "$cancel_job" +control_scenario assert-qb \ + --endpoint "$(qb_endpoint archive-2)" \ + --info-hash "$cancel_hash" \ + --absent >/dev/null +if compose_node archive-2 exec -T qbittorrent \ + test -e /downloads/fixture.bin; then + printf 'precommit cancellation left a materialized target file\n' >&2 + exit 1 +fi +assert_qb cache-1 "$cancel_hash" '0' + +restart_job=$(control_scenario drive \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-1 \ + --info-hash "$cancel_hash" \ + --through source_stage) +compose_control restart control +sleep 2 +control_scenario resume --job-id "$restart_job" +assert_qb archive-1 "$cancel_hash" '0' + +delete_torrent archive-2 "$cancel_hash" +replay_job=$(control_scenario drive \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-2 \ + --info-hash "$cancel_hash" \ + --through source_stage) +compose_node archive-2 stop client +sleep 1 +control_scenario advance \ + --job-id "$replay_job" \ + --action syncthing_transfer >/dev/null +compose_node archive-2 start client +control_scenario resume --job-id "$replay_job" +control_scenario assert-qb \ + --endpoint "$(qb_endpoint archive-2)" \ + --info-hash "$cancel_hash" \ + --selected '0' \ + --no-downloaded >/dev/null + +delete_torrent archive-2 "$cancel_hash" +stale_source_job=$(control_scenario create-paused \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-2 \ + --info-hash "$cancel_hash") +compose_node cache-1 exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/filePrio \ + --data-urlencode "hash=$cancel_hash" \ + --data-urlencode id=0 \ + --data-urlencode priority=0 >/dev/null +control_scenario expect-failure \ + --job-id "$stale_source_job" \ + --contains 'source resource changed after job confirmation' +compose_node cache-1 exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/filePrio \ + --data-urlencode "hash=$cancel_hash" \ + --data-urlencode id=0 \ + --data-urlencode priority=1 >/dev/null +compose_node cache-1 exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/recheck \ + --data-urlencode "hashes=$cancel_hash" >/dev/null +assert_qb cache-1 "$cancel_hash" '0' + +target_appeared_job=$(control_scenario create-paused \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-2 \ + --info-hash "$cancel_hash") +compose_node archive-2 exec -T qbittorrent /bin/sh -c \ + "printf '%s' '$cancel_torrent' | base64 -d > /tmp/appeared.torrent" +compose_node archive-2 exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/add \ + -F torrents=@/tmp/appeared.torrent \ + -F savepath=/downloads \ + -F stopped=true >/dev/null +wait_qb_present archive-2 "$cancel_hash" +control_scenario expect-failure \ + --job-id "$target_appeared_job" \ + --contains 'target resource appeared after job confirmation' +assert_qb cache-1 "$cancel_hash" '0' +delete_torrent archive-2 "$cancel_hash" + +printf '%s\n' \ + 'complex matrix passed: selective merges, union eviction, hostile retention,' \ + 'queued/partial cancellation, restart/replay, and hostile state changes' diff --git a/e2e/scripts/up.sh b/e2e/scripts/up.sh index ddb5d16..5f90798 100755 --- a/e2e/scripts/up.sh +++ b/e2e/scripts/up.sh @@ -24,3 +24,6 @@ done if [[ "${E2E_RUN_TRANSFER:-0}" == "1" ]]; then "$E2E_ROOT/scripts/archive-happy.sh" fi +if [[ "${E2E_RUN_COMPLEX:-0}" == "1" ]]; then + "$E2E_ROOT/scripts/complex-matrix.sh" +fi diff --git a/src/archive_clients/daemon.py b/src/archive_clients/daemon.py index b480c19..1edaf54 100644 --- a/src/archive_clients/daemon.py +++ b/src/archive_clients/daemon.py @@ -964,6 +964,21 @@ class ArchiveClientDaemon: acknowledgement.error.message = "job step is invalid" else: acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_ACCEPTED + elif command.WhichOneof("payload") == "cancel_job": + if self.jobs is None: + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED + acknowledgement.error.code = common_pb2.ERROR_CODE_UNAVAILABLE + acknowledgement.error.message = ( + "job executor is unavailable" + ) + elif not command.cancel_job.job_id: + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED + acknowledgement.error.code = ( + common_pb2.ERROR_CODE_INVALID_ARGUMENT + ) + acknowledgement.error.message = "cancel job ID is required" + 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 diff --git a/tests/test_daemon.py b/tests/test_daemon.py index 67d3f25..a56de4d 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -67,6 +67,16 @@ class DaemonTransportTests(unittest.IsolatedAsyncioTestCase): control_pb2.COMMAND_ACK_STATUS_ACCEPTED, ) + cancel = control_pb2.Command(command_id=str(uuid4())) + cancel.cancel_job.job_id = definition.job_id + acknowledgement = daemon._initial_acknowledgement( + cancel, set(), False + ) + self.assertEqual( + acknowledgement.status, + control_pb2.COMMAND_ACK_STATUS_ACCEPTED, + ) + async def test_ensure_route_is_durable_and_duplicate_replays_updates(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory)