rich/MeshCore
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Bound nRF52 repeater OTA proof memory and correct RAK3401 bridge GPS contract

Browse Source
This commit is contained in:
mikecarper committed 2026-10-21 09:53:23 -07:00
1 parent 8c4fb8d505
commit 38f68596f7
7 files changed
+172 -34
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

build_legacy.sh
+3 -3
View File
@@ -2173,11 +2173,11 @@ is_rak_gps_retaining_ota_target() {
if [ -n "${NRF52_OTA_SENSOR_PROFILE:-}" ]; then
case "$target_lc" in
rak_4631_*bridge_rs232_serial1*) return 1 ;;
rak_3401_*bridge_rs232_serial1*|rak_4631_*bridge_rs232_serial1*) return 1 ;;
rak_3401_*|rak_4631_*) return 0 ;;
esac
fi
# Serial1 is the RAK12501 UART on RAK4631. The Serial1 RS-232 bridge owns
# Serial1 is the RAK12501 UART on RAK3401/RAK4631. The RS-232 bridge owns
# that port instead, so its reduced profile intentionally uses the INA-only
# recipe and must not promise or require the WisBlock GPS provider.
case "$target_lc" in
@@ -4036,7 +4036,7 @@ apply_lora_ota_no_external_sensors_profile() {
# costs less than 5 KiB and covers their most common external telemetry use.
# The ordinary sibling remains fully sensor-enabled.
# GPS availability follows the exact target recipe. In particular, the
# RAK4631 Serial1 RS-232 bridge intentionally omits GPS because both need the
# RAK3401/RAK4631 Serial1 RS-232 bridge omits GPS because both need the
# same UART; other reduced RAK profiles below retain compatible GPS paths.
if ! is_rak_i2c_voltage_monitor_ota_target "$env_name"; then
omitted_sensor_flags+=" ${voltage_monitor_flags}"
scripts/check_firmware_ram.py
+3 -3
View File
@@ -307,9 +307,9 @@ def requirements(platform, defines, target):
leaf_limit = integer(defines, "OTA_PROOFGEN_SCRATCH",
16384 if platform == "ESP32_PLATFORM" else
8192 if "OTA_SD_STORE" in defines else 4096)
bounded_default = (platform == "ESP32_PLATFORM"
and (("COMPANION_RADIO_FULL" in defines and "BOARD_HAS_PSRAM" not in defines)
or qualified_repeater)
bounded_default = (((platform == "ESP32_PLATFORM" and "COMPANION_RADIO_FULL" in defines
and "BOARD_HAS_PSRAM" not in defines)
or (platform in {"ESP32_PLATFORM", "NRF52_PLATFORM"} and qualified_repeater))
and 4 <= leaf_limit <= 16384)
bounded = integer(defines, "OTA_BOUNDED_SOURCE_PROOFS", int(bounded_default))
if bounded not in (0, 1):
src/helpers/ota/OtaMemoryPolicy.h
+5 -5
View File
@@ -1,12 +1,12 @@
#pragma once
// Non-PSRAM Full Companions and qualified ESP32 repeaters retain the complete source
// geometry without caching every Merkle leaf. An explicit 0 keeps the legacy
// Non-PSRAM ESP32 Full Companions and qualified ESP32/nRF52 repeaters retain
// the complete source geometry without caching every Merkle leaf. An explicit 0 keeps the legacy
// cache for matched qualification builds.
#ifndef OTA_BOUNDED_SOURCE_PROOFS
#if defined(ESP32_PLATFORM) && \
((defined(COMPANION_RADIO_FULL) && !defined(BOARD_HAS_PSRAM)) || \
(defined(MESH_CLIENT_REPEATER_ONLY) && MESH_CLIENT_REPEATER_ONLY)) && \
#if ((defined(ESP32_PLATFORM) && defined(COMPANION_RADIO_FULL) && !defined(BOARD_HAS_PSRAM)) || \
((defined(ESP32_PLATFORM) || defined(NRF52_PLATFORM)) && \
defined(MESH_CLIENT_REPEATER_ONLY) && MESH_CLIENT_REPEATER_ONLY)) && \
(!defined(OTA_PROOFGEN_SCRATCH) || \
(OTA_PROOFGEN_SCRATCH >= 4 && OTA_PROOFGEN_SCRATCH <= 16384))
#define OTA_BOUNDED_SOURCE_PROOFS 1
test/fixtures/ota_source_proofs/test.cpp
Vendored
+12 -9
View File
@@ -1,5 +1,6 @@
using Workspace = SourceMerkleProof<4096, OTA_LEAVES_FRAG>;
static constexpr uint32_t MaximumSourceBlocks = OTA_PROOFGEN_SCRATCH / 4;
using Workspace = SourceMerkleProof<MaximumSourceBlocks, OTA_LEAVES_FRAG>;
static_assert(sizeof(Workspace) <= 1536, "bounded allocation includes getter scratch");
class LeafSource : public MotaSource {
@@ -49,6 +50,7 @@ static void grouped_equivalence() {
2047,2048,2049,4095,4096}) counts.push_back(count);
size_t checked = 0;
for (uint32_t count : counts) {
if (count > MaximumSourceBlocks) continue;
LeafSource source(count);
source.quantum = count <= 65 ? uint16_t(std::array<int, 5>{1,3,17,127,128}[count % 5]) : 127;
Workspace workspace;
@@ -81,7 +83,8 @@ static void grouped_equivalence() {
++checked;
}
}
printf("PASS: byte-identical grouped proofs (%zu blocks, max4096; workspace=%zu)\n", checked, sizeof(Workspace));
printf("PASS: byte-identical grouped proofs (%zu blocks, max%u; workspace=%zu)\n",
checked, unsigned(MaximumSourceBlocks), sizeof(Workspace));
}
static void helper_faults() {
@@ -89,7 +92,7 @@ static void helper_faults() {
Workspace workspace;
assert(!workspace.bind(nullptr, 0, 17, 65, source.root));
assert(!workspace.bind(&source, 0, 17, 0, source.root));
assert(!workspace.bind(&source, 0, 17, 4097, source.root));
assert(!workspace.bind(&source, 0, 17, MaximumSourceBlocks + 1, source.root));
assert(!workspace.bind(&source, 0, UINT32_MAX, 65, source.root));
assert(workspace.bind(&source, 0, 17, 65, source.root));
assert(workspace.prepare() == Workspace::Step::Pending);
@@ -143,14 +146,14 @@ public:
};
static void manager_lifetimes() {
TrackedSource a(4096, 0x73), b(65, 0x56);
TrackedSource a(MaximumSourceBlocks, 0x73), b(65, 0x56);
a.host_encoding = b.host_encoding = Source::HostEncoding::Good;
a.fail_payload = true;
Sent sent;
{
OtaManager manager; manager.begin(123, send, &sent);
assert(manager.add_source(&a) && manager.add_source(&b));
assert(request(manager, a, 4095));
assert(request(manager, a, MaximumSourceBlocks - 1));
assert(allocated == 0); // receiving a request never allocates a full leaf table
sent.accept_data = sent.accept_proof = false;
manager.serviceEgress(); // cache host-generated DATA
@@ -205,21 +208,21 @@ static void manager_lifetimes() {
sent.accept_proof = true;
manager.serviceEgress();
assert(!manager.pendingServeJobs());
verify_v2(sent, a, 4095, true);
verify_v2(sent, a, MaximumSourceBlocks - 1, true);
manager.set_clock(5000000); manager.loop();
assert(allocated == 0); // idle timer releases the only workspace without an encoder scratch
sent.accept_data = false;
assert(request(manager, a, 4095));
assert(request(manager, a, MaximumSourceBlocks - 1));
fail_next = true;
manager.serviceEgress(); manager.serviceEgress();
assert(!manager.pendingServeJobs() && !fail_next && allocated == 0);
a.fail_leaves = true;
assert(request(manager, a, 4095));
assert(request(manager, a, MaximumSourceBlocks - 1));
manager.serviceEgress(); manager.serviceEgress();
assert(!manager.pendingServeJobs());
manager.clear_sources(); assert(allocated == 0);
a.fail_leaves = false;
assert(manager.add_source(&a) && request(manager, a, 4095));
assert(manager.add_source(&a) && request(manager, a, MaximumSourceBlocks - 1));
manager.serviceEgress(); manager.serviceEgress();
assert(allocated == sizeof(Workspace));
assert(manager.remove_source(&a)); assert(allocated == 0 && !manager.pendingServeJobs());
test/test_firmware_ram.py
+25 -1
View File
@@ -667,7 +667,7 @@ int main() {
}, "board_companion_radio_full")
def test_compact_tables_keep_capacity_and_reserve_active_ota_allocations(self):
for platform, leaf_limit in (("NRF52_PLATFORM", 4096), ("ESP32_PLATFORM", 1536),
for platform, leaf_limit in (("NRF52_PLATFORM", 1536), ("ESP32_PLATFORM", 1536),
("RP2040_PLATFORM", 4096), ("STM32_PLATFORM", 4096)):
with self.subTest(platform=platform):
flags = {"ENABLE_OTA": 1, "MESH_CLIENT_REPEATER_ONLY": 1,
@@ -720,6 +720,30 @@ int main() {
if name != "ota_generic_source_buffers":
self.assertEqual(selected[name], amount)
def test_nrf52_repeater_proofs_keep_sensor_profiles_limits_and_other_reserves(self):
for full_sensors in (False, True):
for store in ({}, {"OTA_RAK_AUTO_STORE": 1}, {"OTA_SD_STORE": 1}):
flags = {"ENABLE_OTA": 1, "MESH_CLIENT_REPEATER_ONLY": 1,
"ENV_INCLUDE_BME680_BSEC": int(full_sensors), **store}
for scratch in (None, 4, 4096, 8192, 16383, 16384, 32768):
if scratch is not None:
flags["OTA_PROOFGEN_SCRATCH"] = scratch
with self.subTest(full_sensors=full_sensors, store=store, scratch=scratch):
legacy = ram.requirements("NRF52_PLATFORM", {
**flags, "OTA_BOUNDED_SOURCE_PROOFS": 0}, "RAK_3401_repeater")
selected = ram.requirements("NRF52_PLATFORM", flags, "RAK_3401_repeater")
leaf_limit = scratch if scratch is not None else (8192 if "OTA_SD_STORE" in store else 4096)
workspace = 1536 if 4 <= leaf_limit <= 16384 else leaf_limit
self.assertEqual(selected["components"]["ota_generic_source_buffers"],
workspace + 2048 + 32)
for name, amount in legacy["components"].items():
if name != "ota_generic_source_buffers":
self.assertEqual(selected["components"][name], amount)
self.assertEqual(legacy["required_heap_bytes"] - selected["required_heap_bytes"],
leaf_limit - workspace)
self.assertEqual(legacy["required_contiguous_bytes"],
selected["required_contiguous_bytes"])
def test_combined_rak_oled_bound_does_not_change_other_display_budgets(self):
defines = {
"RAK_4631": 1, "ETHERNET_ENABLED": 1, "OTA_RAK_AUTO_STORE": 1,
test/test_ota_source_proofs.py
+32 -13
View File
@@ -23,29 +23,37 @@ static_assert(OTA_PROOFGEN_SCRATCH / 4 == 4095, "odd byte limit lost usable leaf
#endif
"""
base = [os.environ.get("CXX", "g++"), "-std=c++17", "-fsyntax-only", "-x", "c++", "-",
"-I", str(ROOT / "src"), "-DESP32_PLATFORM=1"]
for role in ("COMPANION_RADIO_FULL", "MESH_CLIENT_REPEATER_ONLY"):
"-I", str(ROOT / "src")]
for platform, role in (("ESP32_PLATFORM", "COMPANION_RADIO_FULL"),
("ESP32_PLATFORM", "MESH_CLIENT_REPEATER_ONLY"),
("NRF52_PLATFORM", "MESH_CLIENT_REPEATER_ONLY")):
for scratch in (None, 0, 1, 3, 4, 5, 16383, 16384, 16385, 32768):
bounded = scratch is None or 4 <= scratch <= 16384
flags = [f"-D{role}=1"]
flags = [f"-D{platform}=1", f"-D{role}=1"]
if scratch is not None:
flags.append(f"-DOTA_PROOFGEN_SCRATCH={scratch}")
with self.subTest(role=role, scratch=scratch):
with self.subTest(platform=platform, role=role, scratch=scratch):
result = subprocess.run([*base, *flags, f"-DEXPECT_BOUNDED={int(bounded)}"],
input=source, text=True, capture_output=True, timeout=30)
self.assertEqual(result.returncode, 0, result.stderr)
# Repeater proof buffers are internal even when the board has PSRAM.
result = subprocess.run([*base, "-DMESH_CLIENT_REPEATER_ONLY=1",
result = subprocess.run([*base, "-DESP32_PLATFORM=1", "-DMESH_CLIENT_REPEATER_ONLY=1",
"-DBOARD_HAS_PSRAM=1", "-DEXPECT_BOUNDED=1"],
input=source, text=True, capture_output=True, timeout=30)
self.assertEqual(result.returncode, 0, result.stderr)
for flags in ([], ["-DMESH_CLIENT_REPEATER_ONLY=0"],
["-DCOMPANION_RADIO_FULL=1", "-DBOARD_HAS_PSRAM=1"],
["-DMESH_CLIENT_REPEATER_ONLY=1", "-DOTA_BOUNDED_SOURCE_PROOFS=0"]):
with self.subTest(legacy=flags):
result = subprocess.run([*base, *flags, "-DEXPECT_BOUNDED=0"], input=source,
text=True, capture_output=True, timeout=30)
self.assertEqual(result.returncode, 0, result.stderr)
for platform in ("ESP32_PLATFORM", "NRF52_PLATFORM", "RP2040_PLATFORM", "STM32_PLATFORM"):
cases = [[], ["-DMESH_CLIENT_REPEATER_ONLY=0"],
["-DCOMPANION_RADIO_FULL=1", "-DBOARD_HAS_PSRAM=1"],
["-DMESH_CLIENT_REPEATER_ONLY=1", "-DOTA_BOUNDED_SOURCE_PROOFS=0"]]
if platform != "ESP32_PLATFORM":
cases.append(["-DCOMPANION_RADIO_FULL=1"])
if platform in ("RP2040_PLATFORM", "STM32_PLATFORM"):
cases.append(["-DMESH_CLIENT_REPEATER_ONLY=1"])
for flags in cases:
with self.subTest(platform=platform, legacy=flags):
result = subprocess.run([*base, f"-D{platform}=1", *flags, "-DEXPECT_BOUNDED=0"],
input=source, text=True, capture_output=True, timeout=30)
self.assertEqual(result.returncode, 0, result.stderr)
for scratch in (0, 3, 16385):
with self.subTest(forced_invalid=scratch):
result = subprocess.run([*base, f"-DOTA_PROOFGEN_SCRATCH={scratch}",
@@ -111,6 +119,17 @@ int main() {
self.assertEqual(ran.returncode, 0, ran.stdout + ran.stderr)
def test_grouped_equivalence_faults_and_manager_lifetimes(self):
self.check_grouped_equivalence(["-DESP32_PLATFORM=1", "-DCOMPANION_RADIO_FULL=1"])
def test_nrf52_full_and_reduced_profiles_keep_maximum_source_and_ownership(self):
for sensors in (0, 1):
with self.subTest(full_sensors=bool(sensors)):
self.check_grouped_equivalence([
"-DNRF52_PLATFORM=1", "-DMESH_CLIENT_REPEATER_ONLY=1",
f"-DENV_INCLUDE_BME680_BSEC={sensors}",
])
def check_grouped_equivalence(self, flags):
# Reuse the existing source/wire/allocator harness, retaining its legacy
# tests unchanged in their own suite. Only its helpers precede this point.
prefix = (ROOT / "test/fixtures/ota_cold_buffers/test.cpp").read_text().split(
@@ -126,7 +145,7 @@ int main() {
os.environ.get("CXX", "g++"), "-std=c++17", "-O1", "-Wall", "-Wextra", "-Werror",
"-Wno-unused-parameter", "-Wno-unused-function", "-Wno-sign-compare",
"-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie",
"-DESP32_PLATFORM=1", "-DCOMPANION_RADIO_FULL=1",
*flags,
"-I", str(ROOT / "src"), "-I", str(ROOT / "test/mocks"), str(source),
*[str(ROOT / "src/helpers/ota" / name) for name in files], str(ROOT / "src/Utils.cpp"),
"-Wl,--wrap=malloc", "-Wl,--wrap=free", "-o", str(binary),
test/test_rak_rs232_gps_contract.py
+92
View File
@@ -0,0 +1,92 @@
#!/usr/bin/env python3
"""Exercise the real release contract without invoking PlatformIO or building."""
import configparser
import os
from pathlib import Path
import re
import subprocess
import unittest
ROOT = Path(os.environ.get('MESHCORE_TEST_ROOT', Path(__file__).resolve().parents[1]))
SCRIPT = Path(os.environ.get('MESHCORE_BUILD_CONTRACT_SCRIPT', ROOT / 'build_legacy.sh'))
PROBE = r'''
set -euo pipefail
pio() { echo 'forbidden PlatformIO invocation' >&2; return 99; }
platformio() { pio "$@"; }
source "$1"
target=$2
NRF52_OTA_SENSOR_PROFILE=$3
BUILD_PROFILE_FOR_TARGET=auto
PIO_ENV_PLATFORM_BY_NAME[$target]=NRF52_PLATFORM
PIO_ENV_OTA_BY_NAME[$target]=1
PIO_ENV_FLAGS_BY_NAME=()
BUILD_CAPABILITIES=()
BUILD_EXPECTATIONS=()
BUILD_APPLICATION_EXPECTATIONS=()
BUILD_REDUCTIONS=()
PLATFORMIO_BUILD_FLAGS=''
PLATFORMIO_BUILD_UNFLAGS=''
declare_build_capability_contract "$target" NRF52_PLATFORM
apply_lora_ota_no_external_sensors_profile "$target"
printf 'EXPECT:%s\n' "${BUILD_EXPECTATIONS[@]}"
printf 'FLAGS:%s\n' "$PLATFORMIO_BUILD_FLAGS"
printf 'REDUCTION:%s\n' "${BUILD_REDUCTIONS[@]}"
'''
def contract(target, profile):
result = subprocess.run(['bash', '-c', PROBE, 'rak-gps-contract', str(SCRIPT), target, profile],
cwd=ROOT, text=True, capture_output=True, timeout=10)
if result.returncode:
raise AssertionError(result.stdout + result.stderr)
fields = {'EXPECT': [], 'FLAGS': [], 'REDUCTION': []}
for line in result.stdout.splitlines():
prefix, sep, value = line.partition(':')
if sep and prefix in fields:
fields[prefix].append(value)
return fields
class RakRs232GpsContractTest(unittest.TestCase):
def test_serial1_both_sensor_profiles_do_not_promise_or_enable_gps(self):
for board in ('3401', '4631'):
target = 'RAK_' + board + '_repeater_bridge_rs232_serial1'
config = configparser.ConfigParser(interpolation=None, inline_comment_prefixes=(';',))
config.read(ROOT / ('variants/rak' + board + '/platformio.ini'))
recipe = config['env:' + target]['build_flags']
self.assertRegex(recipe, r'-D\s*WITH_RS232_BRIDGE=Serial1\b')
self.assertIn('-UENV_INCLUDE_GPS', recipe)
# Target-local UART ownership must override the inherited GPS
# provider before either sensor-profile overlay is applied.
gps_overrides = re.findall(r'-(?:D\s*ENV_INCLUDE_GPS(?:=\d+)?|U\s*ENV_INCLUDE_GPS)\b', recipe)
self.assertTrue(gps_overrides)
self.assertNotRegex(gps_overrides[-1], r'-D\s*ENV_INCLUDE_GPS(?:=1)?$')
for profile in ('full', 'reduced'):
with self.subTest(board=board, sensor_profile=profile):
result = contract(target, profile)
self.assertFalse(any(value.startswith('sensor.gps=') for value in result['EXPECT']))
flags = ' '.join(result['FLAGS'])
self.assertNotRegex(flags, r'-D\s*ENV_INCLUDE_GPS(?:=1)?(?:\s|$)')
for sensor in ('INA219', 'INA226', 'INA260', 'INA3221'):
self.assertTrue(any(value.startswith('sensor.' + sensor.lower() + '=')
for value in result['EXPECT']))
if profile == 'reduced':
self.assertIn('-DENV_INCLUDE_' + sensor + '=1', flags)
self.assertNotIn('-UENV_INCLUDE_' + sensor, flags)
if profile == 'reduced':
self.assertIn('-UENV_INCLUDE_BME280', flags)
self.assertTrue(any('GPS conflicts with RS-232 on Serial1' in value
for value in result['REDUCTION']))
def test_compatible_gps_profiles_still_require_the_linked_provider(self):
for target in ('RAK_3401_repeater', 'RAK_4631_repeater',
'RAK_4631_repeater_bridge_rs232_serial2'):
for profile in ('full', 'reduced'):
with self.subTest(target=target, sensor_profile=profile):
result = contract(target, profile)
self.assertIn('sensor.gps=meshcore.capability.rak_wisblock_gps.v1', result['EXPECT'])
if __name__ == '__main__':
unittest.main()