diff --git a/e2e/README.md b/e2e/README.md index e2b3b93..f3d8147 100644 --- a/e2e/README.md +++ b/e2e/README.md @@ -52,6 +52,32 @@ restores the target root's original mode even on an aborted run. The scheduler pause/resume endpoints exist only on the loopback test adapter; normal control and Telegram orchestration always use the automatic scheduler. +Set `E2E_RUN_ADVERSARIAL=1` to run the deferred Phase 8 v2/fault tranche in +one pass. The scenario performs archive → eviction → unarchive round trips for +a pure-v2 torrent and a hybrid torrent, checking that hybrid placements retain +both identities. It then: + +- leaves a Syncthing command durable while archive-2 is offline, waits beyond + the five-second E2E stall threshold, verifies the job retains overall + progress while `STALLED`, and reconnects the client to finish it; +- recreates archive-2 with `client-exhausted.toml`, whose impossible free-space + reserve exercises the real capacity guard without consuming disk, and + verifies a precise precommit failure with the source intact; and +- runs a separate `protocol-probe` container in the control network namespace. + The probe sends job event 2 before event 1, requires a snapshot request and + stale-state error, sends event 1 twice, and replays event 2 to prove + reordered convergence and duplicate idempotency over the real WebSocket. + +The adversarial script restores the normal archive-2 client configuration and +automatic scheduler through an exit trap. This tranche is intentionally +checked in before its first execution; run it later with: + +```bash +E2E_RUN_ADVERSARIAL=1 \ +ARCHIVE_CONTROL_SOURCE=/path/to/mogic-bot \ +./e2e/scripts/up.sh +``` + 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/archive-2/compose.yaml b/e2e/archive-2/compose.yaml index b3c5a12..bdfa2b8 100644 --- a/e2e/archive-2/compose.yaml +++ b/e2e/archive-2/compose.yaml @@ -53,7 +53,7 @@ services: qbittorrent: condition: service_healthy volumes: - - ./config/client.toml:/etc/archive-control/client.toml:ro + - "${ARCHIVE_2_CLIENT_CONFIG:-./config/client.toml}:/etc/archive-control/client.toml:ro" - ./secrets:/run/secrets:ro - ./runtime/client-state:/var/lib/archive-control - ./runtime/client-backups:/var/backups/archive-control @@ -68,4 +68,3 @@ networks: archive-control-e2e: external: true name: archive-control-e2e - diff --git a/e2e/archive-2/config/client-exhausted.toml b/e2e/archive-2/config/client-exhausted.toml new file mode 100644 index 0000000..c20740c --- /dev/null +++ b/e2e/archive-2/config/client-exhausted.toml @@ -0,0 +1,35 @@ +client_id = "archive-2" +display_name = "E2E Archive 2 (exhausted reserve)" +role = "archive" +control_endpoint = "ws://control:8765/archive_control" +shared_token_file = "/run/secrets/archive_control_token" +state_db = "/var/lib/archive-control/client.db" +backup_dir = "/var/backups/archive-control" + +[connection] +registration_timeout = "5s" +heartbeat_interval = "2s" +offline_timeout = "10s" +reconnect_initial = "1s" +reconnect_max = "5s" +reconnect_reset_after = "10s" +reconnect_jitter = false + +[jobs] +# Deliberately impossible reserve used only by the Phase 8 disk-exhaustion +# scenario. This exercises the real capacity guard without filling a volume. +free_space_reserve_bytes = 9223372036854775807 + +[qbittorrent] +endpoint = "http://qb-archive-2:8080" +username = "admin" +password_file = "/run/secrets/qb_password" +api_root = "/downloads" +local_root = "/data/qb" + +[syncthing] +endpoint = "http://syncthing-archive-2:8384" +api_key_file = "/run/secrets/syncthing_api_key" +api_root = "/sync" +local_root = "/data/sync" +advertised_addresses = ["tcp://syncthing-archive-2:22000"] diff --git a/e2e/control/compose.yaml b/e2e/control/compose.yaml index bb1f79d..520005e 100644 --- a/e2e/control/compose.yaml +++ b/e2e/control/compose.yaml @@ -23,6 +23,16 @@ services: profiles: [tools] depends_on: [control] + protocol-probe: + image: archive-clients:e2e + entrypoint: ["python", "/e2e/scenarios/protocol_ordering.py"] + network_mode: service:control + profiles: [tools] + depends_on: [control] + volumes: + - ../scenarios:/e2e/scenarios:ro + - ./secrets:/run/secrets:ro + networks: archive-control-e2e: external: true diff --git a/e2e/control/config/config.json b/e2e/control/config/config.json index 96d4bb7..7d6bc13 100644 --- a/e2e/control/config/config.json +++ b/e2e/control/config/config.json @@ -11,6 +11,7 @@ "offline_timeout": "10s", "command_ack_timeout": "2s", "command_max_attempts": 3, + "stall_after": "5s", "route_policy": "eager_mesh", "route_setup_timeout": "5m", "test_http": { diff --git a/e2e/scenarios/adversarial_jobs.py b/e2e/scenarios/adversarial_jobs.py new file mode 100755 index 0000000..0c91f6e --- /dev/null +++ b/e2e/scenarios/adversarial_jobs.py @@ -0,0 +1,213 @@ +#!/usr/bin/env python3 +"""Assertions used by the Phase 8 v2, stall, and capacity scenarios.""" + +from __future__ import annotations + +import argparse +import json +import time + +from complex_jobs import ( + CONTROL, + get_json, + post_json, + qb_json, + selection_indices, +) + + +def command_qb_hash(args: argparse.Namespace) -> None: + deadline = time.monotonic() + args.timeout + last = None + while time.monotonic() < deadline: + records = qb_json(args.endpoint, "/torrents/info") + matches = [ + item for item in records + if item.get("name") == args.name + ] + last = matches + if len(matches) == 1: + value = str(matches[0].get("hash", "")) + if len(value) not in {40, 64}: + raise RuntimeError(f"qBittorrent exposed an invalid hash: {value}") + print(value) + return + if len(matches) > 1: + raise RuntimeError(f"qBittorrent name lookup is ambiguous: {matches}") + time.sleep(0.25) + raise RuntimeError(f"qBittorrent torrent did not appear: {last}") + + +def command_wait_client(args: argparse.Namespace) -> None: + deadline = time.monotonic() + args.timeout + last = None + while time.monotonic() < deadline: + clients = get_json(f"{CONTROL}/clients") + matches = [ + item for item in clients + if item.get("client_id") == args.client_id + ] + last = matches + if ( + len(matches) == 1 + and bool(matches[0].get("connected")) == args.connected + ): + print(json.dumps({ + "client_id": args.client_id, + "connected": args.connected, + "generation": matches[0].get("generation"), + }, sort_keys=True)) + return + time.sleep(0.25) + raise RuntimeError(f"client connection state did not converge: {last}") + + +def command_wait_state(args: argparse.Namespace) -> None: + deadline = time.monotonic() + args.timeout + last = None + while time.monotonic() < deadline: + last = get_json(f"{CONTROL}/jobs/{args.job_id}") + if last["state"] == args.state: + sequence = int(last.get("last_event_sequence", 0)) + if sequence < args.minimum_sequence: + time.sleep(0.1) + continue + progress = (last.get("latest_event") or {}).get("progress") or {} + overall = float(progress.get("overallFractionComplete", 0)) + if overall < args.minimum_overall: + raise RuntimeError( + f"job lost overall progress while {args.state}: {last}" + ) + print(json.dumps({ + "job_id": args.job_id, + "last_event_sequence": sequence, + "overall_fraction_complete": overall, + "state": last["state"], + }, sort_keys=True)) + return + if last["state"] in { + "JOB_STATE_CANCELLED", + "JOB_STATE_FAILED", + "JOB_STATE_SUCCEEDED", + }: + raise RuntimeError( + f"job reached {last['state']} while waiting for {args.state}" + ) + time.sleep(0.1) + raise RuntimeError(f"job state did not converge: {last}") + + +def command_mark_stalled(args: argparse.Namespace) -> None: + deadline = time.monotonic() + args.timeout + last = None + while time.monotonic() < deadline: + outcomes = post_json(f"{CONTROL}/scheduler/advance", {}) + last = outcomes + if any( + item.get("job_id") == args.job_id + and item.get("action") == "marked_stalled" + for item in outcomes + ): + args.state = "JOB_STATE_STALLED" + args.minimum_sequence = 1 + command_wait_state(args) + return + time.sleep(0.25) + raise RuntimeError(f"scheduler did not mark the job stalled: {last}") + + +def command_assert_placement(args: argparse.Namespace) -> None: + expected_files = [ + int(item) for item in args.files.split(",") if item + ] + placements = get_json(f"{CONTROL}/placements") + matches = [ + item for item in placements + if item.get("client_id") == args.client_id + and item.get("state") == "PLACEMENT_STATE_PRESENT" + and ( + (args.v1 and item.get("info_hash_v1_hex") == args.v1) + or (args.v2 and item.get("info_hash_v2_hex") == args.v2) + ) + ] + if len(matches) != 1: + raise RuntimeError(f"placement identity is missing or ambiguous: {matches}") + placement = matches[0] + actual_files = selection_indices(placement["verified_files"]) + if ( + actual_files != expected_files + or (args.v1 and placement.get("info_hash_v1_hex") != args.v1) + or (args.v2 and placement.get("info_hash_v2_hex") != args.v2) + ): + raise RuntimeError(f"placement identity/selection mismatch: {placement}") + print(json.dumps({ + "client_id": args.client_id, + "files": actual_files, + "info_hash_v1_hex": placement.get("info_hash_v1_hex", ""), + "info_hash_v2_hex": placement.get("info_hash_v2_hex", ""), + }, sort_keys=True)) + + +def command_scheduler(args: argparse.Namespace) -> None: + result = post_json(f"{CONTROL}/scheduler/{args.action}", {}) + expected = args.action == "resume" + if result.get("automatic_scheduler") is not expected: + raise RuntimeError(f"scheduler gate did not {args.action}: {result}") + print(json.dumps(result, sort_keys=True)) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser() + commands = parser.add_subparsers(dest="command", required=True) + + qb_hash = commands.add_parser("qb-hash") + qb_hash.add_argument("--endpoint", required=True) + qb_hash.add_argument("--name", required=True) + qb_hash.add_argument("--timeout", type=float, default=30) + qb_hash.set_defaults(run=command_qb_hash) + + client = commands.add_parser("wait-client") + client.add_argument("--client-id", required=True) + client.add_argument( + "--connected", action=argparse.BooleanOptionalAction, default=True + ) + client.add_argument("--timeout", type=float, default=30) + client.set_defaults(run=command_wait_client) + + state = commands.add_parser("wait-state") + state.add_argument("--job-id", required=True) + state.add_argument("--state", required=True) + state.add_argument("--minimum-sequence", type=int, default=0) + state.add_argument("--minimum-overall", type=float, default=0) + state.add_argument("--timeout", type=float, default=30) + state.set_defaults(run=command_wait_state) + + stalled = commands.add_parser("mark-stalled") + stalled.add_argument("--job-id", required=True) + stalled.add_argument("--minimum-overall", type=float, default=0.01) + stalled.add_argument("--timeout", type=float, default=15) + stalled.set_defaults(run=command_mark_stalled) + + placement = commands.add_parser("assert-placement") + placement.add_argument("--client-id", required=True) + placement.add_argument("--v1", default="") + placement.add_argument("--v2", default="") + placement.add_argument("--files", required=True) + placement.set_defaults(run=command_assert_placement) + + scheduler = commands.add_parser("scheduler") + scheduler.add_argument( + "--action", choices=("pause", "resume"), required=True + ) + scheduler.set_defaults(run=command_scheduler) + return parser + + +def main() -> int: + args = build_parser().parse_args() + args.run(args) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/e2e/scenarios/protocol_ordering.py b/e2e/scenarios/protocol_ordering.py new file mode 100755 index 0000000..7e7dd4a --- /dev/null +++ b/e2e/scenarios/protocol_ordering.py @@ -0,0 +1,220 @@ +#!/usr/bin/env python3 +"""Exercise duplicate and reordered job events across a real WebSocket.""" + +from __future__ import annotations + +import asyncio +import json +import time +import urllib.request +from pathlib import Path +from uuid import uuid4 + +from websockets.asyncio.client import connect + +from archive_clients.protocol import decode, encode, new_envelope +from archive_control.v1 import client_pb2, common_pb2, control_pb2, job_pb2 + + +CONTROL_HTTP = "http://127.0.0.1:18081/test/v1" +CONTROL_WS = "ws://127.0.0.1:8765/archive_control" +CLIENT_ID = "protocol-probe" + + +def http_json(method: str, path: str, value: object | None = None): + encoded = ( + json.dumps(value, separators=(",", ":")).encode() + if value is not None + else None + ) + request = urllib.request.Request( + f"{CONTROL_HTTP}{path}", + data=encoded, + method=method, + headers={"Content-Type": "application/json"} if encoded else {}, + ) + with urllib.request.urlopen(request, timeout=15) as response: + return json.load(response) + + +def registration(): + envelope = new_envelope() + request = envelope.register_request + request.protocol_version.CopyFrom(envelope.protocol_version) + request.client.client_id = CLIENT_ID + request.client.display_name = "E2E protocol ordering probe" + request.client.role = common_pb2.CLIENT_ROLE_CACHE + request.connection_instance_id = str(uuid4()) + request.shared_token = Path( + "/run/secrets/archive_control_token" + ).read_text(encoding="utf-8").strip() + request.capabilities.max_envelope_bytes = 1024 * 1024 + return envelope + + +def acknowledgement(command_id: str): + envelope = new_envelope() + envelope.command_ack.command_id = command_id + envelope.command_ack.status = ( + control_pb2.COMMAND_ACK_STATUS_ACCEPTED + ) + return envelope + + +def heartbeat_ack(sequence: int): + envelope = new_envelope() + envelope.heartbeat_ack.sequence = sequence + return envelope + + +def job_event( + job_id: str, + *, + event_id: str, + sequence: int, + revision: int, + event_type: int, + state: int, +): + envelope = new_envelope() + event = envelope.job_event + event.event_id = event_id + event.job_id = job_id + event.sequence = sequence + event.job_revision = revision + event.type = event_type + event.state = state + event.occurred_at.GetCurrentTime() + if state == job_pb2.JOB_STATE_FAILED: + event.error.code = common_pb2.ERROR_CODE_PRECONDITION_FAILED + event.error.message = "protocol ordering probe terminal event" + return envelope + + +async def receive_payload(websocket, expected: set[str], timeout: float = 15): + deadline = time.monotonic() + timeout + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise RuntimeError(f"timed out waiting for {sorted(expected)}") + envelope = decode(await asyncio.wait_for(websocket.recv(), remaining)) + payload = envelope.WhichOneof("payload") + if payload == "heartbeat": + await websocket.send( + encode(heartbeat_ack(envelope.heartbeat.sequence)) + ) + continue + if payload in expected: + return envelope + + +async def main() -> None: + async with connect( + CONTROL_WS, + ping_interval=None, + proxy=None, + max_size=1024 * 1024, + ) as websocket: + await websocket.send(encode(registration())) + response = await receive_payload(websocket, {"register_response"}) + if ( + response.register_response.status + != client_pb2.REGISTRATION_STATUS_ACCEPTED + ): + raise RuntimeError("protocol probe registration was rejected") + + lease = await asyncio.to_thread( + http_json, + "POST", + "/protocol/order-probe", + {"client_id": CLIENT_ID}, + ) + if not lease.get("delivered"): + raise RuntimeError("protocol probe command was not delivered") + assigned = await receive_payload(websocket, {"command"}) + if assigned.command.WhichOneof("payload") != "assign_job": + raise RuntimeError("protocol probe received the wrong command") + if assigned.command.command_id != lease["command_id"]: + raise RuntimeError("protocol probe received the wrong command ID") + await websocket.send( + encode(acknowledgement(assigned.command.command_id)) + ) + + job_id = lease["job_id"] + first = job_event( + job_id, + event_id=str(uuid4()), + sequence=1, + revision=0, + event_type=control_pb2.JOB_EVENT_TYPE_ASSIGNED, + state=job_pb2.JOB_STATE_PREPARING, + ) + second = job_event( + job_id, + event_id=str(uuid4()), + sequence=2, + revision=1, + event_type=control_pb2.JOB_EVENT_TYPE_FAILED, + state=job_pb2.JOB_STATE_FAILED, + ) + + second_frame = encode(second) + await websocket.send(second_frame) + observed: set[str] = set() + while observed != {"snapshot", "error"}: + envelope = await receive_payload( + websocket, {"command", "protocol_error"} + ) + if envelope.WhichOneof("payload") == "protocol_error": + error = envelope.protocol_error + if error.offending_message_id != second.message_id: + raise RuntimeError("gap error identified the wrong envelope") + if ( + "expected event sequence 1, received 2" + not in error.error.message + ): + raise RuntimeError("gap error did not explain the ordering fault") + observed.add("error") + continue + if ( + envelope.command.WhichOneof("payload") + != "request_job_snapshot" + or list(envelope.command.request_job_snapshot.job_ids) + != [job_id] + ): + raise RuntimeError("gap did not request the expected snapshot") + await websocket.send( + encode(acknowledgement(envelope.command.command_id)) + ) + observed.add("snapshot") + + first_frame = encode(first) + await websocket.send(first_frame) + await websocket.send(first_frame) + await websocket.send(second_frame) + + deadline = time.monotonic() + 15 + last = None + while time.monotonic() < deadline: + last = await asyncio.to_thread( + http_json, "GET", f"/jobs/{job_id}" + ) + if ( + last["state"] == "JOB_STATE_FAILED" + and int(last["last_event_sequence"]) == 2 + and int(last["revision"]) == 1 + ): + print(json.dumps({ + "duplicate_sequence": 1, + "gap_sequence": 2, + "job_id": job_id, + "recovered_terminal_state": last["state"], + "snapshot_requested": True, + }, sort_keys=True)) + return + await asyncio.sleep(0.1) + raise RuntimeError(f"protocol ordering probe did not converge: {last}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/e2e/scripts/phase8-adversarial.sh b/e2e/scripts/phase8-adversarial.sh new file mode 100755 index 0000000..e638859 --- /dev/null +++ b/e2e/scripts/phase8-adversarial.sh @@ -0,0 +1,239 @@ +#!/usr/bin/env bash +set -euo pipefail + +source "$(dirname "$0")/lib.sh" + +v2_hash=8650a1a074039673ce156d4fdede8e38051108ddc90936eea818a20be2de5bfc +v2_qb_prefix=${v2_hash:0:40} +v2_torrent=ZDQ6aW5mb2Q5OmZpbGUgdHJlZWQwOmQ2Omxlbmd0aGkxOGUxMTpwaWVjZXMgcm9vdDMyOkzwzP0tijSmHJ/EFMYDoUaQHYtd6SNXTmo3yoKC8M+SZWUxMjptZXRhIHZlcnNpb25pMmU0Om5hbWU2OnYyLmJpbjEyOnBpZWNlIGxlbmd0aGkxNjM4NGVlZQ== +hybrid_v1=b77eea6da964b9661e082d567218ddf26609af56 +hybrid_v2=a7ca521665677a5bdace1bd4be32dbe0e43e25a3c578bb5e98efc76fb2bd9eeb +hybrid_torrent=ZDQ6aW5mb2Q5OmZpbGUgdHJlZWQwOmQ2Omxlbmd0aGkyMmUxMTpwaWVjZXMgcm9vdDMyOnAzSygNaQ3iIRKqDJWOXW4HIsea/KNFhE1SN+3/qvJSZWU2Omxlbmd0aGkyMmUxMjptZXRhIHZlcnNpb25pMmU0Om5hbWUxMDpoeWJyaWQuYmluMTI6cGllY2UgbGVuZ3RoaTE2Mzg0ZTY6cGllY2VzMjA6xQCrcqTOQ+LNUqZJC7p47gKZ/GhlZQ== +stall_hash=330be0cb7c2201135a2de63b28e77993745ff688 +stall_torrent=ZDQ6aW5mb2Q2Omxlbmd0aGkzMWU0Om5hbWUxMTpmaXh0dXJlLmJpbjEyOnBpZWNlIGxlbmd0aGkxNjM4NGU2OnBpZWNlczIwOlFlB05MtIUU4oem2MNz5LJW/GI6ZWU= + +control_complex() { + compose_control exec -T control \ + python /e2e/scenarios/complex_jobs.py "$@" +} + +control_adversarial() { + compose_control exec -T control \ + python /e2e/scenarios/adversarial_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 +} + +delete_hash() { + 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 +} + +install_torrent() { + local node=$1 + local name=$2 + local encoded=$3 + compose_node "$node" exec -T qbittorrent /bin/sh -c \ + "printf '%s' '$encoded' | base64 -d > /tmp/phase8.torrent" + compose_node "$node" exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/add \ + -F torrents=@/tmp/phase8.torrent \ + -F savepath=/downloads \ + -F stopped=true >/dev/null + local qb_hash + qb_hash=$(control_adversarial qb-hash \ + --endpoint "$(qb_endpoint "$node")" \ + --name "$name") + compose_node "$node" exec -T qbittorrent curl -fsS \ + -X POST http://127.0.0.1:8080/api/v2/torrents/recheck \ + --data-urlencode "hashes=$qb_hash" >/dev/null + control_complex assert-qb \ + --endpoint "$(qb_endpoint "$node")" \ + --info-hash "$qb_hash" \ + --selected '0' >/dev/null + printf '%s' "$qb_hash" +} + +assert_qb_by_name() { + local node=$1 + local name=$2 + local qb_hash + qb_hash=$(control_adversarial qb-hash \ + --endpoint "$(qb_endpoint "$node")" \ + --name "$name") + control_complex assert-qb \ + --endpoint "$(qb_endpoint "$node")" \ + --info-hash "$qb_hash" \ + --selected '0' \ + --no-downloaded >/dev/null +} + +restore_archive2=false +restore_scheduler=false +restore_archive2_client() { + if [[ "$restore_archive2" == true ]]; then + compose_node archive-2 up -d --force-recreate client >/dev/null + restore_archive2=false + fi +} +restore_test_environment() { + restore_archive2_client + if [[ "$restore_scheduler" == true ]]; then + control_adversarial scheduler --action resume >/dev/null 2>&1 || true + restore_scheduler=false + fi +} +trap restore_test_environment EXIT + +for node in cache-1 cache-2 archive-1 archive-2; do + for hash in \ + "$v2_hash" "$v2_qb_prefix" "$hybrid_v1" "$hybrid_v2" "$stall_hash"; do + delete_hash "$node" "$hash" + done + compose_node "$node" exec -T qbittorrent \ + rm -f /downloads/v2.bin /downloads/hybrid.bin /downloads/fixture.bin +done + +compose_node cache-1 exec -T --user 1001:1001 qbittorrent /bin/sh -c \ + 'printf "phase8-v2-content\n" > /downloads/v2.bin' +v2_source_qb_hash=$(install_torrent cache-1 v2.bin "$v2_torrent") +control_complex transfer \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-1 \ + --info-hash "$v2_hash" \ + --selected '0' \ + --expected-selection '0' \ + --expected-delta '0' +assert_qb_by_name archive-1 v2.bin +control_adversarial assert-placement \ + --client-id archive-1 \ + --v2 "$v2_hash" \ + --files '0' +control_complex evict \ + --cache-client cache-1 \ + --info-hash "$v2_hash" +control_complex assert-qb \ + --endpoint "$(qb_endpoint cache-1)" \ + --info-hash "$v2_source_qb_hash" \ + --absent >/dev/null +control_complex transfer \ + --operation unarchive \ + --source-client archive-1 \ + --target-client cache-2 \ + --info-hash "$v2_hash" \ + --selected '0' \ + --expected-selection '0' \ + --expected-delta '0' +assert_qb_by_name cache-2 v2.bin +control_adversarial assert-placement \ + --client-id cache-2 \ + --v2 "$v2_hash" \ + --files '0' + +compose_node cache-1 exec -T --user 1001:1001 qbittorrent /bin/sh -c \ + 'printf "phase8-hybrid-content\n" > /downloads/hybrid.bin' +hybrid_source_qb_hash=$(install_torrent cache-1 hybrid.bin "$hybrid_torrent") +control_complex transfer \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-2 \ + --info-hash "$hybrid_v1" \ + --selected '0' \ + --expected-selection '0' \ + --expected-delta '0' +assert_qb_by_name archive-2 hybrid.bin +control_adversarial assert-placement \ + --client-id archive-2 \ + --v1 "$hybrid_v1" \ + --v2 "$hybrid_v2" \ + --files '0' +control_complex evict \ + --cache-client cache-1 \ + --info-hash "$hybrid_v1" +control_complex assert-qb \ + --endpoint "$(qb_endpoint cache-1)" \ + --info-hash "$hybrid_source_qb_hash" \ + --absent >/dev/null +control_complex transfer \ + --operation unarchive \ + --source-client archive-2 \ + --target-client cache-2 \ + --info-hash "$hybrid_v1" \ + --selected '0' \ + --expected-selection '0' \ + --expected-delta '0' +assert_qb_by_name cache-2 hybrid.bin +control_adversarial assert-placement \ + --client-id cache-2 \ + --v1 "$hybrid_v1" \ + --v2 "$hybrid_v2" \ + --files '0' + +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 fixture.bin "$stall_torrent" >/dev/null +stall_job=$(control_complex drive \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-2 \ + --info-hash "$stall_hash" \ + --through source_stage) +compose_node archive-2 stop client +control_adversarial wait-client \ + --client-id archive-2 \ + --no-connected +control_complex advance \ + --job-id "$stall_job" \ + --action syncthing_transfer >/dev/null +sleep 6 +control_adversarial mark-stalled \ + --job-id "$stall_job" \ + --minimum-overall 0.2 +compose_node archive-2 start client +control_adversarial wait-client --client-id archive-2 +control_complex resume --job-id "$stall_job" +assert_qb_by_name archive-2 fixture.bin + +delete_hash archive-2 "$stall_hash" +compose_node archive-2 exec -T qbittorrent rm -f /downloads/fixture.bin +ARCHIVE_2_CLIENT_CONFIG=./config/client-exhausted.toml \ + compose_node archive-2 up -d --force-recreate client >/dev/null +restore_archive2=true +control_adversarial wait-client --client-id archive-2 +disk_job=$(control_complex create-paused \ + --operation archive \ + --source-client cache-1 \ + --target-client archive-2 \ + --info-hash "$stall_hash") +control_complex expect-failure \ + --job-id "$disk_job" \ + --contains 'insufficient free space' +control_complex assert-qb \ + --endpoint "$(qb_endpoint cache-1)" \ + --info-hash "$stall_hash" \ + --selected '0' >/dev/null +restore_archive2_client +control_adversarial wait-client --client-id archive-2 + +control_adversarial scheduler --action pause >/dev/null +restore_scheduler=true +compose_control run --rm protocol-probe +control_adversarial scheduler --action resume >/dev/null +restore_scheduler=false + +printf '%s\n' \ + 'phase 8 adversarial matrix passed: v2/hybrid round trips, recoverable stall,' \ + 'bounded disk exhaustion, and duplicate/reordered WebSocket job events' diff --git a/e2e/scripts/up.sh b/e2e/scripts/up.sh index 5f90798..acae2e9 100755 --- a/e2e/scripts/up.sh +++ b/e2e/scripts/up.sh @@ -27,3 +27,6 @@ fi if [[ "${E2E_RUN_COMPLEX:-0}" == "1" ]]; then "$E2E_ROOT/scripts/complex-matrix.sh" fi +if [[ "${E2E_RUN_ADVERSARIAL:-0}" == "1" ]]; then + "$E2E_ROOT/scripts/phase8-adversarial.sh" +fi diff --git a/tests/test_jobs.py b/tests/test_jobs.py index baa8736..cb51cb6 100644 --- a/tests/test_jobs.py +++ b/tests/test_jobs.py @@ -32,6 +32,36 @@ class CompleteSyncthing: class ClientJobHappyPathTests(unittest.TestCase): + def test_capacity_guard_fails_before_data_movement(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + store = ClientStore(root / "client.db") + store.initialize() + executor = ClientJobExecutor( + client_id="archive-1", + qbittorrent=Mock(), + store=store, + qb_root=root, + qb_api_root=Path("/downloads"), + route_path=lambda _: root, + syncthing_transport=Mock(), + sparse_supported=True, + free_space_reserve_bytes=100, + ) + before = {path.name for path in root.iterdir()} + with patch( + "archive_clients.jobs.shutil.disk_usage", + return_value=Mock(free=109), + ), self.assertRaisesRegex( + JobExecutionError, + "109 bytes available, 110 bytes required including reserve", + ): + executor._require_space(root, 10) + self.assertEqual( + {path.name for path in root.iterdir()}, + before, + ) + def test_archive_and_unarchive_five_step_execution(self): for operation in ( job_pb2.JOB_OPERATION_ARCHIVE, diff --git a/tests/test_resources.py b/tests/test_resources.py index 4497148..e46574f 100644 --- a/tests/test_resources.py +++ b/tests/test_resources.py @@ -12,6 +12,48 @@ from archive_control.v1 import resource_pb2 class ResourceTests(unittest.TestCase): + def test_pure_v2_identity_and_single_file_tree_normalization(self): + info = { + b"file tree": { + b"": { + b"length": 3, + b"pieces root": b"x" * 32, + }, + }, + b"meta version": 2, + b"name": b"v2.bin", + b"piece length": 16384, + } + metainfo_bytes = encode({b"info": info}) + decoded = decode_metainfo(metainfo_bytes) + self.assertEqual(decoded.info_hash_v1_hex, "") + self.assertEqual( + decoded.info_hash_v2_hex, + hashlib.sha256(encode(info)).hexdigest(), + ) + self.assertEqual(decoded.files[0].path, "v2.bin") + + normalized = normalize_resource( + { + "hash": decoded.info_hash_v2_hex, + "name": "v2.bin", + "state": "stoppedUP", + }, + [{ + "index": 0, + "name": "v2.bin", + "size": 3, + "completed": 3, + "priority": 1, + }], + metainfo_bytes, + ) + self.assertEqual( + normalized.summary.resource_id.info_hash_v2_hex, + decoded.info_hash_v2_hex, + ) + self.assertFalse(normalized.summary.resource_id.info_hash_v1_hex) + def test_hybrid_identity_and_selection_normalization(self): info = { b"file tree": {