fix(mqtt): typed results for client operations, and stop advancing state on failure (F06)

`connect()`, `reconnect()`, `disconnect()`, `softDisconnect()` and `forceStop()`
all returned void, so every caller in the bridge treated "asked" as "done":

- `setupSlot()` marked a slot activated after `connect()` whatever happened. A
  failed `esp_mqtt_client_start()` therefore consumed one of the scarce
  active-slot positions, handed the slot to a reconnect ladder that is gated on
  activation, and was never retried by the deferred-setup path.
- `reconnect()` explicitly proceeded after `esp_mqtt_set_config()` failed,
  reconnecting on the previous configuration — the renewed token in the buffer,
  the old one on the wire.
- the renewal path recorded the new expiry before the bounce succeeded, so a
  failed bounce left the live session on the old credential with the next
  renewal not due for a whole token lifetime.
- `softDisconnect()` logged its timeout and told its caller nothing.
- `disconnect()` waited for the DISCONNECTED event with no bound, on the very
  task whose stop acknowledgement the shutdown waits for.

Now every one of them returns `esp_err_t`, a failed configuration transaction
aborts rather than starting or reconnecting on a half-updated config, and
`disconnect()`'s wait is bounded (it still stops the client, and reports
ESP_ERR_TIMEOUT when the event never arrived).

Bridge consequences:

- a failed start leaves the slot unactivated, so the existing deferred-setup
  retry revisits it and it holds no active-slot position;
- a failed renewal bounce re-arms the renewal instead of recording it, so the
  next maintenance pass retries;
- a reconnect that fails *locally* rolls back the backoff advance made for it.
  The ladder and the breaker bound broker and network faults; an uninitialised
  client or an uncommitted config transaction is neither, and inflating the
  ladder for it was how a local fault could trip a breaker meant for a broker.
This commit is contained in:
agessaman committed 2026-09-09 17:40:45 -07:00
1 parent ca986d40b5
commit abe838bf58
4 files changed
+194 -69

No files matched your search

