Merge pull request #46 from agessaman/fix/mqtt-review-followups

Fix MQTT client issues with JWT renewal, memory allocation, and NTP handling

.gitignore

@@ -32,3 +32,7 @@ platformio.local.ini
 .build-wt-*/
 .wt-*/
 scripts/__pycache__/*
+
+# Reduced-TLS mbedTLS archives, ~6 MB per arch, fetched by scripts/fetch_mbedtls_4k.sh.
+# Not committed because they must be rebuilt for every espressif32 platform bump.
+.mbedtls-4k/

docs/mbedtls-tls-footprint.md

@@ -136,6 +136,36 @@ fix was `rm -rf` the package and `pio pkg install` to re-download stock. A prepe
 search path keeps the change scoped to one env, because the linker takes each archive member
 from the first archive that satisfies an undefined symbol.
 
+### How the archives are distributed
+
+They are not committed: ~6 MB per architecture, and they have to be rebuilt for every
+`platformio/espressif32` bump, so committing them would grow history permanently and go
+stale silently. Instead they are published as a release asset and fetched on demand:
+
+```
+scripts/fetch_mbedtls_4k.sh esp32s3            # download + verify against the manifest
+MESHCORE_REDUCED_TLS=1 pio run -e Heltec_v3_repeater_observer_mqtt
+```
+
+- `scripts/mbedtls_4k_manifest.txt` — per-arch sha256 of each archive. **Update it on every
+  platform bump**, together with the published asset.
+- `scripts/fetch_mbedtls_4k.sh` — downloads into `.mbedtls-4k/<arch>/` (gitignored) and
+  verifies. `MBEDTLS_4K_LOCAL=<dir>` copies from a local build tree instead of downloading.
+- `scripts/mbedtls_4k.py` — wired into `esp32_base.extra_scripts`, but a **no-op unless
+  `MESHCORE_REDUCED_TLS=1`**, so ordinary builds need no artifact and behave as before.
+
+The opt-in path is deliberately loud, because both ways this can go wrong produce a
+firmware that looks correct and silently lacks the change:
+
+| failure | what happens without a guard | guard |
+|---|---|---|
+| directory missing or partial | linker ignores an unusable `-L` and resolves mbedTLS from the framework | pre-build: hard error |
+| archives stale after a platform bump | links the wrong build | pre-build: sha256 vs manifest |
+| `-L` present but outranked | framework archives win, flag is inert | post-link: `firmware.map` must resolve every `libmbed*.a` to `.mbedtls-4k/` |
+
+That last one is the reason the post-link check exists rather than trusting the flag: a
+build flag reaching the compiler proves nothing about what got linked.
+
 ## How to verify it worked
 
 1. `strings`/`grep` the new `sdkconfig.h` for the four settings.

lib/PsychicMqttClient/src/PsychicMqttClient.cpp

@@ -433,8 +433,18 @@ void PsychicMqttClient::connect()
         }
     }
 
-    ESP_ERROR_CHECK_WITHOUT_ABORT(esp_mqtt_client_start(_client));
-    ESP_LOGI(TAG, "MQTT client started.");
+    esp_err_t start_result = esp_mqtt_client_start(_client);
+    ESP_ERROR_CHECK_WITHOUT_ABORT(start_result);
+    if (start_result == ESP_OK)
+    {
+        _started = true;
+        ESP_LOGI(TAG, "MQTT client started.");
+    }
+    else
+    {
+        // Reporting success here hides the one state reconnect() cannot recover from.
+        ESP_LOGE(TAG, "MQTT client failed to start: %s", esp_err_to_name(start_result));
+    }
 }
 
 void PsychicMqttClient::reconnect()
@@ -489,9 +499,40 @@ void PsychicMqttClient::disconnect()
     }
 
     esp_mqtt_client_stop(_client);
+    _started = false;
     ESP_LOGI(TAG, "MQTT client stopped.");
 }
 
+void PsychicMqttClient::softDisconnect(unsigned long timeout_ms)
+{
+    if (_client == nullptr)
+    {
+        ESP_LOGW(TAG, "MQTT client not started.");
+        return;
+    }
+
+    if (!_connected)
+    {
+        // Nothing to close; leaving the task alone is the whole point.
+        return;
+    }
+
+    ESP_LOGI(TAG, "Disconnecting MQTT transport (client task retained).");
+    _stopMqttClient = false;
+    esp_mqtt_client_disconnect(_client);
+
+    unsigned long waited = 0;
+    while (!_stopMqttClient && waited < timeout_ms)
+    {
+        vTaskDelay(10 / portTICK_PERIOD_MS);
+        waited += 10;
+    }
+    if (!_stopMqttClient)
+    {
+        ESP_LOGW(TAG, "softDisconnect: no DISCONNECTED event in %lums", timeout_ms);
+    }
+}
+
 void PsychicMqttClient::forceStop()
 {
     if (_client == nullptr)
@@ -506,6 +547,7 @@ void PsychicMqttClient::forceStop()
     }
     ESP_ERROR_CHECK_WITHOUT_ABORT(esp_mqtt_client_stop(_client));
     _connected = false;
+    _started = false;
     ESP_LOGI(TAG, "MQTT client forcefully stopped.");
 }
 

lib/PsychicMqttClient/src/PsychicMqttClient.h

@@ -368,6 +368,29 @@ public:
      */
     void disconnect();
 
