Merge branch 'mqtt-bridge-implementation-flex' into add-rflab-mqtt-preset

Author: agessaman

.github/workflows/build-observer-firmwares.yml

@@ -47,6 +47,7 @@ jobs:
     runs-on: ubuntu-latest
     outputs:
       matrix: ${{ steps.split.outputs.matrix }}
+      build_number: ${{ steps.buildnum.outputs.n }}
     steps:
       - name: Clone Repo
         uses: actions/checkout@v4
@@ -66,6 +67,27 @@ jobs:
               | jq -cs .)
           echo "matrix=$MATRIX" >> "$GITHUB_OUTPUT"
 
+      - name: Compute observer build number
+        id: buildnum
+        shell: bash
+        run: |
+          # Per-base published-build counter. N increments once per release and
+          # resets to 1 when FIRMWARE_VERSION (the MeshCore base version) changes.
+          # Read-only here (off the published counter); the release job is the sole
+          # writer and only writes on a successful build, so a failed build doesn't
+          # burn a number. First run / 404 -> empty -> N=1.
+          COUNTER_URL="https://observer.gessaman.com/observer-build-counter.json"
+          CUR=$(curl -fsSL "$COUNTER_URL" 2>/dev/null || echo '{}')
+          PREV_BASE=$(echo "$CUR" | jq -r '.baseVersion // ""')
+          PREV_BUILD=$(echo "$CUR" | jq -r '.build // 0')
+          if [ "$PREV_BASE" = "$FIRMWARE_VERSION" ]; then
+            N=$((PREV_BUILD + 1))
+          else
+            N=1
+          fi
+          echo "Base $FIRMWARE_VERSION; previous build $PREV_BUILD (base $PREV_BASE) -> N=$N"
+          echo "n=$N" >> "$GITHUB_OUTPUT"
+
   # Build one shard (several envs) per runner. build.sh emits both the app
   # .bin and the ESP32 -merged.bin into out/ for each env.
   build:
@@ -100,6 +122,9 @@ jobs:
         uses: ./.github/actions/setup-build-environment
 
       - name: Build Shard ${{ matrix.shard.idx }}
+        env:
+          # Stamp the per-base build number into the embedded version (v1.16.0.N).
+          FIRMWARE_BUILD_NUMBER: ${{ needs.enumerate.outputs.build_number }}
         run: /usr/bin/env bash build.sh build-firmware ${{ matrix.shard.envs }}
 
       - name: Upload Shard Artifact
@@ -112,7 +137,7 @@ jobs:
   # Collect all shard outputs, publish the rolling release, then point the
   # flasher at the new build by bumping the hash + notes in its config.json.
   release:
-    needs: build
+    needs: [enumerate, build]
     runs-on: ubuntu-latest
     steps:
       - name: Clone Repo
@@ -134,6 +159,8 @@ jobs:
         run: |
           mkdir -p out
           find artifacts -type f -name '*.bin' -exec cp -f {} out/ \;
+          # Per-env partition-table signatures (for the slim manifest's OTA gate).
+          find artifacts -type f -name '*.partsig' -exec cp -f {} out/ \;
           echo "Collected binaries:"; ls -1 out
 
       - name: Compute Short SHA
@@ -153,8 +180,10 @@ jobs:
               --notes "Rolling build of all *_observer_mqtt firmwares. The git short hash is embedded in each asset filename."
           fi
 
