dandri/HaloKeymind
1
0
Fork 0
mirror of https://github.com/mikecarper/MeshCore.git synced 2026-10-20 12:41:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/keymindCascade' into keymindCascade

Browse Source
This commit is contained in:
mikecarper
2026-09-30 23:38:05 -07:00
parent 301ef3758d 0f8f1fb9f3
commit 51bc7f1e14
10 changed files with 1174 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/lora_ota_automation.md
+19
View File
@@ -964,6 +964,25 @@ bypasses signature/install checks. It does **not** fix later block-retry
stalls on old firmware or relays beyond the controller's direct radio reach;
the firmware update is the permanent fix.
For a short path, `--legacy-relay-reboot-recovery` offers a lower-airtime
alternative after that same exact `0/0` timeout. It requires one managed relay
whose saved normal-channel and active temporary-channel routes are each **zero
or one hop**, plus `--source-shares-controller`. Before any reboot, the runner
checks the relay and destination identities, the exact failed manifest, and
at least eight minutes of remaining transfer and target TempRadio time. It
asks for a separate confirmation at the failure (or requires both this flag
and `--yes` in unattended use). The relay's `reboot` CLI intentionally sends
no reply: the runner returns the shared source/controller to normal radio,
authenticates the relay's exact key there, re-arms only the remaining portion
of the target's TempRadio lease, then proves the relay, destination and same
manifest on the temporary channel before one pull retry. If the normal path
cannot be proved after reboot, it stops instead of assuming the reboot worked;
the other temporary leases still expire. A reboot also clears the relay's
temporary radio setting, so manually rebooting it during a transfer without
this re-arming sequence will break the temporary path. When both legacy
recovery options are enabled, the reboot choice is offered first; declining
it leaves the direct-cache workaround available on a zero-hop route.
Read-only and replay-safe transmissions retry up to three times. Three retries
or 90 seconds, whichever comes first, opens a 10-second stop-or-continue
prompt when stdin is an interactive terminal. Continue remains the default on
docs/rak3401_1.17.1.7_legacy24.md
+272
View File
@@ -0,0 +1,272 @@
# RAK3401: 1.16.07 to 1.17.1.7 without a bootloader update
Qualified on MercerWoodMesh on 27 September 2026: all 15 application updates
passed over LoRa, from the exact supplied 1.16.07 image, without a bootloader
update. Use `RAK3401-1.16.07-to-1.17.1.7-legacy24-mOTA.zip` and verify it against
the accompanying `.zip.sha256` file before extracting. Follow the exact
before/after hashes in its `CHAIN.csv`; this is not a single-step stock update.
## Exact compatibility boundary
This is an application-only migration for target `2FA509C1`, hardware
`RAK_3401_repeater_lora_ota_no_external_sensors`. It is not a generic RAK4631
package. The installed application must have body hash `71F4026CBE4B8B74`
(`1.16.07`, full image SHA-256
`4301fc63ebd661c70f9bc40e7eca44d3cb8a358cb3c0075e700d9676a57478cb`).
The bootloader remains `0.9.2-OTAFIX2.4`, with ABI 2 and codec mask `0x4`.
Its application staging ceiling is `0xD4000`. Every package is a codec-2
CRLE in-place delta. DEFLATE is used only for transport by compatible
intermediate applications; the old bootloader still applies CRLE.
The destination is built from source commit
`2d03e098cf75d58adcf35fe3d4dbbeaf0af114e1` with the explicit
`MESHCORE_NRF52_LEGACY_FLASH_OTA=1` build selection. This preserves the
ordinary S140 linker and disables hybrid flash/RAM staging and bootloader
self-update. It is deliberately different from the stock hybrid 1.17.1.7
release. The endpoint has:
- Version: `v1.17.1.7-halo-keymind-cascade-legacy24-2d03e098`
- Image: 521,496 bytes
- Body hash: `D437D1F17AF5B9F3`
- SHA-256: `c9e63a94f58f881b1853fc5d9d348cd75a6cdd454c5b572341d652a6141e3157`
Do not replace this endpoint with the stock ZIP. Do not send a bootloader or
SoftDevice update. The ZIP/UF2 files used to construct and qualify the chain
are not themselves LoRa update containers.
Future application updates for this node must also be qualified against the
unchanged OTAFIX2.4 contract. Keep the legacy build selection; do not assume
that a later stock release is an interchangeable target for this chain.
## What "smallest" means
The search compares an explicit, hash-pinned set of application images and
every allowed page-aligned in-place workspace for each increasing-version
transition. The starting bootstrap is fixed: its 89,844-byte package is the
only admitted first edge. Synthetic bridge version numbers are real compiled
versions below 1.17.1.7; their binaries are not relabeled after compilation.
The `bytes` objective minimizes ideal serialized MeshCore traffic first,
then install count. The `packets` objective minimizes ideal originated
packets first, then serialized traffic. The older `transport` objective
minimizes install count before traffic and is not equivalent to either.
Accounting includes manifests, block requests, proof requests/responses,
DATA fragments, and MeshCore framing. It uses the actual pinned encoder for
compressed blocks. It excludes discovery, retries, admin/control commands,
other mesh traffic, flood fan-out, and LoRa preamble/coding overhead. A fixed linear relay path
can be accounted for separately. These are finite-search minima, not a proof
that no possible custom firmware or protocol could ever use fewer bytes.
The ideal compressed counts also assume negotiation remains in that mode.
An older receiver can fall back to raw DATA after stalls; the physical-test
transmission totals include that extra traffic.
Best lossless byte count is not a guarantee of the fastest real radio link.
Use the fastest preset that remains reliable over the actual route. SF5 /
500 kHz is a direct bench measurement, not a recommendation to switch an
unreachable field node or unconfigured relay chain to that preset.
## Selected route
The completed search covers 52 images and 47,920 flash geometries. The
packet-first route has 15 packages, totaling 2,340,160 container bytes. Its
ideal direct-link traffic is **12,621 packets / 1,808,961 serialized MeshCore
bytes**. These totals include traffic in both directions, not just DATA.
Fifteen is also the fewest installs among the searched candidates.
The byte-first route uses the same sequence of application images, with
four different patch workspaces: 12,625 packets / 1,808,787 bytes. Saving
174 bytes costs four packets. The delivered packages use the packet-first
choice; do not substitute packages from the comparison route.
| Step | Resulting version | Container bytes | Ideal packets | Ideal MeshCore bytes |
| ---: | --- | ---: | ---: | ---: |
| 1 | 1.16.7.9 | 89,844 | 878 | 101,203 |
| 2 | 1.16.7.10 | 62,838 | 551 | 70,031 |
| 3 | 1.16.8.7 | 106,125 | 496 | 70,134 |
| 4 | 1.16.9.105 | 149,542 | 728 | 104,992 |
| 5 | 1.16.9.110 | 112,540 | 489 | 67,887 |
| 6 | 1.16.9.111 | 71,070 | 268 | 35,453 |
| 7 | 1.16.9.116 | 138,545 | 692 | 101,308 |
| 8 | 1.16.9.122 | 177,536 | 995 | 148,671 |
| 9 | 1.16.10.1 | 258,398 | 1,477 | 223,042 |
| 10 | 1.16.11.0 | 245,426 | 1,450 | 219,953 |
| 11 | 1.16.12.1 | 217,608 | 1,167 | 172,794 |
| 12 | 1.16.13.1 | 199,395 | 1,024 | 150,005 |
| 13 | 1.16.14.0 | 199,527 | 989 | 143,160 |
| 14 | 1.16.15.0 | 139,188 | 618 | 86,721 |
| 15 | 1.17.1.7 | 172,578 | 799 | 113,607 |
Some bridge versions also exist as different binaries. `CHAIN.csv`'s exact
body hashes and package checksums, not this version-only table, determine
which package is valid at each step. The first two transfers use legacy
1 KiB blocks; subsequent transfers use negotiated 2 KiB blocks with
transport DEFLATE and the unchanged CRLE bootloader format.
The second transfer also omits explicit proof requests: that historical
receiver waits for the current sender's proactive Merkle proofs. The original
1.16.07 receiver does not have this optimization.
## Measured MercerWoodMesh replay
All 15 selected packages passed in order on the Pi's debug-wired RAK3401,
starting with the exact 1.16.07 image and OTAFIX2.4. Every application payload
crossed LoRa from the Heltec T096. There were no debug writes, application
substitutions, failed installs, or resumed downloads between the 15 steps.
The direct bench used 909.950 MHz, BW 500 kHz, SF5, CR5, continuous receive,
and a 128-symbol source preamble. Observations for both radios totaled:
- 14,069 transmitted frames, including discovery and retries.
- 22 minutes 31 seconds of measured download time.
- 47 minutes 35 seconds summed active test-step time, including configuration,
checks, installs, and reboots; deliberate gaps between runs are excluded.
These are bench measurements, not a field-link completion-time promise. The
control commands were sent over USB; a remote field console adds LoRa control
traffic. `BENCHMARK.csv` separates each step's ideal cost from its observed
frame count and timing. Actual serialized radio bytes were not measured.
Independent debug-port readback matched the complete final application,
unchanged bootloader, and unchanged MBR/SoftDevice. UICR matched the normal
post-startup baseline throughout this replay. The initial pre-first-boot
UICR dump differs because normal Nordic/RAK startup configures the reset pins
and 3.3 V regulator; this is documented in the evidence, not a bootloader
upgrade.
After a fresh reboot, the watchdog reported `on`. Both radios returned to
910.525 MHz / BW 62.5 kHz / SF7 / CR5, and a nonce-matched normal-channel trace
through the exact target key prefix completed in 423 ms. The inaccessible
field node has not been touched. The lab RAK remains on this repeater image;
the Pi's `mctomqtt.service` remains stopped because it previously owned that
serial port. Radios other than the RAK3401 and sending Heltec T096 were left
untouched.
## Field preparation
All destination commands below can be sent through the authenticated remote
repeater console; USB/Bluetooth at the destination is not required. Keep a
separate control connection available while the sender owns its companion's
serial port. A local
Full Companion radio attached to the sender runs the bundled `motatool`
seeder. Resolve ports by device identity, not a remembered `ttyACM` number.
The qualified source is the Pi's Heltec T096, USB serial
`651F8E496197F882`, running Full Companion
`v1.17.1.8-halo-keymind-cascade-dev-d4b3bb56`. It supports the extended
2 KiB transport and host-backed DEFLATE. The destination is RAK3401 USB
serial `0B81C9C68D8D01B4`. Older companion firmware or an older `motatool`
is not an interchangeable sender for this bundle.
Before any change, record `ver`, `ota self`, `ota status`,
`get bootloader.ver`, `get system.watchdog`, `get radio.rxps`,
`get powersaving`, `get rxdelay`, `get af`, and `ota config`. Match the exact
starting body, hardware, bootloader, and signing-key policy. An unsigned
bundle is not suitable for a node requiring signed packages.
Choose a stable-power maintenance window long enough for all 15 installs;
do not begin an install while the destination is at risk of losing power.
Disable the destination system watchdog with `set system.watchdog off`.
Allow its one-time hardware reset and wait at least 90 seconds; reconnect
and prove that it reports `off`. A quick software reboot is not a substitute.
Use continuous receive on the destination and sender throughout the bridge
chain (`set radio.rxps off`). Record and disable a second scanning radio
profile on the sender if it would take the radio away from the OTA channel.
Record `rxdelay`, `af`, power-saving, reach, and relay settings before changing
them. Use `hops 0` only when there are no intermediate relays. Obey applicable
airtime/duty restrictions and restore all saved settings when finished.
The direct bench also uses `set rxdelay 0` and `set af 0` on both radios,
`ota speed 1` on the sender, and `set powersaving off` where supported.
The earliest applications do not recognize the last command; continuous
receive is still selected through `set radio.rxps off`. The source has
`set radio2 off` throughout the transfer. Wait for configuration replies to
finish before handing its serial connection to the seeder.
For the qualified direct bench configuration only, the destination uses
`tempradio 909.950,500,5,5,120`; the tested current Full Companion accepts
`tempradio 909.950,500,5,5,120,128` for an explicit 128-symbol preamble.
Historical destination parsers do not accept the sixth argument. The sender
and destination must be on the same live radio preset. Do not assume the
temporary preset survives a bridge reboot.
The newer historical receivers performed poorly with a 16-symbol source
preamble in this test. Use the qualified longer setting. Also avoid frequent
`ota status` or `ota stats` polling: these versions rehash the complete running
application while answering, which can interrupt receive processing. The lab
runner observes the sender's local log and checks the receiver near completion
or at a 60-second interval. Set a field maintenance lease appropriate to the
actual preset and route; the 120-minute value above is a bench setting.
## Applying a qualified bundle
Verify the outer archive hash and every entry in `SHA256SUMS.txt`. Verify
each `.mota` using the included, hash-pinned `motatool`. `CHAIN.csv` is the
authoritative ordered list of packages, manifest IDs, and before/after hashes.
On the Pi, use `tools/motatool-linux-aarch64` (make it executable after
extracting if necessary). The x86-64 binary is for Linux/WSL, not native
Windows. The provenance images, tools, and evidence are not sent over LoRa;
only the selected `.mota` in the current step's serving directory is sent.
For every row, including after an interrupted run:
1. Prove that `ota self` matches that row's exact before hash and that the
bootloader reports `apply OK`. Do not select a row by version alone.
2. Reconcile `ota status`. Resume only that row's known partial MID. If it is
already ready, do not redownload it. Stop on an unrelated or ambiguous
session. Never send a blind `ota cancel`.
3. Reapply the agreed radio/receive guardrails after reboot. Start the sender
with only this row's package in its serving directory:
`motatool serve --dir STEP_DIRECTORY --serial COMPANION_PORT --baud 115200 -v`.
4. Run `ota ls` until the expected version is advertised. The old listing
shows a version, not a MID. Use `ota pull EXPECTED_MID flash` and poll
`ota status` at a restrained interval.
5. Require `ready to install` with the exact MID. Stop the seeder gracefully
with Ctrl-C. Recheck watchdog `off`, then send `ota install`.
6. Allow up to five minutes for the application to return. Require the exact
after hash, expected version, and `bootloader: apply OK` before continuing.
Stop if any check fails; do not try the next package against another base.
A retained previous container may be re-discovered after boot. If the exact
previous MID is verifying, wait for it to become ready or idle. Only after
re-proving the current running hash may that attached, completed previous
MID be cancelled. An idle `ota cancel` does not prove that flash was erased.
After the final hash is proved, restore the destination, sender, and relays
to their recorded normal settings and confirm normal-channel reachability.
Re-enable the system watchdog and require it to report `on`. The current
endpoint supports `normalradio`; coordinate channel restoration so that the
remote node does not become unreachable during the handoff.
The endpoint requires a temporary radio channel for OTA operations, so a
normal-channel `ota ls` is not a reachability test. Use an ordinary mesh trace
or authenticated remote command after restoring normal radio.
The existing `rak3401_mota_chain.py` pins older release archives. It must not
be pointed at this new bundle or have its safety checks bypassed.
## Evidence and reproducibility
`build_rak3401_legacy24_bundle.py` checks every base/target hash, hardware
identity, increasing version, block capability, flash workspace, and staging
margin. It reconstructs every target with both zero- and erased-flash-filled
scratch space, and applies every package with the actual OTAFIX2.4 release
implementation compiled as a flash simulator. It remeasures transport with
the pinned encoder. Its output remains explicitly offline-only until a
separate physical qualification record is attached.
`ACCOUNTING-CORRECTION.json` records the proactive-proof accounting correction.
It changed the ideal totals only; all package checksums and the selected
hardware-test sequence remained unchanged.
The archive includes the 52 frozen search images, all 47,920 measured
geometries, encoder binaries, per-step simulator logs, and raw hardware-test
records. Only the selected `.mota` containers are served over LoRa. Lab
scripts in `evidence/` are tied to the Pi's USB/debug fixture and are not a
remote field deployment controller.
Regression checks: 534 LoRa OTA tests ran successfully with one opt-in
dependency-install test skipped; all 11 linker-selection tests passed.
All PlatformIO builds must be serialized, including builds in separate
checkout directories that share the same PlatformIO package installation.
scripts/nrf52_internal_bootloader_link.py
+31 -20
View File
@@ -28,13 +28,17 @@ requested = (
or env["PIOENV"] in direct_targets
)
# A field node on OTAFIX2.4 cannot receive a hybrid flash/RAM handoff or
# replace its own bootloader. Keep its ordinary application OTA path and
# linker when explicitly building a legacy-compatible destination image.
legacy_flash = os.environ.get("MESHCORE_NRF52_LEGACY_FLASH_OTA")
if legacy_flash not in (None, "0", "1"):
raise RuntimeError("MESHCORE_NRF52_LEGACY_FLASH_OTA must be 0 or 1")
if legacy_flash == "1":
if not requested:
raise RuntimeError("legacy flash OTA requires a qualified internal nRF52 OTA target")
if requested:
# The macro is injected here so direct PIO and build.sh synthetic aliases
# cannot silently differ. Package admission is runtime-capacity based.
env.AppendUnique(CPPDEFINES=[
"OTA_INTERNAL_BOOTLOADER_UPDATE",
"OTA_HYBRID_RAM_STORE",
])
board = env.BoardConfig()
# Most variants inherit the framework's top-level ``build.ldscript``.
# A few otherwise ordinary Adafruit nRF52840 boards (notably Keepteen LT1)
@@ -60,18 +64,25 @@ if requested:
f"S140 v6/v7 non-ExtraFS linker; found mcu={mcu or 'none'} "
f"linker={current or 'none'}"
)
mota_linker = Path(env["PROJECT_DIR"]) / "boards" / (
"nrf52840_s140_%s_mota64.ld" % suffix
)
if not mota_linker.is_file():
raise RuntimeError("missing mOTA hybrid linker: %s" % mota_linker)
if legacy_flash == "1":
print("nRF52 legacy flash OTA: ordinary linker; no hybrid RAM or bootloader self-update")
else:
mota_linker = Path(env["PROJECT_DIR"]) / "boards" / (
"nrf52840_s140_%s_mota64.ld" % suffix
)
if not mota_linker.is_file():
raise RuntimeError("missing mOTA hybrid linker: %s" % mota_linker)
# The framework builder normally resolves LDSCRIPT_PATH from BoardConfig.
# Set both so this remains deterministic whether PlatformIO invokes this
# pre-script before or after that framework step.
board.update("build.ldscript", str(mota_linker))
env.Replace(LDSCRIPT_PATH=str(mota_linker))
print(
"nRF52 shared-slot bootloader update: %s; fixed 64 KiB mOTA RAM arena; runtime EndF headroom required"
% mota_linker.name
)
# Direct PIO and build.sh aliases must select the same feature/map pair.
env.AppendUnique(CPPDEFINES=[
"OTA_INTERNAL_BOOTLOADER_UPDATE",
"OTA_HYBRID_RAM_STORE",
])
# The framework normally resolves LDSCRIPT_PATH from BoardConfig. Set
# both so ordering relative to that framework step cannot change it.
board.update("build.ldscript", str(mota_linker))
env.Replace(LDSCRIPT_PATH=str(mota_linker))
print(
"nRF52 shared-slot bootloader update: %s; fixed 64 KiB mOTA RAM arena; runtime EndF headroom required"
% mota_linker.name
)
test/test_nrf52_internal_bootloader_link.py
+35 -1
View File
@@ -62,13 +62,15 @@ class FakeEnvironment:
raise AssertionError(f"unexpected fake-SCons call: {values!r}")
def run_selector(*, pioenv, board_values, project=ROOT, requested=False):
def run_selector(*, pioenv, board_values, project=ROOT, requested=False, legacy_flash=None):
board = FakeBoardConfig(board_values)
environment = FakeEnvironment(project, pioenv, board)
process_environment = (
{"MESHCORE_NRF52_INTERNAL_BOOTLOADER_UPDATE": "1"}
if requested else {}
)
if legacy_flash is not None:
process_environment["MESHCORE_NRF52_LEGACY_FLASH_OTA"] = legacy_flash
output = io.StringIO()
with mock.patch.dict(os.environ, process_environment, clear=True):
with contextlib.redirect_stdout(output):
@@ -134,6 +136,38 @@ def c_integer(path, name):
class Nrf52InternalBootloaderSelectorTest(unittest.TestCase):
def test_legacy_flash_preserves_linker_for_direct_and_release_builds(self):
for requested in (False, True):
with self.subTest(requested=requested):
environment, output = run_selector(
pioenv=V6_TARGET, requested=requested, legacy_flash="1",
board_values={"build.mcu": "nrf52840",
"build.ldscript": "boards/nrf52840_s140_v6.ld"},
)
self.assertEqual(environment.cppdefines, [])
self.assertEqual(environment.board.updates, [])
self.assertEqual(environment.replacements, {})
self.assertIn("no hybrid RAM or bootloader self-update", output)
def test_legacy_flash_rejects_unqualified_target_and_typo(self):
for target, value, error in (
("RAK_3401_repeater", "1", "qualified internal"),
(V6_TARGET, "yes", "must be 0 or 1"),
):
with self.subTest(target=target, value=value):
with self.assertRaisesRegex(RuntimeError, error):
run_selector(pioenv=target, legacy_flash=value,
board_values={"build.mcu": "nrf52840"})
def test_legacy_flash_still_validates_mcu_and_memory_map(self):
for mcu, linker in (("nrf52832", "nrf52840_s140_v6.ld"),
("nrf52840", "nrf52840_s140_v6_mota64.ld"),
("nrf52840", "custom_nrf52840_s140_v6.ld")):
with self.subTest(mcu=mcu, linker=linker):
with self.assertRaisesRegex(RuntimeError, "normal S140 v6/v7"):
run_selector(pioenv=V6_TARGET, legacy_flash="1",
board_values={"build.mcu": mcu, "build.ldscript": linker})
def test_direct_v6_target_selects_hybrid_linker_and_both_features(self):
environment, output = run_selector(
pioenv=V6_TARGET,
tools/lora_ota/build_rak3401_legacy24_bundle.py
+178
View File
@@ -0,0 +1,178 @@
#!/usr/bin/env python3
"""Package and independently validate a RAK3401 application-only legacy route.
This builder does not qualify hardware. Its output remains an offline-validated
candidate until the exact chain has passed the MercerWoodMesh radio test.
"""
from __future__ import annotations
import argparse
import csv
import hashlib
import json
from pathlib import Path
import shutil
import subprocess
import tempfile
import rak3401_route_search as search
import build_rak3401_compact_bundle as validation
START_SHA = "4301fc63ebd661c70f9bc40e7eca44d3cb8a358cb3c0075e700d9676a57478cb"
BOOTSTRAP_SHA = "8364257a2b3a219905e870fad6fbb2040a96ca4b4bb7201b2867534cc2b45530"
BOOTSTRAP_PACKAGE_SHA = "74e7195efff24fa2fc75fa0cc7fb15803273b5e7e02c4df2391645ea3e882dfa"
# Built from OTAFIX2.4 release source d73de8372e89b8ef352747c8bc7a1aaeab80fbfe.
# A different host build needs a separately reviewed provenance update.
APPLY_SIM_SHA = "b35d1a5c7a646aee382d6ca14f147408dde5ae690812e286a64e36e3d8a46a96"
ENDPOINT_VERSION = "1.17.1.7"
ENDPOINT_SHA = "c9e63a94f58f881b1853fc5d9d348cd75a6cdd454c5b572341d652a6141e3157"
def require(condition: bool, message: str) -> None:
if not condition:
raise ValueError(message)
def run(command: list[str], log: Path) -> None:
completed = subprocess.run(command, capture_output=True, text=True, timeout=300)
log.write_text(completed.stdout + completed.stderr, encoding="utf-8")
require(completed.returncode == 0, f"Command failed; see {log}")
def validate_manifest(manifest, source: dict, target: dict, block: int) -> None:
"""Bind a generated container to its exact application-only transition."""
require(not manifest.is_full and not manifest.is_bootloader and manifest.codec_id == 2,
"Only CRLE in-place application deltas are allowed")
require(manifest.target_id == search.EXPECTED_TARGET_ID
and manifest.hw_id.rstrip(b"\0") == b"RAK_3401", "Wrong manifest hardware")
require(manifest.base_hash.hex().lower() == str(source["body_hash"]).lower(), "Wrong base hash")
require(manifest.image_hash.hex() == target["sha256"], "Wrong target hash")
require(manifest.fw_version == search.motalib.pack_version(target["version"]), "Wrong target version")
require(manifest.image_size == int(target["size"]), "Wrong target size")
require(manifest.block_size == block, "Wrong signed block geometry")
def validate_workspace(source: dict, target: dict, memory: int, container_size: int,
expected_margin: int) -> int:
stage = search.align_down(search.STAGE_CEILING - container_size)
margin = stage - search.APP_BASE - memory
require(memory % 4096 == 0, "Workspace must be page aligned")
require(margin >= 0 and margin == expected_margin, "Legacy staging overlap")
require(memory >= search.align_up(max(int(source["size"])+8192, int(target["size"]))),
"Unsafe workspace")
return margin
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
for flag in ("inventory", "route", "bootstrap", "motatool", "detools", "apply-sim", "output"):
parser.add_argument("--" + flag, type=Path, required=True)
args = parser.parse_args()
require(not args.output.exists(), "Output must be a new directory")
require(search.sha256_file(args.bootstrap) == BOOTSTRAP_PACKAGE_SHA, "Wrong physical bootstrap package")
require(search.sha256_file(args.apply_sim) == APPLY_SIM_SHA, "Unqualified bootloader simulator")
route = json.loads(args.route.read_text(encoding="utf-8"))
require(route.get("status") == "reachable" and route.get("search_complete") is True,
"Route must be reachable and complete within its declared inventory")
images = search.load_inventory(args.inventory)
require(images[0]["sha256"] == START_SHA, "Wrong deployed starting image")
require(images[1]["sha256"] == BOOTSTRAP_SHA, "Wrong pinned bootstrap image")
require(images[-1]["version"] == ENDPOINT_VERSION, "Wrong endpoint version")
require(images[-1]["sha256"] == ENDPOINT_SHA,
"Endpoint must be the application-only legacy24 build, not the hybrid stock image")
nodes = route["nodes"]
require(all(type(node) is int and 0 <= node < len(images) for node in nodes), "Invalid route node")
require(nodes[0:2] == [0, 1] and nodes[-1] == len(images)-1,
"Route does not join the pinned start to the endpoint")
require(len(route["steps"]) == len(nodes)-1, "Inconsistent route length")
args.output.mkdir(parents=True)
output = args.output.resolve()
(output / "motas").mkdir()
logs = output / "validation"
logs.mkdir()
rows = []
tools = {}
with tempfile.TemporaryDirectory(prefix="legacy24-build-") as temporary:
work = Path(temporary)
frozen = []
for record in images:
payload = Path(record["path"]).read_bytes()
require(hashlib.sha256(payload).hexdigest() == record["sha256"],
"Inventory changed while freezing images")
path = work / (record["sha256"] + ".bin")
path.write_bytes(payload)
frozen.append(path)
for name in ("motatool", "detools", "apply_sim"):
original = getattr(args, name).resolve()
# detools can be a Python launcher whose adjacent venv is required.
tools[name] = {"path": str(original), "sha256": search.sha256_file(original)}
for number, (source, target, step) in enumerate(zip(nodes, nodes[1:], route["steps"]), 1):
require((source, target) == (step["source_node"], step["target_node"]), "Broken chain adjacency")
require(search.is_forward_progress(images, source, target), "Non-increasing update version")
require(step["expected_target_version"] == images[target]["version"], "Route version pin mismatch")
memory = int(str(step["inplace_memory"]), 0)
block = search.source_block_size(images, source)
require(block == int(step["block_size"]), "Block size exceeds running receiver contract")
name = f"step-{number:02d}__{images[source]['version']}-to-{images[target]['version']}.mota"
package = output / "motas" / name
if number == 1:
require(step["reuse_baseline_package"] is True, "Bootstrap must be byte-identical")
shutil.copyfile(args.bootstrap, package)
require(package.stat().st_size == 89844, "Wrong bootstrap container size")
else:
require(not step["reuse_baseline_package"], "Unexpected bootstrap reuse")
run([str(args.motatool), "build", "--base", str(frozen[source]),
"--fw", str(frozen[target]), "--patch-type", "in-place",
"--inplace-memory", hex(memory), "--segment-size", "4096",
"--block-size", str(block), "--out", str(package)], logs / f"{number:02d}-build.log")
run([str(args.motatool), "verify", str(package)], logs / f"{number:02d}-verify.log")
parsed = search.motalib.parse_container(package.read_bytes())
manifest = parsed.manifest
validate_manifest(manifest, images[source], images[target], block)
require(manifest.image_hash.hex() == step["expected_target_sha256"], "Wrong route target hash")
require(package.stat().st_size == int(step["expected_container_size"]), "Package size changed from search")
margin = validate_workspace(images[source], images[target], memory,
package.stat().st_size, int(step["expected_staging_margin"]))
patch = work / "payload.patch"
patch.write_bytes(parsed.payload)
require(validation.patch_geometry(args.detools, patch, name) ==
(memory, 4096, int(images[source]["size"]), int(images[target]["size"])), "Patch geometry mismatch")
base, expected = frozen[source].read_bytes(), frozen[target].read_bytes()
for fill in (0, 255):
rebuilt = validation.apply_in_place(args.detools, base, parsed.payload, len(expected), memory, work, f"step-{number}", fill)
require(rebuilt == expected, "Independent patch reconstruction failed")
run([str(args.apply_sim), str(frozen[source]), str(package), str(frozen[target])], logs / f"{number:02d}-otafix24.log")
if "transport" in step:
row = dict(source=source, target=target, payload=len(parsed.payload),
block_size=block, container=package.stat().st_size)
if search.source_uses_deflate(images, source):
row.update(search.measure_transport_size(args.motatool, patch, block))
require(row["payload_sha256"] == step["transport"]["payload_sha256"], "Encoded payload changed")
require(row["transport_encoder_sha256"] == step["transport"]["encoder_sha256"], "Route used another encoder")
cost = search.edge_transport_cost(row, images, int(route["transport_accounting"]["relay_hops"]))
require({"packets", "origin_mesh_bytes", "linear_path_bytes"}.issubset(step["transport"]),
"Route lacks transport totals")
for key, value in step["transport"].items():
if key not in ("payload_sha256", "encoder_sha256"):
require(cost.get(key) == value, "Transport count changed: " + key)
rows.append(dict(step=number, from_version=images[source]["version"],
to_version=images[target]["version"], mota_file="motas/"+name,
mota_size=package.stat().st_size, mota_sha256=search.sha256_file(package),
base_body_hash=manifest.base_hash.hex().upper(),
target_body_hash=str(images[target]["body_hash"]).upper(),
target_sha256=images[target]["sha256"], target_image_size=manifest.image_size,
inplace_memory=hex(memory), staging_margin=margin, block_size=block,
manifest_id=manifest.merkle_root.hex().upper()))
print(f"Validated {number}/{len(route['steps'])}: {name}", flush=True)
with (output / "CHAIN.csv").open("w", newline="", encoding="ascii") as file:
writer = csv.DictWriter(file, fieldnames=list(rows[0]))
writer.writeheader(); writer.writerows(rows)
(output / "ROUTE.json").write_text(json.dumps(route, indent=2) + "\n", encoding="ascii")
(output / "VALIDATION.json").write_text(json.dumps(dict(status="offline-validated; hardware qualification pending",
bootloader="0.9.2-OTAFIX2.4, unchanged", tools=tools, steps=rows), indent=2) + "\n", encoding="ascii")
print(output)
return 0
if __name__ == "__main__":
raise SystemExit(main())
tools/lora_ota/lora_ota.py
+249 -1
View File
@@ -151,6 +151,7 @@ OTA_MAX_HOPS = 8
# explicit, best-effort legacy workaround, not an acknowledgement of reception.
LEGACY_OTA_CACHE_FRAMES = 192
LEGACY_OTA_CACHE_ROUNDS = 2
LEGACY_RELAY_REBOOT_MIN_REMAINING_SECONDS = 8 * 60
TARGET_RXPS_RECOVERY_FILE = "target-rxps-settings.json"
SOURCE_RXPS_RECOVERY_FILE = "source-rxps-settings.json"
MIN_MESHCLI_VERSION = (1, 6, 0)
@@ -4762,6 +4763,8 @@ def confirm_update(
print(f" TempRadio : {args.temp_radio}")
if getattr(args, "legacy_relay_dedup_workaround", False):
print(" legacy retry: up to 2 x 192 short, non-forwarded TempRadio frames on manifest timeout")
if getattr(args, "legacy_relay_reboot_recovery", False):
print(" legacy reboot: offer one relay reboot on exact 0/0 manifest timeout; re-prove normal and TempRadio paths")
saved_rxps = getattr(args, "target_rxps_saved", None)
rxps_profile = getattr(args, "target_rxps_profile", None)
if isinstance(saved_rxps, RxpsSettings) and saved_rxps.enabled:
@@ -5188,14 +5191,181 @@ def recover_legacy_manifest_timeout(
return True
def relay_route_hops(
controller: Controller,
selector: str,
expected_key: str,
label: str,
) -> int:
contacts = controller._run(["contact_info", selector], label, timeout=5)
matching = [c for c in contacts if contact_matches_selector(c, expected_key)]
if len(matching) != 1 or matching[0].get("public_key", "").lower() != expected_key:
raise OtaError(f"cannot prove exact relay contact for {label}")
hops = matching[0].get("out_path_len")
if type(hops) is not int or hops not in (0, 1):
raise OtaError(f"relay reboot recovery requires a 0- or 1-hop {label}; got {hops!r}")
return hops
def recover_manifest_by_relay_reboot(
controller: Controller,
args: argparse.Namespace,
package: MotaInfo,
seeder: SeederProcess | None,
status: str,
deadline: float,
target_armed_at: float,
relay_public_keys: dict[str, str],
original_radio: RadioSettings,
temp_radio: RadioSettings,
on_normal: Callable[[], None],
on_temp_owned: Callable[[], None],
on_temp_active: Callable[[], None],
) -> bool:
"""One explicitly approved reboot; never assume that a silent reboot arrived."""
if (download_manifest_id(status) != package.manifest_id
or not re.search(r"download: failed \(manifest timeout\) 0/0\b", status, re.I)):
return False
if not getattr(args, "source_shares_controller", False) or len(args.relay_values) != 1:
raise OtaError("relay reboot recovery requires one relay and a shared source/controller")
relay, password = args.relay_values[0]
expected_key = relay_public_keys.get(relay)
if expected_key is None or not re.fullmatch(r"[0-9a-f]{64}", expected_key):
raise OtaError("relay reboot recovery has no verified relay public key")
if getattr(args, "relay_reboot_normal_hops", None) not in (0, 1):
raise OtaError("relay reboot recovery lacks a proven normal-channel 0- or 1-hop route")
relay_route_hops(controller, relay, expected_key, "temporary-channel route")
remaining = target_armed_at + args.temp_values[-1] * 60 - time.monotonic()
if (remaining < LEGACY_RELAY_REBOOT_MIN_REMAINING_SECONDS
or deadline - time.monotonic() < LEGACY_RELAY_REBOOT_MIN_REMAINING_SECONDS):
raise OtaError("too little target TempRadio or transfer time remains for relay reboot recovery")
source_status = source_cli_command(args, "tempradio", bounded=True)
parsed = parse_source_temp_radio_status(source_status)
seconds = re.search(r"\b(\d+)s left\b", source_status)
expected_temp = RadioSettings(*args.temp_values[:4], False)
if (parsed is None or parsed[0] != "active" or parsed[1] is None
or not expected_temp.matches(parsed[1]) or seconds is None
or int(seconds[1]) < LEGACY_RELAY_REBOOT_MIN_REMAINING_SECONDS):
raise OtaError("cannot prove sufficient shared-source TempRadio lease for relay reboot")
target_key = read_remote_public_key_bounded(controller, args.target)
if read_remote_public_key_bounded(controller, relay, password=password) != expected_key:
raise OtaError("relay identity changed before reboot recovery")
current = remote_command_with_seeder(
controller, args.target, "ota status", seeder,
"immediately before relay reboot recovery",
)
if download_manifest_id(current) != package.manifest_id:
raise OtaError("destination session changed before relay reboot recovery")
if not re.search(r"download: failed \(manifest timeout\) 0/0\b", current, re.I):
require_package_session(current, package)
return True
if not args.yes:
if not sys.stdin.isatty():
raise OtaError("relay reboot recovery requires an interactive approval or --yes")
try:
answer = input(
f"Reboot relay {relay} ({args.relay_reboot_normal_hops} normal hops) "
"and re-arm the temporary radio before retrying this manifest? [y/N] "
).strip().lower()
except EOFError:
answer = ""
if answer not in ("y", "yes"):
print("[legacy] relay reboot declined; download remains staged")
return False
if (target_armed_at + args.temp_values[-1] * 60 - time.monotonic()
< LEGACY_RELAY_REBOOT_MIN_REMAINING_SECONDS
or deadline - time.monotonic()
< LEGACY_RELAY_REBOOT_MIN_REMAINING_SECONDS):
raise OtaError("relay reboot approval outlasted the safe TempRadio/transfer window")
source_status = source_cli_command(args, "tempradio", bounded=True)
parsed = parse_source_temp_radio_status(source_status)
seconds = re.search(r"\b(\d+)s left\b", source_status)
if (parsed is None or parsed[0] != "active" or parsed[1] is None
or not expected_temp.matches(parsed[1]) or seconds is None
or int(seconds[1]) < LEGACY_RELAY_REBOOT_MIN_REMAINING_SECONDS):
raise OtaError("shared source TempRadio lease changed before relay reboot")
ensure_seeder_running(seeder, "before relay reboot recovery")
print(f"[legacy] sending one no-reply reboot to {relay}; a reply timeout is expected")
try:
controller.remote_command(
relay, "reboot", password=password, retry=False,
reply_timeout=5, operation_timeout=20,
)
except TransmissionError:
pass # Firmware intentionally sends no reply; prove the next state.
controller.forget_remote_auth(relay)
if not shorten_source_temp_window(args):
raise OtaError("shared source did not return to normal radio after relay reboot")
on_normal()
require_radio_settings(controller.get_radio(), original_radio,
"shared controller after relay reboot")
normal_proof_deadline = min(deadline, time.monotonic() + 90)
if read_remote_public_key_bounded(
controller, relay, password=password, deadline=normal_proof_deadline,
) != expected_key:
raise OtaError("rebooted relay did not return with its exact key on normal radio")
relay_route_hops(controller, relay, expected_key, "recovered normal-channel route")
# Bound the replacement lease by the destination's original lease. A
# second 120-minute lease here would strand this relay after the target
# and original source have returned to normal.
remaining = target_armed_at + args.temp_values[-1] * 60 - time.monotonic()
recovery_minutes = int((remaining - 60) // 60)
if recovery_minutes < 5:
raise OtaError("target TempRadio lease is too short to re-arm the relay")
recovery_command = temp_radio_command_for_minutes(args, recovery_minutes)
armed = []
relay_ack = arm_relay_temp_radio_once(
controller, relay, password, recovery_command, armed, recovery_minutes,
)
on_temp_owned() # A lost local reply can still have changed the radio.
switch_controller_to_temp_radio(controller, args, recovery_command, temp_radio)
on_temp_active()
time.sleep(TEMP_RADIO_SWITCH_DELAY_SECONDS)
require_source_on_temp_after_uncertain_arm(args, temp_radio)
if read_remote_public_key_bounded(controller, relay, password=password) != expected_key:
raise OtaError("relay identity was not recovered on the temporary radio")
if not relay_ack:
print("[legacy] lost relay TempRadio reply reconciled by exact temporary-channel identity")
if read_remote_public_key_bounded(controller, args.target) != target_key:
raise OtaError("destination identity changed during relay reboot recovery")
current = remote_command_with_seeder(controller, args.target, "ota status", seeder,
"after relay reboot recovery")
if download_manifest_id(current) != package.manifest_id:
raise OtaError("destination session changed during relay reboot recovery")
if not re.search(r"download: failed \(manifest timeout\) 0/0\b", current, re.I):
require_package_session(current, package)
return True
if time.monotonic() >= deadline:
raise OtaError("transfer deadline expired during relay reboot recovery")
try:
reply = controller.remote_command(
args.target, f"ota pull {package.manifest_id} flash", retry=False,
)
except TransmissionError:
reply = remote_command_with_seeder(controller, args.target, "ota status", seeder,
"resolving relay reboot recovery pull")
require_package_session(reply, package)
else:
if not re.match(rf"OK (?:pulling|resuming) mid={package.manifest_id}\b", reply, re.I):
raise OtaError(f"relay reboot recovery pull was not confirmed: {reply}")
print("[legacy] relay reboot path re-proven; retried the same manifest once")
return True
def monitor_download(
controller: Controller,
args: argparse.Namespace,
package: MotaInfo,
seeder: SeederProcess | None = None,
reboot_recovery: Callable[[str, float], bool] | None = None,
) -> str:
deadline = time.monotonic() + args.transfer_timeout_minutes * 60
legacy_recoveries = 0
reboot_offered = False
poll_seconds = float(args.poll_seconds)
# A status command is ordinary half-duplex LoRa traffic. Asking immediately
# after `ota pull` can occupy the link for a full reply timeout and was
@@ -5250,6 +5420,13 @@ def monitor_download(
controller, args.target, "ota status", seeder,
"during transfer",
)
if (reboot_recovery is not None and not reboot_offered
and download_manifest_id(status) == package.manifest_id
and re.search(r"download: failed \(manifest timeout\) 0/0\b", status, re.I)):
reboot_offered = True
if reboot_recovery(status, deadline):
last_progress = time.monotonic()
continue
if recover_legacy_manifest_timeout(
controller, args, package, seeder, status, deadline, legacy_recoveries
):
@@ -5279,6 +5456,13 @@ def monitor_download(
controller, args.target, "ota status", seeder, "during transfer"
)
query_seconds = time.monotonic() - query_started
if (reboot_recovery is not None and not reboot_offered
and download_manifest_id(status) == package.manifest_id
and re.search(r"download: failed \(manifest timeout\) 0/0\b", status, re.I)):
reboot_offered = True
if reboot_recovery(status, deadline):
time.sleep(min(first_wait, max(0.0, deadline - time.monotonic())))
continue
if recover_legacy_manifest_timeout(
controller, args, package, seeder, status, deadline, legacy_recoveries
):
@@ -7612,6 +7796,12 @@ def build_parser() -> argparse.ArgumentParser:
"requires one directly reachable relay and --source-shares-controller; "
"adds airtime, never reboots, does not fix legacy block retries"),
)
parser.add_argument(
"--legacy-relay-reboot-recovery", action="store_true",
help=("on exact 0/0 manifest timeout, offer one relay reboot and "
"normal/temporary path requalification before retrying the same MID; "
"requires one 0- or 1-hop relay and --source-shares-controller"),
)
parser.add_argument(
"--source-contact",
metavar="NAME_OR_KEY",
@@ -7811,6 +8001,12 @@ def validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) ->
):
parser.error("--legacy-relay-dedup-workaround requires one --relay, "
"--source-shares-controller, and no additional OTA hops")
if args.legacy_relay_reboot_recovery and (
len(args.relay) != 1 or not args.source_shares_controller
or args.source_already_temp
):
parser.error("--legacy-relay-reboot-recovery requires one --relay, "
"--source-shares-controller, and a managed source TempRadio")
clear_manifest = args.clear_completed_manifest
clear_body_hash = args.clear_completed_on_body_hash
if bool(clear_manifest) != bool(clear_body_hash):
@@ -8684,6 +8880,7 @@ def main(
seeder_attempted = False
source_temp_owned = False
source_temp_cleanup_attempted = False
target_temp_armed_at: float | None = None
source_rxps_saved: RxpsSettings | None = None
source_rxps_changed = False
source_rxps_recovery_path: Path | None = None
@@ -8692,6 +8889,7 @@ def main(
target_rxps_changed = False
armed_relay_values: list[tuple[str, str]] = []
relay_public_keys: dict[str, str] = {}
relay_reboot_preflight_key: str | None = None
relay_timing_settings: list[RelayTimingSettings] = []
owned_ota_hop_settings: list[OtaHopSettings] = []
password = args.password or os.environ.get("MESHCORE_ADMIN_PASSWORD", "")
@@ -8907,6 +9105,15 @@ def main(
args.target_rxps_saved = target_rxps_saved
args.target_rxps_profile = target_rxps_profile
args.ota_hop_settings = prepare_ota_hop_settings(controller, args)
if args.legacy_relay_reboot_recovery:
relay_name, relay_password = args.relay_values[0]
relay_reboot_preflight_key = read_remote_public_key_bounded(
controller, relay_name, password=relay_password,
)
args.relay_reboot_normal_hops = relay_route_hops(
controller, relay_name, relay_reboot_preflight_key,
"normal-channel route before TempRadio",
)
confirm_update(args, target, package)
freq, bandwidth, sf, cr, _minutes = args.temp_values
@@ -9032,7 +9239,18 @@ def main(
)
for relay_name, relay_password in args.relay_values
}
if args.legacy_relay_reboot_recovery:
relay_name, _relay_password = args.relay_values[0]
if relay_public_keys[relay_name] != relay_reboot_preflight_key:
raise OtaError("relay identity changed after reboot-route preflight")
args.relay_reboot_normal_hops = relay_route_hops(
controller, relay_name, relay_public_keys[relay_name],
"normal-channel route before TempRadio",
)
print(f"[legacy] relay reboot fallback eligible: "
f"{args.relay_reboot_normal_hops} normal-channel hops")
target_temp_armed_at = time.monotonic()
target_temp_owned = True
if args.source_shares_controller:
# A lost-reply probe can schedule the shared local source onto the
@@ -9114,7 +9332,35 @@ def main(
seeder_attempted = True
seeder.start()
find_and_start_pull(controller, args, package, seeder)
monitor_download(controller, args, package, seeder)
def mark_recovery_normal() -> None:
nonlocal controller_changed, source_temp_owned
controller_changed = False
source_temp_owned = False
def mark_recovery_temp_owned() -> None:
nonlocal source_temp_owned
# Set before the one-shot local command: a lost acknowledgement
# can still mean the shared radio changed channels.
source_temp_owned = True
def mark_recovery_temp_active() -> None:
nonlocal controller_changed
controller_changed = True
def try_relay_reboot(status: str, deadline: float) -> bool:
assert target_temp_armed_at is not None
return recover_manifest_by_relay_reboot(
controller, args, package, seeder, status, deadline,
target_temp_armed_at, relay_public_keys, original_radio,
temp_radio, mark_recovery_normal, mark_recovery_temp_owned,
mark_recovery_temp_active,
)
monitor_download(
controller, args, package, seeder,
reboot_recovery=(try_relay_reboot if args.legacy_relay_reboot_recovery else None),
)
# A serial source is exclusively owned by motatool. Release that
# ownership before trying to restore its policy through the text CLI.
seeder.stop()
@@ -9275,6 +9521,8 @@ def main(
):
if not controller_changed:
try:
if args.source_shares_controller and args.legacy_relay_reboot_recovery:
source_temp_owned = True
switch_controller_to_temp_radio(
controller, args, temp_command, temp_radio
)
tools/lora_ota/rak3401_route_search.py
+108 -41
View File
@@ -7,12 +7,19 @@ 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 fields ``ota_transport_deflate``,
MeshCore bytes after minimizing package count. ``--objective bytes`` and
``--objective packets`` consider every feasible forward edge and minimize total
traffic first, even when an additional install saves airtime. Inventory fields
``ota_transport_deflate``,
``ota_fetch_pipeline``, and ``transport_max_block_bytes`` belong to the
currently running/source image; the pinned first bootstrap is always legacy
raw. The explicit maximum controls signed-container block geometry independently
of whether DEFLATE is allowed; either DEFLATE or 2 KiB geometry negotiates the
171-byte v2 wire profile.
``ota_proactive_proofs`` distinguishes later legacy receivers that accept a
modern seeder's unsolicited proof from the original receiver that explicitly
requests each proof. If supplied for one image, it must be supplied for all.
"""
from __future__ import annotations
@@ -45,6 +52,7 @@ AVAILABLE_PAGES = (STAGE_CEILING - APP_BASE) // PAGE
EXPECTED_TARGET_ID = 0x2FA509C1
EXPECTED_HARDWARE = "RAK_3401"
TRANSPORT_CAPABILITY = "ota_transport_deflate"
PROACTIVE_PROOF_CAPABILITY = "ota_proactive_proofs"
TRANSPORT_PIPELINE_FIELD = "ota_fetch_pipeline"
TRANSPORT_MAX_BLOCK_FIELD = "transport_max_block_bytes"
TRANSPORT_CAPABILITY_ALIAS = "transport_inflate"
@@ -168,7 +176,7 @@ def _common_transport_cost(block_count: int) -> tuple[int, int, int]:
)
def legacy_transport_cost(payload_size: int) -> dict[str, int | str]:
def legacy_transport_cost(payload_size: int, proactive_proofs: bool = False) -> dict[str, int | str]:
"""Ideal clean legacy transfer, including requests, DATA, and proofs."""
blocks = _block_lengths(payload_size, LEGACY_BLOCK_SIZE)
block_count = len(blocks)
@@ -178,13 +186,15 @@ def legacy_transport_cost(payload_size: int) -> dict[str, int | str]:
)
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.
# 9-byte OTA_REQ frame. Original receivers also immediately request the
# proof; later legacy receivers wait for a modern seeder's proactive proof.
data_bytes = payload_size + 11 * data_packets
request_bytes = 11 * block_count
proof_request_bytes = 9 * block_count
proof_requests = 0 if proactive_proofs else block_count
proof_request_bytes = 9 * proof_requests
manifest_bytes = common_bytes - proof_bytes
return {
"profile": "legacy-160-raw",
"profile": "legacy-160-proactive-proof" if proactive_proofs else "legacy-160-raw",
"block_size": LEGACY_BLOCK_SIZE,
"payload_bytes": payload_size,
"wire_bytes": payload_size,
@@ -194,12 +204,12 @@ def legacy_transport_cost(payload_size: int) -> dict[str, int | str]:
"manifest_packets": common_packets - block_count,
"request_pipeline": 1,
"request_packets": block_count,
"proof_request_packets": block_count,
"proof_request_packets": proof_requests,
"proof_packets": block_count,
"origin_mesh_bytes": (
common_bytes + data_bytes + request_bytes + proof_request_bytes
),
"packets": common_packets + data_packets + 2 * block_count,
"packets": common_packets + data_packets + block_count + proof_requests,
"manifest_bytes": manifest_bytes,
"block_request_bytes": request_bytes,
"proof_request_bytes": proof_request_bytes,
@@ -369,6 +379,10 @@ def load_inventory(path: Path) -> list[dict[str, object]]:
isinstance(record, dict) and TRANSPORT_MAX_BLOCK_FIELD in record
for record in images
)
strict_proof_contract = any(
isinstance(record, dict) and PROACTIVE_PROOF_CAPABILITY in record
for record in images
)
edge_policy = document.get("edge_policy", {})
strict_version_ranks = (
isinstance(edge_policy, dict)
@@ -468,6 +482,14 @@ def load_inventory(path: Path) -> list[dict[str, object]]:
f"{TRANSPORT_CAPABILITY}"
)
record[TRANSPORT_CAPABILITY] = capability
if strict_proof_contract and PROACTIVE_PROOF_CAPABILITY not in record:
raise RouteSearchError(f"image {index} {PROACTIVE_PROOF_CAPABILITY} is required")
proactive_proofs = record.get(PROACTIVE_PROOF_CAPABILITY, False)
if type(proactive_proofs) is not bool:
raise RouteSearchError(f"image {index} {PROACTIVE_PROOF_CAPABILITY} must be boolean")
if index == 0 and proactive_proofs:
raise RouteSearchError("the deployed first receiver requires explicit proof requests")
record[PROACTIVE_PROOF_CAPABILITY] = proactive_proofs
raw_max_block = record.get(TRANSPORT_MAX_BLOCK_FIELD)
if raw_max_block is None:
if strict_block_contract:
@@ -943,6 +965,26 @@ def minimum_hop_pairs(
}
def endpoint_route_pairs(
rows: list[dict[str, object]], images: list[dict[str, object]]
) -> set[tuple[int, int]]:
"""All start-to-endpoint edges, including routes with extra installs."""
pairs = feasible_forward_pairs(rows, images)
outgoing: dict[int, list[int]] = {}
for source, target in pairs:
outgoing.setdefault(source, []).append(target)
reachable = {0}
for source in range(len(images)):
if source in reachable:
reachable.update(outgoing.get(source, []))
leads_to_endpoint = {len(images) - 1}
for source in range(len(images) - 2, -1, -1):
if any(target in leads_to_endpoint for target in outgoing.get(source, [])):
leads_to_endpoint.add(source)
return {(source, target) for source, target in pairs
if source in reachable and target in leads_to_endpoint}
def job_needs_transport_stats(
job: Job,
images: list[dict[str, object]],
@@ -950,7 +992,7 @@ def job_needs_transport_stats(
relevant_pairs: set[tuple[int, int]] | None = None,
) -> bool:
return (
objective == "transport"
objective in ("transport", "bytes", "packets")
and source_uses_deflate(images, job[0])
and (relevant_pairs is None or (job[0], job[1]) in relevant_pairs)
)
@@ -1237,7 +1279,9 @@ def edge_transport_cost(
block_size,
)
else:
cost = legacy_transport_cost(payload)
cost = legacy_transport_cost(
payload, source > 0 and images[source].get(PROACTIVE_PROOF_CAPABILITY, False)
)
result = dict(cost)
result["linear_path_bytes"] = linear_path_bytes(
int(cost["origin_mesh_bytes"]), int(cost["packets"]), relay_hops
@@ -1248,9 +1292,10 @@ def edge_transport_cost(
def select_route(rows: list[dict[str, object]], images: list[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"):
if objective not in ("container", "transport", "bytes", "packets"):
raise RouteSearchError(f"unknown route objective: {objective}")
if objective == "transport" and not complete:
traffic = objective != "container"
if traffic and not complete:
raise RouteSearchError(
"transport objective requires complete geometry before shortest-DAG sizing"
)
@@ -1267,13 +1312,12 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]],
"container": baseline_size, "stage_start": baseline_stage,
"margin": baseline_margin, "feasible": True,
}
if objective == "transport":
if traffic:
baseline_row["payload"] = payload_size_from_container(baseline_size)
all_feasible_pairs = feasible_forward_pairs(rows, images)
relevant_pairs = (
minimum_hop_pairs(rows, images)
if objective == "transport" else all_feasible_pairs
)
relevant_pairs = (minimum_hop_pairs(rows, images) if objective == "transport"
else endpoint_route_pairs(rows, images) if traffic
else all_feasible_pairs)
edges: dict[tuple[int, int], dict[str, object]] = {(0, 1): baseline_row}
for row in rows:
source = int(row["source"])
@@ -1292,12 +1336,11 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]],
key = source, target
if key not in relevant_pairs:
continue
if objective == "transport":
transport_bytes = int(
edge_transport_cost(row, images, relay_hops)["linear_path_bytes"]
)
if traffic:
cost = edge_transport_cost(row, images, relay_hops)
rank = (
transport_bytes, int(row["container"]),
int(cost["packets"]) if objective == "packets" else 0,
int(cost["linear_path_bytes"]), int(row["container"]),
-int(row["margin"]), int(row["memory"]),
)
else:
@@ -1306,9 +1349,11 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]],
if old is None:
edges[key] = row
else:
if objective == "transport":
if traffic:
old_cost = edge_transport_cost(old, images, relay_hops)
old_rank = (
int(edge_transport_cost(old, images, relay_hops)["linear_path_bytes"]),
int(old_cost["packets"]) if objective == "packets" else 0,
int(old_cost["linear_path_bytes"]),
int(old["container"]), -int(old["margin"]), int(old["memory"]),
)
else:
@@ -1331,10 +1376,15 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]],
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"])
if traffic else int(row["container"])
)
candidate = (best[source][0] + 1,
best[source][1] + edge_bytes,
increment = (1, edge_bytes)
if objective == "bytes":
increment = (edge_bytes, 1)
elif objective == "packets":
increment = (int(edge_transport_cost(row, images, relay_hops)["packets"]), edge_bytes)
candidate = (best[source][0] + increment[0],
best[source][1] + increment[1],
best[source][2] + [target])
if target not in best or candidate < best[target]:
best[target] = candidate
@@ -1344,17 +1394,18 @@ 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"
)
objective_text = {
"transport": "minimum packages, then minimum ideal linear-path serialized MeshCore bytes",
"container": "minimum packages, then minimum total container bytes",
"bytes": "minimum ideal linear-path serialized MeshCore bytes, then minimum packages",
"packets": "minimum ideal packets, then minimum ideal linear-path serialized MeshCore bytes",
}[objective]
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(all_feasible_pairs),
"objective": objective_text}
if objective == "transport":
if traffic:
transport_encoder = transport_encoder_for_rows(
rows, images, relevant_pairs
)
@@ -1364,6 +1415,7 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]],
"supported_block_sizes": [LEGACY_BLOCK_SIZE, DEFLATE_BLOCK_SIZE],
"comparison_baseline_block_size": LEGACY_BLOCK_SIZE,
"source_capability_field": TRANSPORT_CAPABILITY,
"source_proactive_proof_field": PROACTIVE_PROOF_CAPABILITY,
"source_pipeline_field": TRANSPORT_PIPELINE_FIELD,
"source_max_block_field": TRANSPORT_MAX_BLOCK_FIELD,
"source_profile_matrix": dict(TRANSPORT_PROFILE_MATRIX),
@@ -1373,13 +1425,18 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]],
"adaptive requests, manifest, exact per-block proofs, and MeshCore framing"
),
"excluded": (
"discovery, retries, flood fan-out, and radio-dependent LoRa PHY coding/preamble"
"discovery, retries, admin/control commands, 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),
reachable = {0}
for source, target in sorted(all_feasible_pairs):
if source in reachable:
reachable.add(target)
result = {**common, "status": "unreachable", "reachable_nodes": sorted(reachable),
"endpoint_node": endpoint,
"endpoint_incoming_feasible": sorted(
source for source, target in all_feasible_pairs
@@ -1398,7 +1455,7 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]],
"expected_staging_margin": int(row["margin"]),
"expected_target_sha256": images[target]["sha256"],
"expected_target_version": images[target].get("version", "unknown")}
if objective == "transport":
if traffic:
cost = edge_transport_cost(row, images, relay_hops)
step["transport"] = {
key: cost[key] for key in (
@@ -1424,8 +1481,11 @@ def select_route(rows: list[dict[str, object]], images: list[dict[str, object]],
"shortest_package_count": hop_distance[endpoint],
"shortest_route_count": route_count[endpoint],
"selected_total_bytes": selected_container_bytes, "steps": steps}
if objective == "transport":
result["selected_total_transport_bytes"] = best[endpoint][1]
if traffic:
result["selected_total_transport_bytes"] = sum(int(step["transport"]["linear_path_bytes"]) for step in steps)
result["selected_total_packets"] = sum(int(step["transport"]["packets"]) for step in steps)
result["selected_total_linear_path_packets"] = result["selected_total_packets"] * (relay_hops + 1)
result["selected_package_count"] = len(steps)
write_atomic_text(
output, json.dumps(result, indent=2, sort_keys=True) + "\n"
)
@@ -1440,8 +1500,8 @@ def main(argv: list[str] | None = None) -> int:
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",
"--objective", choices=("container", "transport", "bytes", "packets"), default="container",
help="container/transport minimize packages first; bytes/packets minimize traffic first",
)
parser.add_argument(
"--motatool", metavar="PATH",
@@ -1565,6 +1625,9 @@ def main(argv: list[str] | None = None) -> int:
int(row["memory"]),
)
cache[key] = row
if done % 500 == 0:
write_csv(args.work_dir / "geometry.csv",
project_cache(cache, images, required & cache.keys()))
if done % 100 == 0 or done == len(futures):
print(
f"geometry_measured={done}/{len(geometry_remaining)}",
@@ -1584,13 +1647,14 @@ def main(argv: list[str] | None = None) -> int:
geometry_complete = geometry_completed == required
complete = geometry_complete
if args.objective == "transport":
if args.objective in ("transport", "bytes", "packets"):
if not geometry_complete:
raise RouteSearchError(
"transport objective requires complete geometry before "
"shortest-DAG sizing"
)
relevant_pairs = minimum_hop_pairs(rows, images)
relevant_pairs = (minimum_hop_pairs(rows, images) if args.objective == "transport"
else endpoint_route_pairs(rows, images))
transport_remaining = [
job for job in jobs
if not cache_satisfies_job(
@@ -1642,6 +1706,9 @@ def main(argv: list[str] | None = None) -> int:
f"transport regeneration changed geometry for {key}"
)
cache[key] = row
if done % 100 == 0:
write_csv(args.work_dir / "geometry.csv",
project_cache(cache, images, geometry_completed))
if done % 100 == 0 or done == len(futures):
print(
f"transport_measured={done}/"
tools/lora_ota/test_build_rak3401_legacy24_bundle.py
+53
View File
@@ -0,0 +1,53 @@
from copy import deepcopy
from types import SimpleNamespace
import unittest
import build_rak3401_legacy24_bundle as bundle
class Legacy24BundleTests(unittest.TestCase):
def setUp(self):
self.source = dict(body_hash="0123456789ABCDEF", size=413792)
self.target = dict(sha256="ab" * 32, version="1.16.11.0", size=393800)
self.manifest = SimpleNamespace(
is_full=False, is_bootloader=False, codec_id=2,
target_id=bundle.search.EXPECTED_TARGET_ID, hw_id=b"RAK_3401\0\0",
base_hash=bytes.fromhex(self.source["body_hash"]),
image_hash=bytes.fromhex(self.target["sha256"]),
fw_version=bundle.search.motalib.pack_version(self.target["version"]),
image_size=self.target["size"], block_size=2048,
)
def test_exact_application_delta(self):
bundle.validate_manifest(self.manifest, self.source, self.target, 2048)
def test_rejects_other_image_types_and_every_identity_mismatch(self):
for field, value in (
("is_full", True), ("is_bootloader", True), ("codec_id", 3),
("target_id", 0), ("hw_id", b"RAK_4631"),
("base_hash", b"\0" * 8), ("image_hash", b"\0" * 32),
("fw_version", 1), ("image_size", 1), ("block_size", 4096),
):
with self.subTest(field=field):
manifest = deepcopy(self.manifest)
setattr(manifest, field, value)
with self.assertRaises(ValueError):
bundle.validate_manifest(manifest, self.source, self.target, 2048)
def test_exact_zero_margin_is_valid(self):
self.assertEqual(bundle.validate_workspace(self.source, self.target, 0x72000, 245426, 0), 0)
def test_workspace_constraints(self):
for memory, size, margin in (
(0x73000, 245426, -4096), # Overlaps the stored package.
(0x72001, 245426, 0), # Not page aligned.
(0x72000, 245426, 4096), # Different from the searched geometry.
(0x64000, 245426, 57344), # Insufficient source plus two-page headroom.
):
with self.subTest(memory=memory, size=size, margin=margin):
with self.assertRaises(ValueError):
bundle.validate_workspace(self.source, self.target, memory, size, margin)
if __name__ == "__main__":
unittest.main()
tools/lora_ota/test_lora_ota_legacy_dedup.py
+146
View File
@@ -158,6 +158,152 @@ class LegacyManifestRecoveryTests(unittest.TestCase):
self.assertEqual([c.args[-1] for c in recover.call_args_list], [0, 1, 2])
self.assertEqual({c.args[-2] for c in recover.call_args_list}, {1300})
def test_monitor_offers_reboot_only_for_exact_zero_block_timeout(self):
self.args.transfer_timeout_minutes = 20
self.args.poll_seconds = 1
self.args.reply_timeout = 1
fetching = f"download: fetching 0/40 id={self.package.manifest_id}"
ready = f"download: ready to install 40/40 id={self.package.manifest_id}"
self.controller.remote_command.side_effect = [fetching, self.status, ready]
reboot = mock.Mock(return_value=True)
with mock.patch.object(ota, "initial_status_wait_seconds", return_value=0), \
mock.patch.object(ota, "recover_legacy_manifest_timeout", return_value=False):
self.assertEqual(
ota.monitor_download(self.controller, self.args, self.package,
reboot_recovery=reboot), ready,
)
reboot.assert_called_once_with(self.status, 1300)
class RelayRebootRecoveryTests(unittest.TestCase):
def setUp(self):
self.package = ota.parse_mota(mota_blob(firmware(b"reboot" * 100, VERSION_NEW)))
self.relay = "ab" * 32
self.target = "cd" * 32
self.status = ("OTA | download: failed (manifest timeout) 0/0 "
f"id={self.package.manifest_id}")
self.args = argparse.Namespace(
source_shares_controller=True, relay_values=[(self.relay, "secret")],
relay_reboot_normal_hops=1, target=self.target,
temp_values=(909.5, 125, 7, 5, 120), yes=True,
)
self.controller = mock.Mock()
self.controller._run.return_value = [{
"public_key": self.relay, "out_path_len": 1,
}]
self.controller.get_radio.return_value = ota.RadioSettings(910.525, 62.5, 7, 5, False)
self.controller.remote_command.side_effect = [
self.status, ota.TransmissionError("no reboot reply"), self.status,
f"OK pulling mid={self.package.manifest_id} -> flash",
]
self.on_normal = mock.Mock()
self.on_temp = mock.Mock()
self.on_temp_active = mock.Mock()
self.addCleanup(mock.patch.stopall)
mock.patch.object(ota.time, "sleep").start()
mock.patch.object(ota.time, "monotonic", return_value=100).start()
self.source_cli = mock.patch.object(
ota, "source_cli_command",
return_value="TempRadio active: 909.500,125.00,7,5 7000s left,preamble=32",
).start()
self.shorten = mock.patch.object(ota, "shorten_source_temp_window", return_value=True).start()
self.read_key = mock.patch.object(
ota, "read_remote_public_key_bounded",
side_effect=[self.target, self.relay, self.relay, self.relay, self.target],
).start()
self.arm = mock.patch.object(ota, "arm_relay_temp_radio_once", return_value=True).start()
self.switch = mock.patch.object(ota, "switch_controller_to_temp_radio").start()
mock.patch.object(ota, "require_source_on_temp_after_uncertain_arm").start()
self.output = contextlib.redirect_stdout(io.StringIO())
self.output.__enter__()
self.addCleanup(self.output.__exit__, None, None, None)
def recover(self, status=None, armed_at=100, deadline=1300):
return ota.recover_manifest_by_relay_reboot(
self.controller, self.args, self.package, None,
self.status if status is None else status, deadline, armed_at,
{self.relay: self.relay}, self.controller.get_radio.return_value,
ota.RadioSettings(909.5, 125, 7, 5, False),
self.on_normal, self.on_temp, self.on_temp_active,
)
def test_one_hop_reboot_rearms_bounded_lease_and_same_mid(self):
self.assertTrue(self.recover())
commands = [c.args[1] for c in self.controller.remote_command.call_args_list]
self.assertEqual(commands, [
"ota status", "reboot", "ota status",
f"ota pull {self.package.manifest_id} flash",
])
self.controller.forget_remote_auth.assert_called_once_with(self.relay)
self.on_normal.assert_called_once()
self.on_temp.assert_called_once()
self.on_temp_active.assert_called_once()
self.assertEqual(
self.arm.call_args.args[3], "tempradio 909.5,125,7,5,119",
)
self.assertEqual(self.read_key.call_count, 5)
def test_direct_relay_is_also_eligible(self):
self.args.relay_reboot_normal_hops = 0
self.controller._run.return_value[0]["out_path_len"] = 0
self.assertTrue(self.recover())
def test_recent_progress_avoids_reboot(self):
self.controller.remote_command.side_effect = [
f"download: fetching 1/40 id={self.package.manifest_id}",
]
self.assertTrue(self.recover())
self.assertEqual([c.args[1] for c in self.controller.remote_command.call_args_list],
["ota status"])
self.on_normal.assert_not_called()
def test_wrong_status_or_route_cannot_reboot(self):
self.assertFalse(self.recover(status=self.status.replace("0/0", "1/40")))
self.controller.remote_command.assert_not_called()
self.controller._run.return_value[0]["out_path_len"] = 2
with self.assertRaisesRegex(ota.OtaError, "0- or 1-hop"):
self.recover()
self.controller.remote_command.assert_not_called()
def test_short_lease_fails_before_reboot(self):
with self.assertRaisesRegex(ota.OtaError, "too little"):
self.recover(armed_at=-7000)
self.controller.remote_command.assert_not_called()
def test_declined_prompt_does_not_reboot(self):
self.args.yes = False
with mock.patch.object(ota.sys.stdin, "isatty", return_value=True), \
mock.patch("builtins.input", return_value="n"):
self.assertFalse(self.recover())
self.assertEqual([c.args[1] for c in self.controller.remote_command.call_args_list],
["ota status"])
def test_normal_path_failure_does_not_rearm_or_pull(self):
self.read_key.side_effect = [self.target, self.relay, ota.TransmissionError("offline")]
with self.assertRaisesRegex(ota.TransmissionError, "offline"):
self.recover()
self.on_normal.assert_called_once()
self.on_temp.assert_not_called()
self.on_temp_active.assert_not_called()
self.arm.assert_not_called()
commands = [c.args[1] for c in self.controller.remote_command.call_args_list]
self.assertEqual(commands, ["ota status", "reboot"])
def test_parser_requires_one_managed_shared_relay(self):
parser = ota.build_parser()
common = ["update.mota", "target", "--legacy-relay-reboot-recovery"]
for flags in ([], ["--relay", "relay"], ["--source-shares-controller"],
["--relay", "a", "--relay", "b", "--source-shares-controller"],
["--relay", "a", "--source-shares-controller", "--source-already-temp"]):
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
ota.validate_args(parser.parse_args(common + flags), parser)
args = parser.parse_args(common + [
"--relay", "relay", "--source-shares-controller",
"--controller-tcp", "127.0.0.1", "--source-tcp", "127.0.0.1",
"--source-cli-tcp", "127.0.0.1",
])
ota.validate_args(args, parser)
if __name__ == "__main__":
unittest.main()
tools/lora_ota/test_rak3401_route_search.py
+83
View File
@@ -29,6 +29,89 @@ def firmware(body: bytes, version: str) -> bytes:
class RouteSearchTests(unittest.TestCase):
def test_legacy_proactive_proofs_remove_one_request_per_block(self) -> None:
normal = search.legacy_transport_cost(61 * 1024)
proactive = search.legacy_transport_cost(61 * 1024, True)
self.assertEqual(normal['packets'] - proactive['packets'], 61)
self.assertEqual(normal['origin_mesh_bytes'] - proactive['origin_mesh_bytes'], 61 * 9)
self.assertEqual(proactive['proof_request_packets'], 0)
self.assertEqual(proactive['data_packets'], normal['data_packets'])
self.assertEqual(proactive['proof_packets'], normal['proof_packets'])
def test_proactive_proof_inventory_is_explicit_and_boolean(self) -> None:
with tempfile.TemporaryDirectory() as directory:
path, _ = self.write_inventory(Path(directory), [b'0', b'1', b'2'])
original = json.loads(path.read_text())
for values in ((False, True, True), (False, 'true', True), (False, True, None), (True, True, True)):
data = json.loads(json.dumps(original))
for record, value in zip(data['images'], values):
if value is not None:
record[search.PROACTIVE_PROOF_CAPABILITY] = value
path.write_text(json.dumps(data))
if values == (False, True, True):
images = search.load_inventory(path)
row = dict(source=1,target=2,container=search.container_size(2048),payload=2048)
self.assertEqual(search.edge_transport_cost(row,images,0)['proof_request_packets'],0)
row['source'] = 0
self.assertEqual(search.edge_transport_cost(row,images,0)['proof_request_packets'],2)
else:
with self.assertRaises(search.RouteSearchError):search.load_inventory(path)
def test_endpoint_routes_keep_longer_paths_but_exclude_dead_ends(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
manifest, _ = self.write_inventory(root, [b"0", b"1", b"2", b"3", b"4", b"5"])
images = search.load_inventory(manifest)
rows = [dict(source=s, target=t, feasible=True) for s,t in
[(1,5),(1,2),(2,5),(1,3),(4,5)]]
self.assertEqual(search.endpoint_route_pairs(rows,images), {(0,1),(1,5),(1,2),(2,5)})
self.assertEqual(search.minimum_hop_pairs(rows,images), {(0,1),(1,5)})
def test_traffic_first_can_choose_more_installs(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"], baseline_size=baseline,
)
images = search.load_inventory(manifest)
def edge(source: int, target: int, payload: int) -> dict[str, object]:
return dict(source=source, target=target, memory=0x98000,
payload=payload, container=search.container_size(payload),
stage_start=0xC0000, margin=4096, feasible=True)
rows = [edge(1, 3, 10000), edge(1, 2, 500), edge(2, 3, 500)]
shortest = search.select_route(rows, images, baseline, root / "short.json",
True, objective="transport")
self.assertEqual(shortest["nodes"], [0, 1, 3])
for objective in ("bytes", "packets"):
result = search.select_route(rows, images, baseline, root / (objective + ".json"),
True, objective=objective)
self.assertEqual(result["nodes"], [0, 1, 2, 3])
self.assertEqual(result["shortest_package_count"], 2)
self.assertEqual(result["selected_package_count"], 3)
self.assertLess(result["selected_total_transport_bytes"], shortest["selected_total_transport_bytes"])
self.assertLess(result["selected_total_packets"], shortest["selected_total_packets"])
def test_packet_objective_counts_proofs_and_requests(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], transport_block_sizes=[1024, 2048, 1024])
images = search.load_inventory(manifest)
rows = []
for memory, payload, wire, packets in [(0x98000, 4096, 300, 2), (0x99000, 1800, 400, 3)]:
rows.append(dict(source=1, target=2, memory=memory, payload=payload,
container=search.container_size(payload, 2048), block_size=2048,
stage_start=0xC0000, margin=100, feasible=True,
payload_sha256=f"{memory:064x}", transport_encoder_sha256="e"*64,
v2_wire_bytes=wire, v2_deflate_bytes=wire, v2_deflate_blocks=(payload+2047)//2048,
v2_data_packets=packets))
expected=min(rows,key=lambda row:(search.edge_transport_cost(row,images,0)['packets'],search.edge_transport_cost(row,images,0)['linear_path_bytes']))
result=search.select_route(rows,images,baseline,root/'packets.json',True,objective='packets')
self.assertEqual(result['steps'][1]['inplace_memory'], f"0x{expected['memory']:X}")
def write_inventory(
self, root: Path, payloads: list[bytes], *, baseline_size: int = 100,
versions: list[str] | None = None,