diff --git a/MQTT_OWNERSHIP.md b/MQTT_OWNERSHIP.md new file mode 100644 index 00000000..d1a81d79 --- /dev/null +++ b/MQTT_OWNERSHIP.md @@ -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. diff --git a/STABILITY_TESTABILITY_HANDOFF.md b/STABILITY_TESTABILITY_HANDOFF.md index 8de24ec7..e0a0f8fc 100644 --- a/STABILITY_TESTABILITY_HANDOFF.md +++ b/STABILITY_TESTABILITY_HANDOFF.md @@ -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. diff --git a/src/helpers/MQTTLifecycle.h b/src/helpers/MQTTLifecycle.h new file mode 100644 index 00000000..f7b4a5c4 --- /dev/null +++ b/src/helpers/MQTTLifecycle.h @@ -0,0 +1,319 @@ +#pragma once + +#include + +// 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 diff --git a/test/test_mqtt_lifecycle/test_mqtt_lifecycle.cpp b/test/test_mqtt_lifecycle/test_mqtt_lifecycle.cpp new file mode 100644 index 00000000..b5dec2f3 --- /dev/null +++ b/test/test_mqtt_lifecycle/test_mqtt_lifecycle.cpp @@ -0,0 +1,450 @@ +#include + +#include +#include + +#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 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 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(); +}