From bdefcb999512b15f221f13c2b9ea2faab0fb4bc8 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Wed, 2 Sep 2026 22:48:39 -0700 Subject: [PATCH] Add exact LoRa OTA route cost objective --- tools/lora_ota/rak3401_route_search.py | 749 +++++++++++++++++++- tools/lora_ota/test_rak3401_route_search.py | 334 ++++++++- 2 files changed, 1042 insertions(+), 41 deletions(-) diff --git a/tools/lora_ota/rak3401_route_search.py b/tools/lora_ota/rak3401_route_search.py index 1618e4440..f1e9dea3d 100755 --- a/tools/lora_ota/rak3401_route_search.py +++ b/tools/lora_ota/rak3401_route_search.py @@ -4,6 +4,12 @@ Cache entries are keyed by source SHA, target SHA, and workspace size rather than inventory indices. A cache generated for an older inventory can therefore be migrated safely when new candidate images are inserted. + +The default secondary objective remains container size. ``--objective transport`` +instead measures the exact live motatool encoder and minimizes clean serialized +MeshCore bytes. Inventory boolean ``ota_transport_deflate`` and integer +``ota_fetch_pipeline`` belong to the currently running/source image; the pinned +first bootstrap is always legacy raw. """ from __future__ import annotations @@ -16,6 +22,7 @@ import os from pathlib import Path import re import shutil +import subprocess import sys import tempfile @@ -34,10 +41,25 @@ FIXED_MEMORY = 0x98000 AVAILABLE_PAGES = (STAGE_CEILING - APP_BASE) // PAGE EXPECTED_TARGET_ID = 0x2FA509C1 EXPECTED_HARDWARE = "RAK_3401" +TRANSPORT_CAPABILITY = "ota_transport_deflate" +TRANSPORT_PIPELINE_FIELD = "ota_fetch_pipeline" +TRANSPORT_CAPABILITY_ALIAS = "transport_inflate" +TRANSPORT_PIPELINE_ALIAS = "fetch_pipeline" +BLOCK_SIZE = 1024 +LEGACY_DATA_BYTES = 160 +V2_DATA_BYTES = 171 +MANIFEST_BYTES = 197 +MANIFEST_FRAGMENT_BYTES = 176 +RAK3401_PIPELINE_MAX = 4 +OTA_MAX_HOPS = 8 +TRANSPORT_FIELDS = [ + "payload_sha256", "transport_encoder_sha256", "v2_wire_bytes", + "v2_deflate_bytes", "v2_deflate_blocks", "v2_data_packets", +] FIELDS = [ "source", "target", "source_sha256", "target_sha256", "memory", "payload", "container", "stage_start", "margin", "feasible", "error", -] +] + TRANSPORT_FIELDS class RouteSearchError(RuntimeError): @@ -56,6 +78,189 @@ def container_size(payload_size: int) -> int: return payload_size + 210 + 4 * ((payload_size + 1023) // 1024) +def payload_size_from_container(total: int) -> int: + """Invert the fixed mOTA geometry used by the pinned bootstrap package.""" + if total <= 210: + raise RouteSearchError("container is too small to contain a payload") + max_blocks = (total - 210 + BLOCK_SIZE - 1) // BLOCK_SIZE + for blocks in range(1, max_blocks + 1): + payload = total - 210 - 4 * blocks + if payload > 0 and (payload + BLOCK_SIZE - 1) // BLOCK_SIZE == blocks: + return payload + raise RouteSearchError(f"container size {total} has no valid payload geometry") + + +def _block_lengths(payload_size: int) -> list[int]: + if payload_size <= 0: + raise RouteSearchError("transport payload must be positive") + full, tail = divmod(payload_size, BLOCK_SIZE) + result = [BLOCK_SIZE] * full + if tail: + result.append(tail) + return result + + +def proof_sibling_count(block_count: int, block_index: int) -> int: + """Match the promote-odd Merkle proof emitted by MeshCore.""" + if block_count <= 0 or not 0 <= block_index < block_count: + raise RouteSearchError("invalid block for Merkle proof sizing") + siblings = 0 + count = block_count + index = block_index + while count > 1: + if not (count % 2 == 1 and index == count - 1): + siblings += 1 + index //= 2 + count = (count + 1) // 2 + return siblings + + +def _common_transport_cost(block_count: int) -> tuple[int, int, int]: + """Return origin-serialized Mesh bytes, packets, and proof-response bytes.""" + manifest_fragments = ( + MANIFEST_BYTES + MANIFEST_FRAGMENT_BYTES - 1 + ) // MANIFEST_FRAGMENT_BYTES + # GET_MANIFEST is 7 OTA bytes; each MANIFEST has a 7-byte OTA header. + # Every source frame also has MeshCore's header + path-length bytes. + manifest_source_bytes = 7 + 2 + MANIFEST_BYTES + 9 * manifest_fragments + proof_source_bytes = sum( + 10 + 4 * proof_sibling_count(block_count, block) + for block in range(block_count) + ) + return ( + manifest_source_bytes + proof_source_bytes, + 1 + manifest_fragments + block_count, + proof_source_bytes, + ) + + +def legacy_transport_cost(payload_size: int) -> dict[str, int | str]: + """Ideal clean legacy transfer, including requests, DATA, and proofs.""" + blocks = _block_lengths(payload_size) + block_count = len(blocks) + data_packets = sum( + (length + LEGACY_DATA_BYTES - 1) // LEGACY_DATA_BYTES + for length in blocks + ) + common_bytes, common_packets, proof_bytes = _common_transport_cost(block_count) + # DATA is 9 OTA header + 2 Mesh bytes. Each legacy block also needs one + # 9-byte OTA_REQ frame and one 7-byte OTA_REQ_PROOF frame. + data_bytes = payload_size + 11 * data_packets + request_bytes = 11 * block_count + proof_request_bytes = 9 * block_count + manifest_bytes = common_bytes - proof_bytes + return { + "profile": "legacy-160-raw", + "payload_bytes": payload_size, + "wire_bytes": payload_size, + "deflate_bytes": 0, + "deflate_blocks": 0, + "data_packets": data_packets, + "manifest_packets": common_packets - block_count, + "request_pipeline": 1, + "request_packets": block_count, + "proof_request_packets": block_count, + "proof_packets": block_count, + "origin_mesh_bytes": ( + common_bytes + data_bytes + request_bytes + proof_request_bytes + ), + "packets": common_packets + data_packets + 2 * block_count, + "manifest_bytes": manifest_bytes, + "block_request_bytes": request_bytes, + "proof_request_bytes": proof_request_bytes, + "proof_response_bytes": proof_bytes, + "data_bytes": data_bytes, + "request_bytes": request_bytes + proof_request_bytes, + } + + +def _v2_request_flights( + block_count: int, pipeline_capacity: int = RAK3401_PIPELINE_MAX +) -> list[int]: + if block_count <= 0 or not 1 <= pipeline_capacity <= RAK3401_PIPELINE_MAX: + raise RouteSearchError("invalid v2 request-flight geometry") + flights: list[int] = [] + remaining = block_count + width = 1 + while remaining: + active = min(width, remaining) + flights.append(active) + remaining -= active + width = min(width + 1, pipeline_capacity) + return flights + + +def v2_transport_cost( + payload_size: int, + wire_bytes: int, + deflate_bytes: int, + deflate_blocks: int, + data_packets: int, + pipeline_capacity: int = RAK3401_PIPELINE_MAX, +) -> dict[str, int | str]: + """Ideal clean negotiated v2 transfer using measured seeder output.""" + blocks = _block_lengths(payload_size) + block_count = len(blocks) + raw_data_packets = sum( + (length + V2_DATA_BYTES - 1) // V2_DATA_BYTES for length in blocks + ) + if not (0 < wire_bytes <= payload_size): + raise RouteSearchError("v2 wire-byte count is inconsistent") + if not (0 <= deflate_bytes <= wire_bytes): + raise RouteSearchError("v2 DEFLATE-byte count is inconsistent") + if not (0 <= deflate_blocks <= block_count): + raise RouteSearchError("v2 DEFLATE-block count is inconsistent") + if not (block_count <= data_packets <= raw_data_packets): + raise RouteSearchError("v2 DATA-packet count is inconsistent") + if (deflate_blocks == 0) != (deflate_bytes == 0): + raise RouteSearchError("v2 DEFLATE totals disagree") + if not 1 <= pipeline_capacity <= RAK3401_PIPELINE_MAX: + raise RouteSearchError("v2 request pipeline is out of range") + + common_bytes, common_packets, proof_bytes = _common_transport_cost(block_count) + flights = _v2_request_flights(block_count, pipeline_capacity) + # DATA: 9-byte existing header + repeated 4-byte stream id + 2 Mesh bytes. + data_bytes = wire_bytes + 15 * data_packets + # One request frame per adaptive flight: 5-byte prefix + 4 bytes/block + 2 Mesh bytes. + request_bytes = sum(7 + 4 * width for width in flights) + manifest_bytes = common_bytes - proof_bytes + return { + "profile": "v2-171-deflate", + "payload_bytes": payload_size, + "wire_bytes": wire_bytes, + "deflate_bytes": deflate_bytes, + "deflate_blocks": deflate_blocks, + "data_packets": data_packets, + "manifest_packets": common_packets - block_count, + "request_pipeline": pipeline_capacity, + "request_packets": len(flights), + "proof_request_packets": 0, + "proof_packets": block_count, + "origin_mesh_bytes": common_bytes + data_bytes + request_bytes, + "packets": common_packets + data_packets + len(flights), + "manifest_bytes": manifest_bytes, + "block_request_bytes": request_bytes, + "proof_request_bytes": 0, + "proof_response_bytes": proof_bytes, + "data_bytes": data_bytes, + "request_bytes": request_bytes, + } + + +def linear_path_bytes(origin_mesh_bytes: int, packets: int, relay_hops: int) -> int: + """Serialized bytes for an ideal single path; real flood fan-out is external.""" + if origin_mesh_bytes < 0 or packets < 0: + raise RouteSearchError("transport byte inputs cannot be negative") + if not 0 <= relay_hops <= OTA_MAX_HOPS: + raise RouteSearchError(f"relay hops must be between 0 and {OTA_MAX_HOPS}") + transmissions = relay_hops + 1 + # Each relay retransmits the frame after appending one one-byte path hash. + return ( + transmissions * origin_mesh_bytes + + packets * relay_hops * transmissions // 2 + ) + + def truth(value: object) -> bool: return value is True or str(value).lower() == "true" @@ -89,8 +294,29 @@ def load_inventory(path: Path) -> list[dict[str, object]]: if not isinstance(images, list) or len(images) < 2: raise RouteSearchError(f"invalid image inventory: {path}") + capability_contract = document.get("capability_contract", {}) + strict_transport_contract = ( + isinstance(capability_contract, dict) + and TRANSPORT_CAPABILITY in capability_contract + ) or any( + isinstance(record, dict) and TRANSPORT_CAPABILITY in record + for record in images + ) + edge_policy = document.get("edge_policy", {}) + strict_version_ranks = ( + isinstance(edge_policy, dict) + and edge_policy.get("require_target_version_rank_gt_source_version_rank") + is True + ) or any( + isinstance(record, dict) and "version_rank" in record + for record in images + ) + result: list[dict[str, object]] = [] seen: set[str] = set() + previous_version_rank = -1 + rank_by_version: dict[str, int] = {} + version_by_rank: dict[int, str] = {} for index, raw in enumerate(images): if not isinstance(raw, dict): raise RouteSearchError(f"inventory image {index} is not an object") @@ -153,6 +379,76 @@ def load_inventory(path: Path) -> list[dict[str, object]]: or not re.fullmatch(r"[0-9a-f]{40}", source_commit) ): raise RouteSearchError(f"image {index} source commit is invalid") + if TRANSPORT_CAPABILITY in record: + capability = record[TRANSPORT_CAPABILITY] + elif strict_transport_contract: + raise RouteSearchError( + f"image {index} {TRANSPORT_CAPABILITY} is required by the " + "inventory capability contract" + ) + else: + # Compatibility for pre-contract inventories. New inventories must + # carry the canonical field explicitly so transport is fail-closed. + capability = record.get(TRANSPORT_CAPABILITY_ALIAS, False) + if not isinstance(capability, bool): + raise RouteSearchError( + f"image {index} {TRANSPORT_CAPABILITY} must be boolean" + ) + alias_capability = record.get(TRANSPORT_CAPABILITY_ALIAS, capability) + if not isinstance(alias_capability, bool) or alias_capability != capability: + raise RouteSearchError( + f"image {index} {TRANSPORT_CAPABILITY_ALIAS} disagrees with " + f"{TRANSPORT_CAPABILITY}" + ) + record[TRANSPORT_CAPABILITY] = capability + raw_pipeline = record.get(TRANSPORT_PIPELINE_FIELD) + if ( + raw_pipeline is None + and not strict_transport_contract + and TRANSPORT_PIPELINE_ALIAS in record + ): + raw_pipeline = record[TRANSPORT_PIPELINE_ALIAS] + if capability and raw_pipeline is None: + raise RouteSearchError( + f"image {index} {TRANSPORT_PIPELINE_FIELD} is required when " + f"{TRANSPORT_CAPABILITY} is true" + ) + if capability: + pipeline = _integer( + raw_pipeline, f"image {index} {TRANSPORT_PIPELINE_FIELD}" + ) + if not 1 <= pipeline <= RAK3401_PIPELINE_MAX: + raise RouteSearchError( + f"image {index} {TRANSPORT_PIPELINE_FIELD} must be between 1 " + f"and {RAK3401_PIPELINE_MAX}" + ) + record[TRANSPORT_PIPELINE_FIELD] = pipeline + elif raw_pipeline is not None and strict_transport_contract: + raise RouteSearchError( + f"image {index} {TRANSPORT_PIPELINE_FIELD} must be null when " + f"{TRANSPORT_CAPABILITY} is false" + ) + else: + record[TRANSPORT_PIPELINE_FIELD] = None + + if "version_rank" not in record and strict_version_ranks: + raise RouteSearchError( + f"image {index} version_rank is required by the inventory edge policy" + ) + version_rank = _integer( + record.get("version_rank", index), f"image {index} version_rank" + ) + if version_rank < previous_version_rank: + raise RouteSearchError("inventory version_rank values must be nondecreasing") + if strict_version_ranks: + known_rank = rank_by_version.setdefault(version, version_rank) + known_version = version_by_rank.setdefault(version_rank, version) + if known_rank != version_rank or known_version != version: + raise RouteSearchError( + "inventory version_rank must map one-to-one with firmware version" + ) + previous_version_rank = version_rank + record["version_rank"] = version_rank if index == 1 and _integer( record.get("baseline_container_size"), "node 1 baseline_container_size", @@ -231,11 +527,88 @@ def cache_key(source_sha: str, target_sha: str, memory: int) -> tuple[str, str, return source_sha.lower(), target_sha.lower(), memory +def transport_stats_complete(row: dict[str, object]) -> bool: + return all(row.get(field) not in (None, "") for field in TRANSPORT_FIELDS) + + +def normalize_transport_stats( + row: dict[str, object], payload_size: int, label: str +) -> dict[str, object]: + present = [row.get(field) not in (None, "") for field in TRANSPORT_FIELDS] + if any(present) and not all(present): + raise RouteSearchError(f"partial transport statistics at {label}") + if not any(present): + return {field: "" for field in TRANSPORT_FIELDS} + + normalized: dict[str, object] = {} + for field in TRANSPORT_FIELDS[:2]: + digest = row.get(field) + if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest): + raise RouteSearchError(f"invalid {field} at {label}") + normalized[field] = digest + for field in TRANSPORT_FIELDS[2:]: + normalized[field] = _integer(row.get(field), f"{field} at {label}") + v2_transport_cost( + payload_size, + int(normalized["v2_wire_bytes"]), + int(normalized["v2_deflate_bytes"]), + int(normalized["v2_deflate_blocks"]), + int(normalized["v2_data_packets"]), + ) + return normalized + + +def measure_transport_size(tool: Path, payload: Path) -> dict[str, object]: + """Ask motatool's live-seeder encoder for exact per-block byte counts.""" + try: + completed = subprocess.run( + [str(tool), "transport-size", "--payload", str(payload), + "--block-size", str(BLOCK_SIZE)], + check=False, capture_output=True, text=True, encoding="utf-8", + timeout=60, + ) + except subprocess.TimeoutExpired as exc: + raise RouteSearchError(f"transport-size tool timed out for {payload}") from exc + except OSError as exc: + raise RouteSearchError(f"cannot run transport-size tool {tool}: {exc}") from exc + if completed.returncode != 0: + detail = (completed.stderr or completed.stdout).strip()[-500:] + raise RouteSearchError( + f"transport-size tool failed for {payload}: {detail or completed.returncode}" + ) + try: + measured = json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise RouteSearchError("transport-size tool returned invalid JSON") from exc + if not isinstance(measured, dict) or measured.get("schema") != 1: + raise RouteSearchError("transport-size tool returned an unsupported result") + expected_payload = payload.stat().st_size + if measured.get("payload_bytes") != expected_payload: + raise RouteSearchError("transport-size tool measured the wrong payload") + if measured.get("block_size") != BLOCK_SIZE: + raise RouteSearchError("transport-size tool used the wrong block size") + expected_blocks = (expected_payload + BLOCK_SIZE - 1) // BLOCK_SIZE + if measured.get("block_count") != expected_blocks: + raise RouteSearchError("transport-size tool returned the wrong block count") + row = { + "payload_sha256": sha256_file(payload), + "transport_encoder_sha256": sha256_file(tool), + "v2_wire_bytes": measured.get("wire_bytes"), + "v2_deflate_bytes": measured.get("deflate_bytes"), + "v2_deflate_blocks": measured.get("deflate_blocks"), + "v2_data_packets": measured.get("data_packets"), + } + return normalize_transport_stats(row, expected_payload, str(payload)) + + def migrate_csv( inventory_path: Path, csv_path: Path ) -> dict[tuple[str, str, int], dict[str, object]]: - """Migrate numeric CSV indices through the inventory that created them.""" + """Migrate cached edges by hash, with numeric-index fallback for old CSVs.""" images = load_inventory(inventory_path) + nodes_by_sha = { + str(image["sha256"]): int(image["node"]) for image in images + } migrated: dict[tuple[str, str, int], dict[str, object]] = {} try: source = csv_path.open(newline="", encoding="ascii") @@ -247,15 +620,18 @@ def migrate_csv( if reader.fieldnames is None or not required_fields.issubset(reader.fieldnames): raise RouteSearchError(f"geometry cache has missing columns: {csv_path}") for line, row in enumerate(reader, 2): - source_node = _integer(row["source"], f"source at {csv_path}:{line}") - target_node = _integer(row["target"], f"target at {csv_path}:{line}") - if not (0 <= source_node < target_node < len(images)): + recorded_source = _integer( + row["source"], f"source at {csv_path}:{line}" + ) + recorded_target = _integer( + row["target"], f"target at {csv_path}:{line}" + ) + if recorded_source >= recorded_target: raise RouteSearchError(f"bad nodes at {csv_path}:{line}") - source_sha = str(images[source_node]["sha256"]) - target_sha = str(images[target_node]["sha256"]) - memory = _integer(row["memory"], f"memory at {csv_path}:{line}") - for column, expected in ( - ("source_sha256", source_sha), ("target_sha256", target_sha) + resolved: list[tuple[int, str]] = [] + for column, recorded_node in ( + ("source_sha256", recorded_source), + ("target_sha256", recorded_target), ): raw_recorded = row.get(column, "") if raw_recorded is None: @@ -264,11 +640,28 @@ def migrate_csv( raise RouteSearchError( f"{column} is invalid at {csv_path}:{line}" ) - recorded = raw_recorded.lower() - if recorded and recorded != expected: - raise RouteSearchError( - f"{column} disagrees with inventory at {csv_path}:{line}" + recorded_sha = raw_recorded.lower() + if recorded_sha: + if not re.fullmatch(r"[0-9a-f]{64}", recorded_sha): + raise RouteSearchError( + f"{column} is invalid at {csv_path}:{line}" + ) + resolved_node = nodes_by_sha.get(recorded_sha) + if resolved_node is None: + raise RouteSearchError( + f"{column} is absent from inventory at {csv_path}:{line}" + ) + resolved.append((resolved_node, recorded_sha)) + else: + if recorded_node >= len(images): + raise RouteSearchError(f"bad nodes at {csv_path}:{line}") + resolved.append( + (recorded_node, str(images[recorded_node]["sha256"])) ) + (source_node, source_sha), (target_node, target_sha) = resolved + if source_node >= target_node: + raise RouteSearchError(f"bad nodes at {csv_path}:{line}") + memory = _integer(row["memory"], f"memory at {csv_path}:{line}") try: payload = int(row.get("payload", "")) total = int(row.get("container", "")) @@ -306,6 +699,9 @@ def migrate_csv( normalized.update(source=source_node, target=target_node, source_sha256=source_sha, target_sha256=target_sha, memory=memory) + normalized.update(normalize_transport_stats( + row, payload, f"{csv_path}:{line}" + )) previous = migrated.get(key) comparable = ("payload", "container", "stage_start", "margin", "feasible", "error") if previous is not None and any( @@ -313,6 +709,16 @@ def migrate_csv( for item in comparable ): raise RouteSearchError(f"conflicting cached geometry for {key}") + if previous is not None and transport_stats_complete(previous): + if transport_stats_complete(normalized) and any( + str(previous.get(item, "")) != str(normalized.get(item, "")) + for item in TRANSPORT_FIELDS + ): + raise RouteSearchError(f"conflicting transport statistics for {key}") + if not transport_stats_complete(normalized): + normalized.update( + {field: previous[field] for field in TRANSPORT_FIELDS} + ) migrated[key] = normalized return migrated @@ -320,6 +726,13 @@ def migrate_csv( Job = tuple[int, int, int, str, str, str, str] +def is_forward_progress( + images: list[dict[str, object]], source: int, target: int +) -> bool: + """Only a strictly newer firmware version may be a route edge.""" + return int(images[target]["version_rank"]) > int(images[source]["version_rank"]) + + def all_jobs(images: list[dict[str, object]]) -> list[Job]: """Enumerate every valid page-aligned workspace for every forward edge.""" jobs: list[Job] = [] @@ -328,12 +741,16 @@ def all_jobs(images: list[dict[str, object]]) -> list[Job]: # Node 1 still runs the fixed-workspace receiver, so measure each of its # possible forward transitions at exactly that legacy workspace. for target in range(2, endpoint + 1): + if not is_forward_progress(images, 1, target): + continue jobs.append((1, target, FIXED_MEMORY, str(images[1]["path"]), str(images[target]["path"]), str(images[1]["sha256"]), str(images[target]["sha256"]))) for source in range(2, endpoint): source_size = int(images[source]["size"]) for target in range(source + 1, endpoint + 1): + if not is_forward_progress(images, source, target): + continue target_size = int(images[target]["size"]) first_page = align_up(max(source_size + 2 * PAGE, target_size)) // PAGE for memory_page in range(first_page, AVAILABLE_PAGES): @@ -343,7 +760,30 @@ def all_jobs(images: list[dict[str, object]]) -> list[Job]: return jobs -def geometry_job(args: tuple[int, int, int, str, str, str, str, str]) -> dict[str, object]: +def job_needs_transport_stats(job: Job, images: list[dict[str, object]], + objective: str) -> bool: + return objective == "transport" and source_uses_v2(images, job[0]) + + +def cache_satisfies_job( + row: dict[str, object] | None, + needs_transport: bool, + expected_encoder_sha256: str | None = None, +) -> bool: + return row is not None and ( + not needs_transport + or not truth(row.get("feasible", False)) + or ( + transport_stats_complete(row) + and ( + expected_encoder_sha256 is None + or row.get("transport_encoder_sha256") == expected_encoder_sha256 + ) + ) + ) + + +def geometry_job(args: tuple[object, ...]) -> dict[str, object]: try: import detools except ImportError as exc: @@ -351,7 +791,18 @@ def geometry_job(args: tuple[int, int, int, str, str, str, str, str]) -> dict[st "detools is required to generate geometry; run this tool in the " "detools pipx environment" ) from exc - source, target, memory, from_raw, to_raw, source_sha, target_sha, patches_raw = args + if len(args) not in (8, 9): + raise RouteSearchError("invalid geometry job") + source, target, memory, from_raw, to_raw, source_sha, target_sha, patches_raw = args[:8] + transport_tool = str(args[8]) if len(args) == 9 and args[8] else "" + source = int(source) + target = int(target) + memory = int(memory) + from_raw = str(from_raw) + to_raw = str(to_raw) + source_sha = str(source_sha) + target_sha = str(target_sha) + patches_raw = str(patches_raw) patch = Path(patches_raw) / f"{source_sha}-{target_sha}-{memory // PAGE:03d}.patch" try: detools.create_patch_filenames( @@ -364,6 +815,11 @@ def geometry_job(args: tuple[int, int, int, str, str, str, str, str]) -> dict[st stage_start = align_down(STAGE_CEILING - total) margin = stage_start - (APP_BASE + memory) error = "" + transport = ( + measure_transport_size(Path(transport_tool), patch) + if transport_tool and margin >= 0 + else {field: "" for field in TRANSPORT_FIELDS} + ) except Exception as exc: # detools reports several backend exception types detail = f"{type(exc).__name__}: {exc}"[-500:].replace("\n", " ") raise RouteSearchError( @@ -374,7 +830,8 @@ def geometry_job(args: tuple[int, int, int, str, str, str, str, str]) -> dict[st return {"source": source, "target": target, "source_sha256": source_sha, "target_sha256": target_sha, "memory": memory, "payload": payload, "container": total, "stage_start": stage_start, "margin": margin, - "feasible": payload >= 0 and margin >= 0, "error": error} + "feasible": payload >= 0 and margin >= 0, "error": error, + **transport} def project_cache(cache: dict[tuple[str, str, int], dict[str, object]], @@ -457,31 +914,127 @@ def write_csv(path: Path, rows: list[dict[str, object]]) -> None: raise +def source_uses_v2(images: list[dict[str, object]], source: int) -> bool: + """Capability is the currently running/source image; bootstrap is raw.""" + return source != 0 and bool(images[source][TRANSPORT_CAPABILITY]) + + +def source_transport_pipeline(images: list[dict[str, object]], source: int) -> int: + """Return the request-flight capacity compiled into the source image.""" + pipeline = images[source][TRANSPORT_PIPELINE_FIELD] + if pipeline is None: + raise RouteSearchError(f"source image {source} has no transport pipeline") + return int(pipeline) + + +def transport_encoder_for_rows( + rows: list[dict[str, object]], images: list[dict[str, object]] +) -> str | None: + """Require one exact live-seeder encoder across comparable measurements.""" + encoders = { + str(row["transport_encoder_sha256"]) + for row in rows + if truth(row.get("feasible", False)) + and source_uses_v2(images, int(row["source"])) + and transport_stats_complete(row) + } + if len(encoders) > 1: + raise RouteSearchError( + "transport statistics mix different motatool encoder binaries" + ) + return next(iter(encoders)) if encoders else None + + +def edge_transport_cost( + row: dict[str, object], images: list[dict[str, object]], relay_hops: int +) -> dict[str, int | str]: + source = int(row["source"]) + payload = int(row.get("payload", 0)) + if payload <= 0: + payload = payload_size_from_container(int(row["container"])) + if source_uses_v2(images, source): + if not transport_stats_complete(row): + raise RouteSearchError( + f"missing transport statistics for capable source edge " + f"{source}->{int(row['target'])} memory=0x{int(row['memory']):X}" + ) + cost = v2_transport_cost( + payload, + int(row["v2_wire_bytes"]), + int(row["v2_deflate_bytes"]), + int(row["v2_deflate_blocks"]), + int(row["v2_data_packets"]), + source_transport_pipeline(images, source), + ) + else: + cost = legacy_transport_cost(payload) + result = dict(cost) + result["linear_path_bytes"] = linear_path_bytes( + int(cost["origin_mesh_bytes"]), int(cost["packets"]), relay_hops + ) + return result + + def select_route(rows: list[dict[str, object]], images: list[dict[str, object]], - baseline_size: int, output: Path, complete: bool) -> dict[str, object]: + baseline_size: int, output: Path, complete: bool, + objective: str = "container", relay_hops: int = 0) -> dict[str, object]: + if objective not in ("container", "transport"): + raise RouteSearchError(f"unknown route objective: {objective}") + if not 0 <= relay_hops <= OTA_MAX_HOPS: + raise RouteSearchError(f"relay hops must be between 0 and {OTA_MAX_HOPS}") endpoint = len(images) - 1 baseline_stage = align_down(STAGE_CEILING - baseline_size) baseline_margin = baseline_stage - (APP_BASE + FIXED_MEMORY) if baseline_margin < 0: raise RouteSearchError("pinned first package does not fit") - edges: dict[tuple[int, int], dict[str, object]] = { - (0, 1): {"source": 0, "target": 1, "memory": FIXED_MEMORY, - "container": baseline_size, "stage_start": baseline_stage, - "margin": baseline_margin, "feasible": True} + baseline_row: dict[str, object] = { + "source": 0, "target": 1, "memory": FIXED_MEMORY, + "container": baseline_size, "stage_start": baseline_stage, + "margin": baseline_margin, "feasible": True, } + if objective == "transport": + baseline_row["payload"] = payload_size_from_container(baseline_size) + edges: dict[tuple[int, int], dict[str, object]] = {(0, 1): baseline_row} for row in rows: - if int(row["source"]) == 0: + source = int(row["source"]) + target = int(row["target"]) + if not (0 <= source < len(images) and 0 <= target < len(images)): + raise RouteSearchError(f"geometry edge {source}->{target} is out of range") + if source == 0: # Older evidence tables measured these irrelevant shortcuts. Keep # them countable for archive verification, but never admit them to # the route graph; node 0 is pinned unconditionally to node 1. continue + if not is_forward_progress(images, source, target): + continue if not truth(row.get("feasible", False)): continue - key = int(row["source"]), int(row["target"]) - rank = (int(row["container"]), -int(row["margin"]), int(row["memory"])) + key = source, target + if objective == "transport": + transport_bytes = int( + edge_transport_cost(row, images, relay_hops)["linear_path_bytes"] + ) + rank = ( + transport_bytes, int(row["container"]), + -int(row["margin"]), int(row["memory"]), + ) + else: + rank = (int(row["container"]), -int(row["margin"]), int(row["memory"])) old = edges.get(key) - if old is None or rank < (int(old["container"]), -int(old["margin"]), int(old["memory"])): + if old is None: edges[key] = row + else: + if objective == "transport": + old_rank = ( + int(edge_transport_cost(old, images, relay_hops)["linear_path_bytes"]), + int(old["container"]), -int(old["margin"]), int(old["memory"]), + ) + else: + old_rank = ( + int(old["container"]), -int(old["margin"]), int(old["memory"]), + ) + if rank < old_rank: + edges[key] = row best: dict[int, tuple[int, int, list[int]]] = {0: (0, 0, [0])} hop_distance = [sys.maxsize] * len(images) @@ -494,8 +1047,12 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]], if source not in best: continue for target, row in outgoing.get(source, []): + edge_bytes = ( + int(edge_transport_cost(row, images, relay_hops)["linear_path_bytes"]) + if objective == "transport" else int(row["container"]) + ) candidate = (best[source][0] + 1, - best[source][1] + int(row["container"]), + best[source][1] + edge_bytes, best[source][2] + [target]) if target not in best or candidate < best[target]: best[target] = candidate @@ -505,11 +1062,33 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]], elif distance == hop_distance[target]: route_count[target] += route_count[source] + objective_text = ( + "minimum packages, then minimum ideal linear-path serialized MeshCore bytes" + if objective == "transport" + else "minimum packages, then minimum total container bytes" + ) common = {"schema": 2, "app_base": f"0x{APP_BASE:X}", "stage_ceiling": f"0x{STAGE_CEILING:X}", "node_count": len(images), "search_complete": complete, "candidate_geometries": len(rows), "feasible_edges": len(edges), - "objective": "minimum packages, then minimum total container bytes"} + "objective": objective_text} + if objective == "transport": + transport_encoder = transport_encoder_for_rows(rows, images) + common["transport_accounting"] = { + "relay_hops": relay_hops, + "source_capability_field": TRANSPORT_CAPABILITY, + "source_pipeline_field": TRANSPORT_PIPELINE_FIELD, + "first_bootstrap_profile": "legacy-160-raw", + "included": ( + "OTA messages, repeated 4-byte v2 stream IDs, 171-byte DATA slicing, " + "adaptive requests, manifest, exact per-block proofs, and MeshCore framing" + ), + "excluded": ( + "discovery, retries, flood fan-out, and radio-dependent LoRa PHY coding/preamble" + ), + } + if transport_encoder is not None: + common["transport_accounting"]["encoder_sha256"] = transport_encoder if endpoint not in best: result = {**common, "status": "unreachable", "reachable_nodes": sorted(best), "endpoint_node": endpoint, @@ -521,17 +1100,41 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]], steps = [] for source, target in zip(nodes, nodes[1:]): row = edges[source, target] - steps.append({"source_node": source, "target_node": target, - "inplace_memory": f"0x{int(row['memory']):X}", - "reuse_baseline_package": source == 0 and target == 1, - "expected_container_size": int(row["container"]), - "expected_staging_margin": int(row["margin"]), - "expected_target_sha256": images[target]["sha256"], - "expected_target_version": images[target].get("version", "unknown")}) + step = {"source_node": source, "target_node": target, + "inplace_memory": f"0x{int(row['memory']):X}", + "reuse_baseline_package": source == 0 and target == 1, + "expected_container_size": int(row["container"]), + "expected_staging_margin": int(row["margin"]), + "expected_target_sha256": images[target]["sha256"], + "expected_target_version": images[target].get("version", "unknown")} + if objective == "transport": + cost = edge_transport_cost(row, images, relay_hops) + step["transport"] = { + key: cost[key] for key in ( + "profile", "payload_bytes", "wire_bytes", "deflate_bytes", + "deflate_blocks", "data_packets", "request_packets", + "manifest_packets", "request_pipeline", "proof_request_packets", + "proof_packets", "packets", "manifest_bytes", "data_bytes", + "block_request_bytes", "proof_request_bytes", + "proof_response_bytes", "origin_mesh_bytes", "linear_path_bytes", + ) + } + if source_uses_v2(images, source): + step["transport"]["payload_sha256"] = row["payload_sha256"] + step["transport"]["encoder_sha256"] = row[ + "transport_encoder_sha256" + ] + steps.append(step) + selected_container_bytes = sum( + int(edges[source, target]["container"]) + for source, target in zip(nodes, nodes[1:]) + ) result = {**common, "status": "reachable", "nodes": nodes, "shortest_package_count": hop_distance[endpoint], "shortest_route_count": route_count[endpoint], - "selected_total_bytes": best[endpoint][1], "steps": steps} + "selected_total_bytes": selected_container_bytes, "steps": steps} + if objective == "transport": + result["selected_total_transport_bytes"] = best[endpoint][1] write_atomic_text( output, json.dumps(result, indent=2, sort_keys=True) + "\n" ) @@ -545,14 +1148,47 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--work-dir", type=Path, required=True) parser.add_argument("--workers", type=int, default=min(8, os.cpu_count() or 1)) parser.add_argument("--no-generate", action="store_true") + parser.add_argument( + "--objective", choices=("container", "transport"), default="container", + help="secondary objective after minimum package count", + ) + parser.add_argument( + "--motatool", metavar="PATH", + help="motatool with `transport-size`; required to measure capable-source edges", + ) + parser.add_argument( + "--relay-hops", type=int, default=0, + help="ideal linear-path relays for transport accounting (default: direct)", + ) args = parser.parse_args(argv) if args.workers < 1: parser.error("--workers must be positive") + if not 0 <= args.relay_hops <= OTA_MAX_HOPS: + parser.error(f"--relay-hops must be between 0 and {OTA_MAX_HOPS}") + transport_tool: Path | None = None + if args.motatool: + discovered = shutil.which(args.motatool) + transport_tool = Path(discovered or args.motatool).resolve() + if not transport_tool.is_file(): + parser.error(f"--motatool is not a regular file: {transport_tool}") args.work_dir.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory( prefix=".frozen-inventory-", dir=args.work_dir ) as frozen_raw: frozen = Path(frozen_raw) + frozen_transport_tool: Path | None = None + frozen_transport_tool_sha256: str | None = None + if transport_tool is not None: + frozen_transport_tool = frozen / ( + "motatool.exe" if transport_tool.suffix.lower() == ".exe" else "motatool" + ) + try: + shutil.copy2(transport_tool, frozen_transport_tool, follow_symlinks=True) + except OSError as exc: + raise RouteSearchError( + f"cannot snapshot transport-size tool {transport_tool}: {exc}" + ) from exc + frozen_transport_tool_sha256 = sha256_file(frozen_transport_tool) inventory_path = snapshot_inventory( args.inventory, frozen / "current" / "images.json" ) @@ -587,19 +1223,40 @@ def main(argv: list[str] | None = None) -> int: for field in comparable ): raise RouteSearchError(f"conflicting reuse inputs for {key}") + if old is not None and transport_stats_complete(old): + if transport_stats_complete(row) and any( + str(old.get(field, "")) != str(row.get(field, "")) + for field in TRANSPORT_FIELDS + ): + raise RouteSearchError( + f"conflicting reuse transport statistics for {key}" + ) + if not transport_stats_complete(row): + row.update({field: old[field] for field in TRANSPORT_FIELDS}) cache[key] = row jobs = all_jobs(images) required = {cache_key(job[5], job[6], job[2]) for job in jobs} remaining = [ job for job in jobs - if cache_key(job[5], job[6], job[2]) not in cache + if not cache_satisfies_job( + cache.get(cache_key(job[5], job[6], job[2])), + job_needs_transport_stats(job, images, args.objective), + frozen_transport_tool_sha256, + ) ] print( f"nodes={len(images)} geometries={len(jobs)} " f"reused={len(required & cache.keys())} remaining={len(remaining)}" ) if remaining and not args.no_generate: + if frozen_transport_tool is None and any( + job_needs_transport_stats(job, images, args.objective) + for job in remaining + ): + parser.error( + "--motatool is required to measure transport costs for capable sources" + ) # Patch payloads are disposable worker scratch. Keep them in a # uniquely owned directory so cleanup can never remove a caller's # pre-existing `work-dir/patches` tree. @@ -609,7 +1266,13 @@ def main(argv: list[str] | None = None) -> int: patches = Path(patches_raw) with ProcessPoolExecutor(max_workers=args.workers) as pool: futures = [ - pool.submit(geometry_job, (*job, str(patches))) + pool.submit( + geometry_job, + (*job, str(patches), + str(frozen_transport_tool) if job_needs_transport_stats( + job, images, args.objective + ) else ""), + ) for job in remaining ] for done, future in enumerate(as_completed(futures), 1): @@ -622,13 +1285,21 @@ def main(argv: list[str] | None = None) -> int: cache[key] = row if done % 100 == 0 or done == len(futures): print(f"measured={done}/{len(remaining)}", flush=True) - completed = required & cache.keys() + completed = { + cache_key(job[5], job[6], job[2]) for job in jobs + if cache_satisfies_job( + cache.get(cache_key(job[5], job[6], job[2])), + job_needs_transport_stats(job, images, args.objective), + frozen_transport_tool_sha256, + ) + } rows = project_cache(cache, images, completed) write_csv(args.work_dir / "geometry.csv", rows) complete = completed == required result = select_route( rows, images, baseline_size, args.work_dir / "route.json", complete=complete, + objective=args.objective, relay_hops=args.relay_hops, ) print(json.dumps(result, indent=2, sort_keys=True)) return 0 if complete else 3 diff --git a/tools/lora_ota/test_rak3401_route_search.py b/tools/lora_ota/test_rak3401_route_search.py index 74d8b7e74..b122a73b5 100644 --- a/tools/lora_ota/test_rak3401_route_search.py +++ b/tools/lora_ota/test_rak3401_route_search.py @@ -32,6 +32,9 @@ class RouteSearchTests(unittest.TestCase): def write_inventory( self, root: Path, payloads: list[bytes], *, baseline_size: int = 100, versions: list[str] | None = None, + transport_capabilities: list[bool] | None = None, + transport_pipelines: list[int | None] | None = None, + version_ranks: list[int] | None = None, ) -> tuple[Path, list[dict[str, object]]]: records = [] for node, payload in enumerate(payloads): @@ -46,12 +49,32 @@ class RouteSearchTests(unittest.TestCase): "sha256": hashlib.sha256(encoded).hexdigest(), "body_hash": hashlib.sha256(payload).digest()[:8].hex(), "version": version, + "version_rank": ( + version_ranks[node] if version_ranks is not None else node + ), } if node == 1: record["baseline_container_size"] = baseline_size + capability = ( + transport_capabilities[node] + if transport_capabilities is not None else False + ) + record[search.TRANSPORT_CAPABILITY] = capability + record[search.TRANSPORT_PIPELINE_FIELD] = ( + transport_pipelines[node] + if transport_pipelines is not None else None + ) records.append(record) manifest = root / "images.json" - manifest.write_text(json.dumps({"images": records}), encoding="ascii") + manifest.write_text(json.dumps({ + "capability_contract": { + search.TRANSPORT_CAPABILITY: "required boolean on every record", + }, + "edge_policy": { + "require_target_version_rank_gt_source_version_rank": True, + }, + "images": records, + }), encoding="ascii") return manifest, records def test_cache_migration_survives_inserted_node_and_remaps_indices(self) -> None: @@ -70,6 +93,8 @@ class RouteSearchTests(unittest.TestCase): writer.writeheader() writer.writerow({ "source": 1, "target": 2, "memory": search.FIXED_MEMORY, + "source_sha256": old[1]["sha256"], + "target_sha256": old[2]["sha256"], "payload": 7, "container": search.container_size(7), "stage_start": search.align_down( @@ -84,7 +109,9 @@ class RouteSearchTests(unittest.TestCase): new_dir, [b"A", b"X", b"B", b"C"], versions=["1.0.0.0", "1.0.0.99", "1.0.0.1", "1.0.0.2"], ) - migrated = search.migrate_csv(old_manifest, cache) + # The recorded numeric nodes no longer identify these images in + # the expanded inventory; authenticated hashes still do. + migrated = search.migrate_csv(new_manifest, cache) key = search.cache_key(str(old[1]["sha256"]), str(old[2]["sha256"]), search.FIXED_MEMORY) self.assertIn(key, migrated) @@ -309,6 +336,287 @@ class RouteSearchTests(unittest.TestCase): self.assertEqual(result["nodes"], [0, 1, 4]) self.assertEqual(result["selected_total_bytes"], 1100) + def test_same_version_variants_are_never_progress_edges(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + manifest, _ = self.write_inventory( + root, [b"0", b"1", b"2-raw", b"2-inflate", b"3"], + versions=[ + "1.0.0.0", "1.0.0.1", "1.0.0.2", "1.0.0.2", "1.0.0.3", + ], + version_ranks=[0, 1, 2, 2, 3], + ) + images = search.load_inventory(manifest) + jobs = search.all_jobs(images) + self.assertFalse(any(job[0] == 2 and job[1] == 3 for job in jobs)) + + def edge(source: int, target: int) -> dict[str, object]: + return { + "source": source, "target": target, "memory": 0x1000, + "container": 100, "stage_start": 0xC0000, + "margin": 100, "feasible": True, + } + + # The only apparent path depends on changing variants without + # advancing the running version. The route must remain unreachable. + result = search.select_route( + [edge(1, 2), edge(2, 3), edge(3, 4)], images, 100, + root / "route.json", True, + ) + self.assertEqual(result["status"], "unreachable") + self.assertNotIn(3, result["reachable_nodes"]) + + def test_transport_cost_accounts_for_every_clean_wire_message(self) -> None: + legacy = search.legacy_transport_cost(1024) + self.assertEqual(legacy["data_packets"], 7) + self.assertEqual(legacy["packets"], 13) + self.assertEqual(legacy["origin_mesh_bytes"], 1355) + self.assertEqual(legacy["manifest_bytes"], 224) + self.assertEqual( + legacy["origin_mesh_bytes"], + sum( + int(legacy[field]) + for field in ( + "manifest_bytes", "data_bytes", "block_request_bytes", + "proof_request_bytes", "proof_response_bytes", + ) + ), + ) + + raw_v2 = search.v2_transport_cost(1024, 1024, 0, 0, 6) + self.assertEqual(raw_v2["data_packets"], 6) + self.assertEqual(raw_v2["packets"], 11) + self.assertEqual(raw_v2["origin_mesh_bytes"], 1359) + self.assertEqual(raw_v2["manifest_bytes"], 224) + + compressed = search.v2_transport_cost(1024, 800, 800, 1, 5) + self.assertEqual(compressed["data_bytes"], 875) + self.assertEqual(compressed["origin_mesh_bytes"], 1120) + self.assertEqual( + search.linear_path_bytes(1120, int(compressed["packets"]), 2), + 3 * 1120 + 3 * int(compressed["packets"]), + ) + + # The receive-inflate compatibility bridges are compiled with a single + # block in flight. Account for their five individual request frames, + # rather than the full RAK3401 receiver's clean [1, 2, 2] flights. + full = search.v2_transport_cost(5 * 1024, 5 * 1024, 0, 0, 30, 4) + bridge = search.v2_transport_cost(5 * 1024, 5 * 1024, 0, 0, 30, 1) + self.assertEqual(full["request_packets"], 3) + self.assertEqual(full["request_bytes"], 41) + self.assertEqual(bridge["request_packets"], 5) + self.assertEqual(bridge["request_bytes"], 55) + self.assertEqual( + int(bridge["origin_mesh_bytes"]) - int(full["origin_mesh_bytes"]), 14 + ) + + def test_published_legacy_chain_and_new_bootstrap_cost_regressions(self) -> None: + old_payloads = [ + 89282, 45974, 105407, 149133, 110666, + 119942, 138883, 173312, 270303, 173854, + ] + old = [search.legacy_transport_cost(size) for size in old_payloads] + self.assertEqual(sum(old_payloads), 1_376_756) + self.assertEqual(sum(int(cost["packets"]) for cost in old), 13_490) + self.assertEqual( + sum(int(cost["origin_mesh_bytes"]) for cost in old), 1_563_957 + ) + + new_bootstrap = search.legacy_transport_cost(64_088) + self.assertEqual(new_bootstrap["data_packets"], 438) + self.assertEqual(new_bootstrap["packets"], 630) + self.assertEqual(new_bootstrap["origin_mesh_bytes"], 72_528) + with self.assertRaisesRegex(search.RouteSearchError, "between 0 and 8"): + search.linear_path_bytes(100, 1, 9) + + def test_proof_sizing_matches_promote_odd_merkle_implementation(self) -> None: + for count in range(1, 33): + leaves = [bytes([index]) * 4 for index in range(count)] + for index in range(count): + self.assertEqual( + search.proof_sibling_count(count, index), + len(search.motalib.merkle_proof(leaves, index)), + ) + + def test_transport_objective_uses_running_source_capability(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + baseline = search.container_size(64) + manifest, _ = self.write_inventory( + root, [b"0", b"1", b"2", b"3", b"4"], + baseline_size=baseline, + transport_capabilities=[True, False, True, False, True], + transport_pipelines=[4, None, 1, None, 4], + ) + images = search.load_inventory(manifest) + + def edge(source: int, target: int, payload: int, + *, compressed: int | None = None) -> dict[str, object]: + row: dict[str, object] = { + "source": source, "target": target, "memory": 0x1000, + "payload": payload, "container": search.container_size(payload), + "stage_start": 0xC0000, "margin": 100, "feasible": True, + } + if compressed is not None: + row.update({ + "payload_sha256": "a" * 64, + "transport_encoder_sha256": "b" * 64, + "v2_wire_bytes": compressed, + "v2_deflate_bytes": compressed, + "v2_deflate_blocks": 1, + "v2_data_packets": (compressed + 170) // 171, + }) + return row + + rows = [ + edge(1, 2, 100), edge(1, 3, 100), + edge(2, 4, 1000, compressed=50), edge(3, 4, 100), + ] + transport = search.select_route( + rows, images, baseline, root / "transport.json", True, + objective="transport", + ) + self.assertEqual(transport["nodes"], [0, 1, 2, 4]) + # Node 0 is marked capable on purpose: bootstrap is nevertheless raw. + self.assertEqual( + transport["steps"][0]["transport"]["profile"], "legacy-160-raw" + ) + # Target node 2's capability does not apply until node 2 is running. + self.assertEqual( + transport["steps"][1]["transport"]["profile"], "legacy-160-raw" + ) + self.assertEqual( + transport["steps"][2]["transport"]["profile"], "v2-171-deflate" + ) + self.assertEqual( + transport["steps"][2]["transport"]["request_pipeline"], 1 + ) + self.assertEqual( + transport["selected_total_bytes"], + sum(step["expected_container_size"] for step in transport["steps"]), + ) + self.assertEqual( + transport["selected_total_transport_bytes"], + sum( + step["transport"]["linear_path_bytes"] + for step in transport["steps"] + ), + ) + + container = search.select_route( + rows, images, baseline, root / "container.json", True, + objective="container", + ) + self.assertEqual(container["nodes"], [0, 1, 3, 4]) + + def test_transport_geometry_rank_includes_per_packet_stream_id_and_framing(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + baseline = search.container_size(64) + manifest, _ = self.write_inventory( + root, [b"0", b"1", b"2"], baseline_size=baseline, + transport_capabilities=[False, True, False], + transport_pipelines=[None, 1, None], + ) + images = search.load_inventory(manifest) + + def edge(memory: int, payload: int, wire: int, packets: int) -> dict[str, object]: + return { + "source": 1, "target": 2, "memory": memory, + "payload": payload, "container": search.container_size(payload), + "stage_start": 0xC0000, "margin": 100, "feasible": True, + "payload_sha256": f"{memory:064x}", + "transport_encoder_sha256": "c" * 64, + "v2_wire_bytes": wire, "v2_deflate_bytes": wire, + "v2_deflate_blocks": 1, "v2_data_packets": packets, + } + + # The 342-byte representation wins despite its larger patch because + # crossing 342 -> 343 adds a third repeated stream ID and frame. + rows = [edge(0x2000, 400, 343, 3), edge(0x3000, 500, 342, 2)] + result = search.select_route( + rows, images, baseline, root / "route.json", True, + objective="transport", + ) + self.assertEqual(result["steps"][1]["inplace_memory"], "0x3000") + self.assertEqual(result["steps"][1]["transport"]["data_packets"], 2) + + def test_transport_measurement_is_bound_to_payload_and_encoder_hashes(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + payload = root / "payload.patch" + payload.write_bytes(b"x" * 300) + tool = root / "motatool" + tool.write_bytes(b"exact encoder binary") + measured = { + "schema": 1, "payload_bytes": 300, "block_size": 1024, + "block_count": 1, "wire_bytes": 200, "deflate_bytes": 200, + "deflate_blocks": 1, "data_packets": 2, + } + completed = mock.Mock(returncode=0, stdout=json.dumps(measured), stderr="") + with mock.patch.object(search.subprocess, "run", return_value=completed) as run: + result = search.measure_transport_size(tool, payload) + self.assertEqual(result["payload_sha256"], search.sha256_file(payload)) + self.assertEqual(result["transport_encoder_sha256"], search.sha256_file(tool)) + self.assertEqual(result["v2_data_packets"], 2) + self.assertIn("transport-size", run.call_args.args[0]) + + def test_partial_or_missing_capable_transport_evidence_is_rejected(self) -> None: + with self.assertRaisesRegex(search.RouteSearchError, "partial transport"): + search.normalize_transport_stats( + {"payload_sha256": "a" * 64}, 100, "test row" + ) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + baseline = search.container_size(64) + manifest, _ = self.write_inventory( + root, [b"0", b"1", b"2"], baseline_size=baseline, + transport_capabilities=[False, True, False], + transport_pipelines=[None, 1, None], + ) + images = search.load_inventory(manifest) + row = { + "source": 1, "target": 2, "memory": 0x1000, + "payload": 100, "container": search.container_size(100), + "stage_start": 0xC0000, "margin": 100, "feasible": True, + } + with self.assertRaisesRegex(search.RouteSearchError, "missing transport"): + search.select_route( + [row], images, baseline, root / "route.json", True, + objective="transport", + ) + + def test_transport_cache_cannot_mix_or_reuse_a_different_encoder(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + baseline = search.container_size(64) + manifest, _ = self.write_inventory( + root, [b"0", b"1", b"2"], baseline_size=baseline, + transport_capabilities=[False, True, False], + transport_pipelines=[None, 1, None], + ) + images = search.load_inventory(manifest) + + def row(memory: int, encoder: str) -> dict[str, object]: + return { + "source": 1, "target": 2, "memory": memory, + "payload": 100, "container": search.container_size(100), + "stage_start": 0xC0000, "margin": 100, "feasible": True, + "payload_sha256": "a" * 64, + "transport_encoder_sha256": encoder, + "v2_wire_bytes": 50, "v2_deflate_bytes": 50, + "v2_deflate_blocks": 1, "v2_data_packets": 1, + } + + first = row(0x1000, "b" * 64) + self.assertTrue(search.cache_satisfies_job(first, True, "b" * 64)) + self.assertFalse(search.cache_satisfies_job(first, True, "c" * 64)) + with self.assertRaisesRegex(search.RouteSearchError, "mix different"): + search.select_route( + [first, row(0x2000, "c" * 64)], images, baseline, + root / "route.json", True, objective="transport", + ) + def test_fixed_outputs_never_follow_preexisting_symlinks(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) @@ -345,6 +653,28 @@ class RouteSearchTests(unittest.TestCase): bad = json.loads(manifest.read_text()) bad["images"][1]["path"] = "missing.bin" cases.append((bad, "missing inventory image")) + bad = json.loads(manifest.read_text()) + bad["images"][1][search.TRANSPORT_CAPABILITY] = "yes" + cases.append((bad, "must be boolean")) + bad = json.loads(manifest.read_text()) + del bad["images"][1][search.TRANSPORT_CAPABILITY] + cases.append((bad, "required by the inventory capability contract")) + bad = json.loads(manifest.read_text()) + bad["images"][1][search.TRANSPORT_CAPABILITY_ALIAS] = True + cases.append((bad, "disagrees")) + bad = json.loads(manifest.read_text()) + bad["images"][1][search.TRANSPORT_CAPABILITY] = True + cases.append((bad, "is required")) + bad = json.loads(manifest.read_text()) + bad["images"][1][search.TRANSPORT_CAPABILITY] = True + bad["images"][1][search.TRANSPORT_PIPELINE_FIELD] = 0 + cases.append((bad, "must be between")) + bad = json.loads(manifest.read_text()) + del bad["images"][1]["version_rank"] + cases.append((bad, "required by the inventory edge policy")) + bad = json.loads(manifest.read_text()) + bad["images"][1]["version_rank"] = 0 + cases.append((bad, "one-to-one")) for number, (document, message) in enumerate(cases): candidate = root / f"bad-{number}.json" candidate.write_text(json.dumps(document), encoding="ascii")