Preserve MQTT TLS connection time and internal DMA headroom

test/test_firmware_ram.py

@@ -42,14 +42,15 @@ def elf32(path, symbols, data=b"", address=0x3F400000):
     return FirmwareElf(path)
 
 
-def esp_fixture(path, modern=False, fragmented=False):
+def esp_fixture(path, modern=False, fragmented=False, internal_length=0x10000):
     address = 0x3F400000
     caps = [0x804, 0x804, 0x404, 0x803, 0x8804]
     data = bytearray(struct.pack("<I", len(caps)))
     symbols = {"soc_memory_region_count": (address, 4)}
     symbols["soc_memory_regions"] = (address + len(data), len(caps) * (20 if modern else 16))
     for index in range(len(caps)):
-        data += struct.pack("<4I", 0x3FFB0000 + index * 0x10000, 0x10000, index, 0)
+        length = internal_length if index == 0 else 0x10000
+        data += struct.pack("<4I", 0x3FFB0000 + index * 0x10000, length, index, 0)
         if modern:
             data += bytes([int(index == 1), 0, 0, 0])
     symbols["soc_memory_types"] = (address + len(data), len(caps) * (16 if modern else 20))
@@ -203,6 +204,83 @@ class FirmwareRamTest(unittest.TestCase):
         }, "v3_repeater")
         self.assertEqual(alternate, combined)
 
