feat(mqtt): add Phase 4 lifecycle ownership + teardown test seams

Fork-owned, host-tested MQTT bridge lifecycle state machine and the narrow
dependency seam to drive it deterministically, plus the cross-core ownership
model. This is the Phase 4 "ownership and teardown test seams" safety net that
must land before the Phase 5 cooperative-shutdown refactor.

- src/helpers/MQTTLifecycle.h: pure state machine
  (Stopped->Starting->Running->StopRequested->Stopping->Stopped) + injected
  Ops seam (clock / task / resource owner / OTA barrier) + Coordinator with a
  bounded stop timeout. No Arduino/FreeRTOS/WiFi deps.
- test/test_mqtt_lifecycle/: 18 GoogleTest cases covering the phase's teardown
  matrix (stop during every activity, callback timing, duplicate/early stop,
  restart, timeout fallback, no-access-after-release) and the OTA-barrier
  scenarios.
- MQTT_OWNERSHIP.md: one owner per mutable runtime domain, current cross-core
  hazards with file:line references, target primitives, and Phase-0-pending
  (hardware-characterization) items.
- STABILITY_TESTABILITY_HANDOFF.md: Phase 4 status and verified premise
  refinements.

Scope: MQTTBridge.cpp is intentionally untouched. The production rewiring
(plain-data snapshot publication, volatile-handshake replacement, cooperative
shutdown) is deferred to Phase 5.
This commit is contained in:
agessaman committed 2026-07-18 20:31:33 -07:00
1 parent 1dfd2af56c
commit 9d5bd6f898
4 files changed
+962 -3

No files matched your search

