ddidderr/lanspread
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

test(peer-cli): verify protocol 8 transfer lifecycle

Browse Source 
Replace legacy metadata and relay expectations with current protocol-8 JSONL
assertions. Scenarios now bind every source to authenticated PeerId and exact
ContentId, prove typed attempt lifecycle order, and fence cancellation,
quarantine, rollback, republishing, and peer-departure outcomes against vacuous
success.

Isolated topologies distinguish direct author pulls from relay and ambient mDNS
substitution. The run log records focused diagnostics and the final fresh-image
S1-S49 acceptance result without presenting Docker-host throughput as a
representative external-LAN measurement.

Test Plan:
- `LANSPREAD_S37_MIN_MIB_PER_S=100 just peer-cli-tests` -- passed S1-S49
- S37 -- passed 2,147,483,656 bytes in 17 chunks at 551.10 MiB/s
- `python3 -m py_compile crates/lanspread-peer-cli/scripts/run_extended_scenarios.py` -- passed
- `ruff check --select F,E9 crates/lanspread-peer-cli/scripts/run_extended_scenarios.py` -- passed
- `git diff --cached --check` -- passed
This commit is contained in:
ddidderr
2026-08-10 14:00:03 +02:00
parent 71dbf27d8b
commit 4a1b08db98
3 changed files with 2441 additions and 515 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/lanspread-peer-cli/README.md
+31 -2
View File
@@ -17,8 +17,12 @@ Useful flags:
- `--games-dir PATH` stores local archives and installs.
- `--state-dir PATH` stores the generated peer identity.
- `--identity-file PATH` loads one existing peer identity strictly. Missing or
invalid files fail startup without repair, quarantine, or fallback.
- `--catalog-db PATH` and `--manifests-dir PATH` select one coherent catalog
authority profile.
- `--fixture GAME_ID` seeds a tiny archive that the fixture unpacker can
install.
install. The selected profile must already authorize that exact fixture.
## Fixture Game Directories
@@ -29,6 +33,27 @@ catalog-backed fake games. Each game includes `version.ini` and a real RAR
archive renamed to `.eti`; `fixture-alpha` and `fixture-bravo` share `ggoo`,
while `fixture-bravo` and `fixture-charlie` share `cnc4`.
The checked-in `catalogs/default` profile authorizes the normal alpha, bravo,
charlie, and persona packages. `catalogs/solid` and `catalogs/multi` are
separate authorities because their `cnctw` packages intentionally contain
different bytes and extracted layouts. `catalogs/unknown` is a source-only
`cod2` profile used to prove that another peer's honest catalog cannot extend
the client's local catalog.
Regenerate and verify the profiles with:
```bash
just fixture-catalogs
just fixture-catalogs-check
```
Both commands use the Rust catalog publisher. Dynamic sparse and many-file
acceptance packages use the same test-only generator through
`just fixture-download-only-catalog` or `just fixture-catalog`; the Python
scenario runner never derives hashes or catalog rows itself. Production
artifacts are a separate corpus and are checked by
`just catalog-check-production`.
## Commands
Every command is a JSON object with `cmd` or `command`; `id` is optional and is
@@ -37,7 +62,7 @@ echoed back on the result or error line.
```json
{"id":"s1","cmd":"status"}
{"id":"p1","cmd":"wait-peers","count":1,"timeout_ms":5000}
{"id":"c1","cmd":"connect","addr":"127.0.0.1:34567"}
{"id":"c1","cmd":"connect","peer_id":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","addr":"127.0.0.1:34567"}
{"id":"g1","cmd":"list-games"}
{"id":"d1","cmd":"download","game_id":"fixture-one","install":true}
{"id":"i1","cmd":"install","game_id":"fixture-one"}
@@ -45,6 +70,10 @@ echoed back on the result or error line.
{"id":"q1","cmd":"shutdown"}
```
`connect` requires the target's `peer_id` and `addr` from the same
`local-peer-ready` event. Address-only connects are rejected because the peer ID
is the TLS identity pin, not descriptive metadata.
The `status` result includes receiver-side `active_operations` and sender-side
`active_outbound_transfers` counts by game ID, which the scenario runner uses to
verify transfer lifecycle cleanup.
crates/lanspread-peer-cli/scripts/run_extended_scenarios.py
+2130 -399
View File
@@ -4,8 +4,9 @@
from __future__ import annotations
import argparse
import base64
import binascii
import hashlib
import ipaddress
import json
import os
import queue
@@ -26,8 +27,15 @@ IMAGE = "lanspread-peer-cli:dev"
NETWORK = "lanspread"
CONTAINER_PREFIX = "lanspread-peer-cli-ext"
CATALOG_DB = "/app/game.db"
CATALOG_MANIFESTS = "/app/manifests"
FIXTURES = REPO / "crates" / "lanspread-peer-cli" / "fixtures"
CATALOG_PROFILES = REPO / "crates" / "lanspread-peer-cli" / "catalogs"
CHUNK_SIZE = 128 * 1024 * 1024
# Runtime liveness needs the 90-second stale threshold plus a 20-second ping
# tick before a departed peer and its remote state are removed. Keep a
# conservative harness margin; graceful shutdown intentionally sends no
# protocol Goodbye.
PEER_DEPARTURE_TIMEOUT_SECONDS = 125
CATALOG_VERSIONS = {
"alienswarm": "20190317",
"bf1942": "20160130",
@@ -59,11 +67,13 @@ class LineWaiter:
class Peer:
runner: "Runner"
name: str
network: str = NETWORK
games_dir: Path | None = None
readonly_games: bool = False
tmpfs_size: str | None = None
fixtures: list[str] = field(default_factory=list)
extra_args: list[str] = field(default_factory=list)
catalog_dir: Path | None = None
process: subprocess.Popen[str] | None = None
output: list[dict[str, Any]] = field(default_factory=list)
@@ -99,7 +109,7 @@ class Peer:
"--rm",
"--init",
"--network",
NETWORK,
self.network,
"--name",
self.container_name,
"-i",
@@ -113,6 +123,17 @@ class Peer:
else:
command.extend(["--tmpfs", f"/games:size={self.tmpfs_size}"])
catalog_db = CATALOG_DB
catalog_manifests = CATALOG_MANIFESTS
if self.catalog_dir is not None:
if not (self.catalog_dir / "game.db").is_file():
raise ScenarioError(f"catalog profile has no game.db: {self.catalog_dir}")
if not (self.catalog_dir / "manifests").is_dir():
raise ScenarioError(f"catalog profile has no manifests/: {self.catalog_dir}")
command.extend(["-v", f"{self.catalog_dir}:/scenario-catalog:ro"])
catalog_db = "/scenario-catalog/game.db"
catalog_manifests = "/scenario-catalog/manifests"
command.extend(
[
IMAGE,
@@ -123,7 +144,9 @@ class Peer:
"--state-dir",
"/state",
"--catalog-db",
CATALOG_DB,
catalog_db,
"--manifests-dir",
catalog_manifests,
]
)
for fixture in self.fixtures:
@@ -243,17 +266,36 @@ class Peer:
return self.send({"cmd": "status"})["data"]
def call_to_play_events(self) -> list[dict[str, Any]]:
return self.send({"cmd": "list-call-to-play"})["data"]["events"]
return self.send({"cmd": "list-call-to-play"})["data"]["view"]["events"]
def publish_call_to_play(self, event: dict[str, Any]) -> None:
self.send({"cmd": "publish-call-to-play", "event": event})
def publish_call_to_play(self, intent: dict[str, Any]) -> dict[str, Any]:
return self.send(
{"cmd": "publish-call-to-play", "intent": intent}
)["data"]["receipt"]
def set_call_to_play_display_name(self, display_name: str) -> None:
self.send(
{
"cmd": "set-call-to-play-display-name",
"display_name": display_name,
}
)
def ready_endpoint(self) -> dict[str, str]:
if self.ready_addr is None or self.peer_id is None:
raise ScenarioError(f"{self.name} is not ready")
return {"peer_id": self.peer_id, "addr": self.ready_addr}
def connect_to(self, other: "Peer") -> None:
if other.ready_addr is None:
raise ScenarioError(f"{other.name} is not ready")
self.send({"cmd": "connect", "addr": other.ready_addr})
self.send({"cmd": "connect", **other.ready_endpoint()})
self.send({"cmd": "wait-peers", "count": 1, "timeout_ms": 10000})
def connect_to_at(self, other: "Peer", addr: str, peer_count: int) -> None:
if other.peer_id is None:
raise ScenarioError(f"{other.name} has no ready peer identity")
self.send({"cmd": "connect", "peer_id": other.peer_id, "addr": addr})
self.send({"cmd": "wait-peers", "count": peer_count, "timeout_ms": 10000})
def shutdown(self) -> None:
if self.process is None:
return
@@ -267,6 +309,30 @@ class Peer:
except subprocess.TimeoutExpired:
self.kill()
def shutdown_gracefully_for_proof(self) -> None:
if self.process is None or self.process.poll() is not None:
raise ScenarioError(
f"{self.name} was not running at the strict graceful-shutdown fence"
)
result = self.send({"cmd": "shutdown"}, timeout=15)
if (
result.get("command") != "shutdown"
or result.get("data", {}).get("stopped") is not True
):
raise ScenarioError(
f"{self.name} returned an invalid graceful-shutdown proof: {result}"
)
try:
exit_code = self.process.wait(timeout=8)
except subprocess.TimeoutExpired as error:
raise ScenarioError(
f"{self.name} did not exit after its graceful-shutdown proof"
) from error
if exit_code != 0:
raise ScenarioError(
f"{self.name} exited {exit_code} after its graceful-shutdown proof"
)
def kill(self) -> None:
subprocess.run(
["docker", "rm", "-f", self.container_name],
@@ -301,6 +367,7 @@ class Runner:
self.games_root = RUN_ROOT / "games"
self.fixture_root = RUN_ROOT / "fixtures"
self.current_peers: list[Peer] = []
self.current_networks: list[str] = []
self.results: list[tuple[str, str]] = []
def log(self, message: str) -> None:
@@ -316,7 +383,7 @@ class Runner:
("S5", self.s5_auto_install_download),
("S6", self.s6_manual_install_uninstall),
("S7", self.s7_duplicate_source_download),
("S8", self.s8_ambiguous_metadata_rejection),
("S8", self.s8_catalog_file_shape_failover),
("S9", self.s9_missing_game),
("S10", self.s10_shutdown_cleanup),
("S11", self.s11_same_identity_reconnect),
@@ -325,7 +392,7 @@ class Runner:
("S14", self.s14_large_multi_peer_chunking),
("S15", self.s15_three_way_version_skew),
("S16", self.s16_catalog_fanout_with_stale),
("S17", self.s17_catalog_conflict_rejection),
("S17", self.s17_catalog_byte_quarantine),
("S18", self.s18_redundant_source_drop),
("S19", self.s19_sole_source_drop),
("S20", self.s20_receiver_write_failure),
@@ -336,7 +403,7 @@ class Runner:
("S25", self.s25_two_downloads_one_client),
("S26", self.s26_duplicate_download_rejection),
("S27", self.s27_self_connect_rejection),
("S28", self.s28_address_change_unit),
("S28", self.s28_reconnect_generation_unit),
("S29", self.s29_empty_peer_participates),
("S30", self.s30_mesh_aggregation),
("S31", self.s31_bootstrapped_peer_source),
@@ -364,6 +431,7 @@ class Runner:
if self.selected and scenario_id.lower() not in self.selected:
continue
self.cleanup_containers()
self.cleanup_scenario_networks()
self.current_peers = []
try:
self.log(f"\n== {scenario_id} ==")
@@ -386,6 +454,7 @@ class Runner:
self.games_root.mkdir(parents=True, exist_ok=True)
self.fixture_root.mkdir(parents=True, exist_ok=True)
self.cleanup_containers()
self.cleanup_scenario_networks()
if self.build_image:
run(["just", "peer-cli-image"], "build peer-cli image")
run(["just", "peer-cli-net"], "prepare peer-cli docker network")
@@ -417,29 +486,96 @@ class Runner:
for peer in reversed(self.current_peers):
peer.shutdown()
self.cleanup_containers()
self.cleanup_scenario_networks()
def create_scenario_network(self, name: str, *, internal: bool = False) -> str:
network = f"{CONTAINER_PREFIX}-{self.run_id}-{name}"
command = ["docker", "network", "create"]
if internal:
command.append("--internal")
command.append(network)
run(command, f"create {name} network")
self.current_networks.append(network)
return network
def cleanup_scenario_networks(self) -> None:
for network in reversed(self.current_networks):
subprocess.run(
["docker", "network", "rm", network],
cwd=REPO,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
self.current_networks = []
def connect_peer_to_network(self, peer: Peer, network: str) -> str:
run(
["docker", "network", "connect", network, peer.container_name],
f"attach {peer.name} to {network}",
)
inspect = run(
[
"docker",
"inspect",
"--format",
f"{{{{(index .NetworkSettings.Networks {json.dumps(network)}).IPAddress}}}}",
peer.container_name,
],
f"inspect {peer.name} address on {network}",
)
address = inspect.stdout.strip()
if not address or peer.ready_addr is None:
raise ScenarioError(f"{peer.name} has no address on {network}")
_, port = peer.ready_addr.rsplit(":", 1)
return f"{address}:{port}"
def peer(
self,
name: str,
*,
network: str = NETWORK,
games_dir: Path | None = None,
readonly_games: bool = False,
tmpfs_size: str | None = None,
fixtures: list[str] | None = None,
extra_args: list[str] | None = None,
catalog_dir: Path | None = None,
) -> Peer:
peer = Peer(
runner=self,
name=name,
network=network,
games_dir=games_dir,
readonly_games=readonly_games,
tmpfs_size=tmpfs_size,
fixtures=fixtures or [],
extra_args=extra_args or [],
catalog_dir=catalog_dir,
).start()
self.current_peers.append(peer)
return peer
def fixture_catalog(
self,
name: str,
game_root: Path,
*,
streamed_install: bool,
) -> Path:
output = self.fixture_root / f"{name}-catalog"
command = ["just", "fixture-catalog", str(output), str(game_root)]
if not streamed_install:
command = [
"just",
"fixture-download-only-catalog",
str(output),
game_root.name,
str(game_root),
]
run(command, f"publish {name} fixture catalog")
return output
def s1_startup_scan(self) -> str:
alpha = self.peer("s1-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
games = alpha.list_games()["local"]
@@ -455,19 +591,30 @@ class Runner:
def s2_direct_connect_handshake(self) -> str:
alpha = self.peer("s2-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
bravo = self.peer("s2-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
connect_many(alpha, [bravo])
alpha.connect_to(bravo)
bravo.connect_to(alpha)
# Poll for library convergence rather than reading once: wait-peers only
# guarantees peer presence, and the mDNS path can upsert a peer with an
# empty library before its snapshot arrives.
wait_peer_has_game(alpha, bravo.peer_id, "bfbc2")
bravo_content_id = catalog_content_id(bravo, "bfbc2")
wait_peer_has_game(alpha, bravo.peer_id, "bfbc2", bravo_content_id)
wait_remote_library_content(
alpha, "bfbc2", bravo_content_id, peer_count=1
)
peers = alpha.list_peers()
if len(peers) != 1 or peers[0]["peer_id"] == alpha.peer_id:
raise ScenarioError(f"bad alpha peers after connect: {peers}")
if peers[0]["game_count"] != 4:
raise ScenarioError(f"expected bravo game_count=4, got {peers}")
# The handshake is bidirectional: bravo must also record alpha's library
# without a separate connect, proving both directions of the exchange.
bravo_view = wait_peer_has_game(bravo, alpha.peer_id, "alienswarm")
# Peer state is pulled by the dialing side; prove each direction with an
# explicit authenticated endpoint.
alpha_content_id = catalog_content_id(alpha, "alienswarm")
bravo_view = wait_peer_has_game(
bravo, alpha.peer_id, "alienswarm", alpha_content_id
)
wait_remote_library_content(
bravo, "alienswarm", alpha_content_id, peer_count=1
)
if bravo_view["game_count"] != 3:
raise ScenarioError(f"expected alpha game_count=3 on bravo, got {bravo_view}")
return "alpha<->bravo exchanged libraries: bravo had 4 games on alpha, alpha had 3 on bravo"
@@ -493,22 +640,29 @@ class Runner:
bravo = self.peer("s4-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
client = self.peer("s4-client")
connect_many(client, [bravo])
content_id = catalog_content_id(bravo, "bfbc2")
wait_peer_has_game(client, bravo.peer_id, "bfbc2", content_id)
wait_remote_library_content(client, "bfbc2", content_id, peer_count=1)
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "bfbc2", "install": False})
client.wait_for(event_is("got-game-files", "bfbc2"), timeout=20, description="got bfbc2", waiter=waiter)
client.wait_for(event_is("download-begin", "bfbc2"), timeout=20, description="begin bfbc2", waiter=waiter)
client.wait_for(event_is("download-finished", "bfbc2"), timeout=60, description="finish bfbc2", waiter=waiter)
assert_chunk_authority(client, "bfbc2", content_id, {bravo.peer_id})
game = wait_local_game(client, "bfbc2", downloaded=True, installed=False)
diff_game_dirs(FIXTURES / "fixture-bravo" / "bfbc2", client.host_games_dir / "bfbc2")
if (client.host_games_dir / "bfbc2" / "local").exists():
raise ScenarioError("bfbc2 local/ exists after install=false")
return f"bfbc2 downloaded with install=false, local state installed={game['installed']}, diff matched"
return (
"bfbc2 exact ContentId matched manifest, raw peer snapshot, remote view, "
f"and every authenticated chunk; install=false state installed={game['installed']}"
)
def s5_auto_install_download(self) -> str:
bravo = self.peer("s5-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
client = self.peer("s5-client")
connect_many(client, [bravo])
waiter = LineWaiter(len(client.output))
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": "cnctw"})
client.wait_for(event_is("download-finished", "cnctw"), timeout=60, description="finish cnctw", waiter=waiter)
client.wait_for(event_is("install-finished", "cnctw"), timeout=30, description="install cnctw", waiter=waiter)
@@ -541,6 +695,10 @@ class Runner:
bravo = self.peer("s7-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
client = self.peer("s7-client")
connect_many(client, [alpha, bravo])
content_id = catalog_content_id(client, "ggoo")
wait_peer_has_game(client, alpha.peer_id, "ggoo", content_id)
wait_peer_has_game(client, bravo.peer_id, "ggoo", content_id)
wait_remote_library_content(client, "ggoo", content_id, peer_count=2)
wait_remote_game(client, "ggoo", peer_count=2)
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "ggoo", "install": False})
@@ -552,63 +710,185 @@ class Runner:
# and nothing was fetched twice.
if count_events(client, "download-finished", "ggoo") != 1:
raise ScenarioError("ggoo did not finish exactly once")
assert_only_chunk_sources(client, "ggoo", {alpha.ready_addr, bravo.ready_addr})
if chunk_sources(client, "ggoo") != {alpha.ready_addr, bravo.ready_addr}:
assert_chunk_authority(client, "ggoo", content_id, {alpha.peer_id, bravo.peer_id})
if chunk_sources(client, "ggoo") != {alpha.peer_id, bravo.peer_id}:
raise ScenarioError(
f"expected both validated sources to serve ggoo, got {chunk_sources(client, 'ggoo')}"
)
assert_no_duplicate_chunks(client, "ggoo")
return "ggoo downloaded once, served from both validated sources with no duplicate chunks"
def s8_ambiguous_metadata_rejection(self) -> str:
dir_a = self.fixture_root / "s8-a"
dir_b = self.fixture_root / "s8-b"
copy_game("ggoo", dir_a)
copy_game("ggoo", dir_b)
with (dir_b / "ggoo" / "ggoo.eti").open("ab") as handle:
def s8_catalog_file_shape_failover(self) -> str:
game_id = "ggoo"
source_a_dir = self.fixture_root / "s8-source-a"
source_b_dir = self.fixture_root / "s8-source-b"
copy_game(game_id, source_a_dir)
copy_game(game_id, source_b_dir)
source_a = self.peer("s8-source-a", games_dir=source_a_dir)
source_b = self.peer("s8-source-b", games_dir=source_b_dir)
bad, good = sorted(
[source_a, source_b], key=lambda peer: peer_id_sort_key(peer.peer_id)
)
# Production distributes the first catalog entry to the first source in
# authenticated (PeerId, address) order. Corrupt that source only after
# startup so this scenario deterministically exercises archive admission
# failure and retry regardless of Docker address or connection order.
with (bad.host_games_dir / game_id / f"{game_id}.eti").open("ab") as handle:
handle.write(b"conflict")
peer_a = self.peer("s8-a", games_dir=dir_a)
peer_b = self.peer("s8-b", games_dir=dir_b)
client = self.peer("s8-client")
connect_many(client, [peer_a, peer_b])
wait_remote_game(client, "ggoo", peer_count=2, version=CATALOG_VERSIONS["ggoo"])
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "ggoo", "install": False})
client.wait_for(event_is("download-failed", "ggoo"), timeout=30, description="ggoo failed", waiter=waiter)
assert_not_exists(client.host_games_dir / "ggoo" / "version.ini")
return "conflicting catalog-version ggoo file sizes emitted download-failed and left no version.ini"
connect_many(client, [bad, good])
content_id = catalog_content_id(client, game_id)
wait_peer_has_game(client, bad.peer_id, game_id, content_id)
wait_peer_has_game(client, good.peer_id, game_id, content_id)
wait_remote_library_content(client, game_id, content_id, peer_count=2)
wait_remote_game(client, game_id, peer_count=2, version=CATALOG_VERSIONS[game_id])
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": game_id, "install": False})
client.wait_for(
event_is("download-finished", game_id),
timeout=60,
description="ggoo oversized-source failover",
waiter=waiter,
)
assert_download_status_trace(
client,
game_id,
start=start,
terminal="download-finished",
)
# The bad source's version.ini is still exact catalog content and is a
# valid successful chunk. Only its oversized archive must be rejected.
assert_chunk_authority(client, game_id, content_id, {bad.peer_id, good.peer_id})
archive_totals = chunk_totals(client, game_id, f"{game_id}.eti")
if set(archive_totals) != {good.peer_id}:
raise ScenarioError(
"oversized source supplied a successful archive chunk instead of failing over: "
f"{archive_totals}"
)
assert_no_duplicate_chunks(client, game_id)
diff_game_dirs(good.host_games_dir / game_id, client.host_games_dir / game_id)
bad_only = self.peer("s8-oversized-only-client")
connect_many(bad_only, [bad])
wait_peer_has_game(bad_only, bad.peer_id, game_id, content_id)
wait_remote_library_content(bad_only, game_id, content_id, peer_count=1)
wait_remote_game(bad_only, game_id, peer_count=1)
failed_start = len(bad_only.output)
failed_waiter = LineWaiter(failed_start)
bad_only.send({"cmd": "download", "game_id": game_id, "install": False})
bad_only.wait_for(
event_is("download-failed", game_id),
timeout=60,
description="oversized-only ggoo failure",
waiter=failed_waiter,
)
assert_download_status_trace(
bad_only,
game_id,
start=failed_start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
)
assert_no_event_since(bad_only, failed_start, "download-finished", game_id)
assert_local_absent(bad_only, game_id)
assert_not_exists(bad_only.host_games_dir / game_id / "version.ini")
assert_not_exists(bad_only.host_games_dir / game_id / "local")
return (
"oversized source failed exact catalog-file admission and the honest source "
"completed; an oversized-only receiver published no version.ini or local/"
)
def s9_missing_game(self) -> str:
game_id = "cod6"
client = self.peer("s9-client")
err = client.send({"cmd": "download", "game_id": "cod2", "install": False}, expect_error=True)
if "no peers have game cod2" not in err["error"]:
raise ScenarioError(f"unexpected missing game error: {err}")
assert_not_exists(client.host_games_dir / "cod2")
return f"missing game command errored '{err['error']}' and created no local directory"
start = len(client.output)
waiter = LineWaiter(start)
queued = client.send(
{"cmd": "download", "game_id": game_id, "install": False}
)
if queued.get("data", {}).get("queued") is not True:
raise ScenarioError(f"missing-source download was not queued: {queued}")
client.wait_for(
event_is("download-failed", game_id),
timeout=20,
description="asynchronous no-source download failure",
waiter=waiter,
)
assert_download_status_trace(
client,
game_id,
start=start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
expect_begin=False,
expect_verification=False,
)
assert_no_event_since(client, start, "download-begin", game_id)
assert_no_event_since(client, start, "download-finished", game_id)
active_events = [
item
for item in client.output[start:]
if item.get("type") == "event"
and item.get("event") == "active-operations-changed"
and any(
operation.get("game_id") == game_id
for operation in item.get("data", {}).get("active_operations", [])
)
]
if active_events:
raise ScenarioError(
f"missing-source download entered active operations: {active_events}"
)
assert_no_active(client, game_id)
assert_local_absent(client, game_id)
game_root = client.host_games_dir / game_id
assert_not_exists(game_root / "version.ini")
assert_not_exists(game_root / "local")
assert_not_exists(game_root)
return (
"missing-source command queued, then emitted download-failed without begin, "
"active-operation, version.ini, local/, or game-root mutation"
)
def s10_shutdown_cleanup(self) -> str:
alpha = self.peer("s10-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
bravo = self.peer("s10-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
connect_many(alpha, [bravo])
connect_bidirectional(alpha, bravo)
wait_remote_game(alpha, "bfbc2", peer_count=1)
bravo.shutdown()
wait_remote_absent(alpha, "bfbc2")
if alpha.list_peers():
raise ScenarioError(f"alpha still has peers after bravo shutdown: {alpha.list_peers()}")
return "bravo graceful shutdown removed peer and bravo-only games from alpha"
wait_departure_topology(
alpha,
expected_peer_ids=set(),
expected_remote_game_ids=set(),
phase="S10 bravo shutdown",
)
return (
"bravo shutdown converged through pinned liveness to an exact empty peer "
"set and remote-library view"
)
def s11_same_identity_reconnect(self) -> str:
alpha = self.peer("s11-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
bravo_dir = FIXTURES / "fixture-bravo"
bravo = self.peer("s11-bravo", games_dir=bravo_dir, readonly_games=True)
connect_many(alpha, [bravo])
connect_bidirectional(alpha, bravo)
wait_remote_game(alpha, "bfbc2", peer_count=1)
first_peer = alpha.list_peers()[0]
first_addr = first_peer["addr"]
first_id = first_peer["peer_id"]
bravo.shutdown()
wait_remote_absent(alpha, "bfbc2")
wait_departure_topology(
alpha,
expected_peer_ids=set(),
expected_remote_game_ids=set(),
phase="S11 old bravo generation shutdown",
)
bravo = self.peer("s11-bravo", games_dir=bravo_dir, readonly_games=True)
connect_many(alpha, [bravo])
connect_bidirectional(alpha, bravo)
wait_remote_game(alpha, "bfbc2", peer_count=1)
peers = alpha.list_peers()
# The real invariant is a single peer entry reusing the same identity (no
# duplicate). The listener address is an OS-assigned ephemeral port: it
@@ -625,10 +905,10 @@ class Runner:
def s12_transfer_serving_gates(self) -> str:
output = run_just_test()
required = [
"local_download_available_gates_on_catalog_operation_and_sentinel",
"get_game_response_respects_serve_gates",
"file_transfer_dispatch_respects_serve_gates",
"local_relative_paths_are_never_transferable",
"wrong_identity_path_or_range_never_authorizes_an_entry",
"chunk_boundaries_are_exact_including_empty_files",
"admission_rejects_every_ineligible_v8_case_before_transfer_registration",
"stream_install_admission_separates_identity_capability_and_valid_payload",
]
missing = [name for name in required if f"{name} ... ok" not in output]
if missing:
@@ -657,21 +937,25 @@ class Runner:
# exceed one CHUNK_SIZE. A 2-chunk file could never trip the check.
file_size = CHUNK_SIZE * 4
create_large_sparse_game(source_dir / game_id, size=file_size)
alpha = self.peer("s14-alpha", games_dir=source_dir)
stage = self.peer("s14-stage")
catalog_dir = self.fixture_catalog(
"s14", source_dir / game_id, streamed_install=False
)
alpha = self.peer("s14-alpha", games_dir=source_dir, catalog_dir=catalog_dir)
stage = self.peer("s14-stage", catalog_dir=catalog_dir)
connect_many(stage, [alpha])
waiter = LineWaiter(len(stage.output))
stage.send({"cmd": "download", "game_id": game_id, "install": False})
stage.wait_for(event_is("download-finished", game_id), timeout=180, description="stage finish", waiter=waiter)
diff_game_dirs(source_dir / game_id, stage.host_games_dir / game_id)
client = self.peer("s14-client")
client = self.peer("s14-client", catalog_dir=catalog_dir)
connect_many(client, [alpha, stage])
wait_remote_game(client, game_id, peer_count=2, version=PERF_GAME_VERSION)
waiter = LineWaiter(len(client.output))
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": game_id, "install": False})
client.wait_for(event_is("download-finished", game_id), timeout=180, description="client finish", waiter=waiter)
diff_game_dirs(source_dir / game_id, client.host_games_dir / game_id)
totals = chunk_totals(client, game_id, f"{game_id}/{game_id}.eti")
totals = chunk_totals(client, game_id, f"{game_id}.eti")
if len(totals) != 2:
raise ScenarioError(f"expected .eti chunks from exactly two peers, got {totals}")
if sum(totals.values()) != file_size:
@@ -682,34 +966,55 @@ class Runner:
return f"{game_id} ({file_size // (1024 * 1024)} MiB) split across two sources, balanced, diff matched: {totals}"
def s15_three_way_version_skew(self) -> str:
specs = [
("s15-a", "20150101"),
("s15-b", "20160101"),
("s15-c", CATALOG_VERSIONS["cnc4"]),
stale_dir = self.fixture_root / "s15-a"
wrong_content_dir = self.fixture_root / "s15-b"
exact_dir = self.fixture_root / "s15-c"
copy_game("cnc4", stale_dir, version="20150101")
copy_game("cnc4", wrong_content_dir, version=CATALOG_VERSIONS["cnc4"])
with (wrong_content_dir / "cnc4" / "cnc4.eti").open("ab") as archive:
archive.write(b"wrong-content-id")
wrong_catalog = self.fixture_catalog(
"s15-wrong-content",
wrong_content_dir / "cnc4",
streamed_install=False,
)
copy_game("cnc4", exact_dir, version=CATALOG_VERSIONS["cnc4"])
peers = [
self.peer("s15-a", games_dir=stale_dir),
self.peer(
"s15-b", games_dir=wrong_content_dir, catalog_dir=wrong_catalog
),
self.peer("s15-c", games_dir=exact_dir),
]
peers = []
for name, version in specs:
game_dir = self.fixture_root / name
copy_game("cnc4", game_dir, version=version)
peers.append(self.peer(name, games_dir=game_dir))
client = self.peer("s15-client")
connect_many(client, peers)
expected_content_id = catalog_content_id(client, "cnc4")
wrong_content_id = catalog_content_id(peers[1], "cnc4")
if wrong_content_id == expected_content_id:
raise ScenarioError("modified s15 fixture did not change ContentId")
wait_peer_without_game(client, peers[0].peer_id, "cnc4")
wait_peer_has_game(client, peers[1].peer_id, "cnc4", wrong_content_id)
wait_peer_has_game(client, peers[2].peer_id, "cnc4", expected_content_id)
wait_remote_library_content(
client, "cnc4", wrong_content_id, peer_count=1
)
wait_remote_library_content(
client, "cnc4", expected_content_id, peer_count=1
)
wait_remote_game(client, "cnc4", peer_count=1, version=CATALOG_VERSIONS["cnc4"])
# Cross-check the RAW advertised versions via list-peers (not the
# catalog-synthesized list-games field), proving the three peers really
# differ and that only the catalog-version peer is aggregated.
for peer, expected_version in zip(peers, ["20150101", "20160101", CATALOG_VERSIONS["cnc4"]]):
advertised = peer_advertised_version(client, peer.peer_id, "cnc4")
if advertised != expected_version:
raise ScenarioError(
f"{peer.name} advertised cnc4 version {advertised}, expected {expected_version}"
)
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "cnc4", "install": False})
client.wait_for(event_is("download-finished", "cnc4"), timeout=60, description="cnc4 finish", waiter=waiter)
assert_only_chunk_sources(client, "cnc4", {peers[2].ready_addr})
assert_chunk_authority(
client, "cnc4", expected_content_id, {peers[2].peer_id}
)
diff_game_dirs(peers[2].host_games_dir / "cnc4", client.host_games_dir / "cnc4")
return "three-way skew exposed only the catalog-version peer and receiver diffed cleanly"
return (
"stale-version peer advertised nothing; wrong-ContentId peer stayed in the raw "
"view but was excluded; every verified chunk came from the exact catalog peer"
)
def s16_catalog_fanout_with_stale(self) -> str:
specs = [
@@ -717,7 +1022,7 @@ class Runner:
("s16-b", CATALOG_VERSIONS["alienswarm"]),
("s16-c", CATALOG_VERSIONS["alienswarm"]),
]
peers = []
peer_dirs = []
for name, version in specs:
game_dir = self.fixture_root / name
copy_game("alienswarm", game_dir, version=version)
@@ -725,127 +1030,248 @@ class Runner:
# catalog-version peers; the stock 120 MiB fixture is a single chunk
# that can only ever come from one source.
inflate_archive_sparse(game_dir / "alienswarm", "alienswarm", CHUNK_SIZE * 2)
peers.append(self.peer(name, games_dir=game_dir))
client = self.peer("s16-client")
peer_dirs.append((name, game_dir))
catalog_dir = self.fixture_catalog(
"s16", peer_dirs[1][1] / "alienswarm", streamed_install=False
)
peers = [
self.peer(name, games_dir=game_dir, catalog_dir=catalog_dir)
for name, game_dir in peer_dirs
]
client = self.peer("s16-client", catalog_dir=catalog_dir)
connect_many(client, peers)
wait_remote_game(client, "alienswarm", peer_count=2, version=CATALOG_VERSIONS["alienswarm"])
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "alienswarm", "install": False})
client.wait_for(event_is("download-finished", "alienswarm"), timeout=180, description="alienswarm finish", waiter=waiter)
assert_only_chunk_sources(client, "alienswarm", {peers[1].ready_addr, peers[2].ready_addr})
totals = chunk_totals(client, "alienswarm", "alienswarm/alienswarm.eti")
if peers[0].ready_addr in totals:
assert_only_chunk_sources(client, "alienswarm", {peers[1].peer_id, peers[2].peer_id})
totals = chunk_totals(client, "alienswarm", "alienswarm.eti")
if peers[0].peer_id in totals:
raise ScenarioError(f"stale peer contributed chunks: {totals}")
if set(totals) != {peers[1].ready_addr, peers[2].ready_addr}:
if set(totals) != {peers[1].peer_id, peers[2].peer_id}:
raise ScenarioError(f"expected .eti to fan out across both B and C, got {totals}")
diff_game_dirs(peers[1].host_games_dir / "alienswarm", client.host_games_dir / "alienswarm")
return f"catalog-version B/C peers split alienswarm.eti while stale A contributed zero; totals={totals}"
def s17_catalog_conflict_rejection(self) -> str:
specs = [
("s17-a", "20150101", False),
("s17-b", CATALOG_VERSIONS["cnc4"], False),
("s17-c", CATALOG_VERSIONS["cnc4"], True),
]
peers = []
for name, version, conflict in specs:
game_dir = self.fixture_root / name
copy_game("cnc4", game_dir, version=version)
if conflict:
with (game_dir / "cnc4" / "cnc4.eti").open("ab") as handle:
handle.write(b"conflict")
peers.append(self.peer(name, games_dir=game_dir))
def s17_catalog_byte_quarantine(self) -> str:
game_id = "cnc4"
source_a_dir = self.fixture_root / "s17-source-a"
source_b_dir = self.fixture_root / "s17-source-b"
copy_game(game_id, source_a_dir)
copy_game(game_id, source_b_dir)
source_a = self.peer("s17-source-a", games_dir=source_a_dir)
source_b = self.peer("s17-source-b", games_dir=source_b_dir)
bad, good = sorted(
[source_a, source_b], key=lambda peer: peer_id_sort_key(peer.peer_id)
)
corrupt_file_same_length(bad.host_games_dir / game_id / f"{game_id}.eti")
client = self.peer("s17-client")
connect_many(client, peers)
connect_many(client, [bad, good])
content_id = catalog_content_id(client, game_id)
wait_peer_has_game(client, bad.peer_id, game_id, content_id)
wait_peer_has_game(client, good.peer_id, game_id, content_id)
wait_remote_library_content(client, game_id, content_id, peer_count=2)
wait_remote_game(client, "cnc4", peer_count=2, version=CATALOG_VERSIONS["cnc4"])
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "cnc4", "install": False})
client.wait_for(event_is("download-failed", "cnc4"), timeout=30, description="cnc4 failed", waiter=waiter)
assert_not_exists(client.host_games_dir / "cnc4" / "version.ini")
return "catalog-version file conflict failed download and left no committed version.ini"
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": game_id, "install": False})
client.wait_for(
event_is("download-finished", game_id),
timeout=60,
description="cnc4 retry after corrupt source",
waiter=waiter,
)
assert_download_status_trace(
client,
game_id,
start=start,
terminal="download-finished",
expect_invalid_source_retry=True,
)
assert_chunk_authority(client, game_id, content_id, {bad.peer_id, good.peer_id})
archive_totals = chunk_totals(client, game_id, f"{game_id}.eti")
if set(archive_totals) != {good.peer_id}:
raise ScenarioError(
"corrupt source supplied a successful archive chunk instead of being quarantined: "
f"{archive_totals}"
)
assert_no_duplicate_chunks(client, game_id)
diff_game_dirs(good.host_games_dir / game_id, client.host_games_dir / game_id)
# A fresh receiver has no runtime quarantine history. With only the
# same-length corrupt source available, catalog verification must fail
# and must not publish either the sentinel or local install state.
bad_only = self.peer("s17-bad-only-client")
connect_many(bad_only, [bad])
wait_peer_has_game(bad_only, bad.peer_id, game_id, content_id)
wait_remote_library_content(bad_only, game_id, content_id, peer_count=1)
wait_remote_game(bad_only, game_id, peer_count=1)
failed_start = len(bad_only.output)
failed_waiter = LineWaiter(failed_start)
bad_only.send({"cmd": "download", "game_id": game_id, "install": False})
bad_only.wait_for(
event_is("download-failed", game_id),
timeout=60,
description="all-bad cnc4 catalog failure",
waiter=failed_waiter,
)
assert_download_status_trace(
bad_only,
game_id,
start=failed_start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
)
assert_no_event_since(bad_only, failed_start, "download-finished", game_id)
assert_local_absent(bad_only, game_id)
assert_not_exists(bad_only.host_games_dir / game_id / "version.ini")
assert_not_exists(bad_only.host_games_dir / game_id / "local")
return (
"same-length corrupt source was quarantined and retried via the honest source; "
"a fresh all-bad receiver failed without version.ini or local/"
)
def s18_redundant_source_drop(self) -> str:
game_id = "bf1942"
source_a_dir = self.fixture_root / "s18-a"
source_b_dir = self.fixture_root / "s18-b"
# Multi-chunk sparse archive so BOTH peers are assigned .eti chunks; a
# single-chunk file could be served entirely by one peer, so killing the
# other would prove nothing. We verify the download SURVIVES a mid-download
# Multi-chunk sparse archive so BOTH peers have more assigned .eti chunks
# than the per-peer stream window; a single-window file could finish before
# the force-kill lands. We verify the download SURVIVES a mid-download
# source kill (every byte still arrives, no download-failed, diff matches).
# Retry-onto-survivor is the mechanism that makes this work and is
# exercised whenever the kill interrupts an unfinished chunk, but the race
# against `docker rm -f` means we cannot deterministically force it, so we
# do not assert it. Sparse zero bytes are identical, so duplicate-source
# majority validation still agrees.
file_size = CHUNK_SIZE * 4
# Waiting for the lower-ID source to fill the stream window makes the
# interruption non-vacuous; the final lower-ID-planned offset must then be
# completed by the survivor. Sparse zero bytes are identical, so both
# sources match the one publisher-derived catalog authority.
file_size = CHUNK_SIZE * 16
create_large_sparse_game(source_a_dir / game_id, size=file_size)
create_large_sparse_game(source_b_dir / game_id, size=file_size)
source_a = self.peer("s18-a", games_dir=source_a_dir)
source_b = self.peer("s18-b", games_dir=source_b_dir)
client = self.peer("s18-client")
catalog_dir = self.fixture_catalog(
"s18", source_a_dir / game_id, streamed_install=False
)
source_a = self.peer("s18-a", games_dir=source_a_dir, catalog_dir=catalog_dir)
source_b = self.peer("s18-b", games_dir=source_b_dir, catalog_dir=catalog_dir)
client = self.peer("s18-client", catalog_dir=catalog_dir)
connect_many(client, [source_a, source_b])
content_id = catalog_content_id(client, game_id)
wait_peer_has_game(client, source_a.peer_id, game_id, content_id)
wait_peer_has_game(client, source_b.peer_id, game_id, content_id)
wait_remote_library_content(client, game_id, content_id, peer_count=2)
wait_remote_game(client, game_id, peer_count=2)
killed, survivor = sorted(
[source_a, source_b], key=lambda peer: peer_id_sort_key(peer.peer_id)
)
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": game_id, "install": False})
client.wait_for(event_is("download-begin", game_id), timeout=20, description="download begin", waiter=waiter)
source_a.kill()
active_streams = wait_outbound_transfer(
killed, game_id, minimum=4, timeout=30
)
assert_no_event_since(client, start, "download-finished", game_id)
assert_no_event_since(client, start, "download-failed", game_id)
killed.kill()
client.wait_for(event_is("download-finished", game_id), timeout=180, description="download finish", waiter=waiter)
assert_download_status_trace(
client,
game_id,
start=start,
terminal="download-finished",
)
# Scan the WHOLE download window (start..) for download-failed. The old
# assert_no_event reused a waiter already advanced past download-finished,
# so it only saw the empty tail and could never fire.
assert_no_event_since(client, start, "download-failed", game_id)
diff_game_dirs(source_b_dir / game_id, client.host_games_dir / game_id)
totals = chunk_totals(client, game_id, f"{game_id}/{game_id}.eti")
if source_b.ready_addr not in totals:
diff_game_dirs(survivor.host_games_dir / game_id, client.host_games_dir / game_id)
assert_chunk_authority(
client, game_id, content_id, {source_a.peer_id, source_b.peer_id}
)
totals = chunk_totals(client, game_id, f"{game_id}.eti")
survivor_id = survivor.peer_id
if survivor_id is None:
raise ScenarioError("surviving source has no authenticated PeerId")
if survivor_id not in totals:
raise ScenarioError(f"surviving source served no .eti chunks: {totals}")
if sum(totals.values()) != file_size:
raise ScenarioError(f"download did not deliver the whole archive ({sum(totals.values())} != {file_size}): {totals}")
# Exact sources are ordered by PeerId. The lower-ID peer receives every
# even archive chunk, so this last even offset was originally assigned to
# `killed`; seeing the survivor complete it proves retry, not just its own
# initial half of the plan.
retried_offset = file_size - (2 * CHUNK_SIZE)
if not any(
item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == game_id
and item["data"].get("relative_path") == f"{game_id}.eti"
and item["data"].get("offset") == retried_offset
and item["data"].get("peer_id") == survivor_id
for item in client.output[start:]
):
raise ScenarioError(
"survivor did not retry the killed source's final planned archive chunk: "
f"offset={retried_offset} totals={totals}"
)
# We deliberately do NOT assert the exact per-source split. The killed
# source can serve a chunk or two before `docker rm -f` lands (a fast-LAN
# race), so requiring totals == {survivor: file_size} would be flaky. The
# robust proof of redundancy is that a source died mid-download yet every
# byte still arrived (sum == file_size plus the diff), the survivor served
# part of it, and no download-failed was emitted.
survivor_bytes = totals[source_b.ready_addr]
survivor_bytes = totals[survivor_id]
return (
f"source killed after begin; all {file_size} bytes delivered "
f"lower-PeerId source killed with {active_streams} outbound streams; "
f"all {file_size} bytes delivered "
f"({survivor_bytes} from the survivor), no download-failed; diff matched; bytes={totals}"
)
def s19_sole_source_drop(self) -> str:
game_id = "bf1942"
source_dir = self.fixture_root / "s19-source"
# Multi-chunk sparse archive force-killed right after download-begin. A
# 120 MiB single-chunk file served by a graceful shutdown could finish
# (~0.15s at LAN speed) before the drop landed, flipping the expected
# failure into a download-finished. With four 128 MiB chunks and a
# forceful kill issued right after download-begin, the source dies with
# the bulk of the transfer still outstanding (an individual chunk may
# complete first, but the full download cannot), so a terminal failure is
# deterministic.
create_large_sparse_game(source_dir / game_id, size=CHUNK_SIZE * 4)
source = self.peer("s19-source", games_dir=source_dir)
client = self.peer("s19-client")
# More chunks than the per-peer stream window ensure pending work remains
# when the force-kill interrupts four positively observed active streams.
create_large_sparse_game(source_dir / game_id, size=CHUNK_SIZE * 8)
catalog_dir = self.fixture_catalog(
"s19", source_dir / game_id, streamed_install=False
)
source = self.peer("s19-source", games_dir=source_dir, catalog_dir=catalog_dir)
client = self.peer("s19-client", catalog_dir=catalog_dir)
connect_many(client, [source])
wait_remote_game(client, game_id, peer_count=1)
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": game_id, "install": False})
client.wait_for(event_is("download-begin", game_id), timeout=20, description="download begin", waiter=waiter)
# Forceful kill (no Goodbye) drops the connection mid-transfer.
active_streams = wait_outbound_transfer(
source, game_id, minimum=4, timeout=30
)
assert_no_event_since(client, start, "download-finished", game_id)
assert_no_event_since(client, start, "download-failed", game_id)
# Forceful kill drops positively active streams without a departure signal.
source.kill()
terminal = client.wait_for(
event_name_in({"download-failed", "download-peers-gone"}, game_id),
event_is("download-failed", game_id),
timeout=120,
description="sole-source drop terminal failure",
waiter=waiter,
)
assert_download_status_trace(
client,
game_id,
start=start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
)
assert_no_event_since(client, start, "download-finished", game_id)
assert_not_exists(client.host_games_dir / game_id / "version.ini")
assert_no_active(client, game_id)
wait_no_active(client, game_id)
assert_local_absent(client, game_id)
return f"sole-source forceful drop mid-transfer -> {terminal['event']}; version.ini absent; no ready row; no active op"
return (
f"sole-source forceful drop with {active_streams} outbound streams -> "
f"{terminal['event']}; version.ini absent; no ready row; no active op"
)
def s20_receiver_write_failure(self) -> str:
source_dir = self.fixture_root / "s20-source"
@@ -854,9 +1280,17 @@ class Runner:
client = self.peer("s20-client", tmpfs_size="32m")
connect_many(client, [source])
wait_remote_game(client, "alienswarm", peer_count=1)
waiter = LineWaiter(len(client.output))
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": "alienswarm", "install": False})
client.wait_for(event_is("download-failed", "alienswarm"), timeout=90, description="download failed", waiter=waiter)
assert_download_status_trace(
client,
"alienswarm",
start=start,
terminal="download-failed",
reason="operation-failed",
)
client.docker_exec("test", "!", "-e", "/games/alienswarm/version.ini")
assert_no_active(client, "alienswarm")
return "32m tmpfs receiver emitted download-failed; /games/alienswarm/version.ini absent; active operations empty"
@@ -866,7 +1300,7 @@ class Runner:
bravo_dir = self.fixture_root / "s21-bravo"
bravo_dir.mkdir(parents=True, exist_ok=True)
bravo = self.peer("s21-bravo", games_dir=bravo_dir)
connect_many(alpha, [bravo])
connect_bidirectional(alpha, bravo)
assert len(alpha.list_peers()) == 1
stage_game_drop(bravo_dir, "cod5")
game = wait_remote_game(alpha, "cod5", peer_count=1)
@@ -877,7 +1311,7 @@ class Runner:
bravo_dir = self.fixture_root / "s22-bravo"
copy_game("cod5", bravo_dir)
bravo = self.peer("s22-bravo", games_dir=bravo_dir)
connect_many(alpha, [bravo])
connect_bidirectional(alpha, bravo)
wait_remote_game(alpha, "cod5", peer_count=1)
shutil.rmtree(bravo_dir / "cod5")
wait_remote_absent(alpha, "cod5")
@@ -891,7 +1325,7 @@ class Runner:
bravo_dir = self.fixture_root / "s23-bravo"
copy_game("cnc4", bravo_dir, version="20160101")
bravo = self.peer("s23-bravo", games_dir=bravo_dir)
connect_many(alpha, [bravo])
connect_bidirectional(alpha, bravo)
wait_remote_absent(alpha, "cnc4", timeout=5)
(bravo_dir / "cnc4" / "version.ini").write_text(CATALOG_VERSIONS["cnc4"], encoding="utf-8")
wait_remote_game(alpha, "cnc4", peer_count=1, version=CATALOG_VERSIONS["cnc4"])
@@ -922,11 +1356,23 @@ class Runner:
source = self.peer("s25-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
client = self.peer("s25-client")
connect_many(client, [source])
waiter = LineWaiter(len(client.output))
start = len(client.output)
bfbc2_waiter = LineWaiter(start)
cnctw_waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": "bfbc2", "install": False})
client.send({"cmd": "download", "game_id": "cnctw", "install": False})
client.wait_for(event_is("download-finished", "bfbc2"), timeout=60, description="bfbc2 finish", waiter=waiter)
client.wait_for(event_is("download-finished", "cnctw"), timeout=60, description="cnctw finish", waiter=waiter)
client.wait_for(
event_is("download-finished", "bfbc2"),
timeout=60,
description="bfbc2 finish",
waiter=bfbc2_waiter,
)
client.wait_for(
event_is("download-finished", "cnctw"),
timeout=60,
description="cnctw finish",
waiter=cnctw_waiter,
)
diff_game_dirs(FIXTURES / "fixture-bravo" / "bfbc2", client.host_games_dir / "bfbc2")
diff_game_dirs(FIXTURES / "fixture-bravo" / "cnctw", client.host_games_dir / "cnctw")
wait_local_game(client, "bfbc2", downloaded=True, installed=False)
@@ -941,8 +1387,11 @@ class Runner:
# operation rather than a near-instant 3 MB transfer that may already be
# finished by the time the second command is read.
create_large_sparse_game(source_dir / game_id, size=CHUNK_SIZE * 2)
source = self.peer("s26-source", games_dir=source_dir)
client = self.peer("s26-client")
catalog_dir = self.fixture_catalog(
"s26", source_dir / game_id, streamed_install=False
)
source = self.peer("s26-source", games_dir=source_dir, catalog_dir=catalog_dir)
client = self.peer("s26-client", catalog_dir=catalog_dir)
connect_many(client, [source])
wait_remote_game(client, game_id, peer_count=1)
waiter = LineWaiter(len(client.output))
@@ -975,7 +1424,9 @@ class Runner:
def s27_self_connect_rejection(self) -> str:
alpha = self.peer("s27-alpha")
err = alpha.send({"cmd": "connect", "addr": alpha.ready_addr}, expect_error=True)
err = alpha.send(
{"cmd": "connect", **alpha.ready_endpoint()}, expect_error=True
)
if "cannot connect peer to itself" not in err["error"]:
raise ScenarioError(f"unexpected self-connect error: {err}")
peers = alpha.list_peers()
@@ -984,29 +1435,59 @@ class Runner:
alpha.status()
return f"self-connect errored '{err['error']}'; peer list stayed empty"
def s28_address_change_unit(self) -> str:
def s28_reconnect_generation_unit(self) -> str:
output = run_just_test()
test_name = "peer_db::tests::address_update_preserves_peer_identity_and_library"
test_name = (
"peer_db::tests::"
"reconnect_gets_new_generation_and_stale_removal_loses_authority"
)
if f"{test_name} ... ok" not in output:
raise ScenarioError(f"S28 unit proof did not run-and-pass in just test output:\n{output}")
return "`just test` passed including address_update_preserves_peer_identity_and_library"
return (
"`just test` passed including reconnect generation fencing against "
"stale removal"
)
def s29_empty_peer_participates(self) -> str:
source = self.peer("s29-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
source_network = self.create_scenario_network(
"s29-source", internal=True
)
source = self.peer(
"s29-alpha",
network=source_network,
games_dir=FIXTURES / "fixture-alpha",
readonly_games=True,
)
empty = self.peer("s29-empty")
observer = self.peer("s29-observer")
connect_many(observer, [empty])
connect_bidirectional(observer, empty)
peers = observer.list_peers()
if len(peers) != 1 or peers[0]["game_count"] != 0:
if (
{peer.get("peer_id") for peer in peers} != {empty.peer_id}
or len(peers) != 1
or peers[0]["game_count"] != 0
):
raise ScenarioError(f"expected empty peer with zero games, got {peers}")
connect_many(empty, [source])
self.connect_peer_to_network(empty, source_network)
empty.connect_to(source)
content_id = catalog_content_id(source, "alienswarm")
wait_peer_has_game(empty, source.peer_id, "alienswarm", content_id)
wait_remote_library_content(empty, "alienswarm", content_id, peer_count=1)
wait_remote_game(empty, "alienswarm", peer_count=1)
waiter = LineWaiter(len(empty.output))
empty.send({"cmd": "download", "game_id": "alienswarm", "install": False})
empty.wait_for(event_is("download-finished", "alienswarm"), timeout=90, description="empty download finish", waiter=waiter)
assert_chunk_authority(empty, "alienswarm", content_id, {source.peer_id})
diff_game_dirs(FIXTURES / "fixture-alpha" / "alienswarm", empty.host_games_dir / "alienswarm")
wait_peer_has_game(observer, empty.peer_id, "alienswarm")
return "observer saw zero-game peer; empty downloaded alienswarm, diff matched, then observer's snapshot for that peer contained alienswarm"
wait_peer_has_game(observer, empty.peer_id, "alienswarm", content_id)
wait_remote_library_content(observer, "alienswarm", content_id, peer_count=1)
wait_remote_game(observer, "alienswarm", peer_count=1)
assert_exact_peer_ids(observer, {empty.peer_id})
return (
"observer remained authenticated only to the formerly empty peer, then saw "
"its exact alienswarm ContentId with peer_count=1 after the peer downloaded "
"directly from an isolated source; raw snapshot, remote view, and chunks agreed"
)
def s30_mesh_aggregation(self) -> str:
dirs = []
@@ -1045,21 +1526,33 @@ class Runner:
return f"client aggregated {len(expected)} IDs from 5 peers with expected peer_count/catalog versions"
def s31_bootstrapped_peer_source(self) -> str:
game_id = "alienswarm"
source = self.peer("s31-alpha", games_dir=FIXTURES / "fixture-alpha", readonly_games=True)
bootstrap = self.peer("s31-bootstrap")
connect_many(bootstrap, [source])
content_id = catalog_content_id(bootstrap, game_id)
wait_peer_has_game(bootstrap, source.peer_id, game_id, content_id)
waiter = LineWaiter(len(bootstrap.output))
bootstrap.send({"cmd": "download", "game_id": "alienswarm", "install": False})
bootstrap.wait_for(event_is("download-finished", "alienswarm"), timeout=90, description="bootstrap finish", waiter=waiter)
diff_game_dirs(FIXTURES / "fixture-alpha" / "alienswarm", bootstrap.host_games_dir / "alienswarm")
bootstrap.send({"cmd": "download", "game_id": game_id, "install": False})
bootstrap.wait_for(event_is("download-finished", game_id), timeout=90, description="bootstrap finish", waiter=waiter)
assert_chunk_authority(bootstrap, game_id, content_id, {source.peer_id})
diff_game_dirs(FIXTURES / "fixture-alpha" / game_id, bootstrap.host_games_dir / game_id)
source.kill()
third = self.peer("s31-third")
connect_many(third, [bootstrap])
wait_peer_has_game(third, bootstrap.peer_id, game_id, content_id)
wait_remote_library_content(third, game_id, content_id, peer_count=1)
assert_exact_peer_ids(third, {bootstrap.peer_id})
waiter = LineWaiter(len(third.output))
third.send({"cmd": "download", "game_id": "alienswarm", "install": False})
third.wait_for(event_is("download-finished", "alienswarm"), timeout=90, description="third finish", waiter=waiter)
diff_game_dirs(FIXTURES / "fixture-alpha" / "alienswarm", third.host_games_dir / "alienswarm")
return "third peer downloaded from bootstrapped client after original source kill; diff matched original"
third.send({"cmd": "download", "game_id": game_id, "install": False})
third.wait_for(event_is("download-finished", game_id), timeout=90, description="third finish", waiter=waiter)
assert_chunk_authority(third, game_id, content_id, {bootstrap.peer_id})
diff_game_dirs(FIXTURES / "fixture-alpha" / game_id, third.host_games_dir / game_id)
return (
"bootstrap downloaded exact content from the original source, then served "
"every verified chunk to a third peer after the original source was killed; "
"diff matched original"
)
def s32_reinstall_after_uninstall(self) -> str:
source = self.peer("s32-bravo", games_dir=FIXTURES / "fixture-bravo", readonly_games=True)
@@ -1108,8 +1601,11 @@ class Runner:
def s34_many_small_files(self) -> str:
source_dir = self.fixture_root / "s34-source"
create_many_small_game(source_dir / "bf1942")
source = self.peer("s34-source", games_dir=source_dir)
client = self.peer("s34-client")
catalog_dir = self.fixture_catalog(
"s34", source_dir / "bf1942", streamed_install=True
)
source = self.peer("s34-source", games_dir=source_dir, catalog_dir=catalog_dir)
client = self.peer("s34-client", catalog_dir=catalog_dir)
connect_many(client, [source])
wait_remote_game(client, "bf1942", peer_count=1)
waiter = LineWaiter(len(client.output))
@@ -1134,19 +1630,32 @@ class Runner:
return f"20 small files plus version.ini each transferred as one coherent chunk; diff matched; chunk events={len(chunks)}"
def s35_unknown_game_filtered(self) -> str:
source = self.peer("s35-source", fixtures=["mystery-game"])
game_id = "cod2"
source = self.peer(
"s35-source",
games_dir=FIXTURES / "fixture-unknown",
readonly_games=True,
catalog_dir=CATALOG_PROFILES / "unknown",
)
client = self.peer("s35-client")
connect_many(client, [source])
# Establish the premise: the source really does advertise mystery-game in
# its raw library snapshot. Without this, wait_remote_absent could pass
# vacuously ("absent because never sent" vs "absent because filtered").
wait_peer_has_game(client, source.peer_id, "mystery-game")
wait_remote_absent(client, "mystery-game")
err = client.send({"cmd": "download", "game_id": "mystery-game", "install": False}, expect_error=True)
# The source has an honest, source-only catalog for cod2. The client uses
# the default reduced profile, where cod2 is intentionally unknown.
content_id = catalog_content_id(source, game_id)
wait_peer_has_game(client, source.peer_id, game_id, content_id)
wait_remote_library_content(client, game_id, content_id, peer_count=1)
wait_remote_absent(client, game_id)
err = client.send(
{"cmd": "download", "game_id": game_id, "install": False},
expect_error=True,
)
if "not in the local catalog" not in err["error"]:
raise ScenarioError(f"unexpected unknown game error: {err}")
assert_not_exists(client.host_games_dir / "mystery-game")
return f"unknown game absent from list-games; download errored '{err['error']}'; no local files"
assert_not_exists(client.host_games_dir / game_id)
return (
f"raw v8 availability retained unknown {game_id}/{content_id}, but the local "
f"catalog excluded it from list-games; download errored '{err['error']}'"
)
def s36_catalog_singleton(self) -> str:
peers = []
@@ -1157,35 +1666,40 @@ class Runner:
peers.append(self.peer(f"s36-{index}", games_dir=game_dir))
client = self.peer("s36-client")
connect_many(client, peers)
content_id = catalog_content_id(client, "cnc4")
wait_peer_has_game(client, peers[0].peer_id, "cnc4", content_id)
for stale in peers[1:]:
wait_peer_without_game(client, stale.peer_id, "cnc4")
wait_remote_library_content(client, "cnc4", content_id, peer_count=1)
wait_remote_game(client, "cnc4", peer_count=1, version=CATALOG_VERSIONS["cnc4"])
waiter = LineWaiter(len(client.output))
client.send({"cmd": "download", "game_id": "cnc4", "install": False})
got = client.wait_for(event_is("got-game-files", "cnc4"), timeout=20, description="got game files", waiter=waiter)
client.wait_for(event_is("download-finished", "cnc4"), timeout=60, description="download finish", waiter=waiter)
catalog_addr = peers[0].ready_addr
if catalog_addr is None:
raise ScenarioError("catalog-version peer had no ready addr")
for item in client.output:
if item.get("type") != "event" or item.get("event") != "download-chunk-finished":
continue
data = item["data"]
if data.get("game_id") == "cnc4" and data.get("peer_addr") != catalog_addr:
raise ScenarioError(f"stale peer contributed chunk: {data}")
assert_chunk_authority(client, "cnc4", content_id, {peers[0].peer_id})
diff_game_dirs(peers[0].host_games_dir / "cnc4", client.host_games_dir / "cnc4")
descs = got["data"]["file_descriptions"]
if not descs:
raise ScenarioError("got-game-files had no descriptors")
return "client reported singleton catalog-version peer; stale peers stayed hidden and sent no chunks; diff matched"
return (
"catalog manifest, raw availability, remote view, and every verified chunk used "
"one exact ContentId/source; four stale peers advertised no transferable game"
)
def s37_single_source_download_throughput(self) -> str:
source_dir = self.fixture_root / "s37-source"
create_large_sparse_game(source_dir / PERF_GAME_ID, size=PERF_GAME_SIZE)
source = self.peer("s37-source", games_dir=source_dir)
client = self.peer("s37-client")
catalog_dir = self.fixture_catalog(
"s37", source_dir / PERF_GAME_ID, streamed_install=False
)
source = self.peer("s37-source", games_dir=source_dir, catalog_dir=catalog_dir)
client = self.peer("s37-client", catalog_dir=catalog_dir)
connect_many(client, [source])
content_id = catalog_content_id(client, PERF_GAME_ID)
wait_peer_has_game(client, source.peer_id, PERF_GAME_ID, content_id)
wait_remote_library_content(
client, PERF_GAME_ID, content_id, peer_count=1
)
wait_remote_game(client, PERF_GAME_ID, peer_count=1, version=PERF_GAME_VERSION)
waiter = LineWaiter(len(client.output))
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "download", "game_id": PERF_GAME_ID, "install": False})
finished = client.wait_for(
event_is("download-finished", PERF_GAME_ID),
@@ -1193,11 +1707,21 @@ class Runner:
description=f"{PERF_GAME_ID} throughput download",
waiter=waiter,
)
assert_download_status_trace(
client,
PERF_GAME_ID,
start=start,
terminal="download-finished",
)
destination_archive = client.host_games_dir / PERF_GAME_ID / f"{PERF_GAME_ID}.eti"
if destination_archive.stat().st_size != PERF_GAME_SIZE:
raise ScenarioError(
f"downloaded archive size mismatch: {destination_archive.stat().st_size} != {PERF_GAME_SIZE}"
)
diff_game_dirs(
source_dir / PERF_GAME_ID,
client.host_games_dir / PERF_GAME_ID,
)
throughput = finished.get("data", {}).get("throughput")
if not throughput:
@@ -1208,6 +1732,89 @@ class Runner:
f"throughput byte count mismatch: {throughput['bytes']} != {expected_bytes}"
)
chunks = [
item["data"]
for item in client.output[start:]
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == PERF_GAME_ID
]
if len(chunks) != 17 or int(throughput["chunks"]) != 17:
raise ScenarioError(
"expected exactly 17 verified chunks "
f"(16 archive + version.ini), got events={len(chunks)} throughput={throughput['chunks']}"
)
assert_chunk_authority(client, PERF_GAME_ID, content_id, {source.peer_id})
archive_path = f"{PERF_GAME_ID}.eti"
expected_paths = {archive_path, "version.ini"}
actual_paths = {item.get("relative_path") for item in chunks}
if actual_paths != expected_paths:
raise ScenarioError(
f"verified chunk paths are not exact: {actual_paths} != {expected_paths}"
)
archive_chunks = [item for item in chunks if item.get("relative_path") == archive_path]
version_chunks = [
item
for item in chunks
if item.get("relative_path") == "version.ini"
]
expected_archive_chunks = [
(archive_path, index * CHUNK_SIZE, CHUNK_SIZE) for index in range(16)
]
actual_archive_chunks = sorted(
(
item.get("relative_path"),
int(item["offset"]),
int(item["length"]),
)
for item in archive_chunks
)
if actual_archive_chunks != expected_archive_chunks:
raise ScenarioError(
"archive chunk path/offset/length tuples are not exact: "
f"{actual_archive_chunks}"
)
expected_version_chunk = [("version.ini", 0, 8)]
actual_version_chunks = [
(
item.get("relative_path"),
int(item["offset"]),
int(item["length"]),
)
for item in version_chunks
]
if len(PERF_GAME_VERSION.encode("utf-8")) != 8:
raise ScenarioError(f"performance fixture version is not 8 bytes: {PERF_GAME_VERSION!r}")
if actual_version_chunks != expected_version_chunk:
raise ScenarioError(f"version.ini was not one exact chunk: {version_chunks}")
assert_no_duplicate_chunks(client, PERF_GAME_ID)
terminal_events = [
item.get("event")
for item in client.output[start:]
if item.get("type") == "event"
and item.get("data", {}).get("game_id") == PERF_GAME_ID
and item.get("event")
in {"download-finished", "download-failed"}
]
if terminal_events != ["download-finished"]:
raise ScenarioError(
f"expected exactly one successful terminal event: {terminal_events}"
)
game = wait_local_game(
client,
PERF_GAME_ID,
downloaded=True,
installed=False,
)
assert_game_state(
game,
downloaded=True,
installed=False,
availability="Ready",
)
wait_no_active(client, PERF_GAME_ID)
# The byte count alone never exercises the rate math. Validate the rate
# fields are positive and mutually consistent so a units/divisor bug
# (MiB vs MB, off-by-1000, zero duration) cannot slip through.
@@ -1228,12 +1835,20 @@ class Runner:
f"mbit/mib ratio {ratio} != expected {expected_ratio}: {throughput}"
)
minimum_mib_per_s = float(os.environ.get("LANSPREAD_S37_MIN_MIB_PER_S", "100"))
if throughput["mib_per_s"] < minimum_mib_per_s:
raise ScenarioError(
f"throughput {throughput['mib_per_s']:.2f} MiB/s is below configured "
f"minimum {minimum_mib_per_s:.2f} MiB/s"
)
return (
f"{PERF_GAME_ID} {format_bytes(PERF_GAME_SIZE)} single-source download: "
f"{throughput['mib_per_s']:.2f} MiB/s, "
f"{throughput['mbit_per_s']:.2f} Mbit/s, "
f"{throughput['duration_ms'] / 1000.0:.3f}s, "
f"{throughput['chunks']} chunks"
f"{throughput['duration_ms'] / 1000.0:.3f}s, exactly 17 verified chunks; "
f"host={os.uname().nodename}, storage={RUN_ROOT}, profile=dynamic-s37, "
f"date={time.strftime('%Y-%m-%d')}"
)
def s38_first_play_launch_settings(self) -> str:
@@ -1272,8 +1887,20 @@ class Runner:
for path in [account_file, language_file, ini_file]:
if not path.is_file():
raise ScenarioError(f"expected installed launch settings file: {path}")
if b"PersonaName = stubplayer\r\n" not in ini_file.read_bytes():
raise ScenarioError("installed SmartSteamEmu.ini did not preserve CRLF stub PersonaName")
if account_file.read_bytes() != b"stubaccount":
raise ScenarioError("installed account_name.txt did not contain the exact stub bytes")
if language_file.read_bytes() != b"english":
raise ScenarioError("installed language.txt did not contain the exact stub bytes")
expected_ini = (
b"[Settings]\r\n"
b"AppId = 240\r\n"
b"PersonaName = stubplayer\r\n"
b"Language = english\r\n"
)
if ini_file.read_bytes() != expected_ini:
raise ScenarioError(
"installed SmartSteamEmu.ini did not contain the exact CRLF stub bytes"
)
first = client.send(
{
@@ -1349,11 +1976,21 @@ class Runner:
def stream_install_cnctw(self, prefix: str) -> tuple[Peer, Peer]:
source_dir = self.fixture_root / f"{prefix}-bravo"
copy_game("cnctw", source_dir, version="20160128")
source = self.peer(f"{prefix}-bravo", games_dir=source_dir)
client = self.peer(f"{prefix}-client")
network = self.create_scenario_network(f"{prefix}-stream", internal=True)
source = self.peer(
f"{prefix}-bravo",
network=network,
games_dir=source_dir,
)
client = self.peer(f"{prefix}-client", network=network)
connect_many(client, [source])
wait_remote_game(client, "cnctw", peer_count=1)
waiter = LineWaiter(len(client.output))
assert_exact_peer_ids(client, {source.peer_id})
content_id = catalog_content_id(client, "cnctw")
wait_peer_has_game(client, source.peer_id, "cnctw", content_id)
wait_remote_library_content(client, "cnctw", content_id, peer_count=1)
wait_remote_game(client, "cnctw", peer_count=1, version="20160128")
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "stream-install", "game_id": "cnctw"})
client.wait_for(
event_is("download-begin", "cnctw"),
@@ -1373,6 +2010,14 @@ class Runner:
description="stream install cnctw",
waiter=waiter,
)
assert_stream_install_lifecycle(
client,
"cnctw",
start=start,
content_id=content_id,
source_peer_id=source.peer_id,
source_addr=source.ready_addr,
)
return source, client
def s39_streamed_install_local_only(self) -> str:
@@ -1425,29 +2070,74 @@ class Runner:
)
def s40_streamed_receiver_not_source(self) -> str:
_source, receiver = self.stream_install_cnctw("s40")
observer = self.peer("s40-observer")
connect_many(observer, [receiver])
receiver_snapshot = wait_peer_has_game(observer, receiver.peer_id, "cnctw")
summary = next(
game
for game in receiver_snapshot.get("games", [])
if game.get("id") == "cnctw"
)
if summary.get("availability") != "LocalOnly" or summary.get("downloaded"):
raise ScenarioError(f"receiver did not advertise cnctw as local-only: {summary}")
source, receiver = self.stream_install_cnctw("s40")
source.shutdown()
wait_remote_absent(observer, "cnctw", timeout=5)
err = observer.send(
{"cmd": "download", "game_id": "cnctw", "install": False},
expect_error=True,
observer_network = self.create_scenario_network(
"s40-observer",
internal=True,
)
if "no peers have game cnctw" not in err["error"]:
raise ScenarioError(f"unexpected local-only download error: {err}")
observer = self.peer("s40-observer", network=observer_network)
assert_exact_peer_ids(observer, set())
remote_view_start = len(observer.output)
receiver_addr = self.connect_peer_to_network(receiver, observer_network)
observer.connect_to_at(receiver, receiver_addr, 1)
assert_exact_peer_ids(observer, {receiver.peer_id})
wait_peer_without_game(observer, receiver.peer_id, "cnctw")
wait_remote_library_without_game(
observer,
"cnctw",
start=remote_view_start,
)
wait_remote_absent(observer, "cnctw", timeout=5)
start = len(observer.output)
waiter = LineWaiter(start)
queued = observer.send(
{"cmd": "download", "game_id": "cnctw", "install": False}
)
if queued.get("data", {}).get("queued") is not True:
raise ScenarioError(f"no-source cnctw download was not queued: {queued}")
observer.wait_for(
event_is("download-failed", "cnctw"),
timeout=20,
description="asynchronous no-source cnctw failure",
waiter=waiter,
)
assert_download_status_trace(
observer,
"cnctw",
start=start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
expect_begin=False,
expect_verification=False,
)
assert_no_event_since(observer, start, "download-begin", "cnctw")
assert_no_event_since(observer, start, "download-finished", "cnctw")
active_events = [
item
for item in observer.output[start:]
if item.get("type") == "event"
and item.get("event") == "active-operations-changed"
and any(
operation.get("game_id") == "cnctw"
for operation in item.get("data", {}).get("active_operations", [])
)
]
if active_events:
raise ScenarioError(
f"no-source cnctw download entered active operations: {active_events}"
)
assert_no_active(observer, "cnctw")
assert_exact_peer_ids(observer, {receiver.peer_id})
wait_peer_without_game(observer, receiver.peer_id, "cnctw")
wait_remote_absent(observer, "cnctw", timeout=5)
assert_local_absent(observer, "cnctw")
assert_not_exists(observer.host_games_dir / "cnctw")
return (
"observer saw receiver's local-only cnctw snapshot, but remote aggregation hid it "
f"and download errored '{err['error']}'"
"isolated observer roster changed from empty to exactly the stream-installed "
"receiver; raw/remote views omitted cnctw, and queued failure had no begin, "
"success, active-operation event, or local mutation"
)
def s41_solid_archive_streamed_install(self) -> str:
@@ -1455,13 +2145,29 @@ class Runner:
source_game = source_dir / "cnctw"
shutil.copytree(FIXTURES / "fixture-solid" / "cnctw", source_game)
source = self.peer("s41-solid-source", games_dir=source_dir)
catalog_dir = CATALOG_PROFILES / "solid"
network = self.create_scenario_network("s41-stream", internal=True)
source = self.peer(
"s41-solid-source",
network=network,
games_dir=source_dir,
catalog_dir=catalog_dir,
)
assert_peer_rar_archive_solid(source, "cnctw")
client = self.peer("s41-solid-client")
client = self.peer(
"s41-solid-client",
network=network,
catalog_dir=catalog_dir,
)
connect_many(client, [source])
assert_exact_peer_ids(client, {source.peer_id})
content_id = catalog_content_id(client, "cnctw")
wait_peer_has_game(client, source.peer_id, "cnctw", content_id)
wait_remote_library_content(client, "cnctw", content_id, peer_count=1)
wait_remote_game(client, "cnctw", peer_count=1, version="20160128")
waiter = LineWaiter(len(client.output))
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "stream-install", "game_id": "cnctw"})
client.wait_for(
event_is("download-finished", "cnctw"),
@@ -1475,6 +2181,13 @@ class Runner:
description="solid stream install cnctw",
waiter=waiter,
)
assert_chunk_authority(
client,
"cnctw",
content_id,
{source.peer_id},
start=start,
)
game = wait_local_game(client, "cnctw", downloaded=False, installed=True)
assert_game_state(
@@ -1525,38 +2238,114 @@ class Runner:
def s42_streamed_install_retries_next_source(self) -> str:
bad_dir = self.fixture_root / "s42-bad-source"
good_dir = self.fixture_root / "s42-good-source"
copy_game("cnctw", bad_dir, version="20160128")
shutil.copytree(FIXTURES / "fixture-solid" / "cnctw", bad_dir / "cnctw")
copy_game("cnctw", good_dir, version="20160128")
network = self.create_scenario_network("s42-stream", internal=True)
bad = self.peer(
"s42-bad-source",
games_dir=bad_dir,
extra_args=["--unrar", "/missing-unrar"],
# Learn and persist both installation identities before assigning the
# mismatching profile. Production orders sources by authenticated
# PeerId, not by Docker address or connection order.
source_a_probe = self.peer("s42-source-a", network=network)
source_b_probe = self.peer("s42-source-b", network=network)
lower_probe, upper_probe = sorted(
[source_a_probe, source_b_probe],
key=lambda peer: peer_id_sort_key(peer.peer_id),
)
good = self.peer("s42-good-source", games_dir=good_dir)
if socket_addr_sort_key(bad.ready_addr) > socket_addr_sort_key(good.ready_addr):
lower_id = lower_probe.peer_id
upper_id = upper_probe.peer_id
lower_name = lower_probe.name
upper_name = upper_probe.name
source_a_probe.shutdown()
source_b_probe.shutdown()
bad = self.peer(lower_name, network=network, games_dir=bad_dir)
good = self.peer(upper_name, network=network, games_dir=good_dir)
if bad.peer_id != lower_id or good.peer_id != upper_id:
raise ScenarioError(
"S42 requires the broken source to sort before the good source; "
f"bad={bad.ready_addr}, good={good.ready_addr}"
"S42 source identities changed across the profile restart: "
f"bad={bad.peer_id}/{lower_id}, good={good.peer_id}/{upper_id}"
)
assert_peer_rar_archive_solid(bad, "cnctw")
if peer_id_sort_key(bad.peer_id) >= peer_id_sort_key(good.peer_id):
raise ScenarioError(
"S42 requires the catalog-mismatched source to have the lower authenticated "
f"PeerId; bad={bad.peer_id}, good={good.peer_id}"
)
client = self.peer("s42-client")
client = self.peer("s42-client", network=network)
connect_many(client, [bad, good])
assert_exact_peer_ids(client, {bad.peer_id, good.peer_id})
content_id = catalog_content_id(client, "cnctw")
source_content_ids = {
catalog_content_id(bad, "cnctw"),
catalog_content_id(good, "cnctw"),
}
if source_content_ids != {content_id}:
raise ScenarioError(
f"S42 sources did not share the client's exact ContentId: "
f"{source_content_ids} != {content_id}"
)
wait_peer_has_game(client, bad.peer_id, "cnctw", content_id)
wait_peer_has_game(client, good.peer_id, "cnctw", content_id)
wait_remote_library_content(client, "cnctw", content_id, peer_count=2)
wait_remote_game(client, "cnctw", peer_count=2, version="20160128")
waiter = LineWaiter(len(client.output))
expected = {
"bin/cnctw-payload.bin": unrar_entry_sha256(
good, "cnctw", "bin/cnctw-payload.bin"
),
"data/cnctw-assets.dat": unrar_entry_sha256(
good, "cnctw", "data/cnctw-assets.dat"
),
}
if client.process is None:
raise ScenarioError("S42 client process was not retained")
client_pid = client.process.pid
first_start = len(client.output)
first_waiter = LineWaiter(first_start)
bad_first_edges = LineWaiter(len(bad.output))
good_first_edges = LineWaiter(len(good.output))
client.send({"cmd": "stream-install", "game_id": "cnctw"})
wait_outbound_transfer_cycle(
bad,
"cnctw",
waiter=bad_first_edges,
phase="first streamed install",
)
wait_outbound_transfer_cycle(
good,
"cnctw",
waiter=good_first_edges,
phase="first streamed install",
)
client.wait_for(
event_is("download-finished", "cnctw"),
timeout=60,
description="retry stream finish cnctw",
waiter=waiter,
description="first retry stream finish cnctw",
waiter=first_waiter,
)
client.wait_for(
event_is("install-finished", "cnctw"),
timeout=30,
description="retry stream install cnctw",
waiter=waiter,
description="first retry stream install cnctw",
waiter=first_waiter,
)
assert_stream_install_lifecycle(
client,
"cnctw",
start=first_start,
content_id=content_id,
source_peer_id=good.peer_id,
source_addr=good.ready_addr,
expect_invalid_source_retry=True,
)
assert_chunk_authority(
client,
"cnctw",
content_id,
{good.peer_id},
start=first_start,
)
game = wait_local_game(client, "cnctw", downloaded=False, installed=True)
@@ -1568,41 +2357,182 @@ class Runner:
)
game_root = client.host_games_dir / "cnctw"
assert_not_exists(game_root / ".local.installing")
assert_not_exists(game_root / ".local.backup")
assert_not_exists(game_root / "version.ini")
assert_not_exists(game_root / "cnctw.eti")
assert_only_chunk_sources(client, "cnctw", {good.ready_addr})
expected = {
"bin/cnctw-payload.bin": unrar_entry_sha256(
good, "cnctw", "bin/cnctw-payload.bin"
),
"data/cnctw-assets.dat": unrar_entry_sha256(
good, "cnctw", "data/cnctw-assets.dat"
),
}
actual = {
first_actual = {
rel: sha256_file(game_root / "local" / rel)
for rel in expected
}
if actual != expected:
raise ScenarioError(f"retry streamed payload hashes mismatched: {actual} != {expected}")
if first_actual != expected:
raise ScenarioError(
f"first retry payload hashes mismatched: {first_actual} != {expected}"
)
streamed_bytes = sum(
first_streamed_bytes = sum(
int(item.get("data", {}).get("length", 0))
for item in client.output
for item in client.output[first_start:]
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == "cnctw"
)
expected_bytes = 3 * 1024 * 1024
if streamed_bytes != expected_bytes:
if first_streamed_bytes != expected_bytes:
raise ScenarioError(
f"retry streamed byte count mismatch: {streamed_bytes} != {expected_bytes}"
f"first retry byte count mismatch: {first_streamed_bytes} != {expected_bytes}"
)
uninstall_start = len(client.output)
uninstall_waiter = LineWaiter(uninstall_start)
client.send({"cmd": "uninstall", "game_id": "cnctw"})
client.wait_for(
event_is("uninstall-finished", "cnctw"),
timeout=30,
description="uninstall before quarantine persistence retry",
waiter=uninstall_waiter,
)
wait_no_active(client, "cnctw")
assert_local_absent(client, "cnctw")
assert_not_exists(game_root / "local")
assert_not_exists(game_root / ".local.installing")
assert_not_exists(game_root / ".local.backup")
if client.process.poll() is not None or client.process.pid != client_pid:
raise ScenarioError("S42 client runtime changed before the quarantine retry")
assert_exact_peer_ids(client, {bad.peer_id, good.peer_id})
wait_peer_has_game(client, bad.peer_id, "cnctw", content_id)
wait_peer_has_game(client, good.peer_id, "cnctw", content_id)
wait_remote_library_content(client, "cnctw", content_id, peer_count=2)
wait_no_outbound_transfer(bad, "cnctw")
wait_no_outbound_transfer(good, "cnctw")
assert_no_outbound_edge_during(
bad,
start=bad_first_edges.seen,
phase="first streamed install post-drain",
timeout=0.5,
)
assert_no_outbound_edge_during(
good,
start=good_first_edges.seen,
phase="first streamed install post-drain",
timeout=0.5,
)
second_bad_start = len(bad.output)
second_good_start = len(good.output)
good_second_edges = LineWaiter(second_good_start)
second_start = len(client.output)
second_waiter = LineWaiter(second_start)
client.send({"cmd": "stream-install", "game_id": "cnctw"})
wait_outbound_transfer_cycle(
good,
"cnctw",
waiter=good_second_edges,
phase="second streamed install",
)
client.wait_for(
event_is("download-finished", "cnctw"),
timeout=60,
description="quarantine persistence stream finish cnctw",
waiter=second_waiter,
)
client.wait_for(
event_is("install-finished", "cnctw"),
timeout=30,
description="quarantine persistence install finish cnctw",
waiter=second_waiter,
)
assert_stream_install_lifecycle(
client,
"cnctw",
start=second_start,
content_id=content_id,
source_peer_id=good.peer_id,
source_addr=good.ready_addr,
)
wait_no_outbound_transfer(bad, "cnctw")
wait_no_outbound_transfer(good, "cnctw")
assert_no_outbound_edge_during(
good,
start=good_second_edges.seen,
phase="second streamed install post-drain",
timeout=0.5,
)
assert_chunk_authority(
client,
"cnctw",
content_id,
{good.peer_id},
start=second_start,
)
second_game = wait_local_game(
client,
"cnctw",
downloaded=False,
installed=True,
)
assert_game_state(
second_game,
downloaded=False,
installed=True,
availability="LocalOnly",
)
assert_not_exists(game_root / ".local.installing")
assert_not_exists(game_root / ".local.backup")
assert_not_exists(game_root / "version.ini")
assert_not_exists(game_root / "cnctw.eti")
second_actual = {
rel: sha256_file(game_root / "local" / rel)
for rel in expected
}
if second_actual != expected:
raise ScenarioError(
f"second retry payload hashes mismatched: {second_actual} != {expected}"
)
second_streamed_bytes = sum(
int(item.get("data", {}).get("length", 0))
for item in client.output[second_start:]
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == "cnctw"
)
if second_streamed_bytes != expected_bytes:
raise ScenarioError(
f"second retry byte count mismatch: {second_streamed_bytes} != {expected_bytes}"
)
wait_no_outbound_transfer(bad, "cnctw")
wait_no_outbound_transfer(good, "cnctw")
bad.shutdown_gracefully_for_proof()
good.shutdown_gracefully_for_proof()
second_bad_edges = [
item
for item in bad.output[second_bad_start:]
if item.get("type") == "event"
and item.get("event") == "outbound-transfer-count-changed"
]
if second_bad_edges:
raise ScenarioError(
"second streamed install admitted the quarantined bad source: "
f"{second_bad_edges}"
)
second_good_edges = [
item
for item in good.output[second_good_start:]
if item.get("type") == "event"
and item.get("event") == "outbound-transfer-count-changed"
]
if len(second_good_edges) != 2:
raise ScenarioError(
"second streamed install did not emit exactly good admission/drain edges: "
f"{second_good_edges}"
)
return (
"broken first source failed without chunks, next source completed whole stream; "
f"good={good.ready_addr}, bad={bad.ready_addr}, bytes={streamed_bytes}"
"first install admitted bad then good exact-content sources; after uninstall, "
"the same runtime skipped the quarantined bad source and admitted only good; "
f"good={good.peer_id}, bad={bad.peer_id}, bytes={second_streamed_bytes}"
)
def s43_streamed_install_rejects_installed_game(self) -> str:
@@ -1617,6 +2547,15 @@ class Runner:
description="already-installed stream rejection",
waiter=waiter,
)
assert_download_status_trace(
client,
"cnctw",
start=start,
terminal="download-failed",
reason="operation-failed",
expect_begin=False,
expect_verification=False,
)
assert_no_event_since(client, start, "install-finished", "cnctw")
assert_no_event_since(client, start, "download-finished", "cnctw")
wait_no_active(client, "cnctw")
@@ -1631,86 +2570,166 @@ class Runner:
return "already-installed cnctw rejected a second streamed install without state drift"
def s44_corrupt_stream_rolls_back(self) -> str:
source_dir = self.fixture_root / "s44-corrupt-source"
copy_game("cnctw", source_dir, version="20160128")
(source_dir / "cnctw" / "cnctw.eti").write_bytes(b"not a rar archive")
source_dir = self.fixture_root / "s44-mismatched-source"
shutil.copytree(FIXTURES / "fixture-solid" / "cnctw", source_dir / "cnctw")
source = self.peer("s44-corrupt-source", games_dir=source_dir)
client = self.peer("s44-client")
network = self.create_scenario_network("s44-stream", internal=True)
source = self.peer(
"s44-mismatched-source",
network=network,
games_dir=source_dir,
)
assert_peer_rar_archive_solid(source, "cnctw")
client = self.peer("s44-client", network=network)
connect_many(client, [source])
assert_exact_peer_ids(client, {source.peer_id})
content_id = catalog_content_id(client, "cnctw")
wait_peer_has_game(client, source.peer_id, "cnctw", content_id)
wait_remote_library_content(client, "cnctw", content_id, peer_count=1)
wait_remote_game(client, "cnctw", peer_count=1, version="20160128")
start = len(client.output)
waiter = LineWaiter(start)
source_edges = LineWaiter(len(source.output))
client.send({"cmd": "stream-install", "game_id": "cnctw"})
wait_outbound_transfer_cycle(
source,
"cnctw",
waiter=source_edges,
phase="all-bad streamed install",
)
client.wait_for(
event_is("download-failed", "cnctw"),
timeout=30,
description="corrupt stream failed",
description="catalog-mismatched stream failed",
waiter=waiter,
)
assert_download_status_trace(
client,
"cnctw",
start=start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
)
assert_no_event_since(client, start, "download-finished", "cnctw")
assert_no_event_since(client, start, "install-finished", "cnctw")
wait_no_active(client, "cnctw")
assert_failed_stream_left_no_local(client, "cnctw")
return "corrupt cnctw archive emitted download-failed and left no local install"
return (
"only source had a valid solid archive whose extracted output mismatched the "
"default catalog; integrity failure left no local install"
)
def s45_sender_disconnect_mid_stream(self) -> str:
source_dir = self.fixture_root / "s45-source"
copy_game("alienswarm", source_dir, version="20190317")
source = self.peer("s45-source", games_dir=source_dir)
client = self.peer("s45-client")
network = self.create_scenario_network("s45-stream", internal=True)
source = self.peer("s45-source", network=network, games_dir=source_dir)
client = self.peer("s45-client", network=network)
connect_many(client, [source])
assert_exact_peer_ids(client, {source.peer_id})
content_id = catalog_content_id(client, "alienswarm")
wait_peer_has_game(client, source.peer_id, "alienswarm", content_id)
wait_remote_library_content(client, "alienswarm", content_id, peer_count=1)
wait_remote_game(client, "alienswarm", peer_count=1, version="20190317")
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "stream-install", "game_id": "alienswarm"})
client.wait_for(
first_chunk = client.wait_for(
event_is("download-chunk-finished", "alienswarm"),
timeout=30,
description="first alienswarm stream chunk before source drop",
waiter=waiter,
)
first_data = first_chunk.get("data", {})
if (
first_data.get("peer_id") != source.peer_id
or first_data.get("content_id") != content_id
):
raise ScenarioError(
f"sender-drop barrier used the wrong source/content: {first_data}"
)
source.kill()
terminal = client.wait_for(
event_name_in({"download-failed", "download-peers-gone"}, "alienswarm"),
client.wait_for(
event_is("download-failed", "alienswarm"),
timeout=60,
description="sender disconnect terminal event",
description="prompt sender transport failure",
waiter=waiter,
)
assert_download_status_trace(
client,
"alienswarm",
start=start,
terminal="download-failed",
reason="verified-catalog-sources-exhausted",
)
assert_no_event_since(client, start, "download-finished", "alienswarm")
assert_no_event_since(client, start, "install-finished", "alienswarm")
assert_chunk_authority(
client,
"alienswarm",
content_id,
{source.peer_id},
start=start,
)
wait_no_active(client, "alienswarm")
assert_failed_stream_left_no_local(client, "alienswarm")
return (
"sender disconnect after first alienswarm chunk rolled back stream; "
f"terminal={terminal['event']}"
"sender disconnect after first alienswarm chunk emitted prompt download-failed "
"and rolled back the stream before stale-peer liveness elapsed"
)
def s46_receiver_cancel_mid_stream(self) -> str:
source_dir = self.fixture_root / "s46-source"
copy_game("alienswarm", source_dir, version="20190317")
source = self.peer("s46-source", games_dir=source_dir)
client = self.peer("s46-client")
network = self.create_scenario_network("s46-stream", internal=True)
source = self.peer("s46-source", network=network, games_dir=source_dir)
client = self.peer("s46-client", network=network)
connect_many(client, [source])
assert_exact_peer_ids(client, {source.peer_id})
content_id = catalog_content_id(client, "alienswarm")
wait_peer_has_game(client, source.peer_id, "alienswarm", content_id)
wait_remote_library_content(client, "alienswarm", content_id, peer_count=1)
wait_remote_game(client, "alienswarm", peer_count=1, version="20190317")
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "stream-install", "game_id": "alienswarm"})
client.wait_for(
first_chunk = client.wait_for(
event_is("download-chunk-finished", "alienswarm"),
timeout=30,
description="first alienswarm stream chunk before receiver cancel",
waiter=waiter,
)
first_data = first_chunk.get("data", {})
if (
first_data.get("peer_id") != source.peer_id
or first_data.get("content_id") != content_id
):
raise ScenarioError(
f"receiver-cancel barrier used the wrong source/content: {first_data}"
)
client.send({"cmd": "cancel-download", "game_id": "alienswarm"})
wait_no_active(client, "alienswarm", timeout=60)
assert_failed_stream_left_no_local(client, "alienswarm")
client.shutdown_gracefully_for_proof()
assert_download_status_trace(
client,
"alienswarm",
start=start,
terminal=None,
)
assert_no_event_since(client, start, "download-finished", "alienswarm")
assert_no_event_since(client, start, "download-failed", "alienswarm")
assert_no_event_since(client, start, "install-finished", "alienswarm")
assert_failed_stream_left_no_local(client, "alienswarm")
assert_chunk_authority(
client,
"alienswarm",
content_id,
{source.peer_id},
start=start,
)
return "receiver cancel after first alienswarm chunk rolled back without failed event"
def s47_multi_archive_streams_in_sorted_order(self) -> str:
@@ -1718,12 +2737,28 @@ class Runner:
source_game = source_dir / "cnctw"
shutil.copytree(FIXTURES / "fixture-multi" / "cnctw", source_game)
source = self.peer("s47-source", games_dir=source_dir)
client = self.peer("s47-client")
catalog_dir = CATALOG_PROFILES / "multi"
network = self.create_scenario_network("s47-stream", internal=True)
source = self.peer(
"s47-source",
network=network,
games_dir=source_dir,
catalog_dir=catalog_dir,
)
client = self.peer(
"s47-client",
network=network,
catalog_dir=catalog_dir,
)
connect_many(client, [source])
assert_exact_peer_ids(client, {source.peer_id})
content_id = catalog_content_id(client, "cnctw")
wait_peer_has_game(client, source.peer_id, "cnctw", content_id)
wait_remote_library_content(client, "cnctw", content_id, peer_count=1)
wait_remote_game(client, "cnctw", peer_count=1, version="20160128")
waiter = LineWaiter(len(client.output))
start = len(client.output)
waiter = LineWaiter(start)
client.send({"cmd": "stream-install", "game_id": "cnctw"})
client.wait_for(
event_is("download-finished", "cnctw"),
@@ -1737,6 +2772,13 @@ class Runner:
description="multi-archive stream install",
waiter=waiter,
)
assert_chunk_authority(
client,
"cnctw",
content_id,
{source.peer_id},
start=start,
)
game = wait_local_game(client, "cnctw", downloaded=False, installed=True)
assert_game_state(
@@ -1752,8 +2794,8 @@ class Runner:
chunk_paths = streamed_chunk_paths(client, "cnctw")
expected_paths = [
"cnctw/.local.installing/order/first.txt",
"cnctw/.local.installing/order/second.txt",
"order/first.txt",
"order/second.txt",
]
if chunk_paths != expected_paths:
raise ScenarioError(f"multi-archive stream order mismatch: {chunk_paths}")
@@ -1768,15 +2810,13 @@ class Runner:
def s48_call_to_play_replication_and_late_join(self) -> str:
alice = self.peer("s48-alice")
bob = self.peer("s48-bob")
bob.connect_to(alice)
alice.set_call_to_play_display_name("Alice")
bob.set_call_to_play_display_name("Bob")
connect_bidirectional(alice, bob)
now = int(time.time() * 1000)
create = {
"id": "s48-create",
"call_id": "s48-call",
"actor_id": "",
"actor_name": "Alice",
"at": now,
create = alice.publish_call_to_play({
"call_id": None,
"action": {
"Create": {
"game_id": "cnctw",
@@ -1785,58 +2825,91 @@ class Runner:
"deadline": now + 600_000,
}
},
}
rsvp = {
"id": "s48-rsvp",
"call_id": "s48-call",
"actor_id": "",
"actor_name": "Bob",
"at": now + 1,
})
call_id = create["call_id"]
wait_call_to_play_events(bob, {create["event_id"]})
rsvp = bob.publish_call_to_play({
"call_id": call_id,
"action": "Rsvp",
}
message = {
"id": "s48-message-event",
"call_id": "s48-call",
"actor_id": "",
"actor_name": "Bob",
"at": now + 2,
})
message = bob.publish_call_to_play({
"call_id": call_id,
"action": {
"SendMessage": {
"message_id": "s48-message",
"text": "I am in",
}
},
}
})
expected = {create["event_id"], rsvp["event_id"], message["event_id"]}
wait_call_to_play_events(alice, expected)
alice.publish_call_to_play(create)
wait_call_to_play_events(bob, {"s48-create"})
bob.publish_call_to_play(rsvp)
bob.publish_call_to_play(message)
wait_call_to_play_events(alice, {"s48-create", "s48-rsvp", "s48-message-event"})
charlie = self.peer("s48-charlie")
charlie.connect_to(alice)
events = wait_call_to_play_events(
charlie,
{"s48-create", "s48-rsvp", "s48-message-event"},
# Keep Alice<->Bob live on the ordinary network so Alice must retain
# Bob's slice throughout the no-relay proof. Charlie starts on an
# internal-only bridge shared with Alice: it has no route to Bob's
# ordinary-network address until Bob is explicitly attached below.
late_join_network = self.create_scenario_network(
"s48-late-join", internal=True
)
if len(events) != 3:
raise ScenarioError(f"late join history contains duplicates: {events}")
alice_late_join_addr = self.connect_peer_to_network(alice, late_join_network)
charlie = self.peer("s48-charlie", network=late_join_network)
charlie.set_call_to_play_display_name("Charlie")
charlie.connect_to_at(alice, alice_late_join_addr, 1)
wait_call_to_play_exact_events(alice, expected)
assert_exact_peer_ids(charlie, {alice.peer_id})
alice_only = wait_call_to_play_exact_events_from_exact_peers(
charlie,
{create["event_id"]},
{alice.peer_id},
)
wait_call_to_play_exact_events(alice, expected)
if any(event.get("id") in {rsvp["event_id"], message["event_id"]}
for event in alice_only):
raise ScenarioError(f"Alice relayed Bob-owned events to Charlie: {alice_only}")
return "live create/RSVP/chat replicated and a late joiner received deduplicated history"
bob_late_join_addr = self.connect_peer_to_network(bob, late_join_network)
charlie.connect_to_at(bob, bob_late_join_addr, 2)
events = wait_call_to_play_exact_events(charlie, expected)
if len(events) != 3:
raise ScenarioError(f"direct-author late-join view contains duplicates: {events}")
authors = {event["id"]: event["author_id"] for event in events}
if authors.get(create["event_id"]) != alice.peer_id or any(
authors.get(event_id) != bob.peer_id
for event_id in [rsvp["event_id"], message["event_id"]]
):
raise ScenarioError(f"direct-author attribution was not preserved: {events}")
bob.shutdown()
wait_call_to_play_departure_with_exact_remaining_peers(
charlie,
{create["event_id"]},
"Bob participant",
{alice.peer_id},
bob.peer_id,
)
alice.shutdown()
wait_call_to_play_departure_with_exact_remaining_peers(
charlie,
set(),
"Alice creator",
set(),
alice.peer_id,
)
return (
"late joiner pulled Alice and Bob directly; full-view replacement removed Bob's "
"participant slice on departure, then removed the call when Alice departed"
)
def s49_terminal_call_to_play_late_join(self) -> str:
alice = self.peer("s49-alice")
bob = self.peer("s49-bob")
bob.connect_to(alice)
alice.set_call_to_play_display_name("Alice")
bob.set_call_to_play_display_name("Bob")
connect_bidirectional(alice, bob)
now = int(time.time() * 1000)
create = {
"id": "s49-create",
"call_id": "s49-call",
"actor_id": "",
"actor_name": "Alice",
"at": now,
create = alice.publish_call_to_play({
"call_id": None,
"action": {
"Create": {
"game_id": "cnctw",
@@ -1845,58 +2918,70 @@ class Runner:
"deadline": now + 600_000,
}
},
}
ready = {
"id": "s49-ready",
"call_id": "s49-call",
"actor_id": "",
"actor_name": "Bob",
"at": now + 1,
})
call_id = create["call_id"]
wait_call_to_play_events(bob, {create["event_id"]})
ready = bob.publish_call_to_play({
"call_id": call_id,
"action": {"Respond": {"ready_at": None}},
}
message = {
"id": "s49-message-event",
"call_id": "s49-call",
"actor_id": "",
"actor_name": "Bob",
"at": now + 2,
})
message = bob.publish_call_to_play({
"call_id": call_id,
"action": {
"SendMessage": {
"message_id": "s49-message",
"text": "Ready to launch",
}
},
}
start = {
"id": "s49-start",
"call_id": "s49-call",
"actor_id": "",
"actor_name": "Alice",
"at": now + 3,
})
participant_ids = {ready["event_id"], message["event_id"]}
wait_call_to_play_events(alice, {create["event_id"], *participant_ids})
start = alice.publish_call_to_play({
"call_id": call_id,
"action": "Start",
}
})
expected = {create["event_id"], *participant_ids, start["event_id"]}
wait_call_to_play_events(bob, expected)
alice.publish_call_to_play(create)
wait_call_to_play_events(bob, {"s49-create"})
bob.publish_call_to_play(ready)
bob.publish_call_to_play(message)
wait_call_to_play_events(alice, {"s49-create", "s49-ready", "s49-message-event"})
alice.publish_call_to_play(start)
wait_call_to_play_events(
bob,
{"s49-create", "s49-ready", "s49-message-event", "s49-start"},
late_join_network = self.create_scenario_network(
"s49-late-join", internal=True
)
charlie = self.peer("s49-charlie")
charlie.connect_to(alice)
events = wait_call_to_play_events(
alice_late_join_addr = self.connect_peer_to_network(alice, late_join_network)
charlie = self.peer("s49-charlie", network=late_join_network)
charlie.set_call_to_play_display_name("Charlie")
charlie.connect_to_at(alice, alice_late_join_addr, 1)
alice_ids = {create["event_id"], start["event_id"]}
wait_call_to_play_exact_events(alice, expected)
assert_exact_peer_ids(charlie, {alice.peer_id})
alice_only = wait_call_to_play_exact_events_from_exact_peers(
charlie,
{"s49-create", "s49-ready", "s49-message-event", "s49-start"},
alice_ids,
{alice.peer_id},
)
if len(events) != 4:
raise ScenarioError(f"terminal late-join history is incomplete: {events}")
wait_call_to_play_exact_events(alice, expected)
if any(event.get("id") in participant_ids for event in alice_only):
raise ScenarioError(
f"Alice relayed Bob-owned terminal-call events to Charlie: {alice_only}"
)
return "late joiner reconstructed terminal call outcome, roster, and chat"
bob_late_join_addr = self.connect_peer_to_network(bob, late_join_network)
charlie.connect_to_at(bob, bob_late_join_addr, 2)
events = wait_call_to_play_exact_events(charlie, expected)
if len(events) != 4:
raise ScenarioError(f"direct-author terminal late-join view is incomplete: {events}")
alice.shutdown()
wait_call_to_play_departure_with_exact_remaining_peers(
charlie,
set(),
"Alice creator",
{bob.peer_id},
alice.peer_id,
)
return (
"late joiner pulled terminal creator and participant slices directly; creator "
"departure removed the complete call despite the live participant author"
)
def run(command: list[str], description: str) -> subprocess.CompletedProcess[str]:
@@ -2023,13 +3108,23 @@ def create_large_sparse_game(root: Path, *, size: int, version: str | None = Non
def inflate_archive_sparse(game_root: Path, game_id: str, size: int) -> None:
"""Replace a copied game's `.eti` with a sparse file of `size` bytes so the
archive spans multiple 128 MiB download chunks. Sparse zero bytes are
identical across copies, so duplicate-source majority validation still
agrees and a `diff` against any catalog-version source still matches."""
identical across copies, so one publisher-derived catalog manifest
authorizes every copy and a `diff` against any source still matches."""
archive = game_root / f"{game_id}.eti"
with archive.open("wb") as handle:
handle.truncate(size)
def corrupt_file_same_length(path: Path) -> None:
"""Flip one payload byte without changing the catalog-owned file shape."""
with path.open("r+b") as handle:
first = handle.read(1)
if not first:
raise ScenarioError(f"cannot corrupt empty fixture file: {path}")
handle.seek(0)
handle.write(bytes([first[0] ^ 0xFF]))
def sha256_file(path: Path) -> str:
hasher = hashlib.sha256()
with path.open("rb") as handle:
@@ -2065,12 +3160,17 @@ def assert_peer_rar_archive_solid(peer: Peer, game_id: str) -> None:
raise ScenarioError(f"RAR archive details were not reported: {game_id}")
def socket_addr_sort_key(addr: str | None) -> tuple[int, int]:
if addr is None:
raise ScenarioError("cannot sort missing peer address")
host, port = addr.rsplit(":", 1)
host = host.removeprefix("[").removesuffix("]")
return (int(ipaddress.ip_address(host)), int(port))
def peer_id_sort_key(peer_id: str | None) -> bytes:
if peer_id is None:
raise ScenarioError("cannot sort missing peer identity")
padded = peer_id.upper() + "=" * (-len(peer_id) % 8)
try:
decoded = base64.b32decode(padded)
except (binascii.Error, ValueError) as error:
raise ScenarioError(f"invalid peer identity {peer_id!r}") from error
if len(decoded) != 32:
raise ScenarioError(f"invalid peer identity length for {peer_id!r}")
return decoded
def format_bytes(size: int) -> str:
@@ -2083,6 +3183,122 @@ def connect_many(client: Peer, peers: list[Peer]) -> None:
client.send({"cmd": "wait-peers", "count": len(peers), "timeout_ms": 15000})
def connect_bidirectional(left: Peer, right: Peer) -> None:
left.connect_to(right)
right.connect_to(left)
wait_peer_visible(left, right.peer_id)
wait_peer_visible(right, left.peer_id)
def catalog_content_id(peer: Peer, game_id: str) -> str:
catalog_dir = peer.catalog_dir or CATALOG_PROFILES / "default"
manifest_path = catalog_dir / "manifests" / f"{game_id}.json"
try:
artifact = json.loads(manifest_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError) as error:
raise ScenarioError(
f"failed to load catalog manifest for {peer.name}/{game_id}: {manifest_path}: {error}"
) from error
content_id = artifact.get("content_id")
if (
not isinstance(content_id, str)
or len(content_id) != 64
or content_id.lower() != content_id
or any(char not in "0123456789abcdef" for char in content_id)
):
raise ScenarioError(
f"catalog manifest has invalid content_id for {peer.name}/{game_id}: {content_id!r}"
)
return content_id
def wait_remote_library_content(
peer: Peer,
game_id: str,
content_id: str,
*,
peer_count: int,
timeout: float = 20,
) -> dict[str, Any]:
deadline = time.monotonic() + timeout
last_view: list[dict[str, Any]] = []
while time.monotonic() < deadline:
latest = next(
(
item
for item in reversed(peer.output)
if item.get("type") == "event"
and item.get("event") == "remote-library-view"
),
None,
)
if latest is not None:
last_view = latest.get("data", {}).get("view", {}).get("games", [])
for row in last_view:
if (
row.get("game_id") == game_id
and row.get("content_id") == content_id
and row.get("peer_count") == peer_count
):
return row
time.sleep(0.2)
raise ScenarioError(
f"{peer.name} never received remote-library-view {game_id}/{content_id} "
f"peer_count={peer_count}; latest={last_view}"
)
def wait_remote_library_without_game(
peer: Peer,
game_id: str,
*,
start: int,
timeout: float = 20,
) -> list[dict[str, Any]]:
deadline = time.monotonic() + timeout
last_view: list[dict[str, Any]] | None = None
while time.monotonic() < deadline:
latest = next(
(
item
for item in reversed(peer.output[start:])
if item.get("type") == "event"
and item.get("event") == "remote-library-view"
),
None,
)
if latest is not None:
last_view = latest.get("data", {}).get("view", {}).get("games", [])
if all(row.get("game_id") != game_id for row in last_view):
return last_view
time.sleep(0.2)
raise ScenarioError(
f"{peer.name} never received a post-connect remote-library-view without "
f"{game_id}; latest={last_view}"
)
def wait_peer_visible(
observer: Peer,
peer_id: str | None,
timeout: float = 20,
) -> dict[str, Any]:
if peer_id is None:
raise ScenarioError("cannot wait for a peer without peer_id")
deadline = time.monotonic() + timeout
last_peers: list[dict[str, Any]] = []
while time.monotonic() < deadline:
last_peers = observer.list_peers()
for peer in last_peers:
if peer.get("peer_id") == peer_id:
return peer
time.sleep(0.2)
raise ScenarioError(
f"{observer.name} never saw peer {peer_id}; peers={last_peers}"
)
def wait_remote_game(
peer: Peer,
game_id: str,
@@ -2122,6 +3338,37 @@ def wait_remote_absent(peer: Peer, game_id: str, timeout: float = 20) -> None:
raise ScenarioError(f"{peer.name} still lists remote {game_id}; rows={last_rows}")
def wait_departure_topology(
observer: Peer,
*,
expected_peer_ids: set[str | None],
expected_remote_game_ids: set[str],
phase: str,
) -> None:
if None in expected_peer_ids:
raise ScenarioError(f"{phase} cannot wait on a missing peer identity")
deadline = time.monotonic() + PEER_DEPARTURE_TIMEOUT_SECONDS
last_peers: list[dict[str, Any]] = []
last_remote: list[dict[str, Any]] = []
while time.monotonic() < deadline:
last_peers = observer.list_peers()
last_remote = observer.list_games()["remote"]
actual_peer_ids = {peer.get("peer_id") for peer in last_peers}
actual_remote_game_ids = {game.get("id") for game in last_remote}
if (
actual_peer_ids == expected_peer_ids
and actual_remote_game_ids == expected_remote_game_ids
):
return
time.sleep(0.4)
raise ScenarioError(
f"{phase} did not converge after the {PEER_DEPARTURE_TIMEOUT_SECONDS}s "
f"liveness allowance: expected peers={expected_peer_ids}, "
f"remote_games={expected_remote_game_ids}; peers={last_peers}, "
f"remote={last_remote}"
)
def wait_local_game(
peer: Peer,
game_id: str,
@@ -2161,6 +3408,28 @@ def wait_no_active(peer: Peer, game_id: str, timeout: float = 20) -> None:
raise ScenarioError(f"{peer.name} still has active operation for {game_id}: {last_active}")
def wait_outbound_transfer(
peer: Peer,
game_id: str,
*,
minimum: int = 1,
timeout: float = 20,
) -> int:
deadline = time.monotonic() + timeout
last_active: dict[str, int] = {}
while time.monotonic() < deadline:
active = peer.status()["active_outbound_transfers"]
last_active = active
count = int(active.get(game_id, 0))
if count >= minimum:
return count
time.sleep(0.02)
raise ScenarioError(
f"{peer.name} never reached {minimum} active outbound transfers for {game_id}: "
f"{last_active}"
)
def wait_no_outbound_transfer(peer: Peer, game_id: str, timeout: float = 20) -> None:
deadline = time.monotonic() + timeout
last_active: dict[str, int] = {}
@@ -2175,6 +3444,47 @@ def wait_no_outbound_transfer(peer: Peer, game_id: str, timeout: float = 20) ->
)
def wait_outbound_transfer_cycle(
peer: Peer,
game_id: str,
*,
waiter: LineWaiter,
phase: str,
timeout: float = 30,
) -> None:
for edge in ["admission", "drain"]:
peer.wait_for(
event_is("outbound-transfer-count-changed"),
timeout=timeout,
description=f"{phase} {peer.name} outbound {edge} edge",
waiter=waiter,
)
wait_no_outbound_transfer(peer, game_id, timeout=timeout)
def assert_no_outbound_edge_during(
peer: Peer,
*,
start: int,
phase: str,
timeout: float = 1.0,
) -> None:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
edges = [
item
for item in peer.output[start:]
if item.get("type") == "event"
and item.get("event") == "outbound-transfer-count-changed"
]
if edges:
raise ScenarioError(
f"{phase} unexpectedly admitted an outbound transfer on {peer.name}: "
f"{edges}"
)
time.sleep(0.05)
def wait_call_to_play_events(
peer: Peer,
expected_ids: set[str],
@@ -2193,6 +3503,104 @@ def wait_call_to_play_events(
)
def wait_call_to_play_exact_events(
peer: Peer,
expected_ids: set[str],
timeout: float = 20,
) -> list[dict[str, Any]]:
deadline = time.monotonic() + timeout
last_events: list[dict[str, Any]] = []
while time.monotonic() < deadline:
events = peer.call_to_play_events()
last_events = events
if {event.get("id") for event in events} == expected_ids:
return events
time.sleep(0.2)
raise ScenarioError(
f"{peer.name} never converged to exact Call to Play replacement {expected_ids}: "
f"{last_events}"
)
def wait_call_to_play_departure_with_exact_remaining_peers(
peer: Peer,
expected_event_ids: set[str],
departed_author: str,
expected_remaining_peer_ids: set[str | None],
departed_peer_id: str | None,
) -> list[dict[str, Any]]:
if None in expected_remaining_peer_ids or departed_peer_id is None:
raise ScenarioError("cannot prove departure liveness with a missing identity")
allowed_peer_ids = {*expected_remaining_peer_ids, departed_peer_id}
deadline = time.monotonic() + PEER_DEPARTURE_TIMEOUT_SECONDS
last_events: list[dict[str, Any]] = []
last_peers: list[dict[str, Any]] = []
while time.monotonic() < deadline:
peers = peer.list_peers()
last_peers = peers
actual_peer_ids = {item.get("peer_id") for item in peers}
if not expected_remaining_peer_ids <= actual_peer_ids:
raise ScenarioError(
f"{departed_author} departure lost a peer that must remain live: "
f"expected {expected_remaining_peer_ids}, found {actual_peer_ids}; "
f"peers={peers}"
)
if not actual_peer_ids <= allowed_peer_ids:
raise ScenarioError(
f"{departed_author} departure observed unexpected authenticated peers: "
f"{actual_peer_ids}; peers={peers}"
)
events = peer.call_to_play_events()
last_events = events
if (
{event.get("id") for event in events} == expected_event_ids
and actual_peer_ids == expected_remaining_peer_ids
):
return events
time.sleep(0.2)
raise ScenarioError(
f"{departed_author} departure did not converge after the "
f"{PEER_DEPARTURE_TIMEOUT_SECONDS}s liveness allowance while peers "
f"{expected_remaining_peer_ids} stayed live: events={last_events}, "
f"peers={last_peers}"
)
def assert_exact_peer_ids(peer: Peer, expected_ids: set[str | None]) -> None:
if None in expected_ids:
raise ScenarioError("cannot assert an exact peer set with a missing identity")
peers = peer.list_peers()
actual_ids = {item.get("peer_id") for item in peers}
if actual_ids != expected_ids:
raise ScenarioError(
f"{peer.name} authenticated unexpected peers: {actual_ids} != "
f"{expected_ids}; peers={peers}"
)
def wait_call_to_play_exact_events_from_exact_peers(
peer: Peer,
expected_event_ids: set[str],
expected_peer_ids: set[str | None],
timeout: float = 20,
) -> list[dict[str, Any]]:
deadline = time.monotonic() + timeout
last_events: list[dict[str, Any]] = []
while time.monotonic() < deadline:
assert_exact_peer_ids(peer, expected_peer_ids)
events = peer.call_to_play_events()
last_events = events
if {event.get("id") for event in events} == expected_event_ids:
assert_exact_peer_ids(peer, expected_peer_ids)
return events
time.sleep(0.2)
raise ScenarioError(
f"{peer.name} never converged to exact Call to Play replacement "
f"{expected_event_ids} while restricted to peers {expected_peer_ids}: "
f"{last_events}"
)
def assert_game_state(
game: dict[str, Any],
*,
@@ -2216,6 +3624,7 @@ def wait_peer_has_game(
observer: Peer,
peer_id: str | None,
game_id: str,
content_id: str,
timeout: float = 20,
) -> dict[str, Any]:
if peer_id is None:
@@ -2229,11 +3638,42 @@ def wait_peer_has_game(
for peer in peers:
if peer.get("peer_id") != peer_id:
continue
if any(game.get("id") == game_id for game in peer.get("games", [])):
if any(
game.get("game_id") == game_id
and game.get("content_id") == content_id
for game in peer.get("games", [])
):
return peer
time.sleep(0.4)
raise ScenarioError(
f"{observer.name} never saw peer {peer_id} advertise {game_id}; peers={last_peers}"
f"{observer.name} never saw peer {peer_id} advertise exact {game_id}/{content_id}; "
f"peers={last_peers}"
)
def wait_peer_without_game(
observer: Peer,
peer_id: str | None,
game_id: str,
timeout: float = 20,
) -> dict[str, Any]:
if peer_id is None:
raise ScenarioError("cannot wait for a peer without peer_id")
deadline = time.monotonic() + timeout
last_peer: dict[str, Any] | None = None
while time.monotonic() < deadline:
for peer in observer.list_peers():
if peer.get("peer_id") != peer_id:
continue
last_peer = peer
if peer.get("library_revision") is None:
break
if all(game.get("game_id") != game_id for game in peer.get("games", [])):
return peer
time.sleep(0.4)
raise ScenarioError(
f"{observer.name} never observed an authoritative snapshot without {game_id} "
f"from {peer_id}; peer={last_peer}"
)
@@ -2267,6 +3707,287 @@ def assert_failed_stream_left_no_local(peer: Peer, game_id: str) -> None:
assert_not_exists(game_root / f"{game_id}.eti")
def assert_download_status_trace(
peer: Peer,
game_id: str,
*,
start: int,
terminal: str | None,
reason: str | None = None,
expect_begin: bool = True,
expect_verification: bool = True,
expect_invalid_source_retry: bool = False,
) -> str:
status_names = {
"download-begin",
"download-progress",
"download-activity-changed",
"download-finished",
"download-failed",
}
status_events = [
item
for item in peer.output[start:]
if item.get("type") == "event"
and item.get("event") in status_names
and item.get("data", {}).get("game_id") == game_id
]
if not status_events:
raise ScenarioError(f"no attempt-keyed download status events for {game_id}")
attempt_ids = {
item.get("data", {}).get("attempt_id") for item in status_events
}
if len(attempt_ids) != 1:
raise ScenarioError(
f"download status crossed attempt IDs for {game_id}: {status_events}"
)
attempt_id = next(iter(attempt_ids))
if (
not isinstance(attempt_id, str)
or not attempt_id.isascii()
or not attempt_id.isdigit()
or (len(attempt_id) > 1 and attempt_id.startswith("0"))
):
raise ScenarioError(f"attempt ID is not a canonical decimal string: {attempt_id!r}")
begins = [item for item in status_events if item.get("event") == "download-begin"]
expected_begin_count = 1 if expect_begin else 0
if len(begins) != expected_begin_count:
raise ScenarioError(
f"unexpected begin count for {game_id}: {len(begins)} != {expected_begin_count}"
)
activities = [
item.get("data", {}).get("activity")
for item in status_events
if item.get("event") == "download-activity-changed"
]
expected_activities: list[str | None] = []
if expect_verification:
expected_activities.append("verifying-downloaded-chunks")
if expect_invalid_source_retry:
expected_activities.append("retrying-invalid-source")
expected_activities.append(None)
if activities != expected_activities:
raise ScenarioError(
f"verification activity trace for {game_id} was {activities}, "
f"expected {expected_activities}"
)
activity_indices = [
index
for index, item in enumerate(status_events)
if item.get("event") == "download-activity-changed"
]
if begins and activity_indices and status_events.index(begins[0]) >= activity_indices[0]:
raise ScenarioError(
f"verification activity preceded download begin for {game_id}: {status_events}"
)
terminals = [
item
for item in status_events
if item.get("event") in {"download-finished", "download-failed"}
]
expected_terminal_count = 0 if terminal is None else 1
if len(terminals) != expected_terminal_count:
raise ScenarioError(
f"unexpected terminal count for {game_id}: {terminals}"
)
if terminal is not None:
observed = terminals[0]
if observed.get("event") != terminal:
raise ScenarioError(
f"download terminal for {game_id} was {observed.get('event')}, expected {terminal}"
)
observed_reason = observed.get("data", {}).get("reason")
if observed_reason != reason:
raise ScenarioError(
f"download terminal reason for {game_id} was {observed_reason!r}, "
f"expected {reason!r}"
)
terminal_index = status_events.index(observed)
if begins and status_events.index(begins[0]) >= terminal_index:
raise ScenarioError(
f"download terminal preceded begin for {game_id}: {status_events}"
)
if activities and not any(
index < terminal_index
and item.get("event") == "download-activity-changed"
and item.get("data", {}).get("activity") is None
for index, item in enumerate(status_events)
):
raise ScenarioError(
f"download activity did not clear before terminal for {game_id}: {status_events}"
)
elif reason is not None:
raise ScenarioError("a reason cannot be expected without a terminal event")
observed_milestones: list[tuple[str, str | None]] = []
for item in status_events:
event = item.get("event")
data = item.get("data", {})
if event == "download-begin":
observed_milestones.append((event, None))
elif event == "download-activity-changed":
observed_milestones.append((event, data.get("activity")))
elif event in {"download-finished", "download-failed"}:
observed_milestones.append((event, data.get("reason")))
expected_milestones: list[tuple[str, str | None]] = []
if expect_begin:
expected_milestones.append(("download-begin", None))
if expect_verification:
expected_milestones.append(
("download-activity-changed", "verifying-downloaded-chunks")
)
if expect_invalid_source_retry:
expected_milestones.append(
("download-activity-changed", "retrying-invalid-source")
)
expected_milestones.append(("download-activity-changed", None))
if terminal is not None:
expected_milestones.append((terminal, reason))
if observed_milestones != expected_milestones:
raise ScenarioError(
f"download status milestones for {game_id} were {observed_milestones}, "
f"expected {expected_milestones}"
)
final_status = status_events[-1]
if terminal is not None:
final_status_is_expected = final_status.get("event") == terminal
else:
final_status_is_expected = (
final_status.get("event") == "download-activity-changed"
and final_status.get("data", {}).get("activity") is None
)
if not final_status_is_expected:
raise ScenarioError(
f"download status continued after settlement for {game_id}: {status_events}"
)
return attempt_id
def assert_stream_install_lifecycle(
peer: Peer,
game_id: str,
*,
start: int,
content_id: str,
source_peer_id: str | None,
source_addr: str | None,
expect_invalid_source_retry: bool = False,
) -> None:
if source_peer_id is None or source_addr is None:
raise ScenarioError("cannot prove streamed lifecycle without an exact source endpoint")
window = peer.output[start:]
failure_events = [
item
for item in window
if item.get("type") == "event"
and item.get("event")
in {"download-failed", "install-failed"}
and item.get("data", {}).get("game_id") == game_id
]
if failure_events:
raise ScenarioError(
f"successful streamed install emitted failure events for {game_id}: "
f"{failure_events}"
)
lifecycle: list[tuple[Any, ...]] = []
expected_chunks = {
"bin/cnctw-payload.bin": 2 * 1024 * 1024,
"data/cnctw-assets.dat": 1024 * 1024,
}
for item in window:
if item.get("type") != "event":
continue
event = item.get("event")
data = item.get("data", {})
if event == "active-operations-changed":
active_operations = data.get("active_operations", [])
matching = [
active
for active in active_operations
if active.get("game_id") == game_id
]
if matching:
if len(matching) != 1:
raise ScenarioError(
f"streamed install published duplicate active entries: {item}"
)
lifecycle.append(("active", matching[0].get("operation")))
elif active_operations == []:
lifecycle.append(("active", "empty"))
continue
if data.get("game_id") != game_id:
continue
if event == "download-chunk-finished":
path = data.get("relative_path")
if (
path not in expected_chunks
or int(data.get("offset", -1)) != 0
or int(data.get("length", -1)) != expected_chunks[path]
or data.get("content_id") != content_id
or data.get("peer_id") != source_peer_id
or data.get("peer_addr") != source_addr
):
raise ScenarioError(
f"streamed install emitted a non-authoritative verified chunk: {data}"
)
lifecycle.append(("chunk", path))
continue
if event == "download-activity-changed":
lifecycle.append(("activity", data.get("activity")))
continue
if event == "download-finished":
throughput = data.get("throughput", {})
if throughput.get("bytes") != 3 * 1024 * 1024 or throughput.get("chunks") != 2:
raise ScenarioError(
f"streamed install throughput was not exact: {throughput}"
)
if event in {
"download-begin",
"download-finished",
"install-finished",
"download-failed",
"install-failed",
}:
lifecycle.append((event,))
expected_lifecycle = [
("active", "Downloading"),
("download-begin",),
("activity", "verifying-downloaded-chunks"),
]
if expect_invalid_source_retry:
expected_lifecycle.append(("activity", "retrying-invalid-source"))
expected_lifecycle.extend([
("chunk", "bin/cnctw-payload.bin"),
("chunk", "data/cnctw-assets.dat"),
("activity", None),
("active", "Installing"),
("download-finished",),
("active", "empty"),
("install-finished",),
])
if lifecycle != expected_lifecycle:
raise ScenarioError(
f"streamed install lifecycle for {game_id} was not exact: "
f"{lifecycle} != {expected_lifecycle}"
)
assert_download_status_trace(
peer,
game_id,
start=start,
terminal="download-finished",
expect_invalid_source_retry=expect_invalid_source_retry,
)
def event_is(event: str, game_id: str | None = None) -> Callable[[dict[str, Any]], bool]:
def predicate(item: dict[str, Any]) -> bool:
if item.get("type") != "event" or item.get("event") != event:
@@ -2312,7 +4033,7 @@ def assert_only_chunk_sources(
data = item["data"]
if data.get("game_id") != game_id:
continue
source = data.get("peer_addr")
source = data.get("peer_id")
seen.add(source)
if source not in allowed:
raise ScenarioError(f"unexpected chunk source for {game_id}: {data}")
@@ -2321,6 +4042,37 @@ def assert_only_chunk_sources(
raise ScenarioError(f"no chunk events recorded for {game_id}")
def assert_chunk_authority(
peer: Peer,
game_id: str,
content_id: str,
allowed_peer_ids: set[str | None],
*,
start: int = 0,
) -> None:
allowed = {peer_id for peer_id in allowed_peer_ids if peer_id is not None}
if not allowed:
raise ScenarioError("no authenticated chunk source identities supplied")
chunks = [
item["data"]
for item in peer.output[start:]
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
and item.get("data", {}).get("game_id") == game_id
]
if not chunks:
raise ScenarioError(f"no verified chunk events recorded for {game_id}")
for chunk in chunks:
if chunk.get("content_id") != content_id:
raise ScenarioError(
f"chunk used wrong content identity for {game_id}: {chunk}"
)
if chunk.get("peer_id") not in allowed:
raise ScenarioError(
f"chunk used unauthenticated/unexpected source for {game_id}: {chunk}"
)
def streamed_chunk_paths(peer: Peer, game_id: str) -> list[str]:
return [
item["data"]["relative_path"]
@@ -2339,13 +4091,13 @@ def chunk_totals(peer: Peer, game_id: str, relative_path: str) -> dict[str, int]
data = item["data"]
if data.get("game_id") != game_id or data.get("relative_path") != relative_path:
continue
totals[data["peer_addr"]] = totals.get(data["peer_addr"], 0) + int(data["length"])
totals[data["peer_id"]] = totals.get(data["peer_id"], 0) + int(data["length"])
return totals
def chunk_sources(peer: Peer, game_id: str) -> set[str]:
return {
item["data"]["peer_addr"]
item["data"]["peer_id"]
for item in peer.output
if item.get("type") == "event"
and item.get("event") == "download-chunk-finished"
@@ -2377,27 +4129,6 @@ def count_events(peer: Peer, event: str, game_id: str) -> int:
)
def peer_advertised_version(
observer: Peer, peer_id: str | None, game_id: str, timeout: float = 20
) -> str | None:
"""Returns the raw `eti_version` a peer advertises for a game in its library
snapshot (list-peers). Unlike the list-games `remote` rows, this value is NOT
synthesized from the local catalog, so it faithfully reports the source.
Polls until the peer's library snapshot is observed."""
if peer_id is None:
raise ScenarioError("cannot read advertised version without peer_id")
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
for peer in observer.list_peers():
if peer.get("peer_id") != peer_id:
continue
for game in peer.get("games", []):
if game.get("id") == game_id:
return game.get("eti_version")
time.sleep(0.4)
raise ScenarioError(f"{observer.name} does not see {game_id} advertised by {peer_id}")
def diff_game_dirs(source: Path, destination: Path) -> None:
source_manifest = manifest(source)
destination_manifest = manifest(destination)