+    def test_mqtt_startup_gate_distinguishes_maximum_slot_stack_buffer_capacity(self):
+        flags = {"WITH_MQTT_BRIDGE": 1}
+        plain = ram.requirements("ESP32_PLATFORM", flags, "v4_companion")
+        psram = ram.requirements("ESP32_PLATFORM", {**flags, "BOARD_HAS_PSRAM": 1},
+                                 "v4_companion")
+        # The potential five-slot allowance is reported separately: additional
+        # clients allocate lazily and must pass the production runtime gate.
+        self.assertEqual(plain["required_heap_bytes"], psram["required_heap_bytes"])
+        self.assertEqual(plain["components"]["mqtt_startup_allowance"], 24576)
+        self.assertEqual(psram["components"]["mqtt_startup_allowance"], 24576)
+        for policy, slots, active, potential in ((plain, 3, 2, 24576), (psram, 6, 5, 48384)):
+            with self.subTest(active=active):
+                profile = policy["mqtt"]
+                self.assertEqual(profile["runtime_slot_count"], slots)
+                self.assertEqual(profile["maximum_active_slots"], active)
+                self.assertEqual(profile["startup_active_slot_allowance"], 2)
+                self.assertEqual(profile["maximum_active_slots_without_psram"], 2)
+                self.assertEqual(profile["maximum_slot_stack_buffer_allowance_bytes"], potential)
+                self.assertFalse(profile["maximum_runtime_load_qualified"])
+                self.assertFalse(profile["physical_validation_performed"])
+        ordinary = ram.requirements("ESP32_PLATFORM", {"BOARD_HAS_PSRAM": 1}, "v4_companion")
+        self.assertNotIn("mqtt", ordinary)
+
+    def test_mqtt_worker_stack_growth_raises_gate_and_contiguous_requirement(self):
+        flags = {"WITH_MQTT_BRIDGE": 1, "BOARD_HAS_PSRAM": 1}
+        base = ram.requirements("ESP32_PLATFORM", flags, "v4_companion")
+        large = ram.requirements("ESP32_PLATFORM", {
+            **flags, "MQTT_TASK_STACK_SIZE": 32768,
+        }, "v4_companion")
+        self.assertEqual(large["required_heap_bytes"] - base["required_heap_bytes"], 24576)
+        self.assertEqual(large["required_contiguous_bytes"], 32768)
+        self.assertEqual(large["mqtt"]["maximum_slot_stack_buffer_allowance_bytes"], 72960)
+        small = ram.requirements("ESP32_PLATFORM", {
+            **flags, "MQTT_TASK_STACK_SIZE": 4096, "MQTT_RUNTIME_SLOT_COUNT": 1,
+        }, "v4_companion")
+        self.assertEqual(small["required_heap_bytes"], base["required_heap_bytes"])
+        self.assertEqual(small["mqtt"]["maximum_active_slots"], 1)
+        for slots in (0, 7):
+            with self.subTest(slots=slots), self.assertRaisesRegex(ValueError, "between 1 and 6"):
+                ram.requirements("ESP32_PLATFORM", {
+                    **flags, "MQTT_RUNTIME_SLOT_COUNT": slots,
+                }, "v4_companion")
+        with self.assertRaisesRegex(ValueError, "positive"):
+            ram.requirements("ESP32_PLATFORM", {**flags, "MQTT_TASK_STACK_SIZE": 0}, "v4_companion")
+
+    def test_mqtt_report_at_startup_boundary_does_not_qualify_maximum_load(self):
+        flags = {"WITH_MQTT_BRIDGE": 1, "BOARD_HAS_PSRAM": 1}
+        policy = ram.requirements("ESP32_PLATFORM", flags, "v4_companion")
+        with tempfile.TemporaryDirectory() as temp:
+            path = Path(temp) / "firmware.elf"
+            report = Path(temp) / "firmware.memory.json"
+            for delta, expected in ((-1, False), (0, True)):
+                with self.subTest(delta=delta), contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
+                    capacity = policy["required_heap_bytes"] + delta
+                    esp_fixture(path, modern=True, internal_length=capacity + 0x4000)
+                    status = ram.check_firmware(path, "ESP32_PLATFORM", "esp32s3", flags,
+                                                "v4_companion", report)
+                    result = json.loads(report.read_text())
+                    self.assertEqual(status == 0, expected)
+                    self.assertEqual(result["passed"], expected)
+                    self.assertEqual(result["available_internal_bytes"], capacity)
+                    self.assertEqual(result["mqtt"]["maximum_slot_stack_buffer_allowance_bytes"], 48384)
+                    self.assertFalse(result["mqtt"]["maximum_runtime_load_qualified"])
+                    self.assertFalse(result["mqtt"]["physical_validation_performed"])
+
+    def test_mqtt_report_admission_thresholds_match_production_guard(self):
+        source = (ROOT / "src/helpers/MQTTConnectionAdmission.h").read_text()
+        profile = ram.requirements("ESP32_PLATFORM", {"WITH_MQTT_BRIDGE": 1}, "v4_companion")["mqtt"]
+        for key, name in (
+            ("dma_admission_free_bytes", "kDmaReserveBytes"),
+            ("dma_admission_contiguous_bytes", "kDmaLargestBytes"),
+            ("tls_internal_admission_free_bytes", "kTlsInternalFreeBytes"),
+            ("tls_internal_admission_contiguous_bytes", "kTlsRecordAllocationBytes"),
+        ):
+            with self.subTest(key=key):
+                self.assertIn(f"{name} = {profile[key]};", source)
+
     def test_combined_rak_budgets_uart_beside_exclusive_ethernet_and_ota(self):
         defines = {
             "RAK_4631": 1, "ETHERNET_ENABLED": 1, "OTA_RAK_AUTO_STORE": 1,
@@ -331,7 +409,7 @@ class FirmwareRamTest(unittest.TestCase):
         self.assertEqual(parts["display_pixels_and_driver"], 2048)
         self.assertEqual(parts["uart_bridge_and_driver"], 2560 + 4096)
         self.assertEqual(parts["wireless_stacks"], 49152)
-        self.assertEqual(parts["mqtt_connections_buffers"], 24576)
+        self.assertEqual(parts["mqtt_startup_allowance"], 24576)
         self.assertEqual(parts["allocation_and_transient_margin"], 16384)
         self.assertEqual(full["required_contiguous_bytes"], 18464)
         for driver, expected in (("SH1106Display", 4096), ("SH1107Display", 4096),

test/test_mqtt_connection_admission.py

@@ -0,0 +1,296 @@
+#!/usr/bin/env python3
+"""Test MQTT memory admission and its production allocator/connection wiring."""
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+from test_mqtt_transport_results import method
+
+ROOT = Path(__file__).resolve().parents[1]
+
+POLICY_PREAMBLE = r'''
+#include <cassert>
+#include <cstddef>
+#include <cstdint>
+#include "helpers/MQTTConnectionAdmission.h"
+namespace Admission = MQTTConnectionAdmission;
+'''
+
+HEAP_PREAMBLE = POLICY_PREAMBLE + r'''
+#define ESP_PLATFORM 1
+#define BOARD_HAS_PSRAM 1
+constexpr uint32_t MALLOC_CAP_INTERNAL = 1, MALLOC_CAP_DMA = 2, MALLOC_CAP_SPIRAM = 4;
+static size_t dma_free = 100000, dma_largest = 20000;
+static size_t heap_queries = 0, psram_allocations = 0, internal_allocations = 0;
+static bool psram_fails = false, internal_fails = false;
+static unsigned char psram_storage[4096], internal_storage[4096];
+size_t heap_caps_get_free_size(uint32_t caps) {
+  assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA));
+  ++heap_queries;
+  return dma_free;
+}
+size_t heap_caps_get_largest_free_block(uint32_t caps) {
+  assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA));
+  ++heap_queries;
+  return dma_largest;
+}
+void* heap_caps_calloc(size_t count, size_t size, uint32_t caps) {
+  assert(count && size && count <= SIZE_MAX / size);
+  if (caps == MALLOC_CAP_SPIRAM) {
+    ++psram_allocations;
+    return psram_fails ? nullptr : psram_storage;
+  }
+  assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA));
+  ++internal_allocations;
+  return internal_fails ? nullptr : internal_storage;
+}
+'''
+
+CONNECTION_PREAMBLE = HEAP_PREAMBLE + r'''
+#include <atomic>
+constexpr int RUNTIME_MQTT_SLOTS = 6;
+struct { size_t size = 0; size_t getPsramSize() const { return size; } } ESP;
+struct Preset { const char* server_url; };
+struct MQTTBridge {
+  struct MQTTSlot {
+    const Preset* preset = nullptr;
+    char broker_uri[128] = {}, host[64] = "broker";
+    uint16_t port = 8883;
+  } _slots[RUNTIME_MQTT_SLOTS];
+  std::atomic<bool> _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {};
+  bool hasPendingSlotConnection() const;
+  bool canStartSlotConnection(int index) const;
+};
+'''
+
+
+class MqttConnectionAdmissionTest(unittest.TestCase):
+    def compile_run(self, source):
+        compiler = shutil.which("g++") or shutil.which("clang++")
+        self.assertIsNotNone(compiler, "a host C++ compiler is required")
+        with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-admission-") as temporary:
+            work = Path(temporary)
+            path, binary = work / "admission.cpp", work / "admission"
+            path.write_text(source, encoding="utf-8")
+            result = subprocess.run(
+                [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                 "-I", str(ROOT / "src"), str(path), "-o", str(binary)],
+                capture_output=True, text=True, timeout=60)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            result = subprocess.run([str(binary)], cwd=work, capture_output=True,
+                                    text=True, timeout=10)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_connection_admission_checks_both_total_and_contiguous_dma_heap(self):
+        self.compile_run(POLICY_PREAMBLE + r'''
+int main() {
+  // A large aggregate heap cannot compensate for a fragmented DMA region.
+  assert(!Admission::canStart(true, true, 100000, 4095));
+  assert(!Admission::canStart(true, true, 16383, 20000));
+  assert(Admission::canStart(true, true, 16384, 4096));
+  assert(Admission::canStart(false, false, 16384, 4096));
+  assert(!Admission::canStart(false, false, 16383, 4096));
+  assert(!Admission::canStart(false, false, 16384, 4095));
+
+  // Without PSRAM, TLS also needs a full record allocation and session budget.
+  assert(!Admission::canStart(false, true, 61439, 20000));
+  assert(!Admission::canStart(false, true, 100000, 17407));
+  assert(Admission::canStart(false, true, 61440, 17408));
+  assert(Admission::canStart(true, false, 16384, 4096));
+}
+''')
+
+    def test_internal_fallback_preserves_dma_reserve_and_rejects_fragmentation(self):
+        self.compile_run(POLICY_PREAMBLE + r'''
+int main() {
+  assert(Admission::canFallback(1600, 16384 + 1600, 1600));
+  assert(!Admission::canFallback(1600, 16384 + 1599, 1600));
+  assert(!Admission::canFallback(1600, 100000, 1599));
+  assert(!Admission::canFallback(1600, 16383, 100000));
+  assert(!Admission::canFallback(1, 0, 100000));
+  assert(!Admission::canFallback(0, 100000, 100000));
+  assert(!Admission::canFallback(SIZE_MAX, SIZE_MAX, SIZE_MAX));
+  assert(Admission::canFallback(SIZE_MAX - 16384, SIZE_MAX, SIZE_MAX));
+}
+''')
+
+    def test_calloc_size_rejects_zero_and_overflow_without_allocator_calls(self):
+        self.compile_run(POLICY_PREAMBLE + r'''
+int main() {
+  size_t total = 123;
+  assert(!Admission::checkedCallocSize(0, 16, total) && total == 0);
+  total = 123;
+  assert(!Admission::checkedCallocSize(16, 0, total) && total == 0);
+  total = 123;
+  assert(!Admission::checkedCallocSize(SIZE_MAX, 2, total) && total == 0);
+  assert(Admission::checkedCallocSize(SIZE_MAX, 1, total) && total == SIZE_MAX);
+  assert(Admission::checkedCallocSize(1, SIZE_MAX, total) && total == SIZE_MAX);
+  assert(Admission::checkedCallocSize(2, SIZE_MAX / 2, total));
+  assert(total == SIZE_MAX - 1);
+  assert(Admission::checkedCallocSize(4, 896, total) && total == 3584);
+}
+''')
+
+    def test_explicit_uri_scheme_takes_precedence_over_port(self):
+        self.compile_run(POLICY_PREAMBLE + r'''
+int main() {
+  assert(Admission::requiresTls("wss://broker:1883/mqtt", 1883));
+  assert(Admission::requiresTls("mqtts://broker:1883", 1883));
+  assert(!Admission::requiresTls("ws://broker:443/mqtt", 443));
+  assert(!Admission::requiresTls("mqtt://broker:8883", 8883));
+  assert(Admission::requiresTls("broker", 443));
+  assert(Admission::requiresTls("broker", 8883));
+  assert(!Admission::requiresTls("broker", 1883));
+  assert(!Admission::requiresTls("broker", 9443));
+  assert(Admission::requiresTls(nullptr, 443));
+  assert(!Admission::requiresTls(nullptr, 0));
+  assert(Admission::requiresTls("", 8883));
+}
+''')
+
+    def test_actual_calloc_fallback_preserves_dma_headroom_and_handles_failure(self):
+        source = HEAP_PREAMBLE + method("src/helpers/bridges/MQTTBridge.cpp",
+                                       "static void* psram_calloc(size_t n, size_t size)")
+        self.compile_run(source + r'''
+int main() {
+  assert(psram_calloc(0, 16) == nullptr);
+  assert(psram_calloc(16, 0) == nullptr);
+  assert(psram_calloc(SIZE_MAX, 2) == nullptr);
+  assert(psram_allocations == 0 && internal_allocations == 0 && heap_queries == 0);
+
+  // A successful external allocation needs no internal fallback or reserve.
+  dma_free = dma_largest = 0;
+  assert(psram_calloc(4, 400) == psram_storage);
+  assert(psram_allocations == 1 && internal_allocations == 0 && heap_queries == 0);
+  psram_fails = true;
+  dma_free = 16384 + 1599; dma_largest = 20000;
+  assert(psram_calloc(4, 400) == nullptr && internal_allocations == 0);
+  dma_free = 100000; dma_largest = 1599;
+  assert(psram_calloc(4, 400) == nullptr && internal_allocations == 0);
+  dma_free = 16383; dma_largest = 20000;
+  assert(psram_calloc(1, 1) == nullptr && internal_allocations == 0);
+
+  dma_free = 16384 + 1600; dma_largest = 1600;
+  assert(psram_calloc(4, 400) == internal_storage && internal_allocations == 1);
+  internal_fails = true;
+  assert(psram_calloc(4, 400) == nullptr && internal_allocations == 2);
+}
+''')
+
+    def test_actual_byte_allocators_use_byte_capable_fallback_and_preserve_failed_resize(self):
+        source = POLICY_PREAMBLE + r'''
+#include <cstring>
+#define ESP_PLATFORM 1
+#define BOARD_HAS_PSRAM 1
+constexpr uint32_t MALLOC_CAP_INTERNAL = 1, MALLOC_CAP_SPIRAM = 4, MALLOC_CAP_8BIT = 8;
+static bool psram_fails = false, internal_fails = false;
+static size_t malloc_calls = 0, realloc_calls = 0, free_calls = 0;
+static unsigned char original[32], psram_storage[64], internal_storage[64];
+static void* freed = nullptr;
+void* allocation(uint32_t caps) {
+  if (caps == MALLOC_CAP_SPIRAM) return psram_fails ? nullptr : psram_storage;
+  // INTERNAL alone can select 32-bit-only IRAM on ESP32-S3.
+  assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_8BIT));
+  return internal_fails ? nullptr : internal_storage;
+}
+void* heap_caps_malloc(size_t size, uint32_t caps) {
+  assert(size == 32);
+  ++malloc_calls;
+  return allocation(caps);
+}
+void* heap_caps_realloc(void* ptr, size_t size, uint32_t caps) {
+  assert(ptr == original && size == 48);
+  ++realloc_calls;
+  return allocation(caps);
+}
+void heap_caps_free(void* ptr) { ++free_calls; freed = ptr; }
+'''
+        for signature in ("static void* psram_malloc(size_t size)",
+                          "static void psram_free(void* ptr)",
+                          "static void* psram_realloc(void* ptr, size_t new_size)"):
+            source += method("src/helpers/bridges/MQTTBridge.cpp", signature)
+        self.compile_run(source + r'''
+int main() {
+  assert(psram_malloc(0) == nullptr && malloc_calls == 0);
+  assert(psram_malloc(32) == psram_storage && malloc_calls == 1);
+  psram_fails = true;
+  assert(psram_malloc(32) == internal_storage && malloc_calls == 3);
+  internal_fails = true;
+  assert(psram_malloc(32) == nullptr && malloc_calls == 5 && free_calls == 0);
+
+  memset(original, 0xa5, sizeof(original));
+  psram_fails = internal_fails = false;
+  assert(psram_realloc(original, 48) == psram_storage && realloc_calls == 1);
+  psram_fails = true;
+  assert(psram_realloc(original, 48) == internal_storage && realloc_calls == 3);
+  internal_fails = true;
+  assert(psram_realloc(original, 48) == nullptr && realloc_calls == 5);
+  assert(free_calls == 0);
+  for (unsigned char byte : original) assert(byte == 0xa5);
+
+  assert(psram_realloc(original, 0) == nullptr);
+  assert(free_calls == 1 && freed == original && realloc_calls == 5);
+  assert(psram_realloc(nullptr, 0) == nullptr && free_calls == 1);
+}
+''')
+
+    def test_actual_preflight_uses_detected_psram_transport_and_dma_caps(self):
+        source = CONNECTION_PREAMBLE + method("src/helpers/bridges/MQTTBridge.cpp",
+            "bool MQTTBridge::hasPendingSlotConnection() const")
+        source += method("src/helpers/bridges/MQTTBridge.cpp",
+                         "bool MQTTBridge::canStartSlotConnection(int index) const")
+        self.compile_run(source + r'''
+int main() {
+  MQTTBridge bridge;
+  assert(!bridge.canStartSlotConnection(-1));
+  assert(!bridge.canStartSlotConnection(RUNTIME_MQTT_SLOTS));
+  assert(heap_queries == 0);
+
+  // BOARD_HAS_PSRAM does not establish that physical PSRAM initialized.
+  ESP.size = 0; dma_free = 61440; dma_largest = 17408;
+  assert(bridge.canStartSlotConnection(0));
+  dma_free = 61439;
+  assert(!bridge.canStartSlotConnection(0));
+  dma_free = 100000; dma_largest = 17407;
+  assert(!bridge.canStartSlotConnection(0));
+  ESP.size = 2097152; dma_free = 16384; dma_largest = 4096;
+  assert(bridge.canStartSlotConnection(0));
+  dma_largest = 4095;
+  assert(!bridge.canStartSlotConnection(0));
+
+  // An explicit plaintext URI overrides a TLS port. The effective URI wins
+  // over the raw host; a preset's URI wins over both custom fields.
+  ESP.size = 0; dma_free = 16384; dma_largest = 4096;
+  strcpy(bridge._slots[0].host, "mqtt://broker:8883");
+  assert(bridge.canStartSlotConnection(0));
+  strcpy(bridge._slots[0].broker_uri, "wss://broker/mqtt");
+  assert(!bridge.canStartSlotConnection(0));
+  const Preset plaintext{"ws://broker:443/mqtt"};
+  bridge._slots[0].preset = &plaintext;
+  assert(bridge.canStartSlotConnection(0));
+  const Preset tls{"mqtts://broker:1883"};
+  bridge._slots[0].preset = &tls;
+  assert(!bridge.canStartSlotConnection(0));
+  ESP.size = 2097152;
+  assert(bridge.canStartSlotConnection(0));
+
+  // Every pending TLS attempt, including the target's own attempt, blocks all
+  // other starts until an event callback clears it. No heap walk is needed.
+  for (int pending = 0; pending < RUNTIME_MQTT_SLOTS; ++pending) {
+    bridge._slot_attempt_pending[pending].store(true, std::memory_order_release);
+    const size_t before = heap_queries;
+    for (int index = 0; index < RUNTIME_MQTT_SLOTS; ++index) {
+      assert(!bridge.canStartSlotConnection(index));
+    }
+    assert(heap_queries == before);
+    bridge._slot_attempt_pending[pending].store(false, std::memory_order_release);
+    assert(bridge.canStartSlotConnection(0));
+  }
+}
+''')
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_mqtt_diagnostics.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""Exercise production MQTT diagnostics with distinct internal and DMA heaps."""
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def method(signature):
+    source = (ROOT / "src/helpers/bridges/MQTTBridge.cpp").read_text(encoding="utf-8")
+    start = source.index(signature)
+    opening = source.index("{", start)
+    depth = 1
+    pos = opening + 1
+    while depth:
+        depth += (source[pos] == "{") - (source[pos] == "}")
+        pos += 1
+    return source[start:pos] + "\n"
+
+
+PREAMBLE = r'''
+#include <cassert>
+#include <cstdint>
+#include <cstdarg>
+#include <cstdio>
+#include <cstring>
+#include "helpers/MQTTReplyFormat.h"
+#include "helpers/MQTTConnectionHealth.h"
+#include "helpers/bridges/MQTTErrorLabels.h"
+constexpr int RUNTIME_MQTT_SLOTS = 5;
+constexpr unsigned MALLOC_CAP_DMA = 1U << 3;
+constexpr unsigned MALLOC_CAP_INTERNAL = 1U << 11;
+static unsigned dma_queries = 0;
+static size_t dma_free = 8128, dma_largest = 1536;
+size_t heap_caps_get_free_size(unsigned caps) {
+  assert(caps == MALLOC_CAP_DMA);
+  ++dma_queries;
+  return dma_free;
+}
+size_t heap_caps_get_largest_free_block(unsigned caps) {
+  assert(caps == MALLOC_CAP_DMA);
+  ++dma_queries;
+  return dma_largest;
+}
+struct {
+  int getFreeHeap() const { return 64000; }
+  int getMaxAllocHeap() const { return 30000; }
+} ESP;
+unsigned uxQueueMessagesWaiting(void*) { return 3; }
+static char debug_line[512];
+void capture_debug(const char* format, ...) {
+  va_list args;
+  va_start(args, format);
+  vsnprintf(debug_line, sizeof(debug_line), format, args);
+  va_end(args);
+}
+#define MQTT_DEBUG_PRINTLN(...) capture_debug(__VA_ARGS__)
+struct MQTTPrefs {};
+struct Client {
+  unsigned long getOutboxSize() const { return 0; }
+  unsigned long getOutboxDrops() const { return 0; }
+  unsigned long getPublishOk() const { return 123; }
+  unsigned long getPublishErr() const { return 4; }
+};
+struct MQTTBridge {
+  struct Slot {
+    Client* client = nullptr;
+    bool enabled = true, connected = true;
+  };
+  Slot _slots[RUNTIME_MQTT_SLOTS];
+  bool _initialized = true;
+  void* _packet_queue_handle = reinterpret_cast<void*>(1);
+  int _queue_count = 3, _skipped_publishes = 0;
+  unsigned long _filtered_packets = 0;
+  static constexpr int MAX_QUEUE_SIZE = 50;
+  bool isSlotReady(int) const { return true; }
+  static void formatMqttStatsReply(char*, size_t);
+  static const char* tlsErrorStr(int32_t);
+  void logMemoryStatus();
+};
+static MQTTBridge* s_mqtt_bridge_instance = nullptr;
+'''
+
+MAIN = r'''
+int main() {
+  // Use the public production method, not the pure table alone. Unknown SDK
+  // codes retain raw-number formatting, and write failures stay distinct.
+  assert(strcmp(MQTTBridge::tlsErrorStr(0x8006), "connection timeout") == 0);
+  assert(strcmp(MQTTBridge::tlsErrorStr(0x8008), "server closed connection") == 0);
+  assert(strcmp(MQTTBridge::tlsErrorStr(0x8018), "TLS write failed") == 0);
+  assert(MQTTBridge::tlsErrorStr(0x800B) == nullptr);
+
+  char reply[160];
+  MQTTBridge::formatMqttStatsReply(reply, sizeof(reply));
+  assert(strcmp(reply, "> (bridge not running)") == 0);
+  assert(dma_queries == 0);
+
+  MQTTBridge bridge;
+  Client clients[RUNTIME_MQTT_SLOTS];
+  for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) bridge._slots[i].client = &clients[i];
+  s_mqtt_bridge_instance = &bridge;
+  MQTTBridge::formatMqttStatsReply(reply, sizeof(reply));
+  assert(strstr(reply, "Free=64000 Max=30000") != nullptr);
+#ifdef ESP_PLATFORM
+  // A large general internal block must not mask the smaller AES DMA heap.
+  assert(strstr(reply, "DMA=8128/1536") != nullptr);
+  assert(dma_queries == 2);
+#else
+  assert(strstr(reply, "DMA=") == nullptr);
+  assert(dma_queries == 0);
+#endif
+  // Normal five-slot stats still fit the fixed LoRa reply budget.
+  assert(strstr(reply, "s5=123/4") != nullptr);
+
+  bridge.logMemoryStatus();
+#ifdef ESP_PLATFORM
+  assert(strstr(debug_line, "DMA=8128/1536") != nullptr);
+#else
+  assert(strstr(debug_line, "DMA=") == nullptr);
+#endif
+
+  // Large counters and new heap fields never write beyond a short reply.
+  struct { char reply[32]; unsigned char guard[8]; } bounded;
+  memset(&bounded, 0xA5, sizeof(bounded));
+  bridge._filtered_packets = 4294967295UL;
+  dma_free = 4294967295UL;
+  dma_largest = 4294967295UL;
+  MQTTBridge::formatMqttStatsReply(bounded.reply, sizeof(bounded.reply));
+  assert(bounded.reply[sizeof(bounded.reply) - 1] == '\0');
+  for (unsigned char byte : bounded.guard) assert(byte == 0xA5);
+  MQTTBridge::formatMqttStatsReply(nullptr, 32);
+  MQTTBridge::formatMqttStatsReply(bounded.reply, 0);
+}
+'''
+
+
+class MqttDiagnosticsTests(unittest.TestCase):
+    def test_production_stats_and_labels(self):
+        compiler = shutil.which("g++")
+        if not compiler:
+            self.skipTest("g++ is unavailable")
+        source = PREAMBLE + "\n".join(method(signature) for signature in (
+            "void MQTTBridge::formatMqttStatsReply(",
+            "const char* MQTTBridge::tlsErrorStr(",
+            "void MQTTBridge::logMemoryStatus(",
+        )) + MAIN
+        with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-diagnostics-") as directory:
+            test_source = Path(directory) / "diagnostics.cpp"
+            test_source.write_text(source, encoding="ascii")
+            for esp in (False, True):
+                with self.subTest(esp=esp):
+                    executable = Path(directory) / ("esp" if esp else "other")
+                    command = [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                               "-I", str(ROOT / "src"), str(test_source), "-o", str(executable)]
+                    if esp:
+                        command.insert(1, "-DESP_PLATFORM=1")
+                    subprocess.run(command, check=True, capture_output=True, text=True)
+                    subprocess.run([str(executable)], check=True, capture_output=True, text=True)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_mqtt_error_labels/test_mqtt_error_labels.cpp

@@ -29,6 +29,12 @@ TEST(MQTTErrorLabels, TlsErrorsMatchTheSdkMeaning) {
   EXPECT_TRUE(labelIs(tlsError(0x8010), "cert chain partly parsed"));   // was "mbedTLS error"
   EXPECT_TRUE(labelIs(tlsError(0x8006), "connection timeout"));
   EXPECT_TRUE(labelIs(tlsError(0x8001), "DNS failed"));
+  EXPECT_TRUE(labelIs(tlsError(0x8012), "SNI hostname rejected"));
+  EXPECT_TRUE(labelIs(tlsError(0x8015), "cert parse failed"));
+  EXPECT_TRUE(labelIs(tlsError(0x8017), "TLS setup failed"));
+  // A hardware AES allocation failure returns -1 through the TLS write path,
+  // which the SDK records as 0x8018 rather than a handshake or auth failure.
+  EXPECT_TRUE(labelIs(tlsError(0x8018), "TLS write failed"));
   EXPECT_TRUE(labelIs(tlsError(0x801A), "TLS handshake failed"));
 }
 

test/test_mqtt_handshake_scheduler.py

@@ -0,0 +1,286 @@
+#!/usr/bin/env python3
+"""Exercise production MQTT scheduling with delayed callbacks and bounded heaps.
+
+The actual maintenance/admission/reconnect methods run with a deterministic
+clock. Hardware, SDK task starts, and token generation are controllable seams.
+"""
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+from test_mqtt_transport_results import method
+
+ROOT = Path(__file__).resolve().parents[1]
+
+PREAMBLE = r'''
+#include <atomic>
+#include <cassert>
+#include <cstdint>
+#include <cstdio>
+#include <cstring>
+#include <vector>
+#include "helpers/MQTTConnectionAdmission.h"
+#include "helpers/MQTTConnectionHealth.h"
+#include "helpers/MQTTConnectionPolicy.h"
+#define ESP_PLATFORM 1
+#define MQTT_DEBUG_PRINTLN(...) ((void)0)
+using esp_err_t = int;
+constexpr int ESP_OK = 0, ESP_ERR_INVALID_ARG = 1, ESP_ERR_INVALID_STATE = 2;
+constexpr int ESP_ERR_NO_MEM = 3, RUNTIME_MQTT_SLOTS = 6, WL_CONNECTED = 1;
+constexpr int MQTT_AUTH_JWT = 1;
+constexpr uint32_t MALLOC_CAP_INTERNAL = 1, MALLOC_CAP_DMA = 2;
+static uint32_t clock_ms = 1000;
+static unsigned long wall_seconds = 1800000000;
+static size_t dma_free = 100000, dma_largest = 20000;
+uint32_t millis() { return clock_ms; }
+unsigned long fake_time(void*) { return wall_seconds; }
+#define time fake_time
+struct { int status() const { return WL_CONNECTED; } } WiFi;
+struct { size_t size = 2097152; size_t getPsramSize() const { return size; } } ESP;
+size_t heap_caps_get_free_size(uint32_t caps) {
+  assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); return dma_free;
+}
+size_t heap_caps_get_largest_free_block(uint32_t caps) {
+  assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); return dma_largest;
+}
+const char* esp_err_to_name(int) { return "mock"; }
+struct Preset {
+  const char* server_url = "wss://broker/mqtt";
+  int auth_type = 0;
+  unsigned long token_lifetime = 3300;
+  bool enforce_exp = false;
+};
+bool mqttPresetEnforcesTokenExp(const Preset* preset) {
+  return preset && preset->enforce_exp;
+}
+struct PsychicMqttClient {
+  bool started = false, live = false;
+  int starts = 0, reconnects = 0, bounces = 0, credentials = 0;
+  bool isStarted() const { return started; }
+  bool connected() const { return live; }
+  int connect() { ++starts; started = true; return ESP_OK; }
+  int reconnect() { ++reconnects; return ESP_OK; }
+  void softDisconnect() {
+    ++bounces; live = false;
+    // Releasing the old internal TLS session creates room for its replacement.
+    dma_free += 44000; dma_largest = 20000;
+  }
+  void setCredentials(const char*, const char*) { ++credentials; }
+};
+struct MQTTBridge {
+  struct MQTTSlot {
+    bool enabled = true, connected = false, initial_connect_done = false;
+    const Preset* preset = nullptr;
+    PsychicMqttClient* client = nullptr;
+    char host[64] = "broker", broker_uri[128] = {}, audience[64] = {};
+    uint16_t port = 8883;
+    char* auth_token = nullptr;
+    unsigned long token_expires_at = 0, last_token_renewal = 0;
+    unsigned long last_reconnect_attempt = 0, connected_at_ms = 0;
+    uint8_t reconnect_backoff = 0, max_backoff_failures = 0;
+    bool circuit_breaker_tripped = false;
+    uint32_t start_failures = 0;
+  } _slots[RUNTIME_MQTT_SLOTS];
+  PsychicMqttClient clients[RUNTIME_MQTT_SLOTS];
+  char tokens[RUNTIME_MQTT_SLOTS][16] = {};
+  bool setup_succeeds[RUNTIME_MQTT_SLOTS] = {true, true, true, true, true, true};
+  bool ready[RUNTIME_MQTT_SLOTS] = {true, true, true, true, true, true};
+  std::atomic<bool> _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _stop_requested{false}, _ntp_synced{true};
+  bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS] = {};
+  void* _identity = this;
+  bool _slots_setup_done = true;
+  unsigned long _last_slot_reconnect_ms = 0;
+  int _max_active_slots = 5, mints = 0;
+  const char* _jwt_username = "jwt";
+  static const unsigned long SLOT_SETUP_RETRY_INTERVAL = 60000;
+  std::vector<int> setups;
+  bool isSlotReady(int index) { return ready[index]; }
+  bool setupSlot(int index) {
+    setups.push_back(index);
+    _slots[index].client = &clients[index];
+    if (!setup_succeeds[index]) return false;
+    if (reconnectSlotClient(index) != ESP_OK) return false;
+    _slots[index].initial_connect_done = true;
+    return true;
+  }
+  bool createSlotAuthToken(int index) {
+    ++mints;
+    strcpy(tokens[index], "fresh");
+    _slots[index].auth_token = tokens[index];
+    _slots[index].token_expires_at = wall_seconds + 3300;
+    return true;
+  }
+  void complete(int index) {
+    _slot_attempt_pending[index].store(false, std::memory_order_release);
+    _slots[index].connected = clients[index].live = true;
+  }
+  void only(int index) {
+    for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) _slots[i].enabled = i == index;
+  }
+  int activatedSlotCount() const;
+  bool canActivateSlot(int index) const;
+  bool hasPendingSlotConnection() const;
+  bool canStartSlotConnection(int index) const;
+  esp_err_t reconnectSlotClient(int index);
+  unsigned long slotTokenLifetime(int index) const;
+  void maintainSlotConnections();
+  void maintainSlotConnection(int, unsigned long, unsigned long, bool, bool&, bool&);
+};
+'''
+
+
+class MqttHandshakeSchedulerTest(unittest.TestCase):
+    def fixture(self):
+        bridge = "src/helpers/bridges/MQTTBridge.cpp"
+        return PREAMBLE + "\n".join(method(bridge, signature) for signature in (
+            "int MQTTBridge::activatedSlotCount() const",
+            "bool MQTTBridge::canActivateSlot(int index) const",
+            "bool MQTTBridge::hasPendingSlotConnection() const",
+            "bool MQTTBridge::canStartSlotConnection(int index) const",
+            "esp_err_t MQTTBridge::reconnectSlotClient(int index)",
+            "unsigned long MQTTBridge::slotTokenLifetime(int index) const",
+            "void MQTTBridge::maintainSlotConnections()",
+            "void MQTTBridge::maintainSlotConnection("))
+
+    def compile_run(self, main):
+        compiler = shutil.which("g++") or shutil.which("clang++")
+        self.assertIsNotNone(compiler, "a host C++ compiler is required")
+        with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-scheduler-") as temporary:
+            work = Path(temporary)
+            source, binary = work / "scheduler.cpp", work / "scheduler"
+            source.write_text(self.fixture() + main, encoding="utf-8")
+            result = subprocess.run(
+                [compiler, "-std=c++17", "-Wall", "-Wextra", "-I", str(ROOT / "src"),
+                 str(source), "-o", str(binary)], capture_output=True, text=True, timeout=60)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            result = subprocess.run([str(binary)], cwd=work, capture_output=True,
+                                    text=True, timeout=10)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_first_start_is_immediate_and_delayed_callback_serializes_following_slots(self):
+        self.compile_run(r'''
+int main() {
+  MQTTBridge bridge;
+  clock_ms = 1000;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups == std::vector<int>{0});
+  assert(bridge._slot_attempt_pending[0]);
+  // The time guard has expired, but the earlier asynchronous handshake is
+  // still pending. Ordinary maintenance must not launch another SDK request.
+  clock_ms = 16000;
+  bridge.maintainSlotConnections();
+  clock_ms = 3600000;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups == std::vector<int>{0});
+  assert(bridge.clients[0].starts == 1 && bridge.clients[0].reconnects == 0);
+  bridge.complete(0);
+  bridge.maintainSlotConnections();
+  assert(bridge.setups == std::vector<int>({0, 1}));
+  assert(bridge._slot_attempt_pending[1]);
+  bridge.complete(1);
+  clock_ms += 14999;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups.size() == 2);
+  ++clock_ms;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups == std::vector<int>({0, 1, 2}));
+  bridge._stop_requested = true;
+  bridge.complete(2);
+  clock_ms += 60000;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups.size() == 3);
+}
+''')
+
+    def test_setup_failure_waits_full_minute_even_when_client_was_allocated(self):
+        self.compile_run(r'''
+int main() {
+  MQTTBridge bridge;
+  bridge.only(0); bridge.setup_succeeds[0] = false;
+  clock_ms = 1000;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups.size() == 1 && bridge._slots[0].client);
+  assert(!bridge._slots[0].initial_connect_done && bridge._slots[0].enabled);
+  assert(!bridge.hasPendingSlotConnection());
+  clock_ms = 1050;
+  bridge.maintainSlotConnections();
+  clock_ms = 60999;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups.size() == 1);
+  clock_ms = 61000; bridge.setup_succeeds[0] = true;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups == std::vector<int>({0, 0}));
+  assert(bridge._slots[0].initial_connect_done && bridge.hasPendingSlotConnection());
+}
+''')
+
+    def test_failed_slot_does_not_starve_healthy_slot_or_consume_active_cap(self):
+        self.compile_run(r'''
+int main() {
+  MQTTBridge bridge;
+  bridge._max_active_slots = 1; bridge.setup_succeeds[0] = false;
+  clock_ms = 1000;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups == std::vector<int>{0});
+  assert(bridge.activatedSlotCount() == 0);
+  clock_ms = 1050;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups == std::vector<int>({0, 1}));
+  assert(bridge.activatedSlotCount() == 1);
+  for (int i = 2; i < RUNTIME_MQTT_SLOTS; ++i) assert(!bridge._slots[i].enabled);
+  bridge.complete(1); clock_ms += 60000;
+  bridge.maintainSlotConnections();
+  assert(!bridge._slots[0].enabled && bridge.setups.size() == 2);
+}
+''')
+
+    def test_low_dma_memory_defers_setup_without_consuming_retry_or_capacity(self):
+        self.compile_run(r'''
+int main() {
+  MQTTBridge bridge;
+  bridge.only(0); clock_ms = 1000; dma_free = 16383;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups.empty() && bridge._slots[0].last_reconnect_attempt == 0);
+  assert(bridge._slots[0].enabled && bridge.activatedSlotCount() == 0);
+  dma_free = 100000; dma_largest = 4095;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups.empty());
+  dma_largest = 4096;
+  bridge.maintainSlotConnections();
+  assert(bridge.setups == std::vector<int>{0} && bridge.activatedSlotCount() == 1);
+}
+''')
+
+    def test_internal_tls_renewal_releases_old_session_before_new_memory_admission(self):
+        self.compile_run(r'''
+int main() {
+  MQTTBridge bridge;
+  bridge.only(0); clock_ms = 70000; ESP.size = 0;
+  const Preset jwt{"wss://broker/mqtt", MQTT_AUTH_JWT, 3300, true};
+  auto& slot = bridge._slots[0];
+  slot.preset = &jwt; slot.client = &bridge.clients[0];
+  slot.initial_connect_done = slot.connected = true;
+  slot.client->started = slot.client->live = true;
+  slot.token_expires_at = wall_seconds + 120;
+  // Before releasing the current TLS session, the full admission check fails.
+  dma_free = 20000; dma_largest = 10000;
+  assert(!bridge.canStartSlotConnection(0));
+  bridge.maintainSlotConnections();
+  assert(bridge.mints == 1 && slot.client->bounces == 1);
+  assert(slot.client->reconnects == 1 && bridge._slot_attempt_pending[0]);
+  assert(slot.client->credentials == 1 && dma_free == 64000);
+
+  // Another pending slot still prevents renewal from tearing down a live link.
+  bridge.complete(0); bridge._slot_attempt_pending[1] = true;
+  slot.token_expires_at = wall_seconds + 120; clock_ms += 60000;
+  bridge.maintainSlotConnections();
+  assert(bridge.mints == 1 && slot.client->bounces == 1);
+}
+''')
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_mqtt_network_timeout.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""Exercise production MQTT configuration for both SDK layouts and RAM profiles.
+
+The real optimizer and adapter setters run on the host. SDK field layouts are
+stand-ins; this checks the timeout written to SDK configuration, not a handshake.
+"""
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+from test_mqtt_transport_results import method
+
+ROOT = Path(__file__).resolve().parents[1]
+
+PREAMBLE = r'''
+#include <cassert>
+#include <cstddef>
+#include "helpers/MQTTConnectionPolicy.h"
+struct esp_mqtt_client_config_t {
+#if ESP_IDF_VERSION_MAJOR == 5
+  struct { int keepalive = 0, message_retransmit_timeout = 0; } session;
+  struct { int timeout_ms = 0; } network;
+  struct { int size = 0, out_size = 0; } buffer;
+#else
+  int keepalive = 0, message_retransmit_timeout = 0;
+  int network_timeout_ms = 0, buffer_size = 0;
+#endif
+};
+struct PsychicMqttClient {
+  esp_mqtt_client_config_t _mqtt_cfg;
+  bool _config_dirty = false;
+  size_t _outbox_limit = 0;
+  PsychicMqttClient& setKeepAlive(int);
+  PsychicMqttClient& setMessageRetransmitTimeout(int);
+  PsychicMqttClient& setBufferSize(int);
+  PsychicMqttClient& setNetworkTimeout(int);
+  PsychicMqttClient& setOutboxLimit(size_t);
+  esp_mqtt_client_config_t* getMqttConfig();
+};
+struct MQTTBridge {
+  void optimizeMqttClientConfig(PsychicMqttClient*, bool);
+};
+'''
+
+MAIN = r'''
+int main() {
+  MQTTBridge bridge;
+  bridge.optimizeMqttClientConfig(nullptr, true);
+  PsychicMqttClient client;
+  // Reconfiguration must restore the connection budget after a shorter value,
+  // for both small user/password and large JWT CONNECT buffers.
+  for (bool needs_large : {false, true, false}) {
+    client.setNetworkTimeout(2500);
+    client._config_dirty = false;
+    bridge.optimizeMqttClientConfig(&client, needs_large);
+    assert(client._config_dirty);
+#if ESP_IDF_VERSION_MAJOR == 5
+    const int timeout = client._mqtt_cfg.network.timeout_ms;
+    const int buffer_size = client._mqtt_cfg.buffer.size;
+    const int keepalive = client._mqtt_cfg.session.keepalive;
+    assert(client._mqtt_cfg.buffer.out_size == buffer_size);
+    assert(client._mqtt_cfg.session.message_retransmit_timeout == 15000);
+#else
+    const int timeout = client._mqtt_cfg.network_timeout_ms;
+    const int buffer_size = client._mqtt_cfg.buffer_size;
+    const int keepalive = client._mqtt_cfg.keepalive;
+    assert(client._mqtt_cfg.message_retransmit_timeout == 15000);
+#endif
+    assert(timeout == 10000);
+    assert(timeout == MQTTConnectionPolicy::kNetworkTimeoutMs);
+#ifdef BOARD_HAS_PSRAM
+    assert(buffer_size == 896 && keepalive == 45);
+    assert(client._outbox_limit == 16384);
+#else
+    assert(buffer_size == (needs_large ? 896 : 512) && keepalive == 75);
+    assert(client._outbox_limit == 8192);
+#endif
+  }
+}
+'''
+
+
+class MqttNetworkTimeoutTest(unittest.TestCase):
+    def fixture(self):
+        adapter = "lib/PsychicMqttClient/src/PsychicMqttClient.cpp"
+        source = PREAMBLE + "#include <initializer_list>\n"
+        source += "\n".join(method(adapter, signature) for signature in (
+            "PsychicMqttClient &PsychicMqttClient::setKeepAlive(",
+            "PsychicMqttClient &PsychicMqttClient::setMessageRetransmitTimeout(",
+            "PsychicMqttClient &PsychicMqttClient::setBufferSize(",
+            "PsychicMqttClient &PsychicMqttClient::setNetworkTimeout(",
+            "PsychicMqttClient &PsychicMqttClient::setOutboxLimit(",
+            "esp_mqtt_client_config_t *PsychicMqttClient::getMqttConfig("))
+        source += method("src/helpers/bridges/MQTTBridge.cpp",
+                         "void MQTTBridge::optimizeMqttClientConfig(")
+        return source + MAIN
+
+    def compile_run(self, source, expect_rejection=False):
+        compiler = shutil.which("g++") or shutil.which("clang++")
+        self.assertIsNotNone(compiler, "a host C++ compiler is required")
+        with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-timeout-") as temporary:
+            work = Path(temporary)
+            path = work / "config.cpp"
+            path.write_text(source, encoding="utf-8")
+            for major in (4, 5):
+                for psram in (False, True):
+                    with self.subTest(idf_major=major, psram=psram):
+                        binary = work / f"config-{major}-{int(psram)}"
+                        command = [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                                   "-Wno-unused-parameter",
+                                   f"-DESP_IDF_VERSION_MAJOR={major}",
+                                   "-I", str(ROOT / "src"), str(path), "-o", str(binary)]
+                        if psram:
+                            command.insert(1, "-DBOARD_HAS_PSRAM=1")
+                        compiled = subprocess.run(command, capture_output=True,
+                                                  text=True, timeout=60)
+                        self.assertEqual(compiled.returncode, 0, compiled.stderr)
+                        result = subprocess.run([str(binary)], cwd=work, capture_output=True,
+                                                text=True, timeout=10)
+                        if expect_rejection:
+                            self.assertNotEqual(result.returncode, 0,
+                                                result.stdout + result.stderr)
+                            self.assertIn("Assertion", result.stderr)
+                        else:
+                            self.assertEqual(result.returncode, 0,
+                                             result.stdout + result.stderr)
+
+    def test_production_optimizer_applies_connection_budget_to_every_config(self):
+        self.compile_run(self.fixture())
+
+    def test_previous_publish_timeout_fails_configuration_regression(self):
+        source = self.fixture()
+        old = source.replace(
+            "client->setNetworkTimeout(MQTTConnectionPolicy::kNetworkTimeoutMs);",
+            "client->setNetworkTimeout(2500);", 1)
+        self.assertNotEqual(old, source)
+        self.compile_run(old, expect_rejection=True)
+
+    def test_production_stop_budget_covers_shared_network_timeout_for_every_slot(self):
+        bridge = ROOT / "src/helpers/bridges/MQTTBridge.cpp"
+        source = bridge.read_text(encoding="utf-8")
+        start = source.index("static const uint32_t MQTT_STOP_TIMEOUT_BASE_MS")
+        end = source.index("static inline uint32_t mqttStopTimeoutForSlots(int slots)")
+        fixture = '#include <cassert>\n#include "helpers/MQTTConnectionPolicy.h"\n'
+        fixture += source[start:end]
+        fixture += method("src/helpers/bridges/MQTTBridge.cpp",
+                          "static inline uint32_t mqttStopTimeoutForSlots(int slots)")
+        fixture += r'''
+int main() {
+  assert(mqttStopTimeoutForSlots(0) == 25000);
+  assert(mqttStopTimeoutForSlots(-1) == 25000);
+  assert(mqttStopTimeoutForSlots(2) == 35000);
+  assert(mqttStopTimeoutForSlots(5) == 65000);
+  assert(mqttStopTimeoutForSlots(6) == 75000);
+  assert(mqttStopTimeoutForSlots(6) > UINT16_MAX);
+  assert(mqttStopTimeoutForSlots(16) == 175000);
+}
+'''
+        self.compile_run(fixture)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_mqtt_transport_results.py

@@ -90,8 +90,10 @@ struct MQTTBridge {
     uint32_t start_failures=0;
   };
   MQTTSlot _slots[RUNTIME_MQTT_SLOTS];
-  volatile bool _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {};
   std::atomic<bool> _stop_requested{false};
+  bool hasPendingSlotConnection() const;
+  bool canStartSlotConnection(int index) const;
   esp_err_t reconnectSlotClient(int index);
 };
 '''
@@ -135,10 +137,19 @@ int main() {
   reconnect_result=ESP_FAIL;
   assert(b.reconnectSlotClient(0)==ESP_FAIL);
   assert(!b._slot_attempt_pending[0] && b._slots[0].start_failures==0);
+  int attempts = reconnect_calls;
   b._slot_attempt_pending[0]=true;
-  assert(b.reconnectSlotClient(0)==ESP_FAIL);
+  assert(b.reconnectSlotClient(0)==ESP_ERR_NO_MEM && reconnect_calls==attempts);
   assert(b._slot_attempt_pending[0] && b._slots[0].start_failures==0);
-  sdk_callback=[&]() { b._slot_attempt_pending[0]=false; };
+  b._slot_attempt_pending[0]=false;
+  b._slot_attempt_pending[1]=true;
+  assert(b.reconnectSlotClient(0)==ESP_ERR_NO_MEM && reconnect_calls==attempts);
+  assert(!b._slot_attempt_pending[0] && b._slot_attempt_pending[1]);
+  b._slot_attempt_pending[1]=false;
+  sdk_callback=[&]() {
+    assert(b._slot_attempt_pending[0]);
+    b._slot_attempt_pending[0]=false;
+  };
   assert(b.reconnectSlotClient(0)==ESP_FAIL && !b._slot_attempt_pending[0]);
   sdk_callback=nullptr;
   c._started=false;
@@ -259,7 +270,7 @@ struct MQTTBridge {
   }
   void maintainSlotConnections() {
     work();
-    if (scenario == 10) _stop_requested = true;
+    if (scenario == 5 || scenario == 10) _stop_requested = true;
   }
   void processPacketQueue() {
     work(); ++queue_calls;
@@ -292,9 +303,9 @@ void vTaskDelay(TickType_t ticks) {
   task_sleeps.push_back(duration);
   elapsed_ms += duration;
   if (elapsed_ms >= stop_at_ms) active_bridge->_stop_requested = true;
-  // The no-stop scenario exits after one complete worker iteration. Stop is
-  // raised during its ordinary idle delay, after all setup and publications.
-  if (scenario == 3 && active_bridge->setups.size() == RUNTIME_MQTT_SLOTS)
+  // The no-stop scenario exits after one complete worker iteration. Startup
+  // only schedules slots; connection maintenance now owns their async starts.
+  if (scenario == 3 && active_bridge->queue_calls == 1)
     active_bridge->_stop_requested = true;
   assert(elapsed_ms < 30000); // bound a missing cooperative exit deterministically
 }
@@ -313,11 +324,11 @@ int main(int argc, char** argv) {
   active_bridge = &bridge;
   switch (scenario) {
     case 0: stop_at_ms = 187; break;          // stop while WiFi settles
-    case 1: stop_at_ms = 1173; break;         // stop during first TLS stagger
-    case 2: stop_at_ms = 6191; break;         // stop during second TLS stagger
+    case 1: stop_at_ms = 1173; break;         // stop during ordinary service
+    case 2: stop_at_ms = 6191; break;         // stop during later ordinary service
     case 3: break;                           // preserve every normal delay
     case 4: bridge._stop_requested = true; break;
-    case 5: break;                           // setup callback requests stop
+    case 5: break;                           // maintenance callback requests stop
     case 6: bridge._ntp_force_requested = true; break;
     case 7: bridge._ntp_sync_pending = true; bridge._ntp_synced = false; break;
     case 8: case 9: case 10: case 11: break;  // stop raised during worker callbacks
@@ -327,28 +338,23 @@ int main(int argc, char** argv) {
   assert(bridge._stop_acked && bridge.late_work == 0);
   assert(bridge.teardown == std::vector<std::string>({
     "ntp", "slot0", "slot1", "slot2", "destroy"}));
+  assert(bridge.setups.empty()); // no blocking startup slot loop remains
   if (scenario == 3) {
-    assert(bridge.setups == std::vector<int>({0, 1, 2}));
-    assert(bridge.setup_times == std::vector<uint64_t>({1000, 6000, 11000}));
+    assert(bridge._slots_setup_done);
     assert(bridge.reconfigures == RUNTIME_MQTT_SLOTS);
     assert(bridge.publishes >= RUNTIME_MQTT_SLOTS && bridge.queue_calls == 1);
   } else {
-    const size_t wanted = scenario >= 8 ? RUNTIME_MQTT_SLOTS
-                          : scenario == 0 || scenario == 4 || scenario >= 6 ? 0
-                          : scenario == 2 ? 2 : 1;
-    assert(bridge.setups.size() == wanted);
-    if (scenario >= 8) {
+    if (scenario >= 8 || scenario == 5) {
       assert(bridge.reconfigures == (scenario == 8 ? 1 : RUNTIME_MQTT_SLOTS));
       assert(bridge.publishes == (scenario == 8 ? 0 : scenario == 9 ? 1 : RUNTIME_MQTT_SLOTS));
       assert(bridge.queue_calls == (scenario == 11 ? 1 : 0));
-    } else {
+    } else if (scenario != 1 && scenario != 2) {
       assert(bridge.reconfigures == 0 && bridge.publishes == 0 && bridge.queue_calls == 0);
     }
-    const uint32_t poll_bound = std::max(uint32_t{20}, tick_ms);
+    const uint32_t poll_bound = std::max(uint32_t{50}, tick_ms);
     assert(std::all_of(task_sleeps.begin(), task_sleeps.end(),
                        [poll_bound](uint32_t duration) { return duration <= poll_bound; }));
     if (scenario == 4) assert(elapsed_ms == 0 && task_sleeps.empty());
-    else if (scenario >= 8) assert(elapsed_ms == 11000);
     else if (scenario >= 5) assert(elapsed_ms == 1000);
     else assert(elapsed_ms >= stop_at_ms && elapsed_ms - stop_at_ms < poll_bound);
   }
@@ -368,6 +374,7 @@ ASYNC_STOP_PREAMBLE = r'''
 #include <cstdint>
 #include <limits>
 #include "helpers/MQTTLifecycle.h"
+#include "helpers/MQTTConnectionPolicy.h"
 #define ESP_PLATFORM 1
 #define MQTT_DEBUG_PRINTLN(...) ((void)0)
 #define pdMS_TO_TICKS(ms) (ms)
@@ -483,7 +490,10 @@ int main() {
     clock_ms = UINT32_MAX - 500; elapsed_ms = 0; sleeps = 0;
     bridge_waiting = &bridge;
     bridge.requestStop();
-    ack_after_ms = at_deadline ? bridge._lifecycle.stopTimeoutMs() : 8000;
+    assert(bridge._lifecycle.stopTimeoutMs() == 45000);
+    // A ten-second blocked publish plus healthy client teardown can exceed
+    // the former 29-second budget for three slots without requiring a kill.
+    ack_after_ms = at_deadline ? bridge._lifecycle.stopTimeoutMs() : 31000;
     bridge.end(); // synchronous OTA barrier joins the existing request
     assert(elapsed_ms == ack_after_ms && sleeps > 0 && bridge.ops.signals == 1);
     assert(!bridge._lifecycle.stopTimedOut());
@@ -525,7 +535,7 @@ class MqttTransportResultsTests(unittest.TestCase):
             result = subprocess.run([str(exe)], text=True, capture_output=True, timeout=5)
             self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
 
-    def test_actual_worker_interrupts_startup_and_stagger_without_late_work(self):
+    def test_actual_worker_interrupts_startup_and_service_without_late_work(self):
         compiler = shutil.which("g++") or shutil.which("c++")
         self.assertIsNotNone(compiler, "a host C++ compiler is required")
         bridge = "src/helpers/bridges/MQTTBridge.cpp"
@@ -555,6 +565,10 @@ class MqttTransportResultsTests(unittest.TestCase):
             "esp_err_t PsychicMqttClient::applyConfig()",
             "esp_err_t PsychicMqttClient::connect()",
             "esp_err_t PsychicMqttClient::reconnect()"))
+        source += method("src/helpers/bridges/MQTTBridge.cpp",
+                         "bool MQTTBridge::hasPendingSlotConnection() const")
+        source += method("src/helpers/bridges/MQTTBridge.cpp",
+                         "bool MQTTBridge::canStartSlotConnection(int index) const")
         source += method("src/helpers/bridges/MQTTBridge.cpp",
                          "esp_err_t MQTTBridge::reconnectSlotClient(int index)") + MAIN
         with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-results-") as temp: