From 6dfecb8879087a9e2827adb34c6df13a3f918e3d Mon Sep 17 00:00:00 2001 From: mikecarper Date: Fri, 16 Oct 2026 17:33:13 -0700 Subject: [PATCH] Reuse outbound block storage for host-compressed OTA data --- src/helpers/ota/OtaManager.cpp | 14 +- src/helpers/ota/OtaManager.h | 13 +- test/fixtures/ota_cold_buffers/test.cpp | 303 ++++++++++++++++++++++-- test/test_ota_cold_buffers.py | 19 +- 4 files changed, 315 insertions(+), 34 deletions(-) diff --git a/src/helpers/ota/OtaManager.cpp b/src/helpers/ota/OtaManager.cpp index de836c6a2..37270a1de 100644 --- a/src/helpers/ota/OtaManager.cpp +++ b/src/helpers/ota/OtaManager.cpp @@ -742,22 +742,22 @@ bool OtaManager::loadActiveServeBlock() { _serve_wire_deflated = false; // A host-backed MotaSource can provide a pre-compressed block without putting an encoder in firmware. - // Other sources may opt into a caller-supplied encoder. Encoded DATA remains - // pinned until every fragment has left. Streaming proofs need no output buffer. - if (job.wire_v2 && job.allow_deflate && (v->read_deflated || _deflate_encode) && !v->scratch) { + // Only device encoders need separate raw input and compressed output. Host + // output can use the existing outbound cache, which remains pinned until + // every DATA fragment has left. Streaming proofs need no output buffer. + if (job.wire_v2 && job.allow_deflate && !v->read_deflated && _deflate_encode && !v->scratch) { v->scratch = ensureScratch(); v->scratch_sz = v->scratch ? OTA_MAX_BLOCK : 0; } _source_buffer_used_at = _now_ms; // A host-precompressed representation is enough to serve DATA: proofs use // authenticated leaves. Do not pull the same raw block over BLE/USB first. - if (job.wire_v2 && job.allow_deflate && v->read_deflated && - v->scratch && v->scratch_sz >= blen) { + if (job.wire_v2 && job.allow_deflate && v->read_deflated) { uint16_t encoded_len = 0; - const bool encoded = v->read_deflated(v->deflate_ctx, job.block, v->scratch, + const bool encoded = v->read_deflated(v->deflate_ctx, job.block, _serve_block, (uint16_t)blen, &encoded_len); if (encoded && encoded_len > 0 && encoded_len < blen) { - _serve_wire_block = v->scratch; + _serve_wire_block = _serve_block; _serve_wire_len = encoded_len; _serve_wire_deflated = true; } diff --git a/src/helpers/ota/OtaManager.h b/src/helpers/ota/OtaManager.h index dfefbd71f..26e640fd1 100644 --- a/src/helpers/ota/OtaManager.h +++ b/src/helpers/ota/OtaManager.h @@ -744,15 +744,17 @@ private: uint8_t _n_src_obj = 0; uint8_t _src_manifest[OTA_SRC_MANIFEST_MAX]; // manifest-minus-leaves of the loaded source mota // Cold source leaves are sized to the selected image, within the unchanged - // capacity limit. Encoded DATA needs at most one logical block. These buffers - // remain pinned until queued egress drains, then return to the heap at idle. + // capacity limit. Device-encoder output needs at most one logical block; + // host-encoded output uses _serve_block. Cold buffers stay pinned until + // queued egress drains, then return to the heap at idle. uint8_t* _src_leaves = nullptr; uint32_t _src_leaves_capacity = 0; uint8_t* _scratch = nullptr; uint32_t _source_buffer_used_at = 0; - // Server-side response descriptors are tiny. The active logical block has its own buffer so proof generation - // or a simultaneous fetch cannot overwrite DATA retained behind radio-queue backpressure. + // Server-side response descriptors are tiny. Active DATA stays cached behind + // radio-queue backpressure. Proofs do not touch that cache; inbound decoding + // may reuse it but invalidates it before the next outbound fragment. struct ServeJob { OtaReplyRoute route; uint8_t mid[4] = {0}; @@ -790,7 +792,8 @@ private: uint8_t _serve_block[OTA_MAX_BLOCK]; uint16_t _serve_block_len = 0; bool _serve_block_loaded = false; - const uint8_t* _serve_wire_block = nullptr; // raw _serve_block or DEFLATE bytes in ServeView::scratch + // Raw/host-encoded _serve_block, or device-encoder output in ServeView::scratch. + const uint8_t* _serve_wire_block = nullptr; uint16_t _serve_wire_len = 0; bool _serve_wire_deflated = false; uint8_t _serve_wire_id[4] = {0}; // SHA-256:4 of this exact encoded representation diff --git a/test/fixtures/ota_cold_buffers/test.cpp b/test/fixtures/ota_cold_buffers/test.cpp index 328a08d27..c7eace22d 100644 --- a/test/fixtures/ota_cold_buffers/test.cpp +++ b/test/fixtures/ota_cold_buffers/test.cpp @@ -91,23 +91,31 @@ static void proofs() { class Source : public MotaSource { public: + enum class HostEncoding { Off, Good, Refused, PartialFailure, Empty, NonShrinking, Oversized }; std::vector mota; MotaManifest manifest; bool fail_read = false; - Source(uint32_t blocks, uint8_t value) { - const uint32_t total = 8 + MOTA_MFL + blocks * 4 + blocks * OTA_MAX_BLOCK + 5; + bool fail_payload = false; + HostEncoding host_encoding = HostEncoding::Off; + uint16_t host_encoded_size = 8; + unsigned host_calls = 0, raw_reads = 0; + Source(uint32_t blocks, uint8_t value, uint16_t final_block_size = OTA_MAX_BLOCK) { + assert(blocks && final_block_size && final_block_size <= OTA_MAX_BLOCK); + const uint32_t payload_size = (blocks - 1) * OTA_MAX_BLOCK + final_block_size; + const uint32_t total = 8 + MOTA_MFL + blocks * 4 + payload_size + 5; mota.resize(total); memcpy(mota.data(), MOTA_MAGIC, 4); wr_u32le(mota.data() + 4, total); uint8_t* mf = mota.data() + 8; mf[0] = MOTA_FORMAT_VER; mf[1] = MFLAG_FULL; mf[2] = HASH_ALGO_SHA256; wr_u32le(mf + 3, 123); wr_u32le(mf + 7, 1); - wr_u32le(mf + 11, blocks * OTA_MAX_BLOCK); wr_u32le(mf + 15, blocks * OTA_MAX_BLOCK); + wr_u32le(mf + 11, payload_size); wr_u32le(mf + 15, payload_size); mf[19] = 11; mf[56] = CODEC_FULL; uint8_t* leaves = mf + MOTA_MFL; uint8_t* payload = leaves + blocks * 4; for (uint32_t i = 0; i < blocks; ++i) { - memset(payload + i * OTA_MAX_BLOCK, uint8_t(value + i), OTA_MAX_BLOCK); - merkle_leaf(leaves + i * 4, payload + i * OTA_MAX_BLOCK, OTA_MAX_BLOCK); + const uint16_t length = i == blocks - 1 ? final_block_size : OTA_MAX_BLOCK; + memset(payload + i * OTA_MAX_BLOCK, uint8_t(value + i), length); + merkle_leaf(leaves + i * 4, payload + i * OTA_MAX_BLOCK, length); } merkle_root(mf + 20, leaves, blocks); memcpy(mota.data() + total - 5, MOTA_TRAILER, 5); @@ -121,13 +129,46 @@ public: d.total_size = mota.size(); d.leaves_off = 8 + MOTA_MFL; d.block_count = manifest.block_count; d.payload_off = manifest.payload - mota.data(); d.payload_size = manifest.payload_size; + d.source_caps = host_encoding == HostEncoding::Off ? 0 : MOTA_SOURCE_CAP_DEFLATE_BLOCK; return true; } bool read(uint8_t index, uint32_t off, uint8_t* out, uint32_t len) override { if (index || (uint64_t)off + len > mota.size()) return false; + const uint32_t payload_off = manifest.payload - mota.data(); + if (off >= payload_off && off < payload_off + manifest.payload_size) { + ++raw_reads; + if (fail_payload) { if (len) out[0] = 0xEE; return false; } + } if (fail_read) { if (len) out[0] = 0xEE; return false; } memcpy(out, mota.data() + off, len); return true; } + bool read_deflated_block(uint8_t index, uint16_t block, uint8_t* out, + uint16_t cap, uint16_t* len) override { + ++host_calls; + assert(index == 0 && block < manifest.block_count && len); + assert(cap == std::min(OTA_MAX_BLOCK, manifest.payload_size - block * OTA_MAX_BLOCK)); + // Deliberately opaque fixture representations: manager framing, hashes, + // proofs and cache ownership are real; this is not a DEFLATE codec test. + if (host_encoding == HostEncoding::Good) { + assert(host_encoded_size > 0 && host_encoded_size < cap); + memset(out, manifest.payload[block * OTA_MAX_BLOCK], host_encoded_size); + *len = host_encoded_size; + return true; + } + if (host_encoding == HostEncoding::PartialFailure) { + memset(out, 0xEE, 17); *len = 17; return false; + } + if (host_encoding == HostEncoding::Empty) { out[0] = 0xEE; *len = 0; return true; } + if (host_encoding == HostEncoding::NonShrinking) { + memset(out, 0xEE, cap); *len = cap; return true; + } + if (host_encoding == HostEncoding::Oversized) { + // Report a bad length without violating the callback's output capacity. + out[0] = 0xEE; *len = cap + 1; return true; + } + *len = 0; + return false; + } }; struct Sent { bool accept_data = true, accept_proof = true, accept_manifest = true; @@ -142,7 +183,7 @@ static bool send(void* opaque, const uint8_t* data, uint16_t len, bool) { } static bool encode(void*, const uint8_t* src, uint16_t, uint8_t* dst, uint16_t cap, uint16_t* len) { - assert(cap == OTA_MAX_BLOCK); memset(dst, src[0], 8); *len = 8; return true; + assert(cap == OTA_MAX_BLOCK && src != dst); memset(dst, src[0], 8); *len = 8; return true; } static bool encode_long(void*, const uint8_t* src, uint16_t, uint8_t* dst, uint16_t cap, uint16_t* len) { @@ -159,6 +200,12 @@ static bool request(OtaManager& manager, const Source& source, uint16_t block = uint8_t wire[MAX_PACKET_PAYLOAD]; return manager.on_message(wire, encode_req(wire, sizeof(wire), req)); } +static bool request_v2(OtaManager& manager, const Source& source, uint16_t block, bool allow_deflate) { + ReqMsg req{}; memcpy(req.manifest_id, source.manifest.merkle_root, 4); + req.block_idx = block; req.want_mask = ota_req_make_v2(0xFFF, allow_deflate, true); + uint8_t wire[MAX_PACKET_PAYLOAD]; + return manager.on_message(wire, encode_req(wire, sizeof(wire), req)); +} static bool manifest_request(OtaManager& manager, const Source& source) { GetManifestMsg req{}; memcpy(req.manifest_id, source.manifest.merkle_root, 4); req.want_mask = 0xFFFF; uint8_t wire[MAX_PACKET_PAYLOAD]; @@ -193,6 +240,132 @@ static void verify(const Sent& sent, const Source& source, bool compressed) { assert(!memcmp(bytes.data(), source.manifest.payload, OTA_MAX_BLOCK)); } +static void verify_v2(const Sent& sent, const Source& source, uint16_t block, + bool compressed, uint16_t encoded_size = 8) { + std::vector bytes; + uint8_t identity[4]{}; + bool proof_seen = false; + const uint16_t raw_size = std::min(OTA_MAX_BLOCK, source.manifest.payload_size - block * OTA_MAX_BLOCK); + const uint16_t expected_size = compressed ? encoded_size : raw_size; + for (const auto& packet : sent.packets) { + DataMsg data{}; ProofMsg proof{}; + if (decode_data(packet.data(), packet.size(), data) + && !memcmp(data.manifest_id, source.manifest.merkle_root, 4)) { + assert(data.block_idx == block); + uint8_t fragment; uint16_t length; bool deflated; + assert(ota_data_v2_unpack_extended(data.frag_off, raw_size, fragment, length, deflated)); + assert(deflated == compressed && length == expected_size); + assert(fragment * OTA_FRAG_DATA_V2 == bytes.size()); + if (bytes.empty()) memcpy(identity, data.data, 4); + else assert(!memcmp(identity, data.data, 4)); + bytes.insert(bytes.end(), data.data + 4, data.data + data.data_len); + } else if (decode_proof(packet.data(), packet.size(), proof) + && !memcmp(proof.manifest_id, source.manifest.merkle_root, 4)) { + assert(proof.block_idx == block); + assert(merkle_verify(source.manifest.payload + block * OTA_MAX_BLOCK, raw_size, block, + proof.proof, proof.n_proof, source.manifest.merkle_root, + source.manifest.block_count)); + proof_seen = true; + } + } + assert(proof_seen && bytes.size() == expected_size); + if (compressed) assert(bytes == std::vector(expected_size, source.manifest.payload[block * OTA_MAX_BLOCK])); + else assert(!memcmp(bytes.data(), source.manifest.payload + block * OTA_MAX_BLOCK, raw_size)); + uint8_t expected[4]; mh4(expected, bytes.data(), bytes.size()); + assert(!memcmp(identity, expected, 4)); +} + +static void host_allocation_and_fallbacks() { + for (bool fail_raw : {false, true}) { + Source source(9, 0x42); source.host_encoding = Source::HostEncoding::Good; + source.host_encoded_size = 400; source.fail_payload = fail_raw; + Sent sent; OtaManager manager; + manager.begin(123, send, &sent); manager.set_transport_deflate_encoder(encode); + assert(manager.add_source(&source) && request(manager, source)); + assert(allocated == 36); + fail_next = true; + manager.serviceEgress(); + assert(fail_next && allocated == 36 && source.host_calls == 1 && source.raw_reads == 0); + drain(manager); verify_v2(sent, source, 0, true, 400); + assert(fail_next && allocated == 36 && source.host_calls == 1); + fail_next = false; manager.clear_sources(); assert(allocated == 0); + } + for (auto mode : {Source::HostEncoding::Refused, Source::HostEncoding::PartialFailure, + Source::HostEncoding::Empty, Source::HostEncoding::NonShrinking, + Source::HostEncoding::Oversized}) { + Source source(3, 0x51); source.host_encoding = mode; + Sent sent; OtaManager manager; + manager.begin(123, send, &sent); manager.set_transport_deflate_encoder(encode); + assert(manager.add_source(&source) && request(manager, source)); + fail_next = true; drain(manager); + assert(fail_next && allocated == 12 && source.host_calls == 1 && source.raw_reads == 1); + verify_v2(sent, source, 0, false); + fail_next = false; manager.clear_sources(); assert(allocated == 0); + } + Source source(3, 0x61); source.host_encoding = Source::HostEncoding::PartialFailure; + source.fail_payload = true; + Sent sent; OtaManager manager; manager.begin(123, send, &sent); + assert(manager.add_source(&source) && request(manager, source)); + fail_next = true; manager.serviceEgress(); + assert(fail_next && sent.packets.empty() && !manager.pendingServeJobs()); + assert(source.host_calls == 1 && source.raw_reads == 1 && allocated == 12); + fail_next = false; source.fail_payload = false; source.host_encoding = Source::HostEncoding::Good; + assert(request(manager, source)); drain(manager); verify_v2(sent, source, 0, true); + manager.clear_sources(); assert(allocated == 0); + puts("PASS: host compression needs leaves only; allocation denial and invalid-output fallback"); +} + +static void host_tail_and_maximum_capacity() { + for (bool maximum : {false, true}) { + const uint32_t blocks = maximum ? OTA_PROOFGEN_SCRATCH / 4 : 3; + Source source(blocks, 0x65, maximum ? OTA_MAX_BLOCK : 37); + source.host_encoding = Source::HostEncoding::Good; + source.fail_payload = true; + const uint16_t block = blocks - 1; + Sent sent; OtaManager manager; manager.begin(123, send, &sent); + assert(manager.add_source(&source) && request(manager, source, block)); + assert(allocated == blocks * 4); + if (maximum) assert(allocated == OTA_PROOFGEN_SCRATCH); + fail_next = true; drain(manager); verify_v2(sent, source, block, true); + assert(fail_next && allocated == blocks * 4 && source.host_calls == 1 && source.raw_reads == 0); + fail_next = false; manager.clear_sources(); assert(allocated == 0); + } + puts("PASS: host short tail and maximum leaf capacity retain original source bounds"); +} + +static void host_backpressure_and_raw_switch() { + Source source(5, 0x43); source.host_encoding = Source::HostEncoding::Good; + source.host_encoded_size = 400; + Sent sent; OtaManager manager; manager.begin(123, send, &sent); + assert(manager.add_source(&source) && request(manager, source)); + sent.accept_data = false; fail_next = true; + manager.serviceEgress(); manager.set_clock(100000); manager.serviceEgress(); + assert(fail_next && allocated == 20 && source.host_calls == 1 && sent.packets.empty()); + sent.accept_data = true; manager.serviceEgress(); // Admit exactly one of three fragments. + assert(sent.packets.size() == 1 && manager.pendingServeJobs()); + source.host_encoding = Source::HostEncoding::PartialFailure; // A pinned representation must not be re-read. + sent.accept_data = false; manager.set_clock(200000); manager.loop(); + assert(source.host_calls == 1 && allocated == 20); + sent.accept_data = true; sent.accept_proof = false; + for (unsigned turn = 0; turn < 10; ++turn) { + manager.set_clock(300000 + turn * 10000); manager.serviceEgress(); + } + assert(manager.pendingServeJobs() && fail_next && source.host_calls == 1 && allocated == 20); + sent.accept_proof = true; drain(manager, 500000); verify_v2(sent, source, 0, true, 400); + assert(fail_next); fail_next = false; + sent.packets.clear(); + assert(request_v2(manager, source, 1, false)); drain(manager, 600000); + assert(source.host_calls == 1 && source.raw_reads == 1 && allocated == 20); + verify_v2(sent, source, 1, false); + source.host_encoding = Source::HostEncoding::Good; source.host_encoded_size = 8; + sent.packets.clear(); + assert(request(manager, source, 2)); drain(manager, 800000); + assert(source.host_calls == 2 && source.raw_reads == 1 && allocated == 20); + verify_v2(sent, source, 2, true); + manager.clear_sources(); assert(allocated == 0); + puts("PASS: host bytes stay pinned through DATA/PROOF backpressure and raw/compressed switches"); +} + static void lifetimes() { Source source(9, 0x42); Sent sent; OtaManager manager; @@ -261,14 +434,7 @@ static void faults_and_capacity() { fail_next = true; assert(request(manager, small, 0, false)); drain(manager); assert(fail_next && allocated == 0); fail_next = false; // raw proof serving allocates nothing } -static void concurrent_receive() { - Source outbound(5, 0x40), incoming(3, 0x70); - Sent sent; OtaManager manager; OtaStoreRam<16384> store; - manager.begin(123, send, &sent); manager.set_fetch_store(&store); - manager.set_transport_deflate_encoder(encode); manager.set_transport_deflate_decoder(decode); - assert(manager.add_source(&outbound) && request(manager, outbound)); - sent.accept_data = false; manager.serviceEgress(); - assert(allocated == 20 + OTA_MAX_BLOCK); +static void begin_receive(OtaManager& manager, const Source& incoming) { assert(manager.pull(incoming.manifest.merkle_root, 123) == OtaManager::PULL_STARTED); for (unsigned f = 0; f < 2; ++f) { ManifestMsg mf{}; memcpy(mf.manifest_id, incoming.manifest.merkle_root, 4); @@ -277,6 +443,23 @@ static void concurrent_receive() { uint8_t wire[MAX_PACKET_PAYLOAD]; manager.on_message(wire, encode_manifest(wire, sizeof(wire), mf)); } assert(manager.fetchState() == OtaManager::FETCHING); +} +static void concurrent_receive(bool host = false) { + Source outbound(5, 0x40), incoming(3, 0x70); + if (host) { + outbound.host_encoding = Source::HostEncoding::Good; + outbound.host_encoded_size = 400; + } + const size_t expected_allocation = 20 + (host ? 0 : OTA_MAX_BLOCK); + Sent sent; OtaManager manager; OtaStoreRam<16384> store; + manager.begin(123, send, &sent); manager.set_fetch_store(&store); + manager.set_transport_deflate_encoder(encode); manager.set_transport_deflate_decoder(decode); + assert(manager.add_source(&outbound) && request(manager, outbound)); + sent.accept_data = host; manager.serviceEgress(); + if (host) assert(sent.packets.size() == 1 && manager.pendingServeJobs()); + sent.accept_data = false; + assert(allocated == expected_allocation); + begin_receive(manager, incoming); for (uint16_t block = 0; block < 3; ++block) { uint8_t representation[12]; memset(representation + 4, uint8_t(0x70 + block), 8); mh4(representation, representation + 4, 8); @@ -287,30 +470,69 @@ static void concurrent_receive() { proof.block_idx = block; proof.n_proof = merkle_gen_proof(incoming.manifest.leaves, 3, block, nullptr, siblings); proof.proof = siblings; assert(manager.on_message(wire, encode_proof(wire, sizeof(wire), proof))); manager.set_clock(100000 + block * 10000); manager.loop(); - assert(allocated == 20 + OTA_MAX_BLOCK && manager.pendingServeJobs()); + assert(allocated == expected_allocation && manager.pendingServeJobs()); } assert(manager.fetchState() == OtaManager::COMPLETE); MotaManifest staged; assert(mota_parse(store.data(), store.staged_size(), staged)); assert(!memcmp(staged.payload, incoming.manifest.payload, incoming.manifest.payload_size)); - sent.accept_data = true; drain(manager, 200000); verify(sent, outbound, true); + sent.accept_data = true; drain(manager, 200000); + if (host) verify_v2(sent, outbound, 0, true, 400); + else verify(sent, outbound, true); + if (host) assert(outbound.host_calls > 1 && outbound.raw_reads == 0); manager.clear_sources(); assert(allocated == 0); + if (host) puts("PASS: host DATA reloads after successful interleaved inbound decoding"); } -static void interleaved_manifest() { +static bool partial_decode_failure(void*, const uint8_t*, uint16_t, uint8_t* out, + uint16_t cap, uint16_t* produced) { + assert(cap == OTA_MAX_BLOCK); memset(out, 0xEE, cap / 2); *produced = cap / 2; + return false; +} +static void host_receive_failure() { + Source outbound(5, 0x44), incoming(3, 0x71); + outbound.host_encoding = Source::HostEncoding::Good; outbound.host_encoded_size = 400; + Sent sent; OtaManager manager; OtaStoreRam<16384> store; + manager.begin(123, send, &sent); manager.set_fetch_store(&store); + manager.set_transport_deflate_decoder(partial_decode_failure); + assert(manager.add_source(&outbound) && request(manager, outbound)); + manager.serviceEgress(); // Already-admitted bytes must keep the same representation after reload. + assert(sent.packets.size() == 1 && manager.pendingServeJobs()); + sent.accept_data = false; + assert(outbound.host_calls == 1 && allocated == 20); + begin_receive(manager, incoming); + uint8_t representation[12]; memset(representation + 4, 0x71, 8); + mh4(representation, representation + 4, 8); + DataMsg data{}; memcpy(data.manifest_id, incoming.manifest.merkle_root, 4); + assert(ota_data_v2_pack_extended(0, 8, data.frag_off)); + data.data = representation; data.data_len = sizeof(representation); + uint8_t wire[MAX_PACKET_PAYLOAD]; + assert(manager.on_message(wire, encode_data(wire, sizeof(wire), data))); + assert(manager.fetchState() == OtaManager::FETCHING); + sent.accept_data = true; drain(manager, 100000); verify_v2(sent, outbound, 0, true, 400); + assert(outbound.host_calls == 2 && outbound.raw_reads == 0 && allocated == 20); + manager.clear_sources(); assert(allocated == 0); + puts("PASS: partial failed inbound decoder cannot leak into pinned host DATA"); +} +static void interleaved_manifest(bool host = false) { Source data_source(9, 0x35), manifest_source(17, 0x75); + if (host) { + data_source.host_encoding = Source::HostEncoding::Good; + data_source.host_encoded_size = 400; + } + const size_t output_allocation = host ? 0 : OTA_MAX_BLOCK; Sent sent; OtaManager manager; manager.begin(123, send, &sent); manager.set_transport_deflate_encoder(encode_long); assert(manager.add_source(&data_source) && manager.add_source(&manifest_source)); assert(request(manager, data_source)); manager.serviceEgress(); // first compressed fragment is admitted; later DATA stays pinned - assert(manager.pendingServeJobs() && allocated == 36 + OTA_MAX_BLOCK); + assert(manager.pendingServeJobs() && allocated == 36 + output_allocation); peak = allocated; assert(manifest_request(manager, manifest_source)); - assert(allocated == 68 + OTA_MAX_BLOCK && peak == allocated); // old leaves were freed before growth + assert(allocated == 68 + output_allocation && peak == allocated); // old leaves were freed before growth sent.accept_data = false; manager.set_clock(100000); manager.serviceEgress(); manager.set_clock(200000); manager.serviceEgress(); assert(!manager.pendingManifestJobs() && manager.pendingServeJobs()); - manager.loop(); assert(allocated == 68 + OTA_MAX_BLOCK); + manager.loop(); assert(allocated == 68 + output_allocation); sent.accept_data = true; drain(manager, 300000); std::vector encoded; uint8_t identity[4]{}; bool proof_seen = false; @@ -336,7 +558,7 @@ static void interleaved_manifest() { manager.clearPendingEgress(); assert(allocated == 0 && manager.servedCount() == 2); sent.packets.clear(); assert(request(manager, data_source)); manager.serviceEgress(); fail_next = true; assert(!manifest_request(manager, manifest_source)); - assert(allocated == OTA_MAX_BLOCK); // failed growth released old leaves, but pinned DATA survives + assert(allocated == output_allocation); // failed growth released old leaves, but pinned DATA survives drain(manager, 900000); encoded.clear(); for (const auto& packet : sent.packets) { @@ -348,6 +570,38 @@ static void interleaved_manifest() { manager.clearPendingEgress(); assert(allocated == 0); manager.set_clock(0xFFFFFF00u); assert(request(manager, data_source)); drain(manager, 0xFFFFFF00u); manager.set_clock(0xFFFFFF00u + 100000u); manager.loop(); assert(allocated == 0); + if (host) puts("PASS: host DATA survives interleaved source growth, failure and rollover"); +} + +struct SelfEncoder { + uint8_t* expected_output; + unsigned calls = 0; +}; +static bool self_encode(void* context, const uint8_t* input, uint16_t length, + uint8_t* output, uint16_t capacity, uint16_t* produced) { + auto& encoder = *static_cast(context); + assert(output == encoder.expected_output && input != output); + ++encoder.calls; + return encode(nullptr, input, length, output, capacity, produced); +} +static bool self_read(void* context, uint32_t offset, uint8_t* output, uint32_t length) { + auto& source = *static_cast(context); + return source.read(0, uint32_t(source.manifest.payload - source.mota.data()) + offset, output, length); +} +static void caller_owned_scratch() { + Source source(3, 0x63); + std::array scratch{}; + SelfEncoder encoder{scratch.data()}; + Sent sent; OtaManager manager; manager.begin(123, send, &sent); + manager.set_transport_deflate_encoder(self_encode, &encoder); + assert(manager.serve_self(source.manifest.manifest_start, MOTA_MFL, source.manifest.leaves, + source.manifest.block_count, scratch.data(), scratch.size(), self_read, &source)); + fail_next = true; assert(request(manager, source)); drain(manager); + assert(fail_next && allocated == 0 && encoder.calls == 1); + verify_v2(sent, source, 0, true); + fail_next = false; manager.clearPendingEgress(); + assert(scratch[0] == 0x63 && allocated == 0); + puts("PASS: caller-owned self-source encoder output remains separate and unallocated"); } int main() { proofs(); assert(allocated == 0); @@ -355,5 +609,12 @@ int main() { faults_and_capacity(); assert(allocated == 0); concurrent_receive(); assert(allocated == 0); interleaved_manifest(); assert(allocated == 0); + host_allocation_and_fallbacks(); assert(allocated == 0); + host_tail_and_maximum_capacity(); assert(allocated == 0); + host_backpressure_and_raw_switch(); assert(allocated == 0); + concurrent_receive(true); assert(allocated == 0); + host_receive_failure(); assert(allocated == 0); + interleaved_manifest(true); assert(allocated == 0); + caller_owned_scratch(); assert(allocated == 0); puts("PASS: streaming proofs and OTA cold-buffer lifetimes"); } diff --git a/test/test_ota_cold_buffers.py b/test/test_ota_cold_buffers.py index c4bc777eb..02c76d096 100644 --- a/test/test_ota_cold_buffers.py +++ b/test/test_ota_cold_buffers.py @@ -1,5 +1,10 @@ #!/usr/bin/env python3 -"""Verify streaming proofs and real OTA cold-buffer lifetimes without PlatformIO.""" +"""Verify real OTA cold allocations, host DATA cache ownership and streaming proofs. + +The fixture uses controlled host representations, encoder/decoder callbacks and +radio admission. It validates manager bytes/hashes/proofs under sanitizers, not +the DEFLATE codec or physical USB/BLE/radio transport. +""" from pathlib import Path import os @@ -12,6 +17,15 @@ from test_serial_wifi_sessions import SANITIZER_FLAGS ROOT = Path(__file__).resolve().parents[1] FIXTURE = ROOT / "test/fixtures/ota_cold_buffers/test.cpp" +HOST_MARKERS = ( + "host compression needs leaves only; allocation denial and invalid-output fallback", + "host short tail and maximum leaf capacity retain original source bounds", + "host bytes stay pinned through DATA/PROOF backpressure and raw/compressed switches", + "host DATA reloads after successful interleaved inbound decoding", + "partial failed inbound decoder cannot leak into pinned host DATA", + "host DATA survives interleaved source growth, failure and rollover", + "caller-owned self-source encoder output remains separate and unallocated", +) class OtaColdBufferTests(unittest.TestCase): @@ -36,7 +50,10 @@ class OtaColdBufferTests(unittest.TestCase): self.assertEqual(built.returncode, 0, built.stdout + built.stderr) checked = subprocess.run([str(binary)], capture_output=True, text=True, timeout=60) self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr) + for marker in HOST_MARKERS: + self.assertEqual(checked.stdout.count("PASS: " + marker), 1, checked.stdout) self.assertIn("PASS: streaming proofs and OTA cold-buffer lifetimes", checked.stdout) + print(f"\n[OTA cold buffers: {platform or 'portable'}]\n{checked.stdout}", flush=True) if __name__ == "__main__":