diff --git a/zephcore/adapters/mqtt/ZephyrMQTTPublisher.c b/zephcore/adapters/mqtt/ZephyrMQTTPublisher.c index a10bf81..8408e01 100644 --- a/zephcore/adapters/mqtt/ZephyrMQTTPublisher.c +++ b/zephcore/adapters/mqtt/ZephyrMQTTPublisher.c @@ -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"); } } diff --git a/zephcore/adapters/mqtt/ZephyrMQTTPublisher.h b/zephcore/adapters/mqtt/ZephyrMQTTPublisher.h index 99c313a..883e1a4 100644 --- a/zephcore/adapters/mqtt/ZephyrMQTTPublisher.h +++ b/zephcore/adapters/mqtt/ZephyrMQTTPublisher.h @@ -10,6 +10,7 @@ #pragma once #include +#include #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); diff --git a/zephcore/app/ObserverMesh.cpp b/zephcore/app/ObserverMesh.cpp index d24dd95..45a8e37 100644 --- a/zephcore/app/ObserverMesh.cpp +++ b/zephcore/app/ObserverMesh.cpp @@ -19,12 +19,7 @@ LOG_MODULE_REGISTER(zephcore_observer, CONFIG_ZEPHCORE_OBSERVER_LOG_LEVEL); #include #include -/* 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 /* 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); } diff --git a/zephcore/app/RepeaterMesh.cpp b/zephcore/app/RepeaterMesh.cpp index 3f63939..621db8a 100644 --- a/zephcore/app/RepeaterMesh.cpp +++ b/zephcore/app/RepeaterMesh.cpp @@ -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. */ diff --git a/zephcore/app/RepeaterUplink.cpp b/zephcore/app/RepeaterUplink.cpp index 2657969..4411a8f 100644 --- a/zephcore/app/RepeaterUplink.cpp +++ b/zephcore/app/RepeaterUplink.cpp @@ -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(_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 */ diff --git a/zephcore/app/main_observer.cpp b/zephcore/app/main_observer.cpp index 6c7a144..6639221 100644 --- a/zephcore/app/main_observer.cpp +++ b/zephcore/app/main_observer.cpp @@ -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)); diff --git a/zephcore/app/observer_creds.cpp b/zephcore/app/observer_creds.cpp index bf0a245..af5ff30 100644 --- a/zephcore/app/observer_creds.cpp +++ b/zephcore/app/observer_creds.cpp @@ -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; }