-          # Upload this build, overwriting same-named assets.
-          gh release upload "$RELEASE_TAG" out/* --clobber
+          # Upload this build, overwriting same-named assets. Exclude the .partsig
+          # files — those are consumed locally by the slim-manifest generator below,
+          # not published as release assets (and they'd break the hash-based prune).
+          gh release upload "$RELEASE_TAG" $(find out -maxdepth 1 -type f ! -name '*.partsig') --clobber
 
           # Prune old assets, but RETAIN the most recent KEEP_BUILDS build hashes
           # (not just the current one). The flasher SPA reads config.json once at
@@ -176,7 +205,10 @@ jobs:
           echo "Retaining build hashes:"; echo "$keep_hashes"
           gh release view "$RELEASE_TAG" --json assets -q '.assets[].name' \
             | while read -r asset; do
-                ah=$(printf '%s' "$asset" | grep -oiE '[0-9a-f]{7,40}(-merged)?\.bin$' | grep -oiE '^[0-9a-f]{7,40}')
+                # `|| true`: a non-firmware asset (no embedded hash) makes grep exit
+                # non-zero, which under `bash -e` would abort the step. Empty ah then
+                # falls through to delete-asset, cleaning up any stray non-.bin asset.
+                ah=$(printf '%s' "$asset" | grep -oiE '[0-9a-f]{7,40}(-merged)?\.bin$' | grep -oiE '^[0-9a-f]{7,40}' || true)
                 if [ -n "$ah" ] && grep -qxF "$ah" <<<"$keep_hashes"; then
                   continue
                 fi
@@ -196,6 +228,24 @@ jobs:
             "${{ steps.sha.outputs.short }}" \
             "$GITHUB_WORKSPACE/firmware-notes.html"
 
+      - name: Generate Slim Per-Variant Manifests + Persist Build Counter
+        env:
+          BUILD_NUMBER: ${{ needs.enumerate.outputs.build_number }}
+        run: |
+          # Derive the slim per-variant manifests (flasher/v/<env>.json) that the
+          # firmware fetches for `ota check`/`ota update`, from the just-updated
+          # config.json (single source of truth), stamping this build's number.
+          # Then persist the counter so the next run increments from here.
+          python3 flasher/scripts/gen-slim-manifests.py \
+            --config flasher/config.json \
+            --out-dir flasher/v \
+            --base-version "$FIRMWARE_VERSION" \
+            --build "$BUILD_NUMBER" \
+            --partsig-dir out
+          printf '{\n  "baseVersion": "%s",\n  "build": %s\n}\n' \
+            "$FIRMWARE_VERSION" "$BUILD_NUMBER" > flasher/observer-build-counter.json
+          echo "Build $FIRMWARE_VERSION.$BUILD_NUMBER"
+
       - name: Unshallow for Changelog
         run: |
           # gen_changelog.py needs the full branch history; deepen only now, AFTER
@@ -233,11 +283,15 @@ jobs:
       - name: Commit & Push Flasher Config
         working-directory: flasher
         run: |
-          if git diff --quiet; then
+          # Stage everything: update-firmware.py edits config.json in place, but the
+          # slim manifests (v/*.json) and the build counter can be NEW files, which
+          # `commit -am` would miss — so add -A and check the staged diff.
+          git add -A
+          if git diff --cached --quiet; then
             echo "No flasher changes to commit."
             exit 0
           fi
           git config user.name "meshcore-bot"
           git config user.email "noreply@gessaman.com"
-          git commit -am "Update observer firmware to ${{ steps.sha.outputs.short }}"
+          git commit -m "Update observer firmware to ${{ steps.sha.outputs.short }} (build ${FIRMWARE_VERSION}.${{ needs.enumerate.outputs.build_number }})"
           git push

MQTT_IMPLEMENTATION.md

@@ -116,6 +116,8 @@ The MQTT bridge uses a slot-based architecture with up to 6 concurrent connectio
 | `meshcore-ca-1` | mqtt1.meshcore.ca:443 | JWT (Ed25519) | WSS |
 | `meshcore-ca-2` | mqtt2.meshcore.ca:443 | JWT (Ed25519) | WSS |
 | `bostonmesh` | mqttmc01.bostonme.sh:443 | JWT (Ed25519) | WSS |
+| `ipnt.uk` | mqtt.ipnt.uk:443 | JWT (Ed25519) | WSS |
+| `flmesh` | mcmqtt.jntconnections.com:443 | JWT (Ed25519) | WSS |
 | `inwmesh` | scope.inwmesh.org:8883 | Username/password (per slot via `mqttN.username` / `mqttN.password`) | MQTT over TLS |
 | `rflab` | mqtt.rflab.io:443 | JWT (Ed25519) | WSS |
 | `custom` | User-configured | Username/Password | MQTT or WSS |
@@ -378,6 +380,8 @@ These settings apply across all MQTT slots:
 - `get mqtt.rx` - Get RX packet uplinking setting (on/off)
 - `get mqtt.tx` - Get TX packet uplinking setting (on/off/advert)
 - `get mqtt.interval` - Get status publish interval
+- `get mqtt.ntp` - Get effective NTP server hostname
+- `get mqtt.ntp.diag` - Probe every configured NTP server for connectivity (does not change the clock; serial console shows each server's reported time, LoRa shows a compact `<server> ok|fail` list)
 - `get mqtt.owner` - Get owner public key (serial console only)
 - `get mqtt.email` - Get owner email address (serial console only)
 
@@ -393,6 +397,7 @@ These settings apply across all MQTT slots:
   - `advert` - Uplink only this node's own advert packets (self-originated)
   - `off` - Disable TX packet uplinking
 - `set mqtt.interval <minutes>` - Set status publish interval (1-60 minutes)
+- `set mqtt.ntp <hostname>` - Set custom NTP server (validated with immediate sync); `none` reverts to default
 - `set mqtt.owner <64-hex-char-public-key>` - Set owner public key
 - `set mqtt.email <email>` - Set owner email address
 
@@ -607,8 +612,13 @@ Minimal raw packet data for map integration.
 - Automatic reconnection with exponential backoff
 
 ### NTP Time Synchronization
-- Automatic time synchronization with NTP servers
-- Periodic time updates (every hour)
+- Automatic time synchronization with NTP servers (required for JWT authentication)
+- Default primary: `pool.ntp.org`; built-in fallbacks (tried sequentially on failure): `time.google.com`, `time.cloudflare.com`, `time.aws.com`, `time.nist.gov`
+- Custom primary via `set mqtt.ntp <hostname>`; `set mqtt.ntp none` reverts to default
+- `set mqtt.ntp` runs an immediate sync (primary only, so a typo fails fast) when WiFi is connected and the bridge is running
+- `get mqtt.ntp` returns the effective primary hostname
+- `get mqtt.ntp.diag` probes every configured server (primary + fallbacks) for connectivity and reports each server's time without changing the system clock — a pure diagnostic
+- Periodic time updates (every hour) on the effective primary only
 - Proper UTC system time handling
 
 ### Authentication

build.sh

@@ -149,10 +149,23 @@ build_firmware() {
   case "$1" in
     *observer*) VARIANT_TAG="-observer" ;;
   esac
-  EMBEDDED_VERSION_STRING="${FIRMWARE_VERSION}${VARIANT_TAG}-${COMMIT_HASH}"
 
-  # add firmware version info to end of existing platformio build flags in environment vars
-  export PLATFORMIO_BUILD_FLAGS="${PLATFORMIO_BUILD_FLAGS} -DFIRMWARE_BUILD_DATE='\"${FIRMWARE_BUILD_DATE}\"' -DFIRMWARE_VERSION='\"${EMBEDDED_VERSION_STRING}\"'"
+  # Observer build number: when CI provides FIRMWARE_BUILD_NUMBER (the per-base
+  # published-build counter), append it as a 4th version component so the node
+  # reports e.g. v1.16.0.5-observer-abcdef and `ota check` can show how many
+  # builds behind it is. Local dev builds leave it unset → no 4th component.
+  # The *filename* (FIRMWARE_VERSION_STRING above) is deliberately left without
+  # the build number so assets stay <env>-v<base>-<hash>.bin.
+  BUILD_NUMBER_SUFFIX=""
+  if [ -n "$FIRMWARE_BUILD_NUMBER" ]; then
+    BUILD_NUMBER_SUFFIX=".${FIRMWARE_BUILD_NUMBER}"
+  fi
+  EMBEDDED_VERSION_STRING="${FIRMWARE_VERSION}${BUILD_NUMBER_SUFFIX}${VARIANT_TAG}-${COMMIT_HASH}"
+
+  # add firmware version info to end of existing platformio build flags in environment vars.
+  # OTA_VARIANT is the env name ($1) — it selects this build's slim per-variant manifest
+  # (<OTA_MANIFEST_BASE>/<OTA_VARIANT>.json) that the observer pull-OTA fetches.
+  export PLATFORMIO_BUILD_FLAGS="${PLATFORMIO_BUILD_FLAGS} -DFIRMWARE_BUILD_DATE='\"${FIRMWARE_BUILD_DATE}\"' -DFIRMWARE_VERSION='\"${EMBEDDED_VERSION_STRING}\"' -DOTA_VARIANT='\"$1\"'"
 
   # disable debug flags if requested
   disable_debug_flags
@@ -175,6 +188,13 @@ build_firmware() {
   cp .pio/build/$1/firmware.uf2 out/${FIRMWARE_FILENAME}.uf2 2>/dev/null || true
   cp .pio/build/$1/firmware.zip out/${FIRMWARE_FILENAME}.zip 2>/dev/null || true
 
+  # Emit the partition-table signature (ESP32) for OTA partition-compatibility
+  # checks. Keyed by env name so the slim-manifest generator can find it; the
+  # firmware computes the same signature at runtime from its flashed table.
+  if [ -f ".pio/build/$1/partitions.bin" ]; then
+    python3 scripts/partition_signature.py ".pio/build/$1/partitions.bin" > "out/$1.partsig" 2>/dev/null || true
+  fi
+
 }
 
 # firmwares containing $1 will be built

docs/cli_commands.md

@@ -1076,6 +1076,33 @@ region save
 
 ---
 
+#### View or change the NTP server (MQTT observer only)
+**Usage:**
+- `get mqtt.ntp`
+- `set mqtt.ntp <hostname>`
+- `set mqtt.ntp none`
+
+**Description:** Sets the primary NTP server used for clock sync (required for JWT MQTT auth). On `set`, the device attempts an immediate sync of the just-configured server (primary only, so a typo fails fast) when WiFi is connected and the MQTT bridge is running.
+
+**Fallbacks:** If the primary fails, the firmware tries `pool.ntp.org`, `time.google.com`, `time.cloudflare.com`, `time.aws.com`, and `time.nist.gov` in order (skipping duplicates).
+
+**Default:** `pool.ntp.org` (when unset or `none`)
+
+---
+
+#### Diagnose NTP server connectivity (MQTT observer only)
+**Usage:**
+- `get mqtt.ntp.diag`
+
+**Description:** Probes every configured NTP server (the custom primary, if set, plus the built-in fallbacks) and reports whether each responds. This is a pure connectivity diagnostic — it does **not** change the system clock.
+
+- **Serial console:** prints a detailed table with each server's reported UTC time (or `FAIL`).
+- **Over LoRa:** returns a compact `<server> ok|fail` list, one per line.
+
+Requires WiFi connected and the MQTT bridge running.
+
+---
+
 #### View or change the speed of the bridge (RS-232 only)
 **Usage:**
 - `get bridge.baud`

examples/simple_repeater/MyMesh.cpp

@@ -1445,6 +1445,23 @@ void MyMesh::loop() {
     MESH_DEBUG_PRINTLN("Radio params restored");
   }
 
+#if defined(WITH_MQTT_BRIDGE) && defined(OTA_MANIFEST_BASE)
+  if (_ota_update_at && millisHasNowPassed(_ota_update_at)) { // deferred `ota update`
+    _ota_update_at = 0;                                       // clear timer
+    // The "Beginning update..." reply has now gone out. Free the bridge for heap
+    // headroom, then flash: otaFromManifest reboots into the new image on success
+    // (so this never returns); on any abort (already up to date, partition change,
+    // download error) it returns and we resume the bridge.
+    Serial.println("OTA: starting update");
+    setBridgeState(false);
+    char ota_reply[160];
+    if (!_cli.getBoard()->otaFromManifest(getFirmwareVer(), false, ota_reply)) {
+      Serial.print("OTA: aborted, resuming bridge - "); Serial.println(ota_reply);
+      setBridgeState(true);
+    }
+  }
+#endif
+
   // is pending dirty contacts write needed?
   if (dirty_contacts_expiry && millisHasNowPassed(dirty_contacts_expiry)) {
     acl.save(_fs);

examples/simple_repeater/MyMesh.h

@@ -116,6 +116,7 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks {
 #endif
   CayenneLPP telemetry;
   unsigned long set_radio_at, revert_radio_at;
+  unsigned long _ota_update_at = 0;  // deferred `ota update` fire time (0 = none scheduled)
   float pending_freq;
   float pending_bw;
   uint8_t pending_sf;
@@ -306,9 +307,33 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks {
     bridge->setSlotPreset(slot, _prefs.mqtt_slot_preset[slot]);
   }
 
+  // Schedule the pull-OTA flash to run from loop() in ~2.5 s, leaving time for the
+  // "Beginning update..." CLI reply (CLI_REPLY_DELAY_MILLIS = 600 ms) to transmit
+  // before the flash blocks the loop and reboots.
+  bool beginDeferredOtaUpdate() override {
+    _ota_update_at = millis() + 2500;
+    if (_ota_update_at == 0) _ota_update_at = 1;  // 0 means "none"
+    return true;
+  }
+
   int getQueueSize() override {
     return bridge ? bridge->getQueueSize() : 0;
   }
+
+  bool isMqttBridgeRunning() override {
+    return bridge && bridge->isRunning();
+  }
+
+  bool syncMqttNtp() override {
+    if (!bridge || !bridge->isRunning()) return false;
+    // Marshal onto the MQTT task (Core 0); this runs on the CLI thread (Core 1).
+    return bridge->requestForcedNtpSync();
+  }
+
+  bool runMqttNtpDiag(char* reply, size_t reply_size, bool verbose) override {
+    if (!bridge || !bridge->isRunning()) return false;
+    return bridge->ntpDiag(reply, reply_size, verbose);
+  }
 #endif
 
   // To check if there is pending work

examples/simple_repeater/main.cpp

@@ -82,6 +82,11 @@ void setup() {
     store.save("_main", the_mesh.self_id);
   }
 
+  // Print the running firmware version at boot so it's visible after an OTA
+  // reboot without having to issue `ver` manually.
+  Serial.print("Firmware: "); Serial.print(FIRMWARE_VERSION);
+  Serial.print(" (built "); Serial.print(FIRMWARE_BUILD_DATE); Serial.println(")");
+
   Serial.print("Repeater ID: ");
   mesh::Utils::printHex(Serial, the_mesh.self_id.pub_key, PUB_KEY_SIZE); Serial.println();
 

examples/simple_room_server/MyMesh.h

@@ -296,5 +296,20 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks {
   int getQueueSize() override {
     return bridge ? bridge->getQueueSize() : 0;
   }
+
+  bool isMqttBridgeRunning() override {
+    return bridge && bridge->isRunning();
+  }
+
+  bool syncMqttNtp() override {
+    if (!bridge || !bridge->isRunning()) return false;
+    // Marshal onto the MQTT task (Core 0); this runs on the CLI thread (Core 1).
+    return bridge->requestForcedNtpSync();
+  }
+
+  bool runMqttNtpDiag(char* reply, size_t reply_size, bool verbose) override {
+    if (!bridge || !bridge->isRunning()) return false;
+    return bridge->ntpDiag(reply, reply_size, verbose);
+  }
 #endif
 };

scripts/partition_signature.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+"""Canonical signature of an ESP32 partition table (from a partitions.bin).
+
+Used to decide OTA partition compatibility: the observer firmware computes the
+same signature at runtime from its *flashed* partition table (via esp_partition),
+and `ota update` refuses only when the target build's signature differs from the
+running device's — i.e. a real partition-table change, not a blanket flag.
+
+The signature MUST be computed identically here and in firmware
+(src/helpers/ESP32Board.cpp). Definition:
+
+  for each partition-table entry: (type, subtype, offset, size)
+  sort by offset ascending
+  format each as "%x:%x:%x:%x" (lowercase hex, no 0x, no padding)
+  join with ","
+
+partitions.bin layout: 32-byte records, each starting with magic 0xAA 0x50; the
+trailing MD5 record (magic 0xEB 0xEB) and 0xFF padding are ignored.
+
+Usage:  partition_signature.py <partitions.bin>   # prints the signature
+"""
+import struct
+import sys
+
+
+def signature(bin_path: str) -> str:
+    data = open(bin_path, "rb").read()
+    entries = []
+    for i in range(0, len(data) - 31, 32):
+        rec = data[i:i + 32]
+        if rec[0:2] != b"\xaa\x50":  # not a partition entry (MD5 record / padding)
+            continue
+        ptype, subtype = rec[2], rec[3]
+        offset, size = struct.unpack("<II", rec[4:12])
+        entries.append((ptype, subtype, offset, size))
+    entries.sort(key=lambda e: e[2])
+    return ",".join("%x:%x:%x:%x" % e for e in entries)
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 2:
+        print("usage: partition_signature.py <partitions.bin>", file=sys.stderr)
+        sys.exit(2)
+    print(signature(sys.argv[1]))