test: expand phase 8 fault matrix

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

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

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'

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

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

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)