dandri/ZephCore
1
0
Fork 0
mirror of https://github.com/liquidraver/ZephCore.git synced 2026-09-01 20:38:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

repeater-observer hybrid fixup

Browse Source
This commit is contained in:
liquidraver
2026-07-10 22:43:26 +02:00
parent 983e7a198c
commit 56e41284e1
7 changed files with 148 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zephcore/adapters/mqtt/ZephyrMQTTPublisher.c
+63 -29
View File
@@ -29,19 +29,29 @@ LOG_MODULE_REGISTER(mqtt_pub, CONFIG_LOG_DEFAULT_LEVEL);
/* ========== Publish queue ========== */
/* Pre-serialized message: topic + JSON payload (both copied at enqueue). */
/* Pre-serialized message: JSON payload + 1-byte index into the two fixed
* topic strings registered at start (see enum mqtt_pub_topic). */
#define PUB_TOPIC_MAX 160
#define PUB_PAYLOAD_MAX 1024
#define PUB_QUEUE_LEN 8
struct pub_msg {
char topic[PUB_TOPIC_MAX];
char payload[PUB_PAYLOAD_MAX];
uint16_t payload_len;
uint8_t topic; /* enum mqtt_pub_topic */
char payload[PUB_PAYLOAD_MAX];
};
K_MSGQ_DEFINE(s_pub_queue, sizeof(struct pub_msg), PUB_QUEUE_LEN, 4);
/* Producer staging slot: callers build JSON directly into this buffer via
* mqtt_publisher_stage() then mqtt_publisher_commit() copies it into the
* queue. Single producer (mesh main thread) — no lock. */
static struct pub_msg s_stage;
/* Consumer drain slot: only the MQTT thread touches it (keeps a ~1 KB
* struct off the thread stack). */
static struct pub_msg s_drain;
/* ========== Module state ========== */
static const struct ObserverCreds *s_creds;
@@ -282,10 +292,11 @@ static void run_poll_loop(void)
}
/* Drain publish queue */
struct pub_msg msg;
while (k_msgq_get(&s_pub_queue, &msg, K_NO_WAIT) == 0) {
int rc = do_publish(&s_client, msg.topic,
msg.payload, msg.payload_len, false);
while (k_msgq_get(&s_pub_queue, &s_drain, K_NO_WAIT) == 0) {
const char *topic = (s_drain.topic == MQTT_PUB_TOPIC_PACKETS)
? s_packets_topic : s_status_topic;
int rc = do_publish(&s_client, topic,
s_drain.payload, s_drain.payload_len, false);
if (rc < 0) {
LOG_WRN("Publish failed: %d", rc);
s_connected = false;
@@ -328,6 +339,21 @@ static void run_poll_loop(void)
#define MQTT_THREAD_STACK_SIZE 12288
#define MQTT_THREAD_PRIORITY 5
/* Escalating retry backoff: DNS/connect/CONNACK failures double the delay up
* to the cap (a broker rejecting credentials won't fix itself in 10 s, and
* every TLS retry costs a full handshake). Reset on successful CONNACK. */
#define RETRY_BACKOFF_MIN_S 5
#define RETRY_BACKOFF_MAX_S 300
static uint32_t s_retry_backoff_s = RETRY_BACKOFF_MIN_S;
static void backoff_sleep(void)
{
LOG_INF("MQTT retry in %u s", s_retry_backoff_s);
k_sleep(K_SECONDS(s_retry_backoff_s));
s_retry_backoff_s = MIN(s_retry_backoff_s * 2, RETRY_BACKOFF_MAX_S);
}
static void mqtt_thread_fn(void *p1, void *p2, void *p3);
K_THREAD_DEFINE(mqtt_pub_thread, MQTT_THREAD_STACK_SIZE,
mqtt_thread_fn, NULL, NULL, NULL,
@@ -359,8 +385,8 @@ static void mqtt_thread_fn(void *p1, void *p2, void *p3)
int rc = resolve_host(s_creds->mqtt_host, s_creds->mqtt_port,
&s_broker_addr);
if (rc < 0) {
LOG_WRN("DNS failed — retry in 10s");
k_sleep(K_SECONDS(10));
LOG_WRN("DNS failed");
backoff_sleep();
continue;
}
@@ -374,8 +400,8 @@ static void mqtt_thread_fn(void *p1, void *p2, void *p3)
rc = mqtt_connect(&s_client);
if (rc < 0) {
LOG_WRN("mqtt_connect failed: %d — retry in 10s", rc);
k_sleep(K_SECONDS(10));
LOG_WRN("mqtt_connect failed: %d", rc);
backoff_sleep();
continue;
}
@@ -400,12 +426,15 @@ static void mqtt_thread_fn(void *p1, void *p2, void *p3)
}
if (!s_connected) {
LOG_WRN("CONNACK not accepted (broker rejected or no response) — retry in 10s");
LOG_WRN("CONNACK not accepted (broker rejected or no response)");
mqtt_disconnect(&s_client, NULL);
k_sleep(K_SECONDS(10));
backoff_sleep();
continue;
}
/* Connected — reset the failure backoff */
s_retry_backoff_s = RETRY_BACKOFF_MIN_S;
/* Publish retained "online" status */
publish_status(&s_client, true);
@@ -418,8 +447,8 @@ static void mqtt_thread_fn(void *p1, void *p2, void *p3)
}
mqtt_disconnect(&s_client, NULL);
LOG_INF("MQTT session ended — retry in 5s");
k_sleep(K_SECONDS(5));
LOG_INF("MQTT session ended");
backoff_sleep();
}
}
@@ -442,23 +471,28 @@ void mqtt_publisher_start(const struct ObserverCreds *creds,
LOG_INF("MQTT publisher ready (client_id=%s)", s_client_id);
}
void mqtt_publisher_enqueue(const char *topic, const char *payload, int payload_len)
char *mqtt_publisher_stage(size_t *size)
{
struct pub_msg msg;
strncpy(msg.topic, topic, sizeof(msg.topic) - 1);
msg.topic[sizeof(msg.topic) - 1] = '\0';
int copy_len = payload_len;
if (copy_len >= (int)sizeof(msg.payload)) {
copy_len = (int)sizeof(msg.payload) - 1;
if (size) {
*size = sizeof(s_stage.payload);
}
memcpy(msg.payload, payload, copy_len);
msg.payload[copy_len] = '\0';
msg.payload_len = (uint16_t)copy_len;
return s_stage.payload;
}
if (k_msgq_put(&s_pub_queue, &msg, K_NO_WAIT) != 0) {
LOG_WRN("Publish queue full — packet dropped");
void mqtt_publisher_commit(enum mqtt_pub_topic topic, int payload_len)
{
if (payload_len <= 0) {
return;
}
if (payload_len >= (int)sizeof(s_stage.payload)) {
payload_len = (int)sizeof(s_stage.payload) - 1;
}
s_stage.payload[payload_len] = '\0';
s_stage.payload_len = (uint16_t)payload_len;
s_stage.topic = (uint8_t)topic;
if (k_msgq_put(&s_pub_queue, &s_stage, K_NO_WAIT) != 0) {
LOG_WRN("Publish queue full — message dropped");
}
}
zephcore/adapters/mqtt/ZephyrMQTTPublisher.h
+17 -5
View File
@@ -10,6 +10,7 @@
#pragma once
#include <stdbool.h>
#include <stddef.h>
#ifdef __cplusplus
extern "C" {
@@ -17,6 +18,14 @@ extern "C" {
struct ObserverCreds;
/* The publisher only ever writes to the two topics registered in
* mqtt_publisher_start(); messages carry a 1-byte index instead of a
* per-message topic string. */
enum mqtt_pub_topic {
MQTT_PUB_TOPIC_STATUS = 0,
MQTT_PUB_TOPIC_PACKETS = 1,
};
/*
* Start the MQTT publisher thread.
* creds — runtime credentials; pointer kept, must remain valid.
@@ -32,12 +41,15 @@ void mqtt_publisher_start(const struct ObserverCreds *creds,
const char *packets_topic);
/*
* Enqueue a pre-serialized JSON payload for publishing.
* topic and payload are copied — safe to pass stack buffers.
* Drops the message silently if the queue is full.
* Safe to call from any thread.
* Zero-copy publish staging (single producer — the mesh main thread ONLY).
* mqtt_publisher_stage() returns the staging payload buffer (capacity in
* *size); the caller builds the JSON directly in it, then calls
* mqtt_publisher_commit() which copies the staged message into the queue.
* Drops the message (with a warning) if the queue is full.
* NOT safe to call from any other thread — the staging slot is unlocked.
*/
void mqtt_publisher_enqueue(const char *topic, const char *payload, int payload_len);
char *mqtt_publisher_stage(size_t *size);
void mqtt_publisher_commit(enum mqtt_pub_topic topic, int payload_len);
/* Returns true when the MQTT session is active. */
bool mqtt_publisher_is_connected(void);
zephcore/app/ObserverMesh.cpp
+25 -19
View File
@@ -19,12 +19,7 @@ LOG_MODULE_REGISTER(zephcore_observer, CONFIG_ZEPHCORE_OBSERVER_LOG_LEVEL);
#include <stdlib.h>
#include <time.h>
/* Forward declaration — implemented in ZephyrMQTTPublisher.c */
extern "C" {
void mqtt_publisher_enqueue(const char *topic, const char *payload, int payload_len);
bool mqtt_publisher_is_connected(void);
void mqtt_publisher_reconnect(void);
}
#include <ZephyrMQTTPublisher.h>
/* Forward declaration — implemented in ZephyrWiFiStation.c */
extern "C" {
@@ -136,9 +131,17 @@ void ObserverMesh::buildStatusJson(const char *status, char *out, size_t out_siz
void ObserverMesh::publishStatus(const char *status)
{
static char json_buf[768];
buildStatusJson(status, json_buf, sizeof(json_buf));
mqtt_publisher_enqueue(_status_topic, json_buf, strlen(json_buf));
/* Gate on connected: without it the 300 s status timer fills the
* publish queue with stale "online" messages while WiFi or the broker
* is down, and they all flush with old timestamps on reconnect. The
* on-connect status is posted via MESH_EVENT_MQTT_CONNECT, after
* CONNACK, so nothing is lost by skipping here. */
if (!mqtt_publisher_is_connected()) return;
size_t json_cap;
char *json_buf = mqtt_publisher_stage(&json_cap);
buildStatusJson(status, json_buf, json_cap);
mqtt_publisher_commit(MQTT_PUB_TOPIC_STATUS, (int)strlen(json_buf));
}
void ObserverMesh::buildTopics()
@@ -185,8 +188,10 @@ void ObserverMesh::enqueuePacket(Packet *pkt)
/* Get current timestamp from RTC */
uint32_t now_epoch = _rtc ? _rtc->getCurrentTime() : 0;
/* Build JSON payload matching meshcoretomqtt packet format */
static char json_buf[1024];
/* Build JSON payload matching meshcoretomqtt packet format, directly
* in the publisher's staging buffer (main-thread-only producer). */
size_t json_cap;
char *json_buf = mqtt_publisher_stage(&json_cap);
struct MeshcorePacketJson pj = {
_prefs.node_name,
_pubkey_hex,
@@ -201,14 +206,14 @@ void ObserverMesh::enqueuePacket(Packet *pkt)
(int)(_last_score * 1000.0f),
hash_hex,
};
int json_len = meshcore_build_packet_json(json_buf, sizeof(json_buf), &pj);
int json_len = meshcore_build_packet_json(json_buf, json_cap, &pj);
if (json_len < 0 || json_len >= (int)sizeof(json_buf)) {
if (json_len < 0 || json_len >= (int)json_cap) {
LOG_WRN("Packet JSON truncated (len=%d)", json_len);
json_len = (int)sizeof(json_buf) - 1;
json_len = (int)json_cap - 1;
}
mqtt_publisher_enqueue(_packets_topic, json_buf, json_len);
mqtt_publisher_commit(MQTT_PUB_TOPIC_PACKETS, json_len);
}
DispatcherAction ObserverMesh::onRecvPacket(Packet *pkt)
@@ -652,7 +657,8 @@ void ObserverMesh::publishSelfAdvert()
Utils::toHex(raw_hex, raw, raw_len);
raw_hex[raw_len * 2] = '\0';
static char json_buf[1024];
size_t json_cap;
char *json_buf = mqtt_publisher_stage(&json_cap);
struct MeshcorePacketJson pj = {
name,
_pubkey_hex,
@@ -667,14 +673,14 @@ void ObserverMesh::publishSelfAdvert()
0, /* score */
"0000000000000000", /* hash (not computed for self-advert) */
};
int json_len = meshcore_build_packet_json(json_buf, sizeof(json_buf), &pj);
int json_len = meshcore_build_packet_json(json_buf, json_cap, &pj);
if (json_len < 0 || json_len >= (int)sizeof(json_buf)) {
if (json_len < 0 || json_len >= (int)json_cap) {
LOG_WRN("publishSelfAdvert: JSON truncated");
return;
}
mqtt_publisher_enqueue(_packets_topic, json_buf, json_len);
mqtt_publisher_commit(MQTT_PUB_TOPIC_PACKETS, json_len);
LOG_INF("Self-advert published (lat=%.6f lon=%.6f)",
(double)_creds->lat_e6 / 1e6, (double)_creds->lon_e6 / 1e6);
}
zephcore/app/RepeaterMesh.cpp
+4 -3
View File
@@ -49,6 +49,7 @@ static void simple_sort(T* arr, int count, Comparator cmp) {
LOG_MODULE_REGISTER(zephcore_repeater, CONFIG_ZEPHCORE_MAIN_LOG_LEVEL);
#if IS_ENABLED(CONFIG_ZEPHCORE_REPEATER_UPLINK) && IS_ENABLED(CONFIG_MQTT_LIB)
#define UPLINK_STATUS_INTERVAL_MS 300000
static RepeaterMesh *s_uplink_mesh;
/* Runs on the WiFi thread; the mesh time-sync module is main-thread-only, so
* flag the trusted sync and let loop() arm suppression + drift envelope. */
@@ -1017,14 +1018,14 @@ void RepeaterMesh::begin(RepeaterDataStore* store) {
_uplink_status_topic, _uplink_packets_topic);
mqtt_publisher_set_connect_cb([]() {
/* Runs on the MQTT publisher thread — DON'T publish here. It would
* race the main-thread status path on the shared static JSON buffer
* race the main-thread producer on the publisher's staging buffer
* and toggle the battery-ADC regulator off-main. Defer to the main
* loop via an atomic flag drained in maintenanceLoop(). */
if (s_uplink_mesh) {
atomic_set(&s_uplink_mesh->_uplink_connect_pending, 1);
}
});
_uplink_next_status_at = futureMillis(300000);
_uplink_next_status_at = futureMillis(UPLINK_STATUS_INTERVAL_MS);
LOG_INF("Repeater uplink active: %s", _uplink_packets_topic);
} else {
LOG_INF("Repeater uplink inactive");
@@ -1385,7 +1386,7 @@ void RepeaterMesh::loop() {
}
if (_uplink_next_status_at && millisHasNowPassed(_uplink_next_status_at)) {
publishUplinkStatus("online");
_uplink_next_status_at = futureMillis(300000);
_uplink_next_status_at = futureMillis(UPLINK_STATUS_INTERVAL_MS);
}
if (atomic_cas(&s_uplink_sntp_pending, 1, 0)) {
/* SNTP set the clock (trusted) — arm suppression + drift envelope. */
zephcore/app/RepeaterUplink.cpp
+14 -9
View File
@@ -134,7 +134,8 @@ void RepeaterMesh::publishUplinkPacket(mesh::Packet *pkt)
uint32_t now_epoch = getRTCClock()->getCurrentTime();
static char json_buf[1024];
size_t json_cap;
char *json_buf = mqtt_publisher_stage(&json_cap);
struct MeshcorePacketJson pj = {
_prefs.node_name,
_uplink_pubkey_hex,
@@ -149,17 +150,20 @@ void RepeaterMesh::publishUplinkPacket(mesh::Packet *pkt)
(int)(_uplink_last_score * 1000.0f),
hash_hex,
};
int json_len = meshcore_build_packet_json(json_buf, sizeof(json_buf), &pj);
int json_len = meshcore_build_packet_json(json_buf, json_cap, &pj);
if (json_len <= 0 || json_len >= (int)sizeof(json_buf)) {
if (json_len <= 0 || json_len >= (int)json_cap) {
return;
}
mqtt_publisher_enqueue(_uplink_packets_topic, json_buf, json_len);
mqtt_publisher_commit(MQTT_PUB_TOPIC_PACKETS, json_len);
}
void RepeaterMesh::publishUplinkStatus(const char *status)
{
if (!isUplinkEnabled()) return;
/* The connected gate matters here: without it the 300 s status timer
* fills the publish queue with stale "online" messages while WiFi or the
* broker is down, and they all flush with old timestamps on reconnect. */
if (!isUplinkEnabled() || !mqtt_publisher_is_connected()) return;
if (_uplink_status_topic[0] == '\0') return;
auto& radio_driver = *static_cast<mesh::LoRaRadioBase *>(_radio);
@@ -170,7 +174,8 @@ void RepeaterMesh::publishUplinkStatus(const char *status)
(double)_prefs.freq, (double)_prefs.bw,
(unsigned)_prefs.sf, (unsigned)_prefs.cr);
static char json_buf[768];
size_t json_cap;
char *json_buf = mqtt_publisher_stage(&json_cap);
struct MeshcoreStatusJson sj = {
status,
now_epoch,
@@ -192,12 +197,12 @@ void RepeaterMesh::publishUplinkStatus(const char *status)
(unsigned)(getReceiveAirTime() / 1000),
(unsigned)radio_driver.getPacketsRecvErrors(),
};
int json_len = meshcore_build_status_json(json_buf, sizeof(json_buf), &sj);
int json_len = meshcore_build_status_json(json_buf, json_cap, &sj);
if (json_len <= 0 || json_len >= (int)sizeof(json_buf)) {
if (json_len <= 0 || json_len >= (int)json_cap) {
return;
}
mqtt_publisher_enqueue(_uplink_status_topic, json_buf, json_len);
mqtt_publisher_commit(MQTT_PUB_TOPIC_STATUS, json_len);
}
#endif /* CONFIG_ZEPHCORE_REPEATER_UPLINK && CONFIG_MQTT_LIB */
zephcore/app/main_observer.cpp
+3 -3
View File
@@ -8,7 +8,7 @@
* LittleFS. Nothing is hardcoded.
*
* Event loop:
* LORA_RX → ObserverMesh::loop() → enqueuePacket() → mqtt_publisher_enqueue()
* LORA_RX → ObserverMesh::loop() → enqueuePacket() → mqtt_publisher_commit()
* CLI_RX → ObserverMesh::handleCLI() (set/get commands)
*/
@@ -352,8 +352,8 @@ int main(void)
* this observer on the map (requires lat/lon to be configured). */
mqtt_publisher_set_connect_cb([]() {
/* Runs on the MQTT publisher thread — defer to the main loop. Publishing
* here would race the periodic-status path on publishStatus()'s shared
* static JSON buffer; publishSelfAdvert() also signs + formats. */
* here would race the main-thread producer on the publisher's staging
* buffer; publishSelfAdvert() also signs + formats. */
k_event_post(&mesh_events, MESH_EVENT_MQTT_CONNECT);
});
k_timer_start(&status_timer, K_SECONDS(300), K_SECONDS(300));
zephcore/app/observer_creds.cpp
+22 -5
View File
@@ -45,24 +45,41 @@ extern "C" bool observer_creds_load(struct ObserverCreds *creds,
extern "C" bool observer_creds_save(const struct ObserverCreds *creds,
const char *base_path)
{
/* Atomic replace (.tmp + fs_sync + fs_rename), same pattern as
* identity/prefs — power loss mid-write must not corrupt the creds. */
char path[96];
char tmp_path[104];
snprintf(path, sizeof(path), "%s/obs_creds", base_path);
if (snprintf(tmp_path, sizeof(tmp_path), "%s.tmp", path) >=
(int)sizeof(tmp_path)) {
return false;
}
fs_unlink(tmp_path);
struct fs_file_t f;
fs_file_t_init(&f);
int rc = fs_open(&f, path, FS_O_WRITE | FS_O_CREATE | FS_O_TRUNC);
int rc = fs_open(&f, tmp_path, FS_O_WRITE | FS_O_CREATE | FS_O_TRUNC);
if (rc < 0) {
LOG_ERR("Cannot open %s for write: %d", path, rc);
LOG_ERR("Cannot open %s for write: %d", tmp_path, rc);
return false;
}
ssize_t n = fs_write(&f, creds, sizeof(*creds));
rc = fs_sync(&f);
fs_close(&f);
if (n != (ssize_t)sizeof(*creds)) {
LOG_ERR("obs_creds write failed (%d/%d)", (int)n,
(int)sizeof(*creds));
if (n != (ssize_t)sizeof(*creds) || rc < 0) {
LOG_ERR("obs_creds write failed (%d/%d sync=%d)", (int)n,
(int)sizeof(*creds), rc);
fs_unlink(tmp_path);
return false;
}
if (fs_rename(tmp_path, path) < 0) {
LOG_ERR("obs_creds rename failed");
fs_unlink(tmp_path);
return false;
}