mirror of
https://github.com/mikecarper/MeshCore.git
synced 2026-10-24 04:32:19 +00:00
fix(mqtt): release nothing after a stop the MQTT task never acknowledged (F01)
The timeout fallback did the two things that cannot be done safely to a task that may still be inside mbedTLS or holding esp-mqtt's API mutex: it deleted that task from the loop task, then force-stopped and deleted the clients it owned, then freed the queue and buffers it can reach. `esp_mqtt_client_stop()` waits on the API mutex and the task's stopped event with no bound, so killing its caller strands the next one; a second stop sees the run flag already clear and returns failure without joining. Nothing in that path established quiescence, and `canFlashAfterStop()` could only withhold OTA afterwards. Changes, all of them about who owns what: - `MQTTLifecycle` gains a `StopUnproven` state (appended, so existing values keep their numbering). `StopTimedOut` now leads there and releases NOTHING; it still fires `ota_release` so the OTA barrier aborts. `acceptsNewWork`, `mayRestart` and `isStopInProgress` are false there, `mayTouchOwnedState` is true, and `end()`'s wait loop still terminates. - `begin()` refuses to start while a stop is unproven. A new start was never proof that the previous clients stopped, and it used to clear the dirty latch that withheld OTA. Because a start is now impossible from `StopUnproven`, clearing that latch on an accepted start is sound. - A late acknowledgement is honoured. The task publishes it only after tearing its clients down, so it proves the same thing whenever it arrives; the deadline is an availability policy, not a statement about what the ack means. `pollLateStopAck()` releases the withheld resources and makes the bridge restartable — a slow stop (a blackholed WSS broker can hold the SDK well past the budget) no longer costs a reboot. - The handshake flags become `std::atomic<bool>` with release/acquire, and the ack moves into the task trampoline immediately before `vTaskDelete(nullptr)`, gated on a `_teardown_complete` flag the loop sets. Published from inside the loop it proved teardown returned, not that the task had stopped executing — and `volatile` ordered none of it. This is the soak campaign's blocker #20, the one item on its list that can corrupt memory. - The force-stop paths are gone with their only caller: `destroySlotClients()` and `teardownSlot()` no longer take a `force` flag. - `get mqtt.status` reports "previous stop unproven, reboot to recover" instead of a bare "not running", latched so it survives end() clearing the singleton. Two lifecycle tests asserted the old contract and are re-encoded, not deleted: a timed-out stop is no longer `Stopped`-and-released, and "repeated failed stops leave the bridge restartable" becomes "repeated *slow* stops recover on their late ack" plus a new case pinning that a never-acknowledged stop stays unusable for the boot. That availability trade is the point of the change.
This commit is contained in:
1 parent
31d9925b05
commit
415c5d6580
4 files changed
+285
-90
No files matched your search
@@ -40,6 +40,13 @@ enum class State : uint8_t {
|
||||
Running,
|
||||
StopRequested,
|
||||
Stopping,
|
||||
// A requested stop passed its deadline without an acknowledgement. The MQTT
|
||||
// task may still be alive inside mbedTLS or holding the SDK's API mutex, so
|
||||
// NOTHING it can reach may be released and the bridge may not restart. This
|
||||
// is deliberately not Stopped: releasing on a stop we cannot prove is the
|
||||
// reviewed teardown-heap-panic path (F01). Appended so the values above keep
|
||||
// their numbering.
|
||||
StopUnproven,
|
||||
};
|
||||
|
||||
// Events driven either by the owner (loop task) or by the MQTT task reporting
|
||||
@@ -135,12 +142,17 @@ inline Result apply(State s, Event e) {
|
||||
r.accepted = true;
|
||||
break;
|
||||
case Event::StopAcknowledged:
|
||||
case Event::StopTimedOut:
|
||||
r.next = State::Stopped;
|
||||
r.effects.release_resources = true;
|
||||
r.effects.ota_release = true;
|
||||
r.accepted = true;
|
||||
break;
|
||||
case Event::StopTimedOut:
|
||||
// Unblock the OTA barrier (so it aborts) but release nothing.
|
||||
r.next = State::StopUnproven;
|
||||
r.effects.ota_release = true;
|
||||
r.accepted = true;
|
||||
break;
|
||||
default:
|
||||
// Duplicate StopRequested is a no-op (idempotent stop).
|
||||
break;
|
||||
@@ -150,16 +162,33 @@ inline Result apply(State s, Event e) {
|
||||
case State::Stopping:
|
||||
switch (e) {
|
||||
case Event::StopAcknowledged:
|
||||
case Event::StopTimedOut:
|
||||
r.next = State::Stopped;
|
||||
r.effects.release_resources = true;
|
||||
r.effects.ota_release = true;
|
||||
r.accepted = true;
|
||||
break;
|
||||
case Event::StopTimedOut:
|
||||
r.next = State::StopUnproven;
|
||||
r.effects.ota_release = true;
|
||||
r.accepted = true;
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
break;
|
||||
|
||||
case State::StopUnproven:
|
||||
// A late acknowledgement proves exactly what a timely one proves: the
|
||||
// task published it only after tearing its clients down. The deadline was
|
||||
// an availability policy, not a statement about what the ack means — so
|
||||
// honour it, release, and let the bridge be restartable again. A start
|
||||
// before that proof is refused (see mayRestart).
|
||||
if (e == Event::StopAcknowledged) {
|
||||
r.next = State::Stopped;
|
||||
r.effects.release_resources = true;
|
||||
r.accepted = true;
|
||||
}
|
||||
break;
|
||||
}
|
||||
return r;
|
||||
}
|
||||
@@ -182,6 +211,11 @@ inline bool mayTouchOwnedState(State s) { return s != State::Stopped; }
|
||||
// A restart (begin()) is safe only from a completed stop.
|
||||
inline bool mayRestart(State s) { return s == State::Stopped; }
|
||||
|
||||
// True while a stop has passed its deadline unproven: the owner must keep every
|
||||
// resource the MQTT task can reach (queue, buffers, clients, its own task) and
|
||||
// must not restart. Cleared only by a late StopAcknowledged.
|
||||
inline bool isStopUnproven(State s) { return s == State::StopUnproven; }
|
||||
|
||||
inline bool isStopInProgress(State s) {
|
||||
return s == State::StopRequested || s == State::Stopping;
|
||||
}
|
||||
@@ -193,6 +227,7 @@ inline const char* stateName(State s) {
|
||||
case State::Running: return "Running";
|
||||
case State::StopRequested: return "StopRequested";
|
||||
case State::Stopping: return "Stopping";
|
||||
case State::StopUnproven: return "StopUnproven";
|
||||
}
|
||||
return "?";
|
||||
}
|
||||
@@ -264,10 +299,16 @@ class Coordinator {
|
||||
bool isStopInProgress() const {
|
||||
return MQTTLifecycle::isStopInProgress(_state);
|
||||
}
|
||||
// A restart is safe from a completed stop. A stop that reached Stopped via
|
||||
// the timeout fallback still allows restart (the bridge is down); only OTA
|
||||
// flashing is withheld after a dirty stop.
|
||||
// A restart is safe only from a PROVEN stop. An unproven stop keeps the
|
||||
// bridge down until the task acknowledges late (or the node reboots): a new
|
||||
// start is not proof that the previous clients stopped, and starting on top
|
||||
// of them is what the review refused to approve. Because a start is
|
||||
// impossible while unproven, clearing the dirty latch on start (below) can no
|
||||
// longer erase an unproven stop's OTA block.
|
||||
bool mayRestart() const { return MQTTLifecycle::mayRestart(_state); }
|
||||
// The owner polls this to decide whether it may release/restart, and calls
|
||||
// onTaskStopped() again if the task acknowledges late.
|
||||
bool isStopUnproven() const { return MQTTLifecycle::isStopUnproven(_state); }
|
||||
// OTA erase/write is permitted only after a CLEAN stop. A timed-out stop
|
||||
// leaves ownership uncertain, so flashing stays blocked until a clean
|
||||
// start/stop cycle clears the latch.
|
||||
|
||||
@@ -223,6 +223,11 @@ static unsigned long s_wifi_connected_at = 0;
|
||||
|
||||
// Last WiFi disconnect reason (from ESP-IDF event). Used for get wifi.status diagnostics.
|
||||
static uint8_t s_wifi_disconnect_reason = 0;
|
||||
// Latched by end() when a stop passes its deadline unacknowledged. end() clears
|
||||
// the diagnostic singleton, so without this `get mqtt.status` could only say
|
||||
// "not running" and an operator would have no way to learn that the bridge is
|
||||
// down for the rest of the boot and why.
|
||||
static bool s_stop_unproven = false;
|
||||
static unsigned long s_wifi_disconnect_time = 0;
|
||||
|
||||
#ifdef MQTT_MEMORY_DEBUG
|
||||
@@ -267,7 +272,9 @@ void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPref
|
||||
if (buf == nullptr || bufsize == 0) return;
|
||||
const char* msgs = (obs && obs->mqtt_status_enabled) ? "on" : "off";
|
||||
if (s_mqtt_bridge_instance == nullptr || !s_mqtt_bridge_instance->_initialized) {
|
||||
snprintf(buf, bufsize, "> msgs: %s (bridge not running)", msgs);
|
||||
snprintf(buf, bufsize, "> msgs: %s (bridge %s)", msgs,
|
||||
s_stop_unproven ? "stopped: previous stop unproven, reboot to recover"
|
||||
: "not running");
|
||||
return;
|
||||
}
|
||||
MQTTBridge* b = s_mqtt_bridge_instance;
|
||||
@@ -456,6 +463,8 @@ void MQTTBridge::applyWifiPowerSave() {
|
||||
#endif
|
||||
}
|
||||
|
||||
bool MQTTBridge::stopUnprovenLatched() { return s_stop_unproven; }
|
||||
|
||||
uint8_t MQTTBridge::getLastWifiDisconnectReason() { return s_wifi_disconnect_reason; }
|
||||
unsigned long MQTTBridge::getLastWifiDisconnectTime() { return s_wifi_disconnect_time; }
|
||||
|
||||
@@ -871,6 +880,18 @@ void MQTTBridge::begin() {
|
||||
return;
|
||||
}
|
||||
|
||||
// A stop that never acknowledged leaves its task possibly still running and
|
||||
// everything it can reach still owned. Starting on top of that is the failure
|
||||
// F01 describes: a new start is not proof the previous clients stopped. Check
|
||||
// once for a late acknowledgement (which releases and clears this), then
|
||||
// refuse. Recovery is the task finishing, or a reboot.
|
||||
pollLateStopAck();
|
||||
if (!_lifecycle.mayRestart()) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT Bridge start refused: previous stop unproven (%s) - reboot to recover",
|
||||
MQTTLifecycle::stateName(_lifecycle.state()));
|
||||
return;
|
||||
}
|
||||
|
||||
// PSRAM diagnostic - helps debug memory fragmentation on boards with external RAM
|
||||
#ifdef BOARD_HAS_PSRAM
|
||||
{
|
||||
@@ -1067,8 +1088,9 @@ void MQTTBridge::begin() {
|
||||
// Clear the cooperative-stop handshake before the new task starts reading it.
|
||||
// deliverStop() leaves _stop_requested latched true after a stop cycle, so a
|
||||
// restart must reset it or the fresh task would self-terminate immediately.
|
||||
_stop_requested = false;
|
||||
_stop_acked = false;
|
||||
_stop_requested.store(false, std::memory_order_relaxed);
|
||||
_stop_acked.store(false, std::memory_order_relaxed);
|
||||
_teardown_complete.store(false, std::memory_order_relaxed);
|
||||
BaseType_t create_result = xTaskCreatePinnedToCore(
|
||||
mqttTask,
|
||||
"MQTTBridge",
|
||||
@@ -1161,14 +1183,14 @@ void MQTTBridge::end() {
|
||||
|
||||
#ifdef ESP_PLATFORM
|
||||
// Wait (bounded) for the task to acknowledge. tick() synthesizes the timeout
|
||||
// fallback if the task never acks. Checking the ack first each iteration means
|
||||
// a stop that completes right as the timeout expires is still treated as clean.
|
||||
// if the task never acks. Checking the ack first each iteration means a stop
|
||||
// that completes right as the timeout expires is still treated as clean.
|
||||
while (_lifecycle.isStopInProgress()) {
|
||||
if (_stop_acked) {
|
||||
_lifecycle.onTaskStopped(); // StopRequested -> Stopped (clean): releaseResources()
|
||||
if (_stop_acked.load(std::memory_order_acquire)) {
|
||||
_lifecycle.onTaskStopped(); // -> Stopped (proven): releaseResources()
|
||||
break;
|
||||
}
|
||||
_lifecycle.tick(); // may fire StopTimedOut -> Stopped (dirty): releaseResources()
|
||||
_lifecycle.tick(); // may fire StopTimedOut -> StopUnproven: releases NOTHING
|
||||
if (!_lifecycle.isStopInProgress()) break;
|
||||
vTaskDelay(pdMS_TO_TICKS(20));
|
||||
}
|
||||
@@ -1182,10 +1204,38 @@ void MQTTBridge::end() {
|
||||
|
||||
// Timezone is inline class storage (_timezone_storage) — nothing to delete.
|
||||
// The shared JSON document's pools were freed by releaseRuntimeBuffers() above.
|
||||
// Not running either way, so diagnostics and publishing stop. What differs is
|
||||
// ownership: after an unproven stop the task may still be alive and every
|
||||
// resource it can reach is deliberately still allocated (nothing was freed
|
||||
// above). begin() refuses until the task acknowledges, so _initialized == false
|
||||
// cannot be turned into a second task over the same state.
|
||||
_initialized = false;
|
||||
_slots_setup_done = false; // Reset so deferred setup runs again on next begin()
|
||||
MQTT_DEBUG_PRINTLN("MQTT Bridge stopped (%s)",
|
||||
_lifecycle.stopTimedOut() ? "forced/timeout - OTA blocked" : "clean");
|
||||
s_stop_unproven = _lifecycle.isStopUnproven();
|
||||
if (_lifecycle.isStopUnproven()) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT Bridge stop UNPROVEN after %lu ms: task did not acknowledge. "
|
||||
"Nothing released, restart refused, OTA blocked - reboot to recover.",
|
||||
(unsigned long)_lifecycle.stopTimeoutMs());
|
||||
} else {
|
||||
MQTT_DEBUG_PRINTLN("MQTT Bridge stopped (clean)");
|
||||
}
|
||||
}
|
||||
|
||||
// A stop whose deadline passed unproven is not necessarily wedged forever: the
|
||||
// MQTT task may simply have been slow (a blackholed WSS broker can hold
|
||||
// esp_mqtt_client_stop() well past the budget). Its acknowledgement means the
|
||||
// same thing whenever it arrives — teardown finished and the task is about to
|
||||
// stop executing — so honour it late: release the resources that were withheld
|
||||
// and let the bridge be restartable again. Called from begin() (the moment it
|
||||
// matters) and from the diagnostics path, both on the loop task.
|
||||
void MQTTBridge::pollLateStopAck() {
|
||||
if (!_lifecycle.isStopUnproven()) return;
|
||||
#ifdef ESP_PLATFORM
|
||||
if (!_stop_acked.load(std::memory_order_acquire)) return;
|
||||
MQTT_DEBUG_PRINTLN("MQTT task acknowledged its stop late - releasing withheld resources");
|
||||
_lifecycle.onTaskStopped(); // StopUnproven -> Stopped: releaseResources()
|
||||
s_stop_unproven = false;
|
||||
#endif
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -1203,38 +1253,27 @@ void MQTTBridge::LifecycleOps::startTask() {
|
||||
}
|
||||
|
||||
void MQTTBridge::LifecycleOps::deliverStop() {
|
||||
// Clear any stale ack before raising the request (same ordering as the NTP
|
||||
// handshake: clear the done-flag, then set the request). The MQTT task polls
|
||||
// _stop_requested at the top of mqttTaskLoop().
|
||||
_b->_stop_acked = false;
|
||||
_b->_stop_requested = true;
|
||||
// Clear the completion flags before raising the request, so the task cannot
|
||||
// observe a stale ack from a previous cycle. Release on the request store
|
||||
// publishes those clears to the MQTT task.
|
||||
_b->_stop_acked.store(false, std::memory_order_relaxed);
|
||||
_b->_teardown_complete.store(false, std::memory_order_relaxed);
|
||||
_b->_stop_requested.store(true, std::memory_order_release);
|
||||
}
|
||||
|
||||
void MQTTBridge::LifecycleOps::releaseResources() {
|
||||
MQTTBridge* b = _b;
|
||||
#ifdef ESP_PLATFORM
|
||||
// stopTimedOut() is set before this effect fires (Coordinator::dispatch), so
|
||||
// it reliably distinguishes a clean ack from the timeout fallback.
|
||||
const bool dirty = b->_lifecycle.stopTimedOut();
|
||||
if (dirty && !b->_stop_acked) {
|
||||
// Reviewed fallback: the task never acknowledged (likely wedged in mbedTLS).
|
||||
// Force-kill it and tear down clients here on Core 1 — the pre-cooperative
|
||||
// behavior — accepting the heap risk. The dirty latch keeps OTA flashing
|
||||
// blocked (canFlashAfterStop() == false) so firmware is never written after
|
||||
// this path.
|
||||
if (b->_mqtt_task_handle != nullptr) {
|
||||
vTaskDelete(b->_mqtt_task_handle);
|
||||
}
|
||||
// force: the task is already gone and the client is presumed wedged, so waiting on a
|
||||
// DISCONNECTED event that may never arrive would hang this task (the app loop) forever.
|
||||
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) b->teardownSlot(i, /*force=*/true);
|
||||
b->destroySlotClients(/*force=*/true);
|
||||
}
|
||||
// Clean path (or a task that acked right at the deadline): the MQTT task
|
||||
// already disconnected/deleted its clients on Core 0 and self-terminated, so
|
||||
// we must NOT touch slots here (that would be a cross-core double-delete).
|
||||
// Just drop our handle reference; FreeRTOS reclaims the self-deleted task's
|
||||
// dynamically-allocated stack/TCB in the idle task.
|
||||
// This effect now fires ONLY after an acknowledged stop — timely or late (see
|
||||
// MQTTLifecycle: StopTimedOut leads to StopUnproven, which releases nothing).
|
||||
// So there is no force-kill branch here any more, and there must not be one:
|
||||
// the task published the ack from its trampoline immediately before
|
||||
// vTaskDelete(nullptr), after destroying its own clients on Core 0. Deleting
|
||||
// that task from here, or tearing its clients down a second time, was the
|
||||
// reviewed use-after-free (F01, soak blocker #20).
|
||||
//
|
||||
// The MQTT task self-terminates; FreeRTOS reclaims its stack/TCB in the idle
|
||||
// task. We only drop our handle reference.
|
||||
b->_mqtt_task_handle = nullptr;
|
||||
|
||||
// Drain and delete the FreeRTOS packet queue (value-copied packets, no
|
||||
@@ -1282,8 +1321,14 @@ void MQTTBridge::mqttTask(void* parameter) {
|
||||
MQTTBridge* bridge = static_cast<MQTTBridge*>(parameter);
|
||||
if (bridge) {
|
||||
bridge->mqttTaskLoop();
|
||||
// Last act before ceasing to execute: publish the stop acknowledgement, but
|
||||
// only if the loop actually completed its ordered teardown. An unexpected
|
||||
// return (mqttTaskLoop() has no other exit) must not tell the owner it is
|
||||
// safe to free the queue, the buffers and the clients.
|
||||
if (bridge->_teardown_complete.load(std::memory_order_acquire)) {
|
||||
bridge->_stop_acked.store(true, std::memory_order_release);
|
||||
}
|
||||
}
|
||||
// Task should never return, but if it does, delete itself
|
||||
vTaskDelete(nullptr);
|
||||
}
|
||||
|
||||
@@ -1371,13 +1416,17 @@ void MQTTBridge::mqttTaskLoop() {
|
||||
// vTaskDelete. Acknowledge LAST so end() only frees the queue/buffers once
|
||||
// this teardown has completed, then self-terminate via the mqttTask()
|
||||
// trampoline (vTaskDelete(nullptr)).
|
||||
if (_stop_requested) {
|
||||
if (_stop_requested.load(std::memory_order_acquire)) {
|
||||
MQTT_DEBUG_PRINTLN("MQTT task: cooperative stop - tearing down clients on Core 0");
|
||||
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
|
||||
teardownSlot(i);
|
||||
}
|
||||
destroySlotClients();
|
||||
_stop_acked = true; // release semantics: set only after teardown is done
|
||||
// Record that the ordered teardown finished, but do NOT publish the ack
|
||||
// here: the owner treats the ack as permission to free everything this
|
||||
// task can reach, and between this point and vTaskDelete(nullptr) the task
|
||||
// is still executing. mqttTask() publishes it as its last act.
|
||||
_teardown_complete.store(true, std::memory_order_release);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -1789,17 +1838,11 @@ void MQTTBridge::releaseSlotAuthToken(int index) {
|
||||
slot.last_token_renewal = 0;
|
||||
}
|
||||
|
||||
void MQTTBridge::destroySlotClients(bool force) {
|
||||
void MQTTBridge::destroySlotClients() {
|
||||
for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
|
||||
MQTTSlot& slot = _slots[i];
|
||||
if (slot.client != nullptr) {
|
||||
// force is deliberately NOT gated on connected(). The state it exists for — a client
|
||||
// that already took its DISCONNECTED callback and is now stuck inside
|
||||
// esp_mqtt_client_stop() — reports not-connected, so gating skipped the stop exactly
|
||||
// when it mattered and left the object to be deleted from under a live IDF task.
|
||||
if (force) {
|
||||
slot.client->forceStop();
|
||||
} else if (slot.client->connected()) {
|
||||
if (slot.client->connected()) {
|
||||
slot.client->disconnect();
|
||||
}
|
||||
#ifdef ESP_PLATFORM
|
||||
@@ -2057,18 +2100,12 @@ bool MQTTBridge::setupSlot(int index) {
|
||||
// the client object alive so a subsequent setupSlot() can reuse its mbedTLS
|
||||
// context. This is called both on reconfigure (preset change) and at shutdown;
|
||||
// destruction of the underlying client happens once in destroySlotClients().
|
||||
void MQTTBridge::teardownSlot(int index, bool force) {
|
||||
void MQTTBridge::teardownSlot(int index) {
|
||||
if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return;
|
||||
MQTTSlot& slot = _slots[index];
|
||||
|
||||
// As in destroySlotClients(): force is not gated on connected(), because the wedged
|
||||
// mid-stop state it exists for already reports not-connected.
|
||||
if (slot.client && (force || slot.client->connected())) {
|
||||
if (force) {
|
||||
slot.client->forceStop();
|
||||
} else {
|
||||
slot.client->disconnect();
|
||||
}
|
||||
if (slot.client && slot.client->connected()) {
|
||||
slot.client->disconnect();
|
||||
#ifdef ESP_PLATFORM
|
||||
vTaskDelay(pdMS_TO_TICKS(50));
|
||||
#else
|
||||
|
||||
@@ -280,15 +280,22 @@ private:
|
||||
NtpDiagResult _ntp_diag_results[kMaxNtpServers];
|
||||
int _ntp_diag_count;
|
||||
|
||||
// Cooperative-shutdown handshake (Phase 5). The loop task (Core 1) raises
|
||||
// Cooperative-shutdown handshake. The loop task (Core 1) raises
|
||||
// _stop_requested through the lifecycle Coordinator; the MQTT task (Core 0)
|
||||
// sees it, tears down its own clients on Core 0 (where the mbedTLS contexts
|
||||
// live), sets _stop_acked LAST, and self-terminates. end() waits for the ack
|
||||
// before freeing the queue/buffers. Plain volatile matches the existing
|
||||
// NTP/reconfigure handshake idiom above; replacing all of these with a command
|
||||
// channel / task notifications is explicitly deferred (see MQTT_OWNERSHIP.md).
|
||||
volatile bool _stop_requested = false;
|
||||
volatile bool _stop_acked = false;
|
||||
// live), records _teardown_complete, and publishes _stop_acked from the task
|
||||
// trampoline immediately before vTaskDelete(nullptr). end() waits for that ack
|
||||
// before anything is freed.
|
||||
//
|
||||
// std::atomic, not volatile: this is a two-flag release/acquire handshake
|
||||
// across cores, and the owner acts on it by freeing memory the other task can
|
||||
// reach. `volatile` orders nothing and was the soak campaign's blocker #20.
|
||||
// Publishing the ack from the trampoline is what makes it mean "this task is
|
||||
// about to cease executing" rather than "teardown returned"; _teardown_complete
|
||||
// gates it so an unexpected return from mqttTaskLoop() cannot claim a clean stop.
|
||||
std::atomic<bool> _stop_requested{false};
|
||||
std::atomic<bool> _stop_acked{false};
|
||||
std::atomic<bool> _teardown_complete{false};
|
||||
|
||||
// Timezone handling.
|
||||
// _timezone_storage is inline class storage (zero heap) that is reconfigured
|
||||
@@ -467,10 +474,11 @@ private:
|
||||
bool ensureSlotClient(int index); // Allocate this slot's persistent client + callbacks on first use
|
||||
bool ensureSlotAuthToken(int index); // Allocate this slot's JWT token buffer on first token creation
|
||||
void releaseSlotAuthToken(int index);// Free the token buffer (only with the client — see MQTTSlot)
|
||||
// force=true stops each client without waiting for its DISCONNECTED event. Only the
|
||||
// dirty-stop fallback passes it: disconnect()'s wait is unbounded, so a client already
|
||||
// wedged in mbedTLS would block the caller — MyMesh::loop() — indefinitely.
|
||||
void destroySlotClients(bool force = false); // Delete all persistent clients (shutdown only)
|
||||
// No force variant: the only caller that ever passed one was the dirty-stop
|
||||
// fallback, and that fallback is gone (F01). A client whose stop cannot be
|
||||
// proven is now left alone rather than force-stopped and deleted under a
|
||||
// possibly-live SDK task.
|
||||
void destroySlotClients(); // Delete all persistent clients (shutdown only)
|
||||
bool setupSlot(int index); // Configure and connect the slot; false = not activated
|
||||
// Single definition of "this slot holds one of the _max_active_slots positions":
|
||||
// it is enabled and has been through a successful setupSlot(). Startup, the
|
||||
@@ -479,7 +487,7 @@ private:
|
||||
int activatedSlotCount() const;
|
||||
bool canActivateSlot(int index) const;
|
||||
// force as in destroySlotClients(): skip the unbounded wait, dirty-stop path only.
|
||||
void teardownSlot(int index, bool force = false); // Disconnect the slot's client (keeps the object alive)
|
||||
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);
|
||||
@@ -536,6 +544,9 @@ private:
|
||||
void logMemoryStatus();
|
||||
void refreshOriginFromPrefs();
|
||||
void applyWifiPowerSave(); // one mapping, applied on every association
|
||||
// Honours a stop acknowledgement that arrived after the deadline: releases the
|
||||
// withheld resources and makes the bridge restartable. Loop task only.
|
||||
void pollLateStopAck();
|
||||
// begin()/end()-scoped PSRAM buffers. Each allocation is independent so a
|
||||
// transient heap shortage degrades to the existing stack fallback instead
|
||||
// of making the bridge unusable.
|
||||
@@ -670,6 +681,13 @@ public:
|
||||
* OTA flashing is withheld until a clean start/stop cycle. Mirrors
|
||||
* MQTTLifecycle::mayBeginFlash(); read on the loop task (Core 1). */
|
||||
bool canFlashAfterStop() const { return _lifecycle.mayBeginFlash(); }
|
||||
// True when a stop passed its deadline without the MQTT task acknowledging:
|
||||
// the bridge is down, nothing was released, and it will not restart until the
|
||||
// task acknowledges late (pollLateStopAck) or the node reboots.
|
||||
bool isStopUnproven() const { return _lifecycle.isStopUnproven(); }
|
||||
// Survives end() clearing the diagnostic singleton, so `get mqtt.status` can
|
||||
// still explain why a stopped bridge will not come back without a reboot.
|
||||
static bool stopUnprovenLatched();
|
||||
|
||||
static unsigned long getWifiConnectedAtMillis();
|
||||
|
||||
|
||||
@@ -174,9 +174,12 @@ TEST(MQTTLifecycle, RestartAfterStop) {
|
||||
|
||||
// --- Timeout / fallback ----------------------------------------------------
|
||||
|
||||
// Handoff: "Timeout/fallback behavior when the MQTT task or client does not
|
||||
// acknowledge." Models the reviewed fallback replacing the abrupt vTaskDelete.
|
||||
TEST(MQTTLifecycle, StopTimeoutFiresReviewedFallback) {
|
||||
// F01: a stop that passes its deadline unacknowledged releases NOTHING. The
|
||||
// MQTT task may still be inside mbedTLS or holding the SDK's API mutex, so the
|
||||
// queue, buffers, clients and the task itself all stay owned. The OTA barrier
|
||||
// is still unblocked — so it can abort — but nothing is freed and the bridge
|
||||
// may not restart.
|
||||
TEST(MQTTLifecycle, StopTimeoutReleasesNothingAndBlocksRestart) {
|
||||
FakeOps ops;
|
||||
L::Coordinator c(ops, kStopTimeoutMs);
|
||||
bringUpToRunning(c);
|
||||
@@ -191,18 +194,70 @@ TEST(MQTTLifecycle, StopTimeoutFiresReviewedFallback) {
|
||||
EXPECT_EQ(L::State::StopRequested, c.state());
|
||||
EXPECT_EQ(0, ops.release_calls);
|
||||
|
||||
// At the deadline the fallback fires: forced release, dirty OTA signal.
|
||||
ops.now = 1000 + kStopTimeoutMs;
|
||||
c.tick();
|
||||
EXPECT_EQ(L::State::Stopped, c.state());
|
||||
|
||||
EXPECT_EQ(L::State::StopUnproven, c.state());
|
||||
EXPECT_TRUE(c.isStopUnproven());
|
||||
EXPECT_TRUE(c.stopTimedOut());
|
||||
EXPECT_EQ(1, ops.release_calls);
|
||||
EXPECT_EQ(0, ops.release_calls) << "an unproven stop must not release";
|
||||
EXPECT_EQ(1, ops.stop_complete_calls);
|
||||
EXPECT_FALSE(ops.last_stop_clean);
|
||||
|
||||
// A late ack after the fallback does not double-release.
|
||||
// Everything the task can reach is still owned, and neither a restart nor a
|
||||
// flash is permitted. A repeated tick does not change that.
|
||||
EXPECT_TRUE(c.mayTouchOwnedState());
|
||||
EXPECT_FALSE(c.mayRestart());
|
||||
EXPECT_FALSE(c.mayBeginFlash());
|
||||
EXPECT_FALSE(c.acceptsNewWork());
|
||||
EXPECT_FALSE(c.isStopInProgress()) << "end()'s wait loop must terminate";
|
||||
|
||||
ops.now += kStopTimeoutMs * 10;
|
||||
c.tick();
|
||||
EXPECT_EQ(L::State::StopUnproven, c.state());
|
||||
EXPECT_EQ(0, ops.release_calls);
|
||||
|
||||
// A start attempt is refused outright: a new start is not proof that the
|
||||
// previous clients stopped.
|
||||
EXPECT_FALSE(c.requestStart());
|
||||
EXPECT_EQ(L::State::StopUnproven, c.state());
|
||||
EXPECT_EQ(1, ops.start_task_calls) << "no second task while unproven";
|
||||
EXPECT_TRUE(c.stopTimedOut()) << "a refused start cannot clear the dirty latch";
|
||||
}
|
||||
|
||||
// The deadline is an availability policy, not a statement about what an ack
|
||||
// means: the task publishes it only after tearing its clients down, so a late
|
||||
// ack is exactly as trustworthy as a timely one. Honouring it releases the
|
||||
// resources and makes the bridge restartable without a reboot.
|
||||
TEST(MQTTLifecycle, LateAckRecoversFromAnUnprovenStop) {
|
||||
FakeOps ops;
|
||||
L::Coordinator c(ops, kStopTimeoutMs);
|
||||
bringUpToRunning(c);
|
||||
|
||||
ops.now = 1000;
|
||||
ASSERT_TRUE(c.requestStop());
|
||||
ops.now = 1000 + kStopTimeoutMs;
|
||||
c.tick();
|
||||
ASSERT_EQ(L::State::StopUnproven, c.state());
|
||||
ASSERT_EQ(0, ops.release_calls);
|
||||
|
||||
ops.now = 1000 + kStopTimeoutMs * 4;
|
||||
ASSERT_TRUE(c.onTaskStopped());
|
||||
|
||||
EXPECT_EQ(L::State::Stopped, c.state());
|
||||
EXPECT_EQ(1, ops.release_calls) << "the proof arrived; release now";
|
||||
EXPECT_TRUE(c.mayRestart());
|
||||
// The OTA barrier already reported dirty and its caller acted on that, so the
|
||||
// flash gate stays shut until a clean start/stop cycle.
|
||||
EXPECT_EQ(1, ops.stop_complete_calls);
|
||||
EXPECT_FALSE(c.mayBeginFlash());
|
||||
EXPECT_TRUE(c.stopTimedOut());
|
||||
|
||||
// Restarting from the proven stop clears the latch, and a second ack is inert.
|
||||
EXPECT_FALSE(c.onTaskStopped());
|
||||
EXPECT_EQ(1, ops.release_calls);
|
||||
ASSERT_TRUE(c.requestStart());
|
||||
EXPECT_FALSE(c.stopTimedOut());
|
||||
}
|
||||
|
||||
TEST(MQTTLifecycle, TickWithoutPendingStopIsNoop) {
|
||||
@@ -371,7 +426,9 @@ TEST(MQTTLifecycle, OtaFlashBlockedUntilCleanStopAcknowledged) {
|
||||
}
|
||||
|
||||
// Handoff OTA barrier: "MQTT stop times out: OTA aborts safely rather than
|
||||
// writing under uncertain ownership."
|
||||
// writing under uncertain ownership." The barrier is still released so the
|
||||
// caller aborts, but flashing stays shut and — unlike the pre-F01 contract —
|
||||
// the bridge cannot be restarted to paper over it.
|
||||
TEST(MQTTLifecycle, OtaAbortsWhenStopTimesOut) {
|
||||
FakeOps ops;
|
||||
L::Coordinator c(ops, kStopTimeoutMs);
|
||||
@@ -382,11 +439,16 @@ TEST(MQTTLifecycle, OtaAbortsWhenStopTimesOut) {
|
||||
ops.now = 500 + kStopTimeoutMs;
|
||||
c.tick();
|
||||
|
||||
EXPECT_EQ(L::State::Stopped, c.state());
|
||||
EXPECT_FALSE(c.mayBeginFlash()); // dirty stop => flashing withheld
|
||||
EXPECT_FALSE(ops.last_stop_clean);
|
||||
EXPECT_EQ(L::State::StopUnproven, c.state());
|
||||
EXPECT_EQ(1, ops.stop_complete_calls); // barrier released...
|
||||
EXPECT_FALSE(ops.last_stop_clean); // ...with "do not flash"
|
||||
EXPECT_FALSE(c.mayBeginFlash());
|
||||
EXPECT_FALSE(c.requestStart()); // and no restart to hide it
|
||||
|
||||
// A fresh clean start/stop cycle clears the latch and re-enables flashing.
|
||||
// Only the task's late acknowledgement makes the bridge usable again, and
|
||||
// only a clean cycle after that re-enables flashing.
|
||||
ASSERT_TRUE(c.onTaskStopped());
|
||||
EXPECT_FALSE(c.mayBeginFlash());
|
||||
ASSERT_TRUE(c.requestStart());
|
||||
EXPECT_FALSE(c.stopTimedOut());
|
||||
ASSERT_TRUE(c.onTaskStarted());
|
||||
@@ -415,8 +477,13 @@ TEST(MQTTLifecycle, NoRestartWhileStopInProgress) {
|
||||
}
|
||||
|
||||
// Handoff OTA barrier: "Repeated failed OTA attempts do not ... leave the
|
||||
// bridge permanently stopped."
|
||||
TEST(MQTTLifecycle, RepeatedFailedStopsLeaveBridgeRestartable) {
|
||||
// bridge permanently stopped." Re-encoded for the F01 contract: what keeps the
|
||||
// bridge usable across repeated slow stops is the task's acknowledgement, not
|
||||
// the deadline. A stop that is merely slow recovers every time; a stop that is
|
||||
// never acknowledged deliberately does NOT, because restarting on top of
|
||||
// clients that may still be running is the failure this state exists to
|
||||
// prevent. That is an availability trade, made knowingly.
|
||||
TEST(MQTTLifecycle, RepeatedSlowStopsRecoverOnTheirLateAck) {
|
||||
FakeOps ops;
|
||||
L::Coordinator c(ops, kStopTimeoutMs);
|
||||
|
||||
@@ -427,19 +494,51 @@ TEST(MQTTLifecycle, RepeatedFailedStopsLeaveBridgeRestartable) {
|
||||
ops.now += 1000;
|
||||
ASSERT_TRUE(c.requestStop());
|
||||
ops.now += kStopTimeoutMs;
|
||||
c.tick(); // times out (dirty)
|
||||
c.tick(); // deadline passes with no ack
|
||||
EXPECT_EQ(L::State::StopUnproven, c.state());
|
||||
EXPECT_FALSE(c.mayRestart()); // not until the task proves it finished
|
||||
EXPECT_EQ(attempt, ops.release_calls);
|
||||
|
||||
ops.now += 1000;
|
||||
ASSERT_TRUE(c.onTaskStopped()); // the slow teardown completes
|
||||
EXPECT_EQ(L::State::Stopped, c.state());
|
||||
EXPECT_TRUE(c.mayRestart()); // never permanently stuck
|
||||
EXPECT_TRUE(c.mayRestart());
|
||||
EXPECT_EQ(attempt + 1, ops.release_calls);
|
||||
}
|
||||
EXPECT_EQ(3, ops.start_task_calls);
|
||||
}
|
||||
|
||||
// A never-acknowledged stop stays unusable for the rest of the boot: no
|
||||
// release, no restart, no flash, however long the owner waits or however many
|
||||
// times it asks.
|
||||
TEST(MQTTLifecycle, NeverAcknowledgedStopStaysUnproven) {
|
||||
FakeOps ops;
|
||||
L::Coordinator c(ops, kStopTimeoutMs);
|
||||
bringUpToRunning(c);
|
||||
|
||||
ASSERT_TRUE(c.requestStop());
|
||||
ops.now += kStopTimeoutMs;
|
||||
c.tick();
|
||||
|
||||
for (int i = 0; i < 5; ++i) {
|
||||
ops.now += 60'000;
|
||||
c.tick();
|
||||
EXPECT_FALSE(c.requestStart());
|
||||
EXPECT_FALSE(c.requestStop());
|
||||
EXPECT_EQ(L::State::StopUnproven, c.state());
|
||||
}
|
||||
EXPECT_EQ(0, ops.release_calls);
|
||||
EXPECT_FALSE(c.mayBeginFlash());
|
||||
EXPECT_TRUE(c.mayTouchOwnedState());
|
||||
}
|
||||
|
||||
// --- Diagnostics -----------------------------------------------------------
|
||||
|
||||
TEST(MQTTLifecycle, StateAndEventNamesAreStable) {
|
||||
EXPECT_STREQ("Stopped", L::stateName(L::State::Stopped));
|
||||
EXPECT_STREQ("Running", L::stateName(L::State::Running));
|
||||
EXPECT_STREQ("StopRequested", L::stateName(L::State::StopRequested));
|
||||
EXPECT_STREQ("StopUnproven", L::stateName(L::State::StopUnproven));
|
||||
EXPECT_STREQ("StartRequested", L::eventName(L::Event::StartRequested));
|
||||
EXPECT_STREQ("StopTimedOut", L::eventName(L::Event::StopTimedOut));
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user