+86 -47
View File
@@ -371,20 +371,20 @@ bool PsychicMqttClient::connected()
return _connected;
}
void PsychicMqttClient::connect()
esp_err_t PsychicMqttClient::connect()
{
#if ESP_IDF_VERSION_MAJOR == 5
if (_mqtt_cfg.broker.address.uri == nullptr)
{
ESP_LOGE(TAG, "MQTT URI not set.");
return;
return ESP_ERR_INVALID_STATE;
}
int desired_buffer = _mqtt_cfg.buffer.size > 0 ? _mqtt_cfg.buffer.size : 1024;
#else
if (_mqtt_cfg.uri == nullptr)
{
ESP_LOGE(TAG, "MQTT URI not set.");
return;
return ESP_ERR_INVALID_STATE;
}
int desired_buffer = _mqtt_cfg.buffer_size > 0 ? _mqtt_cfg.buffer_size : 1024;
#endif
@@ -400,41 +400,44 @@ void PsychicMqttClient::connect()
{
ESP_LOGE(TAG, "Failed to allocate reassembly buffer (%u bytes)", (unsigned)_buffer_capacity);
_buffer_capacity = 0;
return ESP_ERR_NO_MEM;
}
}
if (_client == nullptr)
{
_client = esp_mqtt_client_init(&_mqtt_cfg);
if (_client == nullptr)
{
ESP_LOGE(TAG, "esp_mqtt_client_init failed");
return ESP_ERR_NO_MEM;
}
// Register event handler only once when client is first created
// to avoid memory leak from repeated registrations
esp_mqtt_client_register_event(_client, MQTT_EVENT_ANY, _onMqttEventStatic, this);
_config_dirty = false;
}
else if (_config_dirty)
{
// A failed config update must not be followed by a start: the client
// would come up on a partly-updated configuration, which is how a slot
// connects with the previous broker's credentials. Leave _config_dirty
// set so the next attempt retries the whole transaction.
esp_err_t cfg_result = esp_mqtt_set_config(_client, &_mqtt_cfg);
if (cfg_result != ESP_OK)
{
ESP_LOGE(TAG, "connect(): failed to apply mqtt config: %s", esp_err_to_name(cfg_result));
return cfg_result;
}
_config_dirty = false;
ESP_LOGD(TAG, "connect(): applied mqtt config update");
}
else
{
if (_config_dirty)
{
esp_err_t cfg_result = esp_mqtt_set_config(_client, &_mqtt_cfg);
ESP_ERROR_CHECK_WITHOUT_ABORT(cfg_result);
if (cfg_result == ESP_OK)
{
_config_dirty = false;
ESP_LOGD(TAG, "connect(): applied mqtt config update");
}
else
{
ESP_LOGW(TAG, "connect(): failed to apply mqtt config, will retry");
}
}
else
{
ESP_LOGD(TAG, "connect(): mqtt config unchanged, skipping set_config");
}
ESP_LOGD(TAG, "connect(): mqtt config unchanged, skipping set_config");
}
esp_err_t start_result = esp_mqtt_client_start(_client);
ESP_ERROR_CHECK_WITHOUT_ABORT(start_result);
if (start_result == ESP_OK)
{
_started = true;
@@ -445,76 +448,103 @@ void PsychicMqttClient::connect()
// 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));
}
return start_result;
}
void PsychicMqttClient::reconnect()
esp_err_t PsychicMqttClient::reconnect()
{
if (_client == nullptr)
{
ESP_LOGW(TAG, "MQTT client not initialized, cannot reconnect.");
return;
return ESP_ERR_INVALID_STATE;
}
if (_config_dirty)
{
// Apply config only when mutating setters changed _mqtt_cfg.
// Apply config only when mutating setters changed _mqtt_cfg. A failure
// aborts the reconnect: reconnecting on the previous config was how a
// renewed token silently failed to reach the connection.
esp_err_t cfg_result = esp_mqtt_set_config(_client, &_mqtt_cfg);
ESP_ERROR_CHECK_WITHOUT_ABORT(cfg_result);
if (cfg_result == ESP_OK)
if (cfg_result != ESP_OK)
{
_config_dirty = false;
ESP_LOGD(TAG, "reconnect(): applied mqtt config update");
}
else
{
ESP_LOGW(TAG, "reconnect(): failed to apply mqtt config, reconnecting with previous config");
ESP_LOGE(TAG, "reconnect(): failed to apply mqtt config: %s", esp_err_to_name(cfg_result));
return cfg_result;
}
_config_dirty = false;
ESP_LOGD(TAG, "reconnect(): applied mqtt config update");
}
else
{
ESP_LOGD(TAG, "reconnect(): mqtt config unchanged, skipping set_config");
}
ESP_ERROR_CHECK_WITHOUT_ABORT(esp_mqtt_client_reconnect(_client));
esp_err_t r = esp_mqtt_client_reconnect(_client);
if (r != ESP_OK)
{
ESP_LOGE(TAG, "MQTT reconnect request failed: %s", esp_err_to_name(r));
return r;
}
ESP_LOGI(TAG, "MQTT client reconnect requested.");
return ESP_OK;
}
void PsychicMqttClient::disconnect()
esp_err_t PsychicMqttClient::disconnect(unsigned long timeout_ms)
{
if (_client == nullptr)
{
ESP_LOGW(TAG, "MQTT client not started.");
return;
return ESP_ERR_INVALID_STATE;
}
bool clean = true;
if (_connected)
{
ESP_LOGI(TAG, "Disconnecting MQTT client.");
_stopMqttClient = false;
esp_mqtt_client_disconnect(_client);
// Wait for all disconnect events to be processed
while (!_stopMqttClient)
// Bounded, unlike the original: this runs on the MQTT task whose stop
// acknowledgement the bridge's shutdown waits for, so a lost
// DISCONNECTED event used to wedge the whole teardown.
unsigned long waited = 0;
while (!_stopMqttClient && waited < timeout_ms)
{
vTaskDelay(10 / portTICK_PERIOD_MS);
waited += 10;
}
if (!_stopMqttClient)
{
ESP_LOGW(TAG, "disconnect: no DISCONNECTED event in %lums; stopping anyway", timeout_ms);
clean = false;
}
}
esp_mqtt_client_stop(_client);
_started = false;
ESP_LOGI(TAG, "MQTT client stopped.");
esp_err_t stop_result = esp_mqtt_client_stop(_client);
if (stop_result == ESP_OK)
{
_started = false;
ESP_LOGI(TAG, "MQTT client stopped.");
}
else
{
// The SDK task was not joined. Saying otherwise is what let the caller
// destroy the client from under a live task.
ESP_LOGE(TAG, "esp_mqtt_client_stop failed: %s", esp_err_to_name(stop_result));
return stop_result;
}
return clean ? ESP_OK : ESP_ERR_TIMEOUT;
}
void PsychicMqttClient::softDisconnect(unsigned long timeout_ms)
esp_err_t PsychicMqttClient::softDisconnect(unsigned long timeout_ms)
{
if (_client == nullptr)
{
ESP_LOGW(TAG, "MQTT client not started.");
return;
return ESP_ERR_INVALID_STATE;
}
if (!_connected)
{
// Nothing to close; leaving the task alone is the whole point.
return;
return ESP_OK;
}
ESP_LOGI(TAG, "Disconnecting MQTT transport (client task retained).");
@@ -530,25 +560,34 @@ void PsychicMqttClient::softDisconnect(unsigned long timeout_ms)
if (!_stopMqttClient)
{
ESP_LOGW(TAG, "softDisconnect: no DISCONNECTED event in %lums", timeout_ms);
return ESP_ERR_TIMEOUT;
}
return ESP_OK;
}
void PsychicMqttClient::forceStop()
esp_err_t PsychicMqttClient::forceStop()
{
if (_client == nullptr)
{
ESP_LOGW(TAG, "MQTT client not started.");
return;
return ESP_ERR_INVALID_STATE;
}
if (_connected)
{
ESP_LOGI(TAG, "Forced stop MQTT client.");
}
ESP_ERROR_CHECK_WITHOUT_ABORT(esp_mqtt_client_stop(_client));
esp_err_t r = esp_mqtt_client_stop(_client);
if (r != ESP_OK)
{
// The SDK task was not joined: the caller must not destroy this client.
ESP_LOGE(TAG, "forceStop: esp_mqtt_client_stop failed: %s", esp_err_to_name(r));
return r;
}
_connected = false;
_started = false;
ESP_LOGI(TAG, "MQTT client forcefully stopped.");
return ESP_OK;
}
int PsychicMqttClient::subscribe(const char *topic, int qos)
+39 -8
View File
@@ -349,8 +349,17 @@ public:
* @brief Connects the MQTT client to the server.
*
* @note All parameters must be set before calling this method.
*
* @return ESP_OK when esp_mqtt_client_start() accepted the client.
* ESP_ERR_INVALID_STATE if no URI is set, ESP_ERR_NO_MEM if the
* client or its reassembly buffer could not be allocated, or the
* error from esp_mqtt_set_config()/esp_mqtt_client_start().
*
* A failed configuration update does NOT start the client: starting with a
* half-updated config is how a slot ends up connected with the previous
* broker's credentials.
*/
void connect();
esp_err_t connect();
/**
* @brief Reconnects a previously started MQTT client.
@@ -359,14 +368,28 @@ public:
* re-initiating a connection on an already-started client, especially
* when auto-reconnect is disabled. Updates config before reconnecting
* so credential changes (e.g., refreshed JWT tokens) take effect.
*
* @return ESP_OK when the reconnect was requested, ESP_ERR_INVALID_STATE on
* an uninitialised client, or the error from esp_mqtt_set_config().
* A failed config update aborts the reconnect rather than
* reconnecting with the previous configuration.
*/
void reconnect();
esp_err_t reconnect();
/**
* @brief Disconnects the MQTT client from the server.
* This call might be blocking until the client is stopped cleanly
* @brief Disconnects the MQTT client and stops its task.
*
* Blocks until the DISCONNECTED event arrives or timeout_ms elapses, then
* stops the client either way. The wait used to be unbounded, which turned
* a lost event into a wedged caller — and that caller is the MQTT task
* whose acknowledgement the bridge's shutdown waits on.
*
* @return ESP_OK when the client stopped after a clean disconnect,
* ESP_ERR_TIMEOUT when the DISCONNECTED event never arrived (the
* stop still ran), ESP_ERR_INVALID_STATE on an uninitialised
* client, or the error from esp_mqtt_client_stop().
*/
void disconnect();
esp_err_t disconnect(unsigned long timeout_ms = 10000);
/**
* @brief Closes the transport but leaves the client task running.
@@ -380,8 +403,12 @@ public:
* @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.
*
* @return ESP_OK when the DISCONNECTED event arrived, ESP_ERR_TIMEOUT if it
* did not (the transport may still be open), ESP_ERR_INVALID_STATE
* on an uninitialised client, ESP_OK when already disconnected.
*/
void softDisconnect(unsigned long timeout_ms = 5000);
esp_err_t softDisconnect(unsigned long timeout_ms = 5000);
/**
* @brief True once esp_mqtt_client_start() has succeeded and no stop has run.
@@ -392,10 +419,14 @@ public:
bool isStarted() const { return _started; }
/**
* @brief Forcefully stops the MQTT client and disconnects from the server.
* @brief Stops the MQTT client without waiting for a DISCONNECTED event.
* This does not trigger the onDisconnect callbacks.
*
* @return the result of esp_mqtt_client_stop(), or ESP_ERR_INVALID_STATE on
* an uninitialised client. A non-OK result means the SDK task was
* NOT joined: the caller must not destroy the client after it.
*/
void forceStop();
esp_err_t forceStop();
/**
* @brief Subscribes to a topic. Server must be connected
+65 -13
View File
@@ -2091,7 +2091,19 @@ bool MQTTBridge::setupSlot(int index) {
}
}
slot.client->connect();
// Activation is now conditional on the client actually starting. A failed
// esp_mqtt_set_config()/esp_mqtt_client_start() used to be invisible: the slot
// was marked activated, so it consumed one of the scarce active-slot
// positions, the reconnect ladder (which is gated on activation) governed it,
// and nothing retried the setup. Leaving it unactivated hands it to the
// existing deferred-setup retry in maintainSlotConnections() instead.
const esp_err_t connect_result = slot.client->connect();
if (connect_result != ESP_OK) {
MQTT_DEBUG_PRINTLN("MQTT%d start failed (%s) - will retry", index + 1,
esp_err_to_name(connect_result));
slot.last_reconnect_attempt = millis();
return false;
}
slot.initial_connect_done = true;
return true;
}
@@ -2137,17 +2149,16 @@ void MQTTBridge::teardownSlot(int index) {
// which only stops slots still marked connected: a publishing slot's socket fails first, so
// the guard skips it — measured across a 62 s deauth, five slots, zero stops. An idle slot
// with no traffic to fail on is the one case that could still reach here that way.
void MQTTBridge::reconnectSlotClient(int index) {
if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return;
esp_err_t MQTTBridge::reconnectSlotClient(int index) {
if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return ESP_ERR_INVALID_ARG;
MQTTSlot& slot = _slots[index];
if (slot.client == nullptr) return;
if (slot.client == nullptr) return ESP_ERR_INVALID_STATE;
if (!slot.client->isStarted()) {
MQTT_DEBUG_PRINTLN("MQTT%d start (client was stopped)", index + 1);
slot.client->connect();
return;
return slot.client->connect();
}
slot.client->reconnect();
return slot.client->reconnect();
}
@@ -2304,21 +2315,40 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
// to avoid internal heap leak/fragmentation from destroy/create cycles
MQTT_DEBUG_PRINTLN("MQTT%d token renewal: reconnecting with fresh credentials", index + 1);
MQTT_TRACE_HEAP("renewal:before-bounce", index);
esp_err_t bounce_result;
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();
const esp_err_t soft = slot.client->softDisconnect();
if (soft != ESP_OK) {
MQTT_DEBUG_PRINTLN("MQTT%d renewal: soft disconnect did not complete (%s)",
index + 1, esp_err_to_name(soft));
}
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();
bounce_result = 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();
bounce_result = slot.client->connect();
}
if (bounce_result != ESP_OK) {
// The fresh token is in the buffer but did not reach the
// connection: the config transaction failed, or the client would
// not start. Recording this as a completed renewal would leave the
// live session on the old credential until the broker enforced exp,
// with the next renewal not due for a whole token lifetime. Re-arm
// the renewal instead so the next maintenance pass retries the
// bounce, and leave the reconnect ladder alone — this is a local
// failure, not a broker fault.
MQTT_DEBUG_PRINTLN("MQTT%d renewal bounce failed (%s) - retrying next pass",
index + 1, esp_err_to_name(bounce_result));
slot.last_token_renewal = 0;
return;
}
MQTT_TRACE_HEAP("renewal:after-reconnect", index);
reconnect_attempted = true;
@@ -2439,6 +2469,9 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
static_cast<uint32_t>(now_millis), static_cast<uint32_t>(slot.last_reconnect_attempt),
slot.reconnect_backoff, static_cast<uint8_t>(index))) {
slot.last_reconnect_attempt = now_millis;
const uint8_t backoff_before = slot.reconnect_backoff;
const uint8_t failures_before = slot.max_backoff_failures;
const bool breaker_before = slot.circuit_breaker_tripped;
MQTTConnectionPolicy::BackoffAdvance advance = MQTTConnectionPolicy::advanceBackoff(
slot.reconnect_backoff, slot.max_backoff_failures);
slot.reconnect_backoff = advance.reconnect_backoff;
@@ -2463,7 +2496,18 @@ void MQTTBridge::maintainSlotConnection(int index, unsigned long now_millis, uns
}
// Via the helper: reconnect() is a no-op on a client whose start failed,
// which would back off forever without ever starting it.
reconnectSlotClient(index);
const esp_err_t r = reconnectSlotClient(index);
if (r != ESP_OK) {
// A local failure — uninitialised client, or a config transaction that
// did not commit — is not a broker fault. The ladder and the breaker
// exist to bound broker/network faults, so roll back the advance made
// above; the retry interval still paces the next attempt.
MQTT_DEBUG_PRINTLN("MQTT%d reconnect not accepted locally (%s) - backoff unchanged",
index + 1, esp_err_to_name(r));
slot.reconnect_backoff = backoff_before;
slot.max_backoff_failures = failures_before;
slot.circuit_breaker_tripped = breaker_before;
}
}
}
}
@@ -4367,8 +4411,16 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) {
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();
const esp_err_t soft = _slots[i].client->softDisconnect();
if (soft != ESP_OK) {
MQTT_DEBUG_PRINTLN("MQTT%d soft disconnect did not complete (%s)", i + 1,
esp_err_to_name(soft));
}
const esp_err_t rc = _slots[i].client->reconnect();
if (rc != ESP_OK) {
MQTT_DEBUG_PRINTLN("MQTT%d corrected-clock reconnect failed (%s)", i + 1,
esp_err_to_name(rc));
}
} else {
MQTT_DEBUG_PRINTLN("MQTT%d token re-created, no bounce (broker does not enforce exp)",
i + 1);
+4 -1
View File
@@ -490,7 +490,10 @@ private:
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);
// ESP_OK when the reconnect/start was accepted by the SDK. A local failure
// (uninitialised client, failed config transaction) is not a broker fault and
// must not advance this slot's backoff ladder — see maintainSlotConnection().
esp_err_t 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