+    /**
+     * @brief Closes the transport but leaves the client task running.
+     *
+     * disconnect() ends with esp_mqtt_client_stop(), which ends the client task
+     * and returns its 6 KiB stack to the heap right as a TLS handshake vacates
+     * two 16 KiB mbedTLS record buffers — the stack then lands in that hole and
+     * the largest free block ratchets down. This variant omits the stop, so the
+     * task and its stack stay put. Pair it with reconnect().
+     *
+     * @param timeout_ms how long to wait for the DISCONNECTED event before
+     *                   giving up. Bounded on purpose: disconnect()'s wait is
+     *                   unbounded and a lost event would wedge the caller.
+     */
+    void softDisconnect(unsigned long timeout_ms = 5000);
+
+    /**
+     * @brief True once esp_mqtt_client_start() has succeeded and no stop has run.
+     *
+     * reconnect() silently does nothing on a stopped client, so callers that
+     * want to avoid stop/start must check this and fall back to connect().
+     */
+    bool isStarted() const { return _started; }
+
     /**
      * @brief Forcefully stops the MQTT client and disconnects from the server.
      * This does not trigger the onDisconnect callbacks.
@@ -478,6 +501,7 @@ private:
     bool _connected = false;
     bool _stopMqttClient = false;
     bool _config_dirty = true;
+    bool _started = false;
 
     // Runtime cap on the esp-mqtt outbox for QoS 0 async publishes (bytes).
     // 0 = disabled. Enforced in publish(); not an esp-mqtt config field.

platformio.ini

@@ -64,6 +64,8 @@ platform = platformio/espressif32@6.11.0
 monitor_filters = esp32_exception_decoder
 extra_scripts =
   pre:scripts/generate_webconfig_html.py
+; No-op unless MESHCORE_REDUCED_TLS=1; see docs/mbedtls-tls-footprint.md.
+  pre:scripts/mbedtls_4k.py
   merge-bin.py
 build_flags = ${arduino_base.build_flags}
   -D ESP32_PLATFORM

platformio.local.ini.hold

@@ -1,15 +0,0 @@
-; Local-only override (gitignored) pointing the Heltec V3 observer env at a custom
-; framework whose mbedTLS archives were rebuilt with an asymmetric TLS record buffer:
-;   CONFIG_MBEDTLS_ASYMMETRIC_CONTENT_LEN=y
-;   CONFIG_MBEDTLS_SSL_IN_CONTENT_LEN=16384   (unchanged — a peer may send a 16 KiB record)
-;   CONFIG_MBEDTLS_SSL_OUT_CONTENT_LEN=4096   (was 16384)
-;
-; Expected: ~12 KiB less internal DRAM per TLS connection, ~24 KiB across two broker slots.
-; Built from the shipped sdkconfig verbatim plus those three lines, so the archives differ
-; only by this change. See docs/mbedtls-tls-footprint.md.
-;
-; Absolute path, hence local-only: not committable.
-
-[env:Heltec_v3_repeater_observer_mqtt]
-platform_packages =
-  framework-arduinoespressif32 @ file:///Users/adam/framework-arduinoespressif32-tlsfix

scripts/fetch_mbedtls_4k.sh

@@ -0,0 +1,75 @@
+#!/usr/bin/env bash
+# Fetch the reduced-TLS mbedTLS archives into .mbedtls-4k/<arch>/.
+#
+# These archives are built from the shipped sdkconfig plus three lines
+# (CONFIG_MBEDTLS_ASYMMETRIC_CONTENT_LEN=y, IN_CONTENT_LEN 16384,
+# OUT_CONTENT_LEN 4096) and save ~12 KiB of internal DRAM per TLS connection.
+# See docs/mbedtls-tls-footprint.md for the rationale and the build recipe.
+#
+# They are not committed: ~6 MB per architecture, and they must be rebuilt for
+# every platform bump, so they are published as a release asset keyed on the
+# espressif32 platform version instead.
+#
+#   scripts/fetch_mbedtls_4k.sh [arch]        # default: esp32s3
+#
+# Set MBEDTLS_4K_LOCAL to skip the download and copy from a local build tree:
+#   MBEDTLS_4K_LOCAL=~/mbedtls-4k-esp32s3/staged scripts/fetch_mbedtls_4k.sh
+set -euo pipefail
+
+ARCH="${1:-esp32s3}"
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+DEST="$REPO_ROOT/.mbedtls-4k/$ARCH"
+MANIFEST="$REPO_ROOT/scripts/mbedtls_4k_manifest.txt"
+BASE_URL="${MBEDTLS_4K_BASE_URL:-https://github.com/agessaman/MeshCore/releases/download/mbedtls-4k}"
+
+if [ ! -f "$MANIFEST" ]; then
+  echo "error: missing $MANIFEST" >&2
+  exit 1
+fi
+
+# Manifest lines: <arch> <sha256> <filename>. Blank lines and # comments ignored.
+# The "platform" and "stock:<arch>" lines bind the archives to a framework version;
+# they are the build check's business, not ours, and are not architectures.
+expected="$(awk -v a="$ARCH" '$1 == a && $0 !~ /^#/ {print $2"  "$3}' "$MANIFEST")"
+if [ -z "$expected" ]; then
+  echo "error: no manifest entries for arch '$ARCH'" >&2
+  echo "known arches: $(awk '$0 !~ /^#/ && NF && $1 != "platform" && $1 !~ /^stock:/ {print $1}' \
+    "$MANIFEST" | sort -u | tr '\n' ' ')" >&2
+  exit 1
+fi
+
+mkdir -p "$DEST"
+
+if [ -n "${MBEDTLS_4K_LOCAL:-}" ]; then
+  echo "copying from $MBEDTLS_4K_LOCAL"
+  while read -r _sha name; do
+    cp "$MBEDTLS_4K_LOCAL/$name" "$DEST/$name"
+  done <<< "$expected"
+else
+  TARBALL="mbedtls-4k-$ARCH.tar.gz"
+  echo "downloading $BASE_URL/$TARBALL"
+  tmp="$(mktemp -d)"
+  trap 'rm -rf "$tmp"' EXIT
+  curl -fsSL "$BASE_URL/$TARBALL" -o "$tmp/$TARBALL"
+  tar -xzf "$tmp/$TARBALL" -C "$tmp"
+  while read -r _sha name; do
+    # Accept the archive whether or not the tarball has a leading directory.
+    found="$(find "$tmp" -name "$name" -type f | head -1)"
+    if [ -z "$found" ]; then
+      echo "error: $name missing from $TARBALL" >&2
+      exit 1
+    fi
+    cp "$found" "$DEST/$name"
+  done <<< "$expected"
+fi
+
+# Verify every archive against the manifest. A wrong or truncated archive would
+# otherwise link silently and produce a firmware without the reduced buffers.
+cd "$DEST"
+if command -v shasum >/dev/null 2>&1; then
+  echo "$expected" | shasum -a 256 -c -
+else
+  echo "$expected" | sha256sum -c -
+fi
+
+echo "ok: $ARCH archives verified in $DEST"

scripts/mbedtls_4k.py

@@ -0,0 +1,244 @@
+"""Link the reduced-TLS mbedTLS archives, and prove they were actually linked.
+
+Opt in per build with MESHCORE_REDUCED_TLS=1. Off by default, so an ordinary build
+needs no 6 MB artifact and behaves exactly as before.
+
+    MESHCORE_REDUCED_TLS=1 pio run -e Heltec_v3_repeater_observer_mqtt
+
+The archives lower the mbedTLS outbound record buffer from 16 KiB to 4 KiB, saving
+~12 KiB of internal DRAM per TLS connection on non-PSRAM observers. The inbound
+buffer stays at 16 KiB, so the contiguous allocation a handshake needs is unchanged
+— this buys headroom, it does not move that floor. See docs/mbedtls-tls-footprint.md.
+
+Three failure modes this guards against, each of which produces a firmware that looks
+fine and is silently wrong:
+
+  - a -L pointing at a missing or partial directory. The linker ignores an
+    unusable search path and quietly resolves mbedTLS from the framework instead.
+  - archives that do not match the manifest, e.g. left over from an earlier
+    platform version.
+  - a manifest and archives that agree with each other but not with the installed
+    framework, which is what a platform bump without a rebuild leaves behind. That
+    one links cleanly and drifts on struct layout at runtime.
+
+So the opt-in path verifies every archive by sha256 before the build, checks the
+framework's own mbedTLS archives still fingerprint as the ones these were built
+against, and after the link re-reads firmware.map to confirm every libmbed*.a came
+from our directory.
+"""
+Import("env")
+
+import hashlib
+import os
+import sys
+
+REQUIRED = ("libmbedcrypto.a", "libmbedtls_2.a", "libmbedtls.a", "libmbedx509.a")
+
+
+def _fail(msg):
+    print("\n*** reduced-TLS build failed ***", file=sys.stderr)
+    print(msg, file=sys.stderr)
+    print(
+        "\nFetch the archives with:  scripts/fetch_mbedtls_4k.sh <arch>"
+        "\nOr build without them by unsetting MESHCORE_REDUCED_TLS.",
+        file=sys.stderr,
+    )
+    env.Exit(1)
+
+
+def _sha256(path):
+    h = hashlib.sha256()
+    with open(path, "rb") as fh:
+        for chunk in iter(lambda: fh.read(1 << 20), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def _manifest(project_dir, arch):
+    path = os.path.join(project_dir, "scripts", "mbedtls_4k_manifest.txt")
+    if not os.path.isfile(path):
+        _fail("missing scripts/mbedtls_4k_manifest.txt")
+    wanted = {}
+    stock = {}
+    platform_id = ""
+    with open(path) as fh:
+        for line in fh:
+            line = line.strip()
+            if not line or line.startswith("#"):
+                continue
+            parts = line.split()
+            if len(parts) == 2 and parts[0] == "platform":
+                platform_id = parts[1]
+            elif len(parts) == 3 and parts[0] == "stock:" + arch:
+                stock[parts[2]] = parts[1]
+            elif len(parts) == 3 and parts[0] == arch:
+                wanted[parts[2]] = parts[1]
+    return wanted, stock, platform_id
+
+
+# Where each espressif32 generation keeps the archives we displace. First directory
+# holding all four wins, so this resolves without knowing which platform is in play.
+FRAMEWORK_LIB_DIRS = (
+    ("framework-arduinoespressif32", "tools/sdk/%s/lib"),
+    ("framework-arduinoespressif32-libs", "%s/lib"),
+    ("framework-arduinoespressif32", "tools/esp32-arduino-libs/%s/lib"),
+)
+
+
+def _framework_lib_dir(platform, arch):
+    for package, layout in FRAMEWORK_LIB_DIRS:
+        try:
+            base = platform.get_package_dir(package)
+        except Exception:
+            base = None
+        if not base:
+            continue
+        path = os.path.join(base, *(layout % arch).split("/"))
+        if all(os.path.isfile(os.path.join(path, name)) for name in REQUIRED):
+            return path
+    return None
+
+
+if os.environ.get("MESHCORE_REDUCED_TLS", "") not in ("1", "true", "yes"):
+    Return()
+
+project_dir = env.subst("$PROJECT_DIR")
+arch = env.BoardConfig().get("build.mcu", "")
+if not arch:
+    _fail("could not determine board MCU, so cannot pick an archive set")
+
+staged = os.path.join(project_dir, ".mbedtls-4k", arch)
+if not os.path.isdir(staged):
+    _fail("no archives for %s at %s" % (arch, staged))
+
+wanted, stock, manifest_platform = _manifest(project_dir, arch)
+if not wanted:
+    _fail("manifest has no entries for arch '%s'" % arch)
+
+for name in REQUIRED:
+    archive = os.path.join(staged, name)
+    if not os.path.isfile(archive):
+        _fail("missing %s" % archive)
+    if name not in wanted:
+        _fail("%s is not in the manifest for %s" % (name, arch))
+    actual = _sha256(archive)
+    if actual != wanted[name]:
+        _fail(
+            "%s does not match the manifest\n  expected %s\n  actual   %s\n"
+            "Rebuild it for this platform version, or re-run the fetch script."
+            % (archive, wanted[name], actual)
+        )
+
+# Bind the staged archives to the framework they were built against. The hashes above
+# only prove the staged files are the ones the manifest names; they say nothing about
+# whether the manifest is still current. Bump the platform without rebuilding, and
+# every check above still passes while the link takes mbedTLS built against a
+# different IDF — a struct-layout drift that corrupts silently at runtime. So
+# fingerprint the framework's own archives, the ones being displaced: if those moved,
+# the staged pair is stale by construction.
+platform = env.PioPlatform()
+platform_id = "%s@%s" % (platform.name, platform.version)
+stock_dir = _framework_lib_dir(platform, arch)
+if stock_dir is None:
+    _fail("cannot locate the framework's own mbedTLS archives for %s, so the staged "
+          "ones cannot be tied to a framework version" % arch)
+if not stock:
+    _fail("manifest has no stock:%s fingerprints — it predates the framework binding.\n"
+          "Add these lines for the framework now installed (%s):\n%s"
+          % (arch, platform_id,
+             "\n".join("stock:%s %s %s" % (arch, _sha256(os.path.join(stock_dir, n)), n)
+                       for n in REQUIRED)))
+
+for name in REQUIRED:
+    actual = _sha256(os.path.join(stock_dir, name))
+    if name not in stock:
+        _fail("manifest has no stock:%s entry for %s" % (arch, name))
+    if actual != stock[name]:
+        _fail(
+            "the framework's mbedTLS archives are not the ones these were built against.\n"
+            "  %s\n  manifest %s\n  installed %s\n"
+            "Manifest records %s; installed is %s.\n"
+            "Rebuild the reduced-TLS archives against this framework "
+            "(docs/mbedtls-tls-footprint.md), then update every hash in the manifest."
+            % (os.path.join(stock_dir, name), stock[name], actual,
+               manifest_platform or "no platform", platform_id)
+        )
+
+if manifest_platform and manifest_platform != platform_id:
+    # Hashes are the check; the version string is orientation. Identical archives
+    # under a renamed platform are not a compatibility problem.
+    print("reduced-TLS: manifest says %s, installed is %s — archives match, so this is "
+          "only a stale label" % (manifest_platform, platform_id))
+
+# Prepend so these satisfy mbedTLS symbols ahead of the framework's own copies:
+# the linker takes each archive member from the first archive that resolves it.
+env.Prepend(LIBPATH=[staged])
+print("reduced-TLS: linking mbedTLS from %s (verified)" % staged)
+
+
+def _verify_map(source, target, env):
+    """Confirm every mbedTLS archive in the link came from our directory.
+
+    Fails closed. Anything that stops this from *proving* the link — no map, an
+    unparsable map, a short archive list — is a failure, not a warning. A warning
+    here would leave exactly the hole the check exists to close: an opt-in build
+    that succeeds while silently linking the framework's 16 KiB buffers.
+    """
+    # Derive the map name from PROGNAME rather than hardcoding firmware.map, so a
+    # renamed program cannot leave us inspecting a stale or absent file.
+    map_path = os.path.join(env.subst("$BUILD_DIR"),
+                            env.subst("${PROGNAME}") + ".map")
+    if not os.path.isfile(map_path):
+        legacy = os.path.join(env.subst("$BUILD_DIR"), "firmware.map")
+        map_path = legacy if os.path.isfile(legacy) else map_path
+    if not os.path.isfile(map_path):
+        print("\n*** reduced-TLS: no linker map at %s ***" % map_path,
+              file=sys.stderr)
+        print("Cannot prove the reduced-TLS archives were linked. Ensure the env "
+              "emits a map (-Wl,-Map), or unset MESHCORE_REDUCED_TLS.",
+              file=sys.stderr)
+        env.Exit(1)
+        return
+    # The map records whatever the linker was given, which for a -L hit is a path
+    # relative to the linker's cwd (the project dir). Resolve before comparing, or
+    # every one of our own archives reads as stray.
+    staged_real = os.path.realpath(staged)
+    stray = set()
+    seen = set()
+    with open(map_path, errors="replace") as fh:
+        for line in fh:
+            for token in line.split():
+                if "libmbed" not in token or ".a" not in token:
+                    continue
+                path = token.split("(")[0]
+                base = os.path.basename(path)
+                if not base.startswith("libmbed") or not base.endswith(".a"):
+                    continue
+                seen.add(base)
+                resolved = os.path.realpath(os.path.join(project_dir, path))
+                if os.path.dirname(resolved) != staged_real:
+                    stray.add(path)
+    if stray:
+        print("\n*** reduced-TLS: archives linked from the WRONG place ***",
+              file=sys.stderr)
+        for path in sorted(stray):
+            print("  " + path, file=sys.stderr)
+        env.Exit(1)
+        return
+    # Every required archive must appear. Seeing only some of them means the rest
+    # resolved somewhere this parse did not recognise, which is not proof of anything.
+    missing = [name for name in REQUIRED if name not in seen]
+    if missing:
+        print("\n*** reduced-TLS: %s names only %d of %d archives ***"
+              % (os.path.basename(map_path), len(seen), len(REQUIRED)),
+              file=sys.stderr)
+        print("  missing: " + ", ".join(missing), file=sys.stderr)
+        print("Either the map format changed or mbedTLS was resolved elsewhere; "
+              "the reduced buffers cannot be assumed.", file=sys.stderr)
+        env.Exit(1)
+        return
+    print("reduced-TLS: confirmed all %d archives linked from %s"
+          % (len(REQUIRED), staged))
+
+
+env.AddPostAction("$BUILD_DIR/${PROGNAME}.elf", _verify_map)

scripts/mbedtls_4k_manifest.txt

@@ -0,0 +1,23 @@
+# Reduced-TLS mbedTLS archives (OUT_CONTENT_LEN 4096, IN_CONTENT_LEN 16384).
+# Format: <arch> <sha256> <filename>
+#
+# Rebuild these for every espressif32 platform bump — the archives must match the
+# rest of the framework they link against. Built as of platformio/espressif32@6.11.0
+# from the recipe in docs/mbedtls-tls-footprint.md.
+#
+# Fetch with: scripts/fetch_mbedtls_4k.sh esp32s3
+esp32s3 01629f635b33ffa2c1fdfcd8ac52327cd3a92e8d4e7c9a32d92974e0cfdfe398 libmbedcrypto.a
+esp32s3 07e7a09847589fefc35bc8d2f739d556535d535acd7185a9824b2af8edd5b05a libmbedtls_2.a
+esp32s3 e12bcc8d76a368e819987f266e73c265178d6b6675f05d0e29014bb543c402af libmbedtls.a
+esp32s3 c1e10324e19f6d7763737f72d9032cec832bdfce60c7a7d381a5fcfc7dcad573 libmbedx509.a
+
+# Framework binding, read by scripts/mbedtls_4k.py and ignored by the fetch script.
+# "stock:" lines fingerprint the framework's own archives — the ones the entries above
+# displace. They are what makes "rebuild on every platform bump" enforceable instead of
+# advisory: if the framework's copies move, the entries above are stale by construction
+# and the build stops. Regenerate both sets together, never one alone.
+platform espressif32@6.11.0
+stock:esp32s3 abdaf759ee17aa697468427b55a86c6c0082ac4cdeb643a63d8b3ac2df324633 libmbedcrypto.a
+stock:esp32s3 9a580d2ff7c885e1bf59479a14422e3d10a485ae9083acb5db943df714c0ac35 libmbedtls_2.a
+stock:esp32s3 a0f331245feb5cf4fe8d9f99b03d34d64f10e38d8a8d680e35b34ec82aa3cf9e libmbedtls.a
+stock:esp32s3 3d56277224b066118b6c48a973ae18fdf8078448bf54b482ac5557699e52f223 libmbedx509.a

src/helpers/ESP32WsTransportFix.cpp

@@ -33,13 +33,24 @@
 // Fix: [esp32_base] adds `-Wl,--wrap=esp_transport_ws_init`, so every
 // creation of a WS transport (esp-mqtt does one per wss slot) is routed
 // through __wrap_esp_transport_ws_init below, which replaces the freshly
-// allocated 1024-byte buffer with a (WS_BUFFER_SIZE + 1)-byte one. The
-// out-of-bounds index WS_BUFFER_SIZE then lands on our extra byte and the
-// handshake fails cleanly ("Upgrade" header not found) instead of corrupting
-// the heap. Upstream fixed this in ESP-IDF 5.x, so this file compiles to a
+// allocated 1024-byte buffer with a (WS_BUFFER_SIZE + 1)-byte one, so the
+// out-of-bounds index WS_BUFFER_SIZE lands on our extra byte instead of the
+// heap canary. Upstream fixed this in ESP-IDF 5.x, so this file compiles to a
 // pass-through there and can be deleted (together with the --wrap flag) when
 // the fork moves to Arduino core 3.x.
 //
+// Scope: this stops the heap corruption and NOTHING else. It does not make an
+// oversized response fail the handshake — v4.4's read loop also exits on
+// `header_len < WS_BUFFER_SIZE` going false, and then still accepts the
+// connection if "Sec-WebSocket-Accept:" was inside those first bytes (it comes
+// early, so it usually is). ws_connect() therefore returns success on a partial
+// header block and the unread remainder arrives as the first "payload" read,
+// where the deframer parses HTTP bytes as a frame header. Observed on hardware
+// 2026-08-12: a large Cloudflare CSP header produced exactly that, surfacing as
+// `Invalid MSG_TYPE response: 3`. Only IDF 5.2+ fixes it (it requires the
+// "\r\n\r\n" delimiter, preserves the bytes after it, and fails cleanly when the
+// buffer fills); it cannot be patched here because transport_ws.c is precompiled.
+//
 // transport_ws_t below is copied verbatim from ESP-IDF release/v4.4
 // transport_ws.c (the struct is file-private, so it is not in any shipped
 // header). Source fidelity was verified against the shipped binary: addr2line

src/helpers/MQTTConnectionPolicy.h

@@ -17,6 +17,7 @@ static const uint8_t kMaxFailuresAtMaxBackoff = 3;
 static const uint32_t kDefaultJwtLifetimeSecs = 86400UL;
 static const uint32_t kMaxJwtStaggerSecs = 300UL;
 static const uint32_t kMinimumValidEpoch = 1000000000UL;
+static const uint32_t kJwtReconnectSafetyMarginSecs = 60UL;
 static const uint32_t kJwtClockThreshold = 1735689600UL; // 2025-01-01 UTC
 // A wall clock at or past this instant was set from a real source (NTP or an
 // admin); anything earlier is the firmware's unset-clock default (1715770351,
@@ -157,6 +158,20 @@ static inline bool tokenNeedsRenewal(bool time_synced, uint32_t current_time,
   return current_time >= token_expires_at - renewal_buffer_secs;
 }
 
+// A reconnect may keep its credentials only when their validity is known to
+// outlast the next handshake; uncertainty refreshes them before reconnecting.
+static inline bool canReuseJwtForReconnect(bool time_synced, bool has_token,
+                                           bool force_mint,
+                                           uint32_t current_time,
+                                           uint32_t token_expires_at) {
+  return time_synced &&
+         has_token &&
+         !force_mint &&
+         token_expires_at >= kMinimumValidEpoch &&
+         current_time < token_expires_at &&
+         (token_expires_at - current_time) > kJwtReconnectSafetyMarginSecs;
+}
+
 static inline bool renewalAttemptAllowed(uint32_t now, uint32_t last_attempt) {
   return elapsedMs(now, last_attempt) >= kRenewalThrottleMs;
 }
@@ -196,4 +211,48 @@ static inline SlotActivation classifySlotActivation(int slot, const bool* enable
                               : SlotActivation::OverActiveCap;
 }
 
+// What to do with a slot whose token a clock correction just proved stale. Four
+// outcomes, because esp-mqtt accepts a reconnect request only from
+// MQTT_STATE_WAIT_RECONNECT: asking a live client to reconnect is refused and
+// leaves it running on the stale token, so a live session has to have its
+// transport closed first — and that costs a handshake, which is only worth
+// spending where the broker actually enforces exp.
+enum class StaleTokenAction : uint8_t {
+  Defer,      // no fresh credentials — leave the slot to the backoff ladder
+  Reconnect,  // client is down: start it, or wake one that is waiting
+  Bounce,     // live session the broker will reject: close the transport, then reconnect
+  KeepAlive,  // live session the broker tolerates: stage credentials, keep the handshake
+};
+
+// A failed mint yields Defer even when the slot is down: reconnecting then would
+// re-present the credentials the correction just invalidated. Minting fails for
+// recoverable reasons (allocation pressure), and the ladder retries.
+static inline StaleTokenAction classifyStaleToken(bool minted, bool connected,
+                                                  bool broker_enforces_exp) {
+  if (!minted) return StaleTokenAction::Defer;
+  if (!connected) return StaleTokenAction::Reconnect;
+  return broker_enforces_exp ? StaleTokenAction::Bounce : StaleTokenAction::KeepAlive;
+}
+
+// Which clock to fall back on when no NTP server answered.
+enum class ClockSource : uint8_t {
+  None,    // nothing plausible to work from — stay unsynced
+  System,  // libc already holds a usable time
+  Rtc,     // libc does not, but the RTC does
+};
+
+// System first: a clock SNTP set recently outranks an RTC that may have drifted.
+// The RTC matters on a cold boot, where ESP32RTCClock::begin() seeds libc with a 2024
+// placeholder on power-on while a detected chip already holds real time and
+// AutoDiscoverRTCClock::begin() never copies one into the other. Never while
+// validating a server: that asks whether a specific host answers, and no clock can
+// answer it. Pass rtc_time 0 when the board has no clock to consult.
+static inline ClockSource chooseFallbackClock(bool validating_server, uint32_t system_time,
+                                              uint32_t rtc_time, uint32_t min_valid_epoch) {
+  if (validating_server) return ClockSource::None;
+  if (system_time >= min_valid_epoch) return ClockSource::System;
+  if (rtc_time >= min_valid_epoch) return ClockSource::Rtc;
+  return ClockSource::None;
+}
+
 } // namespace MQTTConnectionPolicy

src/helpers/MQTTPresets.h

@@ -47,6 +47,24 @@ struct MQTTPresetDef {
 // Braces match topic placeholders ({device}/{iata}); never send this string to the broker.
 static const char MQTT_USERPASS_USERNAME_PUBKEY[] = "{pubkey}";
 
+// True when the broker tears down a live session once its JWT passes exp, so the
+// renewal must proactively bounce the connection to present a fresh token.
+//
+// Default true, because getting this wrong the safe way costs a re-handshake and
+// getting it wrong the unsafe way costs an outage. waev is the exception: its
+// operator confirmed (2026-08-11) that their servers do not disconnect on expiry,
+// so a live session there needs only its credentials refreshed for the next
+// reconnect. waev is also the only preset with a short token_lifetime, so it was
+// the only one bouncing often — every ~47 min, and each bounce's re-handshake can
+// cost ~10 KB of contiguous internal DRAM on a non-PSRAM board.
+//
+// Keyed by name rather than a struct field on purpose: adding a field would mean
+// re-ordering a dozen positional initialisers below, where a mistake is silent.
+static inline bool mqttPresetEnforcesTokenExp(const MQTTPresetDef* preset) {
+  if (!preset || !preset->name) return true;   // custom/audience slots: assume enforced
+  return strcmp(preset->name, "waev") != 0;
+}
+
 static inline bool mqttPresetUsesDevicePubkeyUsername(const MQTTPresetDef* preset) {
   return preset && preset->auth_type == MQTT_AUTH_USERPASS &&
          preset->userpass_username &&

src/helpers/bridges/MQTTBridge.cpp

@@ -21,6 +21,7 @@
 
 #ifdef ESP_PLATFORM
 #include <esp_wifi.h>
+#include <esp_sntp.h>
 #include <esp_heap_caps.h>
 #include <freertos/FreeRTOS.h>
 #include <freertos/task.h>
@@ -704,6 +705,7 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg
     _slots[i].last_log_time = 0;
     _slots[i].port = 1883;
     _slot_reconfigure_pending[i] = false;
+    _slot_force_jwt_mint[i] = false;
     _status_publish_pending[i] = false;
   }
 
@@ -768,17 +770,10 @@ void MQTTBridge::allocateRuntimeBuffers() {
       _json_scratch_buffer ? "PSRAM" : "stack fallback");
   #endif
 
-#if defined(WITH_MQTT_NEIGHBORS)
-  // Persistent neighbors JSON buffer, heap-allocated on every board: too large to
-  // keep inline in the bridge object the way the non-PSRAM status/packet buffers
-  // are. psram_malloc() falls back to internal DRAM, so this works without PSRAM.
-  // Unlike status/packet there is no stack fallback — a nullptr simply disables
-  // publishing (requestPublishNeighbors/publishNeighbors both no-op on nullptr).
-  _neighbors_json_buffer = static_cast<char*>(MQTTRuntimeBufferLifecycle::allocateIfMissing(
-      _neighbors_json_buffer, NEIGHBORS_JSON_BUFFER_SIZE, psram_malloc));
-  MQTT_DEBUG_PRINTLN("Neighbors buffer: %s",
-      _neighbors_json_buffer ? "ready" : "unavailable");
-#endif
+  // The neighbors JSON buffer is NOT allocated here — requestPublishNeighbors()
+  // allocates it on first use, so a node with mqtt.neighbors off never pays its
+  // 4 KB. mqtt.neighbors is read live with no bridge restart, so gating on the
+  // pref here would leave a runtime enable with no buffer.
 }
 
 void MQTTBridge::releaseRuntimeBuffers() {
@@ -795,7 +790,7 @@ void MQTTBridge::releaseRuntimeBuffers() {
   _json_scratch_doc.clear();
 
 #if defined(WITH_MQTT_NEIGHBORS)
-  // Paired with the unconditional allocation in allocateRuntimeBuffers().
+  // Paired with the lazy allocation in requestPublishNeighbors(); no-op if never used.
   _neighbors_json_buffer = static_cast<char*>(MQTTRuntimeBufferLifecycle::release(
       _neighbors_json_buffer, psram_free));
   _neighbors_publish_len = 0;
@@ -1608,6 +1603,7 @@ bool MQTTBridge::ensureSlotClient(int index) {
   slot.client->onConnect([this, index](bool sessionPresent) {
     MQTT_DEBUG_PRINTLN("MQTT%d connected", index + 1);
     _slots[index].connected = true;
+    _slot_force_jwt_mint[index] = false;
     // NOTE: reconnect_backoff / max_backoff_failures are NOT reset here.
     // A CONNACK alone doesn't prove the link is healthy — a broker that
     // accepts and then drops within seconds would reset the ladder every
@@ -1654,6 +1650,7 @@ bool MQTTBridge::ensureSlotClient(int index) {
     _slots[index].last_sock_errno = error.esp_transport_sock_errno;
     _slots[index].last_error_time = millis();
     if (error.error_type == MQTT_ERROR_TYPE_CONNECTION_REFUSED) {
+      _slot_force_jwt_mint[index] = true;
       // Broker rejected the MQTT CONNECT itself — not a transport failure.
       // return code: 1=protocol, 2=client-id rejected, 3=server unavailable,
       // 4=bad username/password, 5=not authorized. Codes 3/4/5 point at a
@@ -1775,9 +1772,11 @@ bool MQTTBridge::setupSlot(int index) {
   }
 
   // Reconfigure path: if we're re-applying (e.g. after a preset change), stop
-  // the existing connection cleanly first. The client object (and its mbedTLS
-  // context) is reused; setCredentials / setServer below overwrite the config
-  // fields in place before connect() restarts the ESP-IDF client.
+  // the existing connection cleanly first. The client object is reused, but its
+  // mbedTLS context is NOT — closing the transport destroys the TLS session,
+  // record buffers, and peer certificate, and the next connect() reallocates
+  // them. setCredentials / setServer below overwrite the config fields in place
+  // before connect() restarts the ESP-IDF client.
   if (slot.initial_connect_done) {
     if (slot.client->connected()) {
       slot.client->disconnect();
@@ -1806,6 +1805,8 @@ bool MQTTBridge::setupSlot(int index) {
     slot.max_backoff_failures = 0;
     slot.circuit_breaker_tripped = false;
     slot.last_reconnect_attempt = 0;
+    // The refusal that set this belonged to the credentials being cleared here.
+    _slot_force_jwt_mint[index] = false;
   }
 
   bool uses_jwt = (slot.preset && slot.preset->auth_type == MQTT_AUTH_JWT) || slot.audience[0] != '\0';
@@ -1995,8 +1996,27 @@ void MQTTBridge::teardownSlot(int index) {
   slot.last_reconnect_attempt = 0;
   slot.last_log_time = 0;
   slot.last_deferred_log_ms = 0;
+  // The refusal that set this belonged to the credentials being cleared here.
+  _slot_force_jwt_mint[index] = false;
 }
 
+// A stopped client needs connect(): reconnect() is a documented no-op on one, so reaching
+// it here would strand the slot. The WiFi-transition teardown stops a client while leaving
+// initial_connect_done set, so the ladder does see this state.
+void MQTTBridge::reconnectSlotClient(int index) {
+  if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return;
+  MQTTSlot& slot = _slots[index];
+  if (slot.client == nullptr) return;
+
+  if (!slot.client->isStarted()) {
+    MQTT_DEBUG_PRINTLN("MQTT%d start (client was stopped)", index + 1);
+    slot.client->connect();
+    return;
+  }
+  slot.client->reconnect();
+}
+
+
 void MQTTBridge::maintainSlotConnections() {
   if (!_identity) return;
 
@@ -2136,15 +2156,37 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
                                             (time_synced && old_token_expires_at >= 1000000000 &&
                                              current_time >= (old_token_expires_at - renewal_buffer));
 
-        if (old_token_expired_or_imminent || !slot.client->connected()) {
+        // Only bounce for exp if this broker actually enforces it. A broker that
+        // leaves live sessions alone past expiry needs the fresh token at the next
+        // reconnect, not now, and the bounce's re-handshake is where contiguity goes.
+        const bool exp_forces_bounce =
+            old_token_expired_or_imminent && mqttPresetEnforcesTokenExp(slot.preset);
+        if (!exp_forces_bounce && old_token_expired_or_imminent && slot.client->connected()) {
+          MQTT_DEBUG_PRINTLN("MQTT%d token renewed, no bounce (broker does not enforce exp)",
+              index + 1);
+        }
+        if (exp_forces_bounce || !slot.client->connected()) {
           // Disconnect + reconnect with fresh credentials, reusing existing client
           // to avoid internal heap leak/fragmentation from destroy/create cycles
           MQTT_DEBUG_PRINTLN("MQTT%d token renewal: reconnecting with fresh credentials", index + 1);
-          if (slot.client->connected()) {
-            slot.client->disconnect();  // stops the client internally
+          MQTT_TRACE_HEAP("renewal:before-bounce", index);
+          if (slot.client->isStarted()) {
+            // Keep the esp-mqtt task alive across the handshake. disconnect()
+            // would stop it, returning its 6 KiB stack into the hole the two
+            // 16 KiB mbedTLS record buffers just vacated — which is what walks
+            // the largest free block down 16 KiB at a time on non-PSRAM boards.
+            slot.client->softDisconnect();
+            MQTT_TRACE_HEAP("renewal:after-disconnect", index);
+            slot.client->setCredentials(_jwt_username, slot.auth_token);
+            MQTT_TRACE_HEAP("renewal:after-credentials", index);
+            slot.client->reconnect();
+          } else {
+            // Client was stopped (teardown/reconfigure). reconnect() is a no-op
+            // on a stopped client, so this path must start it.
+            slot.client->setCredentials(_jwt_username, slot.auth_token);
+            slot.client->connect();
           }
-          slot.client->setCredentials(_jwt_username, slot.auth_token);
-          slot.client->connect();  // restart stopped client; reconnect() fails silently on a stopped client
+          MQTT_TRACE_HEAP("renewal:after-reconnect", index);
           reconnect_attempted = true;
           _last_slot_reconnect_ms = now_millis;
           MQTT_DEBUG_PRINTLN("MQTT%d int_heap=%d at token renewal reconnect", index + 1,
@@ -2169,6 +2211,59 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
   // persistent clients (Phase 1), the mbedTLS context is allocated once at
   // startup and the preflight is no longer necessary.
 
+  const auto prepareJwtReconnect = [&](bool force_mint, int backoff_level) {
+    const bool has_token = slot.auth_token && slot.auth_token[0] != '\0';
+    const unsigned long expires_at = slot.token_expires_at;
+    const bool remaining_known = time_synced &&
+        expires_at >= MQTTConnectionPolicy::kMinimumValidEpoch;
+    const unsigned long remaining_secs = current_time < expires_at
+        ? expires_at - current_time
+        : 0;
+    const bool force_mint_after_refusal = _slot_force_jwt_mint[index];
+    force_mint = force_mint || force_mint_after_refusal;
+    const bool reuse_token = MQTTConnectionPolicy::canReuseJwtForReconnect(
+        time_synced, has_token, force_mint, static_cast<uint32_t>(current_time),
+        static_cast<uint32_t>(expires_at));
+    const char* mint_reason = "none";
+    if (!reuse_token) {
+      if (backoff_level < 0) {
+        mint_reason = "circuit-breaker-probe";
+      } else if (force_mint_after_refusal) {
+        mint_reason = "connection-refused";
+      } else if (!time_synced) {
+        mint_reason = "clock-unsynced";
+      } else if (!has_token) {
+        mint_reason = "empty-token";
+      } else if (expires_at < MQTTConnectionPolicy::kMinimumValidEpoch) {
+        mint_reason = "invalid-expiry";
+      } else if (current_time >= expires_at) {
+        mint_reason = "expired";
+      } else {
+        mint_reason = "safety-margin";
+      }
+    }
+    const char* mint_result = reuse_token ? "REUSED" : "FAILED";
+    if (!reuse_token && createSlotAuthToken(index)) {
+      slot.client->setCredentials(_jwt_username, slot.auth_token);
+      mint_result = "OK";
+    }
+    char remaining_text[24];
+    if (remaining_known) {
+      snprintf(remaining_text, sizeof(remaining_text), "%lus", remaining_secs);
+    } else {
+      strncpy(remaining_text, "unknown", sizeof(remaining_text));
+    }
+    if (backoff_level >= 0) {
+      MQTT_DEBUG_PRINTLN("MQTT%d JWT reconnect backoff=%d token=%s mint_reason=%s result=%s remaining=%s",
+          index + 1, backoff_level, reuse_token ? "REUSE" : "MINT", mint_reason,
+          mint_result, remaining_text);
+    } else {
+      MQTT_DEBUG_PRINTLN("MQTT%d JWT circuit-breaker probe token=%s mint_reason=%s result=%s remaining=%s",
+          index + 1, reuse_token ? "REUSE" : "MINT", mint_reason, mint_result,
+          remaining_text);
+    }
+  };
+
   // Periodic probe for circuit-breaker-tripped slots (recovery from transient outages)
   // Attempts a single reconnect every 30 minutes to see if the server has come back
   if (slot.circuit_breaker_tripped && !reconnect_attempted) {
@@ -2185,17 +2280,11 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
           _radio ? _radio->getRadioState() : -1,
           (_radio && _radio->getLastRecvMillis() > 0) ? (_ms->getMillis() - _radio->getLastRecvMillis()) : 0);
       if (slot_uses_jwt) {
-        // Regenerate or refresh token, then reconnect the persistent client.
-        // Reaching the ladder at all means setupSlot() ran, so the client object
-        // and its mbedTLS context are live and no full setup is needed here.
-        if (createSlotAuthToken(index)) {
-          slot.client->setCredentials(_jwt_username, slot.auth_token);
-          MQTT_DEBUG_PRINTLN("MQTT%d circuit breaker probe (fresh token)", index + 1);
-        }
-        slot.client->reconnect();
-      } else {
-        slot.client->reconnect();
+        prepareJwtReconnect(true, -1);
       }
+      // Via the helper: reconnect() is a no-op on a client the WiFi-drop path
+      // stopped, which would probe forever without ever starting it.
+      reconnectSlotClient(index);
       // If the connect callback fires and sets slot.connected = true,
       // it will clear circuit_breaker_tripped via the onConnect handler
     }
@@ -2225,22 +2314,14 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
       reconnect_attempted = true;
       _last_slot_reconnect_ms = now_millis;
       if (slot_uses_jwt) {
-        // Always lightweight reconnect on the persistent client. A stale/expired
-        // token is handled by regenerating it in place and updating credentials
-        // — no teardown is needed because the client and its mbedTLS context
-        // persist for the bridge lifetime.
-        if (createSlotAuthToken(index)) {
-          slot.client->setCredentials(_jwt_username, slot.auth_token);
-          MQTT_DEBUG_PRINTLN("MQTT%d reconnect (fresh token, backoff %d)", index + 1, slot.reconnect_backoff);
-        } else {
-          MQTT_DEBUG_PRINTLN("MQTT%d reconnect (token refresh failed, backoff %d)", index + 1, slot.reconnect_backoff);
-        }
-        slot.client->reconnect();
+        prepareJwtReconnect(false, slot.reconnect_backoff);
       } else {
         // Non-JWT slots — lightweight reconnect on existing client.
         MQTT_DEBUG_PRINTLN("MQTT%d reconnect (non-JWT, backoff %d)", index + 1, slot.reconnect_backoff);
-        slot.client->reconnect();
       }
+      // Via the helper: reconnect() is a no-op on a client the WiFi-drop path
+      // stopped, which would back off forever without ever starting it.
+      reconnectSlotClient(index);
     }
   }
 }
@@ -3625,10 +3706,24 @@ void MQTTBridge::setNeighborsSchedule(NeighborsPhase phase, uint32_t secs_until_
 }
 
 void MQTTBridge::requestPublishNeighbors(const char* json, size_t len) {
-  if (!_neighbors_json_buffer || !json || len == 0) return;
+  if (!json || len == 0) return;
   // Drop a new snapshot while one is still being published (Core 0 clears the
   // flag when done). Acquire pairs with the task loop's release store.
   if (_neighbors_publish_pending.load(std::memory_order_acquire)) return;
+  // Allocating here means a stopped bridge must not: a discovery started before the
+  // stop can finish after it, and releaseRuntimeBuffers() has already run, so the
+  // allocation would be retained with no task left to consume it. isRunning() is the
+  // same flag end() guards on.
+  if (!isRunning()) return;
+  // Allocated on first use so a node with neighbors off never pays the 4 KB.
+  // Cross-core safe: the release store below publishes this pointer, and the task
+  // loop only reads it after the matching acquire load.
+  _neighbors_json_buffer = static_cast<char*>(MQTTRuntimeBufferLifecycle::allocateIfMissing(
+      _neighbors_json_buffer, NEIGHBORS_JSON_BUFFER_SIZE, psram_malloc));
+  if (!_neighbors_json_buffer) {
+    MQTT_DEBUG_PRINTLN("Neighbors buffer unavailable, dropping snapshot");
+    return;
+  }
   if (len >= NEIGHBORS_JSON_BUFFER_SIZE) {
     len = NEIGHBORS_JSON_BUFFER_SIZE - 1;
   }
@@ -3859,15 +3954,25 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) {
   const int kMaxNtpRetriesPerServer = 2;
   for (int s = 0; s < server_count && !ntp_ok; s++) {
     const char* server = servers[s];
-    _ntp_client.setPoolServerName(server);
 
     #ifdef ESP_PLATFORM
+    // Authoritative, not advisory. NTPClient::sendNTPPacket() ignores what
+    // beginPacket() returns, and WiFiUDP leaves remote_ip/remote_port at the previous
+    // destination when a name fails to resolve — so asking an unresolvable host sends
+    // the request to whichever server resolved last, and that server's genuine reply
+    // gets credited to this name. Observed on d4: `set mqtt.ntp bogus.invalid` reported
+    // success with a correct epoch, answered by the pool address left over from boot.
+    // Skipping is what keeps the credit honest; the name that answered is the name
+    // recorded.
     IPAddress resolved_ip;
     if (!WiFi.hostByName(server, resolved_ip)) {
-      MQTT_DEBUG_PRINTLN("WARNING: DNS resolution failed for %s - NTP sync may fail", server);
+      MQTT_DEBUG_PRINTLN("NTP: %s does not resolve — skipping, not attempting a send", server);
+      continue;
     }
     #endif
 
+    _ntp_client.setPoolServerName(server);
+
     for (int attempt = 1; attempt <= kMaxNtpRetriesPerServer && !ntp_ok; attempt++) {
       if (attempt > 1) {
         MQTT_DEBUG_PRINTLN("NTP retry %d/%d on %s...", attempt, kMaxNtpRetriesPerServer, server);
@@ -3891,23 +3996,84 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) {
     for (int s = 0; s < server_count && !ntp_ok; s++) {
       const char* server = servers[s];
       MQTT_DEBUG_PRINTLN("SNTP fallback trying %s...", server);
+      // A plausible clock is not evidence this server answered. The device usually
+      // already holds valid time here — from an earlier sync, or the RTC — so polling
+      // time(nullptr) declared the very first server successful without a packet ever
+      // arriving, stopped the fallback walk there, and refreshed _last_ntp_sync. Worse
+      // on the `set mqtt.ntp` validation path, where a typo is supposed to fail fast.
+      // Wait for SNTP itself to report completion. The status is one-shot — reading
+      // COMPLETED clears it — so drop any result an earlier sync left behind, and do
+      // that *before* starting this one: configTime() returns after sntp_init(), so a
+      // fast reply can complete inside it, and clearing afterwards would erase the
+      // very result being waited for.
+      if (sntp_enabled()) {
+        sntp_stop();
+      }
+      sntp_set_sync_status(SNTP_SYNC_STATUS_RESET);
       configTime(0, 0, server);
       for (int i = 0; i < 20; i++) {
         delay(500);
+        if (sntp_get_sync_status() != SNTP_SYNC_STATUS_COMPLETED) continue;
         epochTime = (unsigned long)time(nullptr);
         if (epochTime >= kMinValidEpoch) {
           ntp_ok = true;
           ntp_server_used = server;
           MQTT_DEBUG_PRINTLN("SNTP fallback succeeded on %s: %lu", server, epochTime);
-          break;
+        } else {
+          MQTT_DEBUG_PRINTLN("SNTP fallback: %s synced an implausible epoch %lu", server, epochTime);
         }
+        break;
       }
     }
   }
   #endif
 
-  if (ntp_ok && ntp_server_used) {
-    configTime(0, 0, ntp_server_used);
+  // No server answered, but the clock itself may still be usable. Requiring a real
+  // SNTP completion above removed something the plausible-clock test was doing by
+  // accident: an RTC-backed device on a network that blocks NTP (UDP/123) while
+  // allowing the broker (443) stayed synced and kept minting JWTs. _ntp_synced gates
+  // slot setup outright, so losing that strands those deployments with no slots at
+  // all. Keep the behaviour, but as its own decision rather than as a claim about a
+  // server that never replied. Not on the validation path — `set mqtt.ntp` asks
+  // whether that server works, and the clock cannot answer for it.
+  if (!ntp_ok) {
+    const unsigned long system_time = (unsigned long)time(nullptr);
+    // On a cold boot with a detected RTC chip these disagree: ESP32RTCClock::begin()
+    // stamps libc with a 2024 placeholder on power-on, AutoDiscoverRTCClock::begin()
+    // never copies the chip into it, and getCurrentTime() reads the chip. Asking libc
+    // alone would reject a board that knows exactly what time it is.
+    const unsigned long rtc_time = _rtc ? (unsigned long)_rtc->getCurrentTime() : 0;
+    const MQTTConnectionPolicy::ClockSource source = MQTTConnectionPolicy::chooseFallbackClock(
+        primary_only, (uint32_t)system_time, (uint32_t)rtc_time, (uint32_t)kMinValidEpoch);
+    if (source != MQTTConnectionPolicy::ClockSource::None) {
+      const bool from_rtc = (source == MQTTConnectionPolicy::ClockSource::Rtc);
+      epochTime = from_rtc ? rtc_time : system_time;
+      ntp_ok = true;
+      MQTT_DEBUG_PRINTLN("No NTP server answered; continuing on the existing %s: %lu",
+          from_rtc ? "RTC" : "system clock", epochTime);
+    }
+  }
+
+  if (ntp_ok) {
+    // Take ownership of the system clock here, before anything reads it. configTime()
+    // only restarts SNTP and returns, and _rtc reaches settimeofday() on exactly one
+    // path: AutoDiscoverRTCClock writes a detected DS3231/RV3028/PCF8563/RX8130CE chip
+    // *instead of* its fallback, so on any board carrying one, libc keeps the pre-sync
+    // time. Everything downstream reads time(nullptr) — the stale-token test below, and
+    // the iat of every JWT minted from here on — so once _ntp_synced is true that call
+    // has to already return the epoch we accepted.
+    struct timeval accepted;
+    accepted.tv_sec = (time_t)epochTime;
+    accepted.tv_usec = 0;
+    settimeofday(&accepted, nullptr);
+
+    // Only when a server supplied the accepted epoch. The fallback above necessarily
+    // points configTime() at each server before knowing whether it replies; this is
+    // the post-acceptance call, and there is nothing to re-point it at when the epoch
+    // came from a local clock.
+    if (ntp_server_used) {
+      configTime(0, 0, ntp_server_used);
+    }
 
     if (_rtc) {
       _rtc->setCurrentTime(epochTime);
@@ -3918,11 +4084,12 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) {
     _last_ntp_sync = millis();
     sync_in_progress = false;
 
-    MQTT_DEBUG_PRINTLN("Time synced: %lu (via %s)", epochTime, ntp_server_used);
+    MQTT_DEBUG_PRINTLN("Time synced: %lu (via %s)", epochTime,
+        ntp_server_used ? ntp_server_used : "existing clock");
 
     // If slots are already set up and the time jumped significantly (e.g., SNTP
-    // initially returned stale RTC time, then a later sync corrected it), tear down
-    // and re-setup all JWT-authenticated slots so they get fresh tokens.
+    // initially returned stale RTC time, then a later sync corrected it), re-issue
+    // credentials for every JWT slot the correction left holding an expired token.
     if (_slots_setup_done && was_ntp_synced) {
       unsigned long current_time = (unsigned long)time(nullptr);
       // Every slot, not _max_active_slots: that is a count of positions, never an
@@ -3939,10 +4106,29 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) {
           // in place and reconnect the persistent client. No teardown needed.
           if (_slots[i].token_expires_at > 0 && current_time > _slots[i].token_expires_at) {
             MQTT_DEBUG_PRINTLN("MQTT%d token stale after time correction, re-creating", i + 1);
-            if (createSlotAuthToken(i)) {
-              _slots[i].client->setCredentials(_jwt_username, _slots[i].auth_token);
+            const MQTTConnectionPolicy::StaleTokenAction action =
+                MQTTConnectionPolicy::classifyStaleToken(
+                    createSlotAuthToken(i), _slots[i].client->connected(),
+                    mqttPresetEnforcesTokenExp(_slots[i].preset));
+            if (action == MQTTConnectionPolicy::StaleTokenAction::Defer) {
+              MQTT_DEBUG_PRINTLN("MQTT%d token refresh failed after time correction, "
+                  "deferring to the reconnect ladder", i + 1);
+              continue;
+            }
+            // Staged only; the config is applied by connect()/reconnect() below.
+            _slots[i].client->setCredentials(_jwt_username, _slots[i].auth_token);
+            if (action == MQTTConnectionPolicy::StaleTokenAction::Reconnect) {
+              // Reuse the transport — the fault is stale credentials, not the transport —
+              // but via the helper, so a stopped client is started rather than no-opped.
+              reconnectSlotClient(i);
+            } else if (action == MQTTConnectionPolicy::StaleTokenAction::Bounce) {
+              MQTT_DEBUG_PRINTLN("MQTT%d bouncing for the corrected-clock token", i + 1);
+              _slots[i].client->softDisconnect();
+              _slots[i].client->reconnect();
+            } else {
+              MQTT_DEBUG_PRINTLN("MQTT%d token re-created, no bounce (broker does not enforce exp)",
+                  i + 1);
             }
-            _slots[i].client->reconnect();
           }
         }
       }

src/helpers/bridges/MQTTBridge.h

@@ -36,6 +36,19 @@ class MeshSNMPAgent;  // Forward declaration
   #define MQTT_DEBUG_PRINTLN(...) {}
 #endif
 
+// Largest-free-block trace around the reconnect lifecycle. On non-PSRAM boards the
+// mbedTLS record buffers are two 16 KiB internal-DRAM blocks, so what matters is the
+// largest contiguous block, not the free total — a soak can show flat free heap while
+// max_alloc walks down. Costs two heap_caps calls per reconnect, so it stays on.
+#if defined(MQTT_DEBUG) && defined(ARDUINO) && defined(ESP32)
+  #define MQTT_TRACE_HEAP(point, idx) \
+    MQTT_DEBUG_PRINTLN("HEAPTRACE slot=%d %s free=%u max=%u", (int)(idx) + 1, point, \
+        (unsigned)heap_caps_get_free_size(MALLOC_CAP_INTERNAL), \
+        (unsigned)heap_caps_get_largest_free_block(MALLOC_CAP_INTERNAL))
+#else
+  #define MQTT_TRACE_HEAP(point, idx) do {} while(0)
+#endif
+
 #ifdef WITH_MQTT_BRIDGE
 
 // Periodic neighbors publication keys off the mesh neighbor cache (sized by
@@ -215,6 +228,10 @@ private:
   // Pending slot reconfigure: set from CLI (Core 1), processed by MQTT task (Core 0)
   volatile bool _slot_reconfigure_pending[RUNTIME_MQTT_SLOTS];
 
+  // A broker refusal can invalidate an otherwise clock-valid JWT. The esp-mqtt
+  // callback sets this and the bridge loop consumes it; byte access is atomic.
+  volatile bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS];
+
   // Pending on-connect status publish: set from the onConnect callback (which
   // runs on the esp-mqtt event task, NOT this bridge task), consumed by the MQTT
   // task (Core 0). publishStatusToSlot() touches the shared status doc/buffer/
@@ -434,6 +451,9 @@ private:
   int activatedSlotCount() const;
   bool canActivateSlot(int index) const;
   void teardownSlot(int index);        // Disconnect the slot's client (keeps the object alive)
+  // Reconnect a slot, starting it instead when the client is stopped (reconnect() is a
+  // no-op on a stopped client). See the definition.
+  void reconnectSlotClient(int index);
   void maintainSlotConnections();      // Maintain all slot connections (token renewal, reconnect)
   void maintainSlotConnection(int index, unsigned long now_millis, unsigned long current_time, bool time_synced, bool& reconnect_attempted, bool& teardown_attempted);
   bool createSlotAuthToken(int index); // Create/renew JWT token for a slot

test/test_mqtt_connection_policy/test_mqtt_connection_policy.cpp

@@ -137,6 +137,22 @@ TEST(MQTTConnectionPolicy, SyncedClockRenewsInvalidExpiredOrImminentTokens) {
   EXPECT_TRUE(Policy::tokenNeedsRenewal(true, expires + 1U, expires, 300U));
 }
 
+TEST(MQTTConnectionPolicy, JwtReconnectReusesOnlyProvenValidCredentials) {
+  const uint32_t now = 1735689600U;
+  const uint32_t usable_expiry = now + Policy::kJwtReconnectSafetyMarginSecs + 1U;
+
+  EXPECT_TRUE(Policy::canReuseJwtForReconnect(true, true, false, now, usable_expiry));
+  EXPECT_FALSE(Policy::canReuseJwtForReconnect(
+      true, true, false, now, Policy::kMinimumValidEpoch - 1U));
+  EXPECT_FALSE(Policy::canReuseJwtForReconnect(true, true, false, now, 0U));
+  EXPECT_FALSE(Policy::canReuseJwtForReconnect(
+      true, true, false, now, now + Policy::kJwtReconnectSafetyMarginSecs));
+  EXPECT_FALSE(Policy::canReuseJwtForReconnect(true, true, false, now, now - 1U));
+  EXPECT_FALSE(Policy::canReuseJwtForReconnect(true, false, false, now, usable_expiry));
+  EXPECT_FALSE(Policy::canReuseJwtForReconnect(false, true, false, now, usable_expiry));
+  EXPECT_FALSE(Policy::canReuseJwtForReconnect(true, true, true, now, usable_expiry));
+}
+
 TEST(MQTTConnectionPolicy, RenewalThrottleHasExactBoundaryAndHandlesRollover) {
   EXPECT_FALSE(Policy::renewalAttemptAllowed(59999U, 0U));
   EXPECT_TRUE(Policy::renewalAttemptAllowed(60000U, 0U));
@@ -242,6 +258,80 @@ TEST(SlotActivation, DisabledAndOutOfRangeSlots) {
   EXPECT_EQ(SlotActivation::Disabled, Policy::classifySlotActivation(0, nullptr, 3, 2));
 }
 
+using Policy::StaleTokenAction;
+
+TEST(StaleToken, ConnectedSlotOnAnExpEnforcingBrokerBounces) {
+  // The broker will reject the stale token, and esp-mqtt refuses reconnect() from
+  // CONNECTED, so the transport has to close first.
+  EXPECT_EQ(StaleTokenAction::Bounce,
+            Policy::classifyStaleToken(/*minted=*/true, /*connected=*/true,
+                                       /*broker_enforces_exp=*/true));
+}
+
+TEST(StaleToken, ConnectedSlotOnATolerantBrokerKeepsItsSession) {
+  // waev leaves live sessions alone past exp. Bouncing would spend a 16 KiB
+  // contiguous handshake to replace a session the broker was not going to drop.
+  EXPECT_EQ(StaleTokenAction::KeepAlive,
+            Policy::classifyStaleToken(true, true, /*broker_enforces_exp=*/false));
+}
+
+TEST(StaleToken, DisconnectedSlotReconnectsRegardlessOfBrokerPolicy) {
+  EXPECT_EQ(StaleTokenAction::Reconnect, Policy::classifyStaleToken(true, false, true));
+  EXPECT_EQ(StaleTokenAction::Reconnect, Policy::classifyStaleToken(true, false, false));
+}
+
+TEST(StaleToken, FailedMintNeverReconnects) {
+  // Reconnecting here would re-present the credentials the correction invalidated.
+  // Every combination defers — including the disconnected one, which is the case
+  // that previously reconnected on the stale token.
+  EXPECT_EQ(StaleTokenAction::Defer, Policy::classifyStaleToken(false, false, true));
+  EXPECT_EQ(StaleTokenAction::Defer, Policy::classifyStaleToken(false, false, false));
+  EXPECT_EQ(StaleTokenAction::Defer, Policy::classifyStaleToken(false, true, true));
+  EXPECT_EQ(StaleTokenAction::Defer, Policy::classifyStaleToken(false, true, false));
+}
+
+using Policy::ClockSource;
+
+// 2026-01-01, the bridge's plausibility floor, and the 2024 placeholder
+// ESP32RTCClock::begin() stamps into libc on a power-on reset.
+static const uint32_t kFloor = 1767225600;
+static const uint32_t kPowerOnPlaceholder = 1715770351;
+static const uint32_t kPlausibleNow = 1786000000;
+
+TEST(FallbackClock, PrefersTheSystemClockWhenItIsUsable) {
+  // A clock SNTP set recently outranks an RTC that may have drifted.
+  EXPECT_EQ(ClockSource::System,
+            Policy::chooseFallbackClock(false, kPlausibleNow, kPlausibleNow - 900, kFloor));
+}
+
+TEST(FallbackClock, FallsBackToTheRtcOnAColdBoot) {
+  // The case the system-clock-only check missed: libc holds the power-on
+  // placeholder while a detected chip holds real time.
+  EXPECT_EQ(ClockSource::Rtc,
+            Policy::chooseFallbackClock(false, kPowerOnPlaceholder, kPlausibleNow, kFloor));
+}
+
+TEST(FallbackClock, NothingUsableStaysUnsynced) {
+  EXPECT_EQ(ClockSource::None,
+            Policy::chooseFallbackClock(false, kPowerOnPlaceholder, kPowerOnPlaceholder, kFloor));
+  // rtc_time 0 is how a board with no clock to consult is passed in.
+  EXPECT_EQ(ClockSource::None,
+            Policy::chooseFallbackClock(false, kPowerOnPlaceholder, 0, kFloor));
+}
+
+TEST(FallbackClock, ServerValidationNeverAcceptsALocalClock) {
+  // `set mqtt.ntp` asks whether that host answers. No clock can answer for it,
+  // however plausible — this is the path where a typo has to fail.
+  EXPECT_EQ(ClockSource::None,
+            Policy::chooseFallbackClock(true, kPlausibleNow, kPlausibleNow, kFloor));
+}
+
+TEST(FallbackClock, TheFloorItselfIsAccepted) {
+  EXPECT_EQ(ClockSource::System, Policy::chooseFallbackClock(false, kFloor, 0, kFloor));
+  EXPECT_EQ(ClockSource::Rtc, Policy::chooseFallbackClock(false, kFloor - 1, kFloor, kFloor));
+  EXPECT_EQ(ClockSource::None, Policy::chooseFallbackClock(false, kFloor - 1, kFloor - 1, kFloor));
+}
+
 int main(int argc, char** argv) {
   ::testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();