+166
View File
@@ -0,0 +1,166 @@
# MQTT Bridge Cross-Core Ownership Model
This document is the Phase 4 deliverable from `STABILITY_TESTABILITY_HANDOFF.md`:
it records **one owner for each mutable runtime domain**, maps every place a
non-owner reads owned state across cores today, and states the target primitive
for each. It is paired with the fork-owned lifecycle test seam
(`src/helpers/MQTTLifecycle.h`, `test/test_mqtt_lifecycle/`).
**Status:** ownership model documented and the lifecycle/teardown test seam
landed. The production rewiring it describes (publishing a plain-data snapshot,
replacing the `volatile` handshakes, cooperative shutdown) is **deferred to
Phase 5** — this document is the plan Phase 5 executes, not a description of
current behavior. Line references are against the tree at the time of writing
and should be re-verified before editing.
## Execution contexts
- **MQTT task — Core 0** (`"MQTTBridge"`, `xTaskCreatePinnedToCore(..., MQTT_TASK_CORE=0)`,
entry `mqttTask` → `mqttTaskLoop`, `MQTTBridge.cpp:923`, `:991`). Owns all
WiFi / MQTT / NTP I/O and every mutation of `_slots[]`, connection state, and
NTP state.
- **Loop task — Core 1** (`MyMesh::loop()`). Runs the CLI, WebConfig `tick()`,
and `AlertReporter::onLoop()`. `examples/simple_repeater/MyMesh.cpp:1529`
notes the bridge loop is *not* called here — it lives on Core 0.
- **Producer / radio context.** Stages raw radio bytes before queue handoff
(`storeRawRadioData` → `_staged_*`, consumed by `queuePacket`; both Core 1, in
guaranteed sequence, `MQTTBridge.h:232-239`).
- **Async TCP context.** WebConfig request parsing and immutable response
handoff only.
## Ownership model (one owner per mutable domain)
| Domain | Owner | Notes |
|--------|-------|-------|
| MQTT clients, slot connection state, publish counters, packet drain, NTP I/O | **MQTT task (Core 0)** | `_slots[]`, `PsychicMqttClient`s, `_ntp_client`, `_last_raw_*` |
| CLI execution, preference persistence, WebConfig batch draining, bridge lifecycle requests | **Loop task (Core 1)** | issues start/stop/reconfigure, drains the WebConfig batch |
| Packet staging before queue handoff | **Producer / radio (Core 1)** | `_staged_*`, no lock needed (sequential) |
| Request parse + immutable response | **Async TCP** | must not touch mutable bridge state |
The rule that follows: **loop/WebConfig/CLI/AlertReporter code must not directly
inspect mutable MQTT slot objects or client counters.** The MQTT task must
publish a plain-data snapshot they can read instead.
## Current cross-core hazards (to resolve in Phase 5)
All of the following run on **Core 1** and read state mutated by **Core 0**
without a lock or a published snapshot.
### 1. Diagnostic reads of live `_slots[]` via the singleton
Four `static` accessors reach the live object through the file-scope
`s_mqtt_bridge_instance` (`MQTTBridge.cpp:201`, set at `begin()` end `:849`,
nulled first in `end()` `:858` — a plain, non-atomic pointer):
- `getSlotStatusSnapshot` (`:299`) — despite its name, built on demand from
live `_slots[slot_index]`. **Refined premise:** the returned `name`/`state`
`const char*`s point at static rodata (string literals / `MQTT_PRESETS[]`),
so they are *not* dangling. The real hazards are (a) reading the
`slot.preset` **pointer value**, which Core 0 can null/reassign mid-read
(`applySlotPreset`, `teardownSlot`), and (b) `slot.client->getPublishOk()`,
a live client pointer Core 0 can `delete` during teardown (UAF window).
- `formatMqttStatusReply` (`:207`), `formatMqttStatsReply` (`:263`),
`formatSlotDiagReply` (`:397`) — same singleton + live `_slots[]` reads.
Consumers: CLI (`CommonCLI_Observer.cpp:726, :728, :791`) and the app-layer
stats JSON (`examples/simple_repeater/MyMesh.cpp:1423`, mirrored in
`examples/simple_room_server/MyMesh.cpp:1050`). **Refined premise:**
`WebConfigServer.cpp` itself reads only a compile-time constant
(`getMaxActiveSlots`, `:479`); the live-bridge web reads are in `buildStatsJson`
in the `MyMesh.cpp` app layer.
### 2. Instance reads that survive `end()`
`AlertReporter` (Core 1, `onLoop`, `AlertReporter.cpp:208`) reads live `_slots[]`
via instance methods it reaches through its own `_bridge` pointer
(`AlertReporter.cpp:281` `isSlotEnabledAndAttempted`, `:285`
`getSlotCurrentOutageStartMs`, `:296`/`:311` `getSlotPresetName`). The stats
JSON also calls `bridge->getQueueSize()` (`MyMesh.cpp:1419`).
**Refined premise (teardown hazard):** `end()` nulls only
`s_mqtt_bridge_instance`. The app's `bridge` pointer and `AlertReporter::_bridge`
are **not** cleared by `end()`, so these instance reads can touch a torn-down
bridge. (`getConnectedBrokers()` at `MQTTBridge.cpp:3676` is defined but has
zero consumers.)
### 3. `volatile` cross-core handshakes
Plain `volatile` flags (no atomics/barriers), Core 1 sets the request, Core 0
clears/processes and writes a "done" flag last, Core 1 spins:
- `_slot_reconfigure_pending[]` — set `MQTTBridge.cpp:2057` (Core 1), read/clear
`:1118` (Core 0).
- `_ntp_force_{requested,done,result}` — request `:3298`, process `:1063-1068`,
spin-wait `:3308-3315`.
- `_ntp_diag_{requested,done}` (+ non-volatile `_ntp_diag_results[]`) — request
`:3344`, process `:1073-1076`, spin-wait `:3347`.
The two blocking waiters (`requestForcedNtpSync` `:3298`, `ntpDiag` `:3344`) spin
on Core 1 while `end()` could tear down the singleton/task concurrently.
### 4. Abrupt teardown, no start guard
- `end()` uses `vTaskDelete(_mqtt_task_handle)` (`MQTTBridge.cpp:862`) — the task
is killed wherever it is (possibly mid-`_slots[]` mutation, or inside
mbedTLS); slot/client cleanup then runs *after* deletion on the caller's
context (`:902-905`). This is the OTA teardown heap-panic path.
- `begin()` (`:626`) has **no double-call guard** — a second `begin()` re-runs
allocation and `xTaskCreatePinnedToCore`, leaking the prior queue/task.
- Lifecycle state is a single `_initialized` bool; there is no state enum.
## Target primitives (Phase 5)
- **Task notifications or a command queue** for one-way lifecycle / reconfigure
/ NTP requests — replacing every `volatile` flag in §3.
- **Immutable published snapshots** for WebConfig, CLI diagnostics, and alerting
— the MQTT task publishes a plain-data `SlotStatusSnapshot` (owned char
buffers + scalars, no live pointers) that §1/§2 consumers read. This also
removes the "instance pointer survives `end()`" hazard because consumers stop
dereferencing the live bridge.
- **Atomics** only for truly independent scalar state.
- **A mutex** only where ownership transfer or snapshot publication cannot
express the operation cleanly.
## Lifecycle contract (the test seam)
`src/helpers/MQTTLifecycle.h` encodes the cooperative lifecycle Phase 5 must
implement, as a pure state machine plus a narrow injected `Ops` seam
(clock / task control / resource owner / OTA barrier). The invariants proven by
`test/test_mqtt_lifecycle/` — and that Phase 5's production wiring must preserve:
- `Stopped → Starting → Running → StopRequested → Stopping → Stopped`.
- Idempotent start and stop; safe restart only from `Stopped` (`mayRestart`).
- New connects/publishes/retries/reconfigurations cease once a stop is requested
(`acceptsNewWork`).
- Resources are released **only** after a stop acknowledgment (or the reviewed
timeout fallback) — never mid-run.
- A late/stale callback consults `mayTouchOwnedState()` and is a no-op once the
owner has released resources.
- Bounded stop timeout → reviewed fallback (models replacing the abrupt
`vTaskDelete`).
- **OTA barrier:** `mayBeginFlash()` is true only after a **clean** stop
acknowledgment; a timed-out stop leaves flashing blocked so OTA aborts rather
than writing under uncertain ownership.
### Phase 0 / hardware-pending items
Per the "derive from code + flag" discipline, these are **not** encoded as
constants and must be characterized on hardware before Phase 5 ships:
- The concrete stop timeout (injected as `Coordinator`'s `stop_timeout_ms`;
must be measured against mbedTLS teardown over `wss` with a down broker).
- Exact callback timing/ordering under a real TLS disconnect.
- Heap / largest-block / task-stack high-water and task/client counts across
start/stop/restart (Phase 7 gates).
## Deferred to Phase 5 (not done here)
- Publishing the plain-data snapshot and repointing the §1/§2 consumers at it.
- Replacing the §3 `volatile` handshakes with a command queue / task
notifications.
- The cooperative shutdown state machine in `MQTTBridge` (`end()` rewrite, the
`begin()` double-call guard) and the OTA teardown barrier.
`MQTTBridge.cpp` was intentionally left untouched in Phase 4 to keep the
merge-sensitive file (≈3.7k lines) free of churn until the Phase 5 change lands
as a single reviewable unit.
+27 -3
View File
@@ -26,8 +26,8 @@ index.
| 1 | PR CI smoke builds + ArduinoJson pin enforcement | Done (build-size gate and ASan/UBSan still pending) |
| 2 | PSRAM restart resource symmetry | Done |
| 3 | MQTT preference migration fixtures | Done (filesystem adapter still lives in `CommonCLI`) |
| 0 | Pre-change lifecycle characterization | Not started — next actionable step |
| 4 | Ownership and teardown test seams | Not started |
| 0 | Pre-change lifecycle characterization | Not started — deterministic behavior encoded in Phase 4; hardware items pending |
| 4 | Ownership and teardown test seams | Seams + ownership doc + teardown tests done; production rewiring deferred to Phase 5 |
| 5 | Cooperative MQTT shutdown | Not started (fixes the OTA teardown panic) |
| — | OTA teardown barrier | Not started — release-critical |
| 6 | Request/queue/connection/publication integration tests | Not started |
@@ -268,11 +268,35 @@ Acceptance criteria:
### Phase 4: Establish ownership and teardown test seams
**Status: Not started.** Verified premise: the loop task, WebConfig, CLI, and
**Status: Seams, ownership doc, and teardown tests landed; production rewiring
deferred to Phase 5 by explicit decision (scope: "seams + tests only").** The
fork-owned pure lifecycle state machine and narrow dependency seam
(`src/helpers/MQTTLifecycle.h`), the teardown-focused test matrix
(`test/test_mqtt_lifecycle/`), and the ownership model (`MQTT_OWNERSHIP.md`) are
in place. `MQTTBridge.cpp` was intentionally left untouched to keep the
merge-sensitive file free of churn until the Phase 5 change lands as one
reviewable unit. The invasive production work — publishing the plain-data
snapshot and repointing consumers at it, replacing the `volatile` handshakes
with a command queue / task notifications, and the cooperative-shutdown `end()`
rewrite with a `begin()` double-call guard — is carried into Phase 5.
Verified premise (with Phase 4 refinements): the loop task, WebConfig, CLI, and
`AlertReporter` currently read the MQTT task's live, mutable slot objects and
client counters cross-core without a lock or a published snapshot.
`getSlotStatusSnapshot()` is built on demand from live state despite its name —
the refactor must actually publish a plain-data snapshot, not assume one exists.
Refinements found while mapping the code (see `MQTT_OWNERSHIP.md` for
file:line references):
- The snapshot's `name`/`state` `const char*`s point at static rodata, so they
are not dangling; the real hazards are reading the mutable `slot.preset`
pointer value (Core 0 can null/reassign it) and `slot.client->getPublishOk()`
(a client pointer Core 0 can `delete` during teardown).
- The live-bridge web reads are in the `MyMesh.cpp` app layer (`buildStatsJson`),
not `WebConfigServer.cpp`, which reads only a compile-time constant.
- `end()` clears only the `s_mqtt_bridge_instance` singleton; the app's `bridge`
pointer and `AlertReporter::_bridge` are not cleared, so instance reads can
touch a torn-down bridge.
This phase is the safety net for cooperative shutdown. It must land before the
shutdown state machine.
+319
View File
@@ -0,0 +1,319 @@
#pragma once
#include <stdint.h>
// Fork-owned, dependency-free MQTT bridge lifecycle state machine and the
// narrow dependency seam used to drive it deterministically in host tests.
//
// This is the Phase 4 "ownership and teardown test seam" from
// STABILITY_TESTABILITY_HANDOFF.md. It is intentionally pure (no Arduino,
// FreeRTOS, WiFi, or PsychicMqttClient dependencies) so the exact
// start/stop/restart contract can be exercised with a fake clock and a
// recording Ops double, the same way MQTTConnectionPolicy.h and
// MQTTRuntimeBufferLifecycle.h are tested.
//
// Scope boundary (Phase 4 vs Phase 5): this header is the SPEC and the test
// seam. It is not yet wired into MQTTBridge. Phase 5 ("Implement cooperative
// MQTT shutdown") supplies FreeRTOS/PsychicMqttClient-backed Ops and replaces
// the abrupt vTaskDelete in MQTTBridge::end() with this cooperative lifecycle.
// See MQTT_OWNERSHIP.md for the ownership model and the migration plan.
//
// Behavior source (Phase 0 discipline): every transition and invariant encoded
// here is derived from the current MQTTBridge.cpp control flow (begin()/end(),
// mqttTaskLoop(), the volatile handshakes) — not from a hardware soak. Values
// that require on-hardware characterization (the concrete stop timeout, exact
// mbedTLS teardown timing) are called out with "Phase 0 TODO" and left as
// injectable parameters rather than guessed constants.
namespace MQTTLifecycle {
// The lifecycle proposed by the handoff:
// Stopped -> Starting -> Running -> StopRequested -> Stopping -> Stopped
//
// StopRequested: a stop has been requested and delivered through the ownership
// channel, but the MQTT task has not yet begun its ordered shutdown.
// Stopping: the MQTT task is performing its ordered client/service
// shutdown. A StopBegan signal is optional; a task may ack directly from
// StopRequested if it does not report the intermediate step.
enum class State : uint8_t {
Stopped = 0,
Starting,
Running,
StopRequested,
Stopping,
};
// Events driven either by the owner (loop task) or by the MQTT task reporting
// its own progress. StopTimedOut is synthesized by the Coordinator when a stop
// is not acknowledged within the bounded timeout (the reviewed fallback).
enum class Event : uint8_t {
StartRequested = 0, // owner asked the bridge to start
StartCompleted, // MQTT task signalled init complete (StartAck)
StartFailed, // init failed on the task (partial-init rollback)
StopRequested, // owner (or OTA barrier) asked the bridge to stop
StopBegan, // MQTT task began its ordered shutdown (optional)
StopAcknowledged, // MQTT task signalled ordered shutdown complete (StopAck)
StopTimedOut, // bounded timeout expired with no StopAck (fallback)
};
// Side effects a transition asks the caller to perform. Naming WHO does WHAT
// keeps the Phase 5 production wiring and the host fakes on one contract:
// - create_task: owner creates/pins the MQTT task.
// - deliver_stop: owner delivers the stop request through the channel.
// - release_resources: owner may now free the queue, runtime buffers, and the
// task handle. Fires only after a completed/forced stop
// or an init-failure rollback — never mid-run.
// - ota_release: the OTA barrier's completion acknowledgment is now
// available (a stop reached a terminal state).
struct Effects {
bool create_task = false;
bool deliver_stop = false;
bool release_resources = false;
bool ota_release = false;
};
struct Result {
State next;
Effects effects;
bool accepted; // false => the event was a no-op in this state (idempotency)
};
// Unsigned subtraction is the standard millis() idiom and stays correct across
// a single 32-bit rollover (mirrors MQTTConnectionPolicy::elapsedMs).
inline uint32_t elapsedMs(uint32_t now, uint32_t then) { return now - then; }
// Pure transition function. For a rejected/no-op event the result reports the
// unchanged state, no effects, and accepted == false.
inline Result apply(State s, Event e) {
Result r{s, Effects{}, false};
switch (s) {
case State::Stopped:
// Only a start is meaningful. A stop while already stopped is a no-op
// (idempotent stop). StartFailed/StopAck cannot occur here.
if (e == Event::StartRequested) {
r.next = State::Starting;
r.effects.create_task = true;
r.accepted = true;
}
break;
case State::Starting:
switch (e) {
case Event::StartCompleted:
r.next = State::Running;
r.accepted = true;
break;
case Event::StartFailed:
// Partial-init rollback: release only what the attempt acquired.
r.next = State::Stopped;
r.effects.release_resources = true;
r.accepted = true;
break;
case Event::StopRequested:
// Stop before full initialization: accept it and let the task ack.
r.next = State::StopRequested;
r.effects.deliver_stop = true;
r.accepted = true;
break;
default:
break;
}
break;
case State::Running:
if (e == Event::StopRequested) {
r.next = State::StopRequested;
r.effects.deliver_stop = true;
r.accepted = true;
}
// A duplicate StartRequested/StartCompleted while Running is a no-op.
break;
case State::StopRequested:
switch (e) {
case Event::StopBegan:
r.next = State::Stopping;
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;
default:
// Duplicate StopRequested is a no-op (idempotent stop).
break;
}
break;
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;
default:
break;
}
break;
}
return r;
}
// New connects/publishes/retries/reconfigurations are permitted only while the
// bridge is actively running (or still coming up). Once a stop is requested,
// all new work must cease (handoff: "Cessation of new connects, publishes,
// retries, and reconfigurations").
inline bool acceptsNewWork(State s) {
return s == State::Starting || s == State::Running;
}
// A late/stale client callback may touch owner-released resources ONLY before
// the owner has freed them. After the terminal Stopped state the queue,
// buffers, and clients may be gone, so a callback arriving then must be a
// no-op. (Callbacks that fire during StopRequested/Stopping run before
// release_resources and are still safe.)
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; }
inline bool isStopInProgress(State s) {
return s == State::StopRequested || s == State::Stopping;
}
inline const char* stateName(State s) {
switch (s) {
case State::Stopped: return "Stopped";
case State::Starting: return "Starting";
case State::Running: return "Running";
case State::StopRequested: return "StopRequested";
case State::Stopping: return "Stopping";
}
return "?";
}
inline const char* eventName(Event e) {
switch (e) {
case Event::StartRequested: return "StartRequested";
case Event::StartCompleted: return "StartCompleted";
case Event::StartFailed: return "StartFailed";
case Event::StopRequested: return "StopRequested";
case Event::StopBegan: return "StopBegan";
case Event::StopAcknowledged: return "StopAcknowledged";
case Event::StopTimedOut: return "StopTimedOut";
}
return "?";
}
// Narrow dependency seam. These are the only dependencies the lifecycle needs
// to be driven deterministically (handoff: "for only the dependencies needed").
// Phase 5 implements this over FreeRTOS + PsychicMqttClient; host tests
// implement it as a recording double with a settable clock.
//
// Dependencies enumerated by the handoff and where they land:
// - Clock/timer -> nowMs()
// - Task start/stop/ack/timeout -> startTask()/deliverStop() + the
// onTaskStarted()/onTaskStopped() callbacks
// into Coordinator, and tick() for timeout
// - Runtime allocator + queue -> folded into releaseResources() here; the
// allocate/free symmetry itself is already
// covered by MQTTRuntimeBufferLifecycle.h,
// and queue behavior by Phase 6
// - MQTT client connect/disconnect + delayed callbacks -> modeled by the
// mayTouchOwnedState() guard (a callback
// decides whether it may touch owned state)
// - OTA coordinator/barrier -> onStopComplete(clean) + mayBeginFlash()
struct Ops {
virtual ~Ops() = default;
virtual uint32_t nowMs() = 0; // monotonic ms (millis())
virtual void startTask() = 0; // create/pin the MQTT task
virtual void deliverStop() = 0; // signal stop through the channel
virtual void releaseResources() = 0; // free queue/buffers/task (post-stop)
// Unblock the OTA barrier's waiter. clean == true after a StopAcknowledged;
// clean == false after a StopTimedOut, so OTA aborts rather than flashing
// under uncertain ownership (handoff OTA barrier: "MQTT stop times out: OTA
// aborts safely rather than writing under uncertain ownership").
virtual void onStopComplete(bool clean) = 0;
};
// Drives the state machine against injected Ops and hosts the bounded stop
// timeout. Idempotent start/stop; safe restart after a completed stop.
class Coordinator {
public:
// Phase 0 TODO: stop_timeout_ms must be characterized against real mbedTLS
// teardown timing over wss with a down broker before Phase 5 ships. It is an
// injected parameter precisely so it is not a guessed constant here.
Coordinator(Ops& ops, uint32_t stop_timeout_ms)
: _ops(ops), _stop_timeout_ms(stop_timeout_ms) {}
State state() const { return _state; }
bool stopTimedOut() const { return _stop_timed_out; }
bool acceptsNewWork() const { return MQTTLifecycle::acceptsNewWork(_state); }
bool mayTouchOwnedState() const {
return MQTTLifecycle::mayTouchOwnedState(_state);
}
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.
bool mayRestart() const { return MQTTLifecycle::mayRestart(_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.
bool mayBeginFlash() const {
return _state == State::Stopped && !_stop_timed_out;
}
bool requestStart() { return dispatch(Event::StartRequested); }
bool requestStop() { return dispatch(Event::StopRequested); }
bool onTaskStarted() { return dispatch(Event::StartCompleted); }
bool onTaskStartFailed() { return dispatch(Event::StartFailed); }
bool onStopBegan() { return dispatch(Event::StopBegan); }
bool onTaskStopped() { return dispatch(Event::StopAcknowledged); }
// Call periodically from the owner. Fires the reviewed timeout fallback if a
// requested stop has not been acknowledged within stop_timeout_ms.
void tick() {
if (MQTTLifecycle::isStopInProgress(_state) &&
elapsedMs(_ops.nowMs(), _stop_request_ms) >= _stop_timeout_ms) {
dispatch(Event::StopTimedOut);
}
}
private:
bool dispatch(Event e) {
const Result r = apply(_state, e);
if (!r.accepted) return false;
_state = r.next;
if (e == Event::StartRequested) {
_stop_timed_out = false; // a fresh start clears the dirty-stop latch
} else if (e == Event::StopRequested) {
_stop_request_ms = _ops.nowMs(); // arm the timeout window
} else if (e == Event::StopTimedOut) {
_stop_timed_out = true;
}
if (r.effects.create_task) _ops.startTask();
if (r.effects.deliver_stop) _ops.deliverStop();
if (r.effects.release_resources) _ops.releaseResources();
if (r.effects.ota_release) _ops.onStopComplete(e == Event::StopAcknowledged);
return true;
}
Ops& _ops;
uint32_t _stop_timeout_ms;
State _state = State::Stopped;
uint32_t _stop_request_ms = 0;
bool _stop_timed_out = false;
};
} // namespace MQTTLifecycle
@@ -0,0 +1,450 @@
#include <gtest/gtest.h>
#include <string>
#include <vector>
#include "helpers/MQTTLifecycle.h"
// Phase 4 teardown-focused test seam for the MQTT bridge lifecycle. Every case
// below maps to a bullet in STABILITY_TESTABILITY_HANDOFF.md's "Required
// teardown-focused tests" list or its "OTA Teardown Barrier" scenarios. The
// behavior encoded here is derived from the current MQTTBridge control flow
// (Phase 0 "derive from code + flag" discipline); it is the contract Phase 5
// must preserve when it wires the real bridge into MQTTLifecycle::Coordinator.
namespace L = MQTTLifecycle;
namespace {
// Recording Ops double with a settable clock and an ordered call log.
struct FakeOps : public L::Ops {
uint32_t now = 0;
int start_task_calls = 0;
int deliver_stop_calls = 0;
int release_calls = 0;
int stop_complete_calls = 0;
bool last_stop_clean = false;
std::vector<std::string> log;
uint32_t nowMs() override { return now; }
void startTask() override {
start_task_calls++;
log.push_back("startTask");
}
void deliverStop() override {
deliver_stop_calls++;
log.push_back("deliverStop");
}
void releaseResources() override {
release_calls++;
log.push_back("release");
}
void onStopComplete(bool clean) override {
stop_complete_calls++;
last_stop_clean = clean;
log.push_back(clean ? "otaClean" : "otaDirty");
}
};
const uint32_t kStopTimeoutMs = 5000;
void bringUpToRunning(L::Coordinator& c) {
ASSERT_TRUE(c.requestStart());
ASSERT_EQ(L::State::Starting, c.state());
ASSERT_TRUE(c.onTaskStarted());
ASSERT_EQ(L::State::Running, c.state());
}
} // namespace
// --- Normal ordering -------------------------------------------------------
// Handoff Phase 0: "Normal begin() -> connect -> end() ordering."
TEST(MQTTLifecycle, NormalStartRunStopCycle) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
EXPECT_EQ(L::State::Stopped, c.state());
ASSERT_TRUE(c.requestStart());
EXPECT_EQ(L::State::Starting, c.state());
ASSERT_TRUE(c.onTaskStarted());
EXPECT_EQ(L::State::Running, c.state());
ASSERT_TRUE(c.requestStop());
EXPECT_EQ(L::State::StopRequested, c.state());
ASSERT_TRUE(c.onTaskStopped());
EXPECT_EQ(L::State::Stopped, c.state());
// create_task, then deliver_stop, then a single post-ack resource release
// and a clean OTA-barrier signal — in that order.
const std::vector<std::string> expected = {"startTask", "deliverStop",
"release", "otaClean"};
EXPECT_EQ(expected, ops.log);
EXPECT_EQ(1, ops.start_task_calls);
EXPECT_EQ(1, ops.deliver_stop_calls);
EXPECT_EQ(1, ops.release_calls);
EXPECT_TRUE(ops.last_stop_clean);
}
// --- Idempotency: "Duplicate stop", "Idempotent start and stop requests" ---
TEST(MQTTLifecycle, StartIsIdempotent) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
ASSERT_TRUE(c.requestStart());
// Second start while Starting is a no-op — no extra task creation.
EXPECT_FALSE(c.requestStart());
ASSERT_TRUE(c.onTaskStarted());
// And a no-op while Running.
EXPECT_FALSE(c.requestStart());
EXPECT_EQ(1, ops.start_task_calls);
}
TEST(MQTTLifecycle, DuplicateStopIsIdempotent) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
bringUpToRunning(c);
ASSERT_TRUE(c.requestStop());
// Duplicate stop requests do not re-deliver the stop.
EXPECT_FALSE(c.requestStop());
EXPECT_FALSE(c.requestStop());
EXPECT_EQ(1, ops.deliver_stop_calls);
ASSERT_TRUE(c.onTaskStopped());
// A stop while already Stopped is also a no-op.
EXPECT_FALSE(c.requestStop());
EXPECT_EQ(1, ops.release_calls);
}
// --- Partial / early lifecycle --------------------------------------------
// Handoff: "stop before full initialization".
TEST(MQTTLifecycle, StopBeforeFullInitialization) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
ASSERT_TRUE(c.requestStart());
EXPECT_EQ(L::State::Starting, c.state());
// Stop arrives before StartCompleted.
ASSERT_TRUE(c.requestStop());
EXPECT_EQ(L::State::StopRequested, c.state());
EXPECT_EQ(1, ops.deliver_stop_calls);
// No resources released until the task acknowledges.
EXPECT_EQ(0, ops.release_calls);
ASSERT_TRUE(c.onTaskStopped());
EXPECT_EQ(L::State::Stopped, c.state());
EXPECT_EQ(1, ops.release_calls);
}
// Handoff Phase 2/Phase 0: "Partial initialization failures ... releases only
// resources owned by that attempt." The rollback releases, but no OTA-barrier
// completion is signalled for a failed start.
TEST(MQTTLifecycle, StartFailureRollsBackResources) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
ASSERT_TRUE(c.requestStart());
ASSERT_TRUE(c.onTaskStartFailed());
EXPECT_EQ(L::State::Stopped, c.state());
EXPECT_EQ(1, ops.release_calls);
EXPECT_EQ(0, ops.stop_complete_calls);
// Recoverable: a subsequent start is accepted.
EXPECT_TRUE(c.mayRestart());
EXPECT_TRUE(c.requestStart());
EXPECT_EQ(2, ops.start_task_calls);
}
// Handoff: "restart after stop".
TEST(MQTTLifecycle, RestartAfterStop) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
bringUpToRunning(c);
ASSERT_TRUE(c.requestStop());
ASSERT_TRUE(c.onTaskStopped());
EXPECT_TRUE(c.mayRestart());
ASSERT_TRUE(c.requestStart());
EXPECT_EQ(L::State::Starting, c.state());
ASSERT_TRUE(c.onTaskStarted());
EXPECT_EQ(L::State::Running, c.state());
EXPECT_EQ(2, ops.start_task_calls);
}
// --- 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) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
bringUpToRunning(c);
ops.now = 1000;
ASSERT_TRUE(c.requestStop());
EXPECT_EQ(L::State::StopRequested, c.state());
// Before the deadline, tick() is inert.
ops.now = 1000 + kStopTimeoutMs - 1;
c.tick();
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_TRUE(c.stopTimedOut());
EXPECT_EQ(1, ops.release_calls);
EXPECT_EQ(1, ops.stop_complete_calls);
EXPECT_FALSE(ops.last_stop_clean);
// A late ack after the fallback does not double-release.
EXPECT_FALSE(c.onTaskStopped());
EXPECT_EQ(1, ops.release_calls);
}
TEST(MQTTLifecycle, TickWithoutPendingStopIsNoop) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
bringUpToRunning(c);
ops.now = 1'000'000; // far past any deadline
c.tick();
EXPECT_EQ(L::State::Running, c.state());
EXPECT_EQ(0, ops.release_calls);
}
TEST(MQTTLifecycle, AckBeforeTimeoutPreemptsFallback) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
bringUpToRunning(c);
ops.now = 100;
ASSERT_TRUE(c.requestStop());
ops.now = 100 + kStopTimeoutMs / 2;
ASSERT_TRUE(c.onTaskStopped());
EXPECT_TRUE(ops.last_stop_clean);
EXPECT_FALSE(c.stopTimedOut());
// A later tick past the original deadline must not fire a second time.
ops.now = 100 + kStopTimeoutMs * 10;
c.tick();
EXPECT_EQ(1, ops.release_calls);
EXPECT_EQ(1, ops.stop_complete_calls);
}
// --- Ownership: no access after release ------------------------------------
// Handoff: "No client, queue, buffer, or task access after its owner releases
// it," and "A completion acknowledgment before the loop task releases queues,
// buffers, or other shared resources."
TEST(MQTTLifecycle, ResourcesReleasedOnlyAfterAcknowledgment) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
bringUpToRunning(c);
EXPECT_TRUE(c.mayTouchOwnedState());
ASSERT_TRUE(c.requestStop());
// Still safe to touch owned state — resources are not yet released.
EXPECT_TRUE(c.mayTouchOwnedState());
EXPECT_EQ(0, ops.release_calls);
ASSERT_TRUE(c.onStopBegan());
EXPECT_EQ(L::State::Stopping, c.state());
EXPECT_TRUE(c.mayTouchOwnedState());
EXPECT_EQ(0, ops.release_calls);
ASSERT_TRUE(c.onTaskStopped());
// Owner has released; nothing may touch owned state now.
EXPECT_FALSE(c.mayTouchOwnedState());
EXPECT_EQ(1, ops.release_calls);
}
// Handoff: "Cessation of new connects, publishes, retries, and
// reconfigurations" once a stop is requested.
TEST(MQTTLifecycle, NewWorkCeasesOnceStopRequested) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
EXPECT_FALSE(c.acceptsNewWork()); // Stopped
ASSERT_TRUE(c.requestStart());
EXPECT_TRUE(c.acceptsNewWork()); // Starting
ASSERT_TRUE(c.onTaskStarted());
EXPECT_TRUE(c.acceptsNewWork()); // Running
ASSERT_TRUE(c.requestStop());
EXPECT_FALSE(c.acceptsNewWork()); // StopRequested
ASSERT_TRUE(c.onStopBegan());
EXPECT_FALSE(c.acceptsNewWork()); // Stopping
ASSERT_TRUE(c.onTaskStopped());
EXPECT_FALSE(c.acceptsNewWork()); // Stopped
}
// --- Stop while doing X (the activity matrix) ------------------------------
// Handoff: "Stop while connecting, connected, publishing, retrying, renewing a
// token, running NTP, and applying a slot change." Each activity resolves to
// the same ownership contract: the stop is delivered, new work ceases, and
// resources survive until the task acknowledges. "connecting" happens during
// Starting (before StartCompleted); the rest are Running-state activities.
TEST(MQTTLifecycle, StopDuringEveryActivityHonorsTheContract) {
struct Activity {
const char* label;
bool running; // true => Running-state activity, false => Starting
};
const Activity activities[] = {
{"connecting", false}, {"connected", true},
{"publishing", true}, {"retrying", true},
{"renewingToken", true}, {"runningNtp", true},
{"applyingSlotChange", true},
};
for (const auto& a : activities) {
SCOPED_TRACE(a.label);
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
ASSERT_TRUE(c.requestStart());
if (a.running) {
ASSERT_TRUE(c.onTaskStarted());
ASSERT_EQ(L::State::Running, c.state());
} else {
ASSERT_EQ(L::State::Starting, c.state());
}
ASSERT_TRUE(c.requestStop());
EXPECT_EQ(1, ops.deliver_stop_calls);
EXPECT_FALSE(c.acceptsNewWork());
EXPECT_EQ(0, ops.release_calls); // not until ack
ASSERT_TRUE(c.onTaskStopped());
EXPECT_EQ(L::State::Stopped, c.state());
EXPECT_EQ(1, ops.release_calls);
EXPECT_TRUE(ops.last_stop_clean);
}
}
// --- Callback delivery timing ----------------------------------------------
// Handoff: "Callback delivered before stop, during stop, after disconnect, and
// after the stop acknowledgment." A callback consults mayTouchOwnedState()
// before touching owned state; only the post-acknowledgment callback is a
// no-op. ("after disconnect" is modeled as StopBegan -> Stopping, i.e. the
// task tearing the client down.)
TEST(MQTTLifecycle, CallbackGuardTracksResourceOwnership) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
bringUpToRunning(c);
EXPECT_TRUE(c.mayTouchOwnedState()); // callback before stop
ASSERT_TRUE(c.requestStop());
EXPECT_TRUE(c.mayTouchOwnedState()); // callback during stop (pre-teardown)
ASSERT_TRUE(c.onStopBegan());
EXPECT_TRUE(c.mayTouchOwnedState()); // callback after disconnect begins
ASSERT_TRUE(c.onTaskStopped());
EXPECT_FALSE(c.mayTouchOwnedState()); // callback after stop-ack => no-op
}
// --- OTA teardown barrier (release-critical) -------------------------------
// Handoff OTA barrier: "firmware erase/write must not begin until MQTT shutdown
// has reached a safe acknowledgment point."
TEST(MQTTLifecycle, OtaFlashBlockedUntilCleanStopAcknowledged) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
bringUpToRunning(c);
EXPECT_FALSE(c.mayBeginFlash()); // Running
ASSERT_TRUE(c.requestStop());
EXPECT_FALSE(c.mayBeginFlash()); // StopRequested — not yet safe
ASSERT_TRUE(c.onStopBegan());
EXPECT_FALSE(c.mayBeginFlash()); // Stopping — still not safe
ASSERT_TRUE(c.onTaskStopped());
EXPECT_TRUE(c.mayBeginFlash()); // clean stop => flashing permitted
EXPECT_TRUE(ops.last_stop_clean);
}
// Handoff OTA barrier: "MQTT stop times out: OTA aborts safely rather than
// writing under uncertain ownership."
TEST(MQTTLifecycle, OtaAbortsWhenStopTimesOut) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
bringUpToRunning(c);
ops.now = 500;
ASSERT_TRUE(c.requestStop());
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);
// A fresh clean start/stop cycle clears the latch and re-enables flashing.
ASSERT_TRUE(c.requestStart());
EXPECT_FALSE(c.stopTimedOut());
ASSERT_TRUE(c.onTaskStarted());
ASSERT_TRUE(c.requestStop());
ASSERT_TRUE(c.onTaskStopped());
EXPECT_TRUE(c.mayBeginFlash());
}
// Handoff OTA barrier: "the bridge must not be restarted while flash writing is
// active" — expressed here as: no restart while a stop is in progress.
TEST(MQTTLifecycle, NoRestartWhileStopInProgress) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
bringUpToRunning(c);
ASSERT_TRUE(c.requestStop());
EXPECT_FALSE(c.mayRestart());
EXPECT_FALSE(c.requestStart()); // rejected while StopRequested
ASSERT_TRUE(c.onStopBegan());
EXPECT_FALSE(c.mayRestart());
EXPECT_FALSE(c.requestStart()); // rejected while Stopping
ASSERT_TRUE(c.onTaskStopped());
EXPECT_TRUE(c.mayRestart());
EXPECT_TRUE(c.requestStart());
}
// Handoff OTA barrier: "Repeated failed OTA attempts do not ... leave the
// bridge permanently stopped."
TEST(MQTTLifecycle, RepeatedFailedStopsLeaveBridgeRestartable) {
FakeOps ops;
L::Coordinator c(ops, kStopTimeoutMs);
for (int attempt = 0; attempt < 3; ++attempt) {
SCOPED_TRACE(attempt);
ASSERT_TRUE(c.requestStart());
ASSERT_TRUE(c.onTaskStarted());
ops.now += 1000;
ASSERT_TRUE(c.requestStop());
ops.now += kStopTimeoutMs;
c.tick(); // times out (dirty)
EXPECT_EQ(L::State::Stopped, c.state());
EXPECT_TRUE(c.mayRestart()); // never permanently stuck
}
EXPECT_EQ(3, ops.start_task_calls);
}
// --- 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("StartRequested", L::eventName(L::Event::StartRequested));
EXPECT_STREQ("StopTimedOut", L::eventName(L::Event::StopTimedOut));
}
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}