test(peer-cli): cover streamed retry fallback

NEXT_STEPS item 5 needs streamed installs to have an explicit retry
policy. The handler already retries whole-stream attempts across the
majority-validated peer set, so add S42 to prove that behavior with the
Docker harness instead of leaving it implicit.

S42 starts two catalog-version-matching `cnctw` sources. The first source
sorts first in retry order but has `--unrar /missing-unrar`, so its stream
attempt fails before sending chunks. The second source then completes a
fresh whole-stream attempt. The scenario asserts local-only installed
state, no root archive or sentinel, no `.local.installing` staging
leftover, chunk events only from the good source, matching streamed byte
count, and SHA-256 payload equality against the good source's `unrar p`.

This pins the current policy: retry the entire stream from another
validated peer, do not preserve partial files across attempts, and do not
promise byte-offset resume.

Test Plan:
- python3 -m py_compile crates/lanspread-peer-cli/scripts/run_extended_scenarios.py
- python3 crates/lanspread-peer-cli/scripts/run_extended_scenarios.py S42
- git diff --check
- git diff --cached --check

Refs: NEXT_STEPS.md item 5

crates/lanspread-peer-cli/scripts/run_extended_scenarios.py

@@ -1,10 +1,11 @@
 #!/usr/bin/env python3
-"""Run the peer-cli scenarios S1-S41 through Docker."""
+"""Run the peer-cli scenarios S1-S42 through Docker."""
 
 from __future__ import annotations
 
 import argparse
 import hashlib
+import ipaddress
 import json
 import os
 import queue
@@ -329,6 +330,7 @@ class Runner:
             ("S39", self.s39_streamed_install_local_only),
             ("S40", self.s40_streamed_receiver_not_source),
             ("S41", self.s41_solid_archive_streamed_install),
+            ("S42", self.s42_streamed_install_retries_next_source),
         ]
 
         for scenario_id, scenario in scenarios:
@@ -1250,6 +1252,95 @@ class Runner:
             f"payload hashes={actual}, bytes={streamed_bytes}"
         )
 
+    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")
+        copy_game("cnctw", good_dir, version="20160128")
+
+        bad = self.peer(
+            "s42-bad-source",
+            games_dir=bad_dir,
+            extra_args=["--unrar", "/missing-unrar"],
+        )
+        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):
+            raise ScenarioError(
+                "S42 requires the broken source to sort before the good source; "
+                f"bad={bad.ready_addr}, good={good.ready_addr}"
+            )
+
+        client = self.peer("s42-client")
+        connect_many(client, [bad, good])
+        wait_remote_game(client, "cnctw", peer_count=2, version="20160128")
+
+        waiter = LineWaiter(len(client.output))
+        client.send({"cmd": "stream-install", "game_id": "cnctw"})
+        client.wait_for(
+            event_is("got-game-files", "cnctw"),
+            timeout=20,
+            description="got retry cnctw files",
+            waiter=waiter,
+        )
+        client.wait_for(
+            event_is("download-finished", "cnctw"),
+            timeout=60,
+            description="retry stream finish cnctw",
+            waiter=waiter,
+        )
+        client.wait_for(
+            event_is("install-finished", "cnctw"),
+            timeout=30,
+            description="retry stream install cnctw",
+            waiter=waiter,
+        )
+
+        game = wait_local_game(client, "cnctw", downloaded=False, installed=True)
+        assert_game_state(
+            game,
+            downloaded=False,
+            installed=True,
+            availability="LocalOnly",
+        )
+        game_root = client.host_games_dir / "cnctw"
+        assert_not_exists(game_root / ".local.installing")
+        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 = {
+            rel: sha256_file(game_root / "local" / rel)
+            for rel in expected
+        }
+        if actual != expected:
+            raise ScenarioError(f"retry streamed payload hashes mismatched: {actual} != {expected}")
+
+        streamed_bytes = sum(
+            int(item.get("data", {}).get("length", 0))
+            for item in client.output
+            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:
+            raise ScenarioError(
+                f"retry streamed byte count mismatch: {streamed_bytes} != {expected_bytes}"
+            )
+
+        return (
+            "broken first source failed without chunks, next source completed whole stream; "
+            f"good={good.ready_addr}, bad={bad.ready_addr}, bytes={streamed_bytes}"
+        )
+
 
 def run(command: list[str], description: str) -> subprocess.CompletedProcess[str]:
     result = subprocess.run(
@@ -1402,6 +1493,14 @@ 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 format_bytes(size: int) -> str:
     return f"{size / 1024 / 1024 / 1024:.2f} GiB"