testkit: a worker-side executor double, and ledger's report tests driven by it

examples/TESTING.md

@@ -156,6 +156,28 @@ A `sleep_for` outside `pump.hpp` is a review-rejectable defect. The test
 binary uses the Qt-owning `main()` (QCoreApplication + `Catch::Session` +
 DeferredDelete drain) copied from `tests/qt/test_qt_websocket.cpp`.
 
+`pump.hpp` covers waiting on the *Qt loop*. Waiting on a **background job**
+has its own answer, and it is not a wait at all:
+
+- `step_executor.hpp` — `StepExecutor`, an `IExecutor` that queues posted
+  tasks and runs them only on `runOne()`/`runAll()`. Substituted for the
+  `ThreadPoolExecutor` a model or App would otherwise own, it turns
+  submit-then-poll into an exact sequence: submit, `CHECK(pending() == 1)`,
+  `runOne()`, assert done. The negative half — "the worker has **not** run
+  yet" — is assertable only this way; against a real pool it can only be
+  sampled. `runAll()` picks up tasks a running task posts, so a chained job
+  runs to completion instead of stranding its own continuation, and is
+  bounded so a self-reposting task fails loudly rather than hanging.
+  It mirrors `morph::testing::StepExecutor` (`tests/test_support.hpp`), which
+  the framework's own suite has always had; the ladder copy exists because
+  that header has no reachable include path from `examples/`.
+
+A test that keeps a real `ThreadPoolExecutor` under an async job — because it
+is covering the production wiring, or the fact that the worker runs on a
+genuinely different thread with no session context — says so at the test case
+and pays the retry loop knowingly. `examples/ledger/tests/test_ledger_reports.cpp`
+keeps exactly one such case and converts the rest.
+
 ## Multi-client stress harness
 
 Testkit components, with the rung that **first needs** each (this ordering
@@ -167,6 +189,7 @@ DoD):
 | `testkit_main.cpp`, `pump.hpp`, `backend_rig.hpp`, `db_fixture.hpp`, `db_fault_fixture.hpp`, **fault proxy + strand interleaver** (pulled forward, round-7) | rung 0/1 |
 | `client_pool.hpp`, `convergence.hpp` | rung 3 |
 | `action_driver.hpp`, `process_pool.hpp`, `offline_rig.hpp` | rung 4 |
+| `step_executor.hpp` | rung 5 |
 
 - `db_fault_fixture.hpp` — holds a real `Lightweight::SqlScopedLock` on a
   second, independent `SqlConnection` to the shared test database, producing

examples/common/CMakeLists.txt

@@ -156,7 +156,8 @@ if(NOT Catch2_FOUND)
 endif()
 
 # ── morph_ladder_testkit: pump/fixtures/rig/fault-proxy/interleaver ─────────
-# strand_interleaver.hpp (DeterministicExecutor), db_fixture.hpp and
+# strand_interleaver.hpp (DeterministicExecutor), step_executor.hpp
+# (StepExecutor), db_fixture.hpp and
 # db_fault_fixture.hpp are fully header-defined and have no .cpp: none is a
 # QObject, none needs MOC, and the library already links a non-empty TU
 # (fault_proxy.cpp), so content-free placeholder TUs would be dead weight.
@@ -200,6 +201,7 @@ add_executable(ladder_common_tests
     testkit/test_event_poller.cpp
     testkit/test_fault_proxy.cpp
     testkit/test_strand_interleaver.cpp
+    testkit/test_step_executor.cpp
     testkit/test_wasm_registration_path_native.cpp
     testkit/test_action_driver.cpp
     testkit/test_offline_rig.cpp

examples/common/testkit/step_executor.hpp

@@ -0,0 +1,129 @@
+// SPDX-License-Identifier: Apache-2.0
+#pragma once
+
+#include <cstddef>
+#include <deque>
+#include <functional>
+#include <morph/core/executor.hpp>
+#include <mutex>
+#include <stdexcept>
+
+/// @file
+/// The ladder testkit's worker-side executor double (examples/TESTING.md,
+/// "Pumping discipline"): the substitute for a real `ThreadPoolExecutor`
+/// underneath a submit-now/compute-later job, so a test asserts the exact
+/// sequence submit -> still pending -> `runOne()` -> done, instead of polling
+/// with a sleep and a retry cap.
+///
+/// This is a deliberate mirror of `morph::testing::StepExecutor`
+/// (`tests/test_support.hpp`) under the ladder's own namespace, not a new
+/// design: same name, same API, same rationale. It is duplicated rather than
+/// shared for exactly the reason `strand_interleaver.hpp`'s
+/// `DeterministicExecutor` is duplicated from that same header --
+/// `tests/test_support.hpp` is a private header for `morph_tests`' own
+/// translation units and has no reachable include path from `examples/`.
+/// That reachability gap, not the absence of the semantics, is what left
+/// every ladder async-job test spinning a real pool (morph#161).
+
+namespace morph::ladder::testkit {
+
+/// @brief An `IExecutor` that queues every posted task and runs them only when
+///        the test explicitly asks, one at a time -- never on its own thread.
+///
+/// Where `DeterministicExecutor` (strand_interleaver.hpp) sits *underneath* a
+/// `StrandExecutor` to control the delivery order of continuations, this sits
+/// where a production `ThreadPoolExecutor` would: it is the worker. Injected
+/// as the executor a model or App posts background work to, it turns
+/// "eventually the job finishes" into a sequence of exact, assertable states:
+///
+/// ```cpp
+/// StepExecutor worker;
+/// const auto jobId = model.execute(SubmitReport{...});
+/// CHECK(worker.pending() == 1);                       // queued, and not run
+/// CHECK(status(jobId) == ReportStatus::Pending);      // asserted, not sampled
+/// REQUIRE(worker.runOne());
+/// CHECK(status(jobId) == ReportStatus::Done);         // no sleep, no cap
+/// ```
+///
+/// The `pending()`/`runOne()` pair is what makes the *negative* half of that
+/// assertable at all: "the worker has not run yet" is an ordinary `CHECK`
+/// here, where against a real pool it can only be sampled and hoped for.
+///
+/// Single-threaded by construction: `post()` appends to a deque under a mutex
+/// (posts can legitimately arrive from other threads -- code under test
+/// posting a continuation from inside a running task -- so `post()` still
+/// honours `IExecutor`'s thread-safe contract, `docs/spec/core/executor.md`),
+/// but every task itself runs synchronously on whichever thread calls
+/// `runOne()`/`runAll()`. Concurrent `runOne()`/`runAll()` calls are not
+/// supported: a test reasoning about exact task ordering drives it from one
+/// thread.
+///
+/// Unlike `ThreadPoolExecutor`, a task's exception is not caught and logged
+/// here: it propagates straight out of `runOne()`/`runAll()` to the caller.
+/// That is deliberate -- the caller is a test, and the exception is often a
+/// `REQUIRE` failure the test needs to see rather than have swallowed.
+class StepExecutor : public ::morph::exec::IExecutor {
+public:
+    /// @brief Enqueues @p task; does not run it.
+    /// @param task Callable to run on a later `runOne()`/`runAll()` call.
+    void post(std::function<void()> task) override {
+        std::scoped_lock const lock{_mtx};
+        _queue.push_back(std::move(task));
+    }
+
+    /// @brief Runs exactly one queued task, oldest first (FIFO).
+    /// @return `true` if a task was run, `false` if the queue was empty --
+    ///         returning rather than throwing so that "nothing more was
+    ///         queued" is an ordinary `CHECK_FALSE`, which is half of what a
+    ///         worker-side double is for.
+    bool runOne() {
+        std::function<void()> task;
+        {
+            std::scoped_lock const lock{_mtx};
+            if (_queue.empty()) {
+                return false;
+            }
+            task = std::move(_queue.front());
+            _queue.pop_front();
+        }
+        task();
+        return true;
+    }
+
+    /// @brief Runs every task currently queued, including ones a running task
+    ///        itself posts -- so a chained job runs to completion rather than
+    ///        stranding its own continuation.
+    ///
+    /// Bounded at @p maxSteps rather than looping until the queue is empty: a
+    /// task that keeps re-posting work to this executor (a bug in the code
+    /// under test, or a harness misuse) would otherwise be an undetectable
+    /// infinite loop, hanging the test process with no assertion failure.
+    /// @param maxSteps Upper bound on tasks run before giving up.
+    /// @return Number of tasks run.
+    /// @throws std::runtime_error if @p maxSteps is reached.
+    std::size_t runAll(std::size_t maxSteps = 10'000) {
+        std::size_t ran = 0;
+        while (ran < maxSteps && runOne()) {
+            ++ran;
+        }
+        if (ran == maxSteps) {
+            throw std::runtime_error(
+                "StepExecutor::runAll: exceeded maxSteps -- a task is likely re-posting "
+                "indefinitely; use runOne() to step through and find it");
+        }
+        return ran;
+    }
+
+    /// @brief Number of tasks currently queued, awaiting a `runOne()`/`runAll()`.
+    /// @return Queue depth.
+    [[nodiscard]] std::size_t pending() const {
+        std::scoped_lock const lock{_mtx};
+        return _queue.size();
+    }
+
+private:
+    mutable std::mutex _mtx;
+    std::deque<std::function<void()>> _queue;
+};
+
+}  // namespace morph::ladder::testkit

examples/common/testkit/test_step_executor.cpp

@@ -0,0 +1,114 @@
+// SPDX-License-Identifier: Apache-2.0
+#include <catch2/catch_test_macros.hpp>
+#include <morph/core/executor.hpp>
+#include <functional>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+#include "step_executor.hpp"
+
+namespace {
+
+using morph::ladder::testkit::StepExecutor;
+
+}  // namespace
+
+TEST_CASE("StepExecutor queues a posted task instead of running it", "[ladder][testkit][executor]") {
+    StepExecutor exec;
+    bool ran = false;
+    exec.post([&] { ran = true; });
+
+    // The whole point: after post() returns, the worker has provably not run.
+    // Against a real pool this is exactly the state that can only be sampled.
+    CHECK(exec.pending() == 1);
+    CHECK_FALSE(ran);
+
+    REQUIRE(exec.runOne());
+    CHECK(ran);
+    CHECK(exec.pending() == 0);
+}
+
+TEST_CASE("StepExecutor::runOne reports an empty queue rather than throwing", "[ladder][testkit][executor]") {
+    StepExecutor exec;
+    CHECK_FALSE(exec.runOne());
+
+    exec.post([] {});
+    REQUIRE(exec.runOne());
+    // "Nothing more was queued" is an ordinary assertion, not a CHECK_THROWS --
+    // which is what distinguishes this from DeterministicExecutor::step().
+    CHECK_FALSE(exec.runOne());
+}
+
+TEST_CASE("StepExecutor runs tasks oldest-first", "[ladder][testkit][executor]") {
+    StepExecutor exec;
+    std::vector<int> order;
+    exec.post([&] { order.push_back(1); });
+    exec.post([&] { order.push_back(2); });
+    exec.post([&] { order.push_back(3); });
+
+    REQUIRE(exec.pending() == 3);
+    REQUIRE(exec.runOne());
+    CHECK(order == std::vector<int>{1});
+    CHECK(exec.pending() == 2);
+
+    CHECK(exec.runAll() == 2);
+    CHECK(order == std::vector<int>{1, 2, 3});
+}
+
+TEST_CASE("StepExecutor::runAll picks up tasks posted by a running task", "[ladder][testkit][executor]") {
+    // A chained job must run to completion rather than stranding its own
+    // continuation -- the property that separates runAll() from "run the
+    // N tasks that happened to be queued when I was called".
+    StepExecutor exec;
+    std::vector<std::string> steps;
+    exec.post([&] {
+        steps.emplace_back("first");
+        exec.post([&] {
+            steps.emplace_back("second");
+            exec.post([&] { steps.emplace_back("third"); });
+        });
+    });
+
+    CHECK(exec.runAll() == 3);
+    CHECK(steps == std::vector<std::string>{"first", "second", "third"});
+    CHECK(exec.pending() == 0);
+}
+
+TEST_CASE("StepExecutor::runAll is bounded against a self-reposting task", "[ladder][testkit][executor]") {
+    StepExecutor exec;
+    int runs = 0;
+    // Deliberate harness misuse: without the bound this hangs the process with
+    // no assertion failure and no diagnostic.
+    std::function<void()> repost;
+    repost = [&] {
+        ++runs;
+        exec.post(repost);
+    };
+    exec.post(repost);
+
+    CHECK_THROWS_AS(exec.runAll(/*maxSteps=*/16), std::runtime_error);
+    CHECK(runs == 16);
+}
+
+TEST_CASE("StepExecutor lets a task's exception reach the test", "[ladder][testkit][executor]") {
+    // Unlike ThreadPoolExecutor, which catches and logs: a swallowed exception
+    // here would be a swallowed REQUIRE failure.
+    StepExecutor exec;
+    exec.post([] { throw std::runtime_error{"boom"}; });
+    CHECK_THROWS_AS(exec.runOne(), std::runtime_error);
+    CHECK(exec.pending() == 0);
+}
+
+TEST_CASE("StepExecutor is usable through the IExecutor interface", "[ladder][testkit][executor]") {
+    // How production code sees it: a model or App holding a
+    // shared_ptr<IExecutor> substitutes this for its ThreadPoolExecutor.
+    StepExecutor exec;
+    ::morph::exec::IExecutor& iface = exec;
+    bool ran = false;
+    iface.post([&] { ran = true; });
+    CHECK(exec.pending() == 1);
+    CHECK_FALSE(ran);
+    REQUIRE(exec.runOne());
+    CHECK(ran);
+}

examples/ledger/include/ledger/models/ledger_model.hpp

@@ -65,6 +65,34 @@ namespace ledger {
 ///        explicitly.
 class LedgerModel {
   public:
+    /// @brief The ordinary shape: owns a one-thread `ThreadPoolExecutor` for
+    ///        `execute(SubmitReport)`'s worker, per `_reportExecutor`'s own
+    ///        default. This is the constructor the bridge registry uses --
+    ///        every non-test caller reaches the model this way.
+    LedgerModel() = default;
+
+    /// @brief Substitutes a caller-supplied executor for the default worker
+    ///        pool. `_reportExecutor`'s comment has always claimed a caller
+    ///        could do this "without this class changing shape"; until this
+    ///        constructor existed there was in fact no way to, which is why
+    ///        `test_ledger_reports.cpp` had to poll a real pool with sleeps
+    ///        (morph#161).
+    ///
+    ///        Nothing else changes: `execute(SubmitReport)` still posts the
+    ///        same task, capturing the same plain values. Passing
+    ///        `morph::ladder::testkit::StepExecutor` makes the worker run
+    ///        exactly when the test says `runOne()`, so "submitted, still
+    ///        Pending, the worker has not run yet" becomes an assertion
+    ///        rather than a sample.
+    /// @param reportExecutor Where `execute(SubmitReport)` posts the report
+    ///        aggregation.
+    /// @throws std::invalid_argument if @p reportExecutor is null -- checked
+    ///         here rather than left to a null dereference inside
+    ///         `execute(SubmitReport)`, which would fire on whichever call
+    ///         first submits a report, arbitrarily far from the construction
+    ///         that caused it.
+    explicit LedgerModel(std::shared_ptr<::morph::exec::IExecutor> reportExecutor);
+
     /// @brief Creates an account in the ledger named by `action.ledgerId`.
     ///        The model's first keyed action -- see the hand-written
     ///        `ModelKeyTraits`/`ActionKeyTraits` specialisations below this
@@ -309,7 +337,10 @@ class LedgerModel {
     ///        A `shared_ptr<IExecutor>` rather than a `ThreadPoolExecutor`
     ///        by value so a caller can substitute a different executor
     ///        (a `MainThreadExecutor`, a deterministic double) without this
-    ///        class changing shape.
+    ///        class changing shape -- reachable through the
+    ///        `explicit LedgerModel(std::shared_ptr<IExecutor>)` constructor
+    ///        above, which is what `test_ledger_reports.cpp` uses to drive
+    ///        the worker deterministically.
     std::shared_ptr<::morph::exec::IExecutor> _reportExecutor =
         std::make_shared<::morph::exec::ThreadPoolExecutor>(1);
 };

examples/ledger/src/models/ledger_model.cpp

@@ -21,7 +21,9 @@
 #include <cstdint>
 #include <functional>
 #include <map>
+#include <memory>
 #include <optional>
+#include <stdexcept>
 #include <string>
 #include <unordered_map>
 #include <unordered_set>
@@ -459,6 +461,16 @@ void LedgerModel::logAction(const Action& action, const Result& result, std::str
     _log->flush();
 }
 
+LedgerModel::LedgerModel(std::shared_ptr<::morph::exec::IExecutor> reportExecutor)
+    : _reportExecutor{std::move(reportExecutor)} {
+    // Checked rather than left to a later null dereference inside
+    // execute(SubmitReport): the crash would happen on whichever call first
+    // submits a report, arbitrarily far from the construction that caused it.
+    if (_reportExecutor == nullptr) {
+        throw std::invalid_argument{"LedgerModel: reportExecutor must not be null"};
+    }
+}
+
 AccountInfo LedgerModel::execute(const OpenAccount& action) {
     const auto* ctx = morph::session::current();
     if (ctx == nullptr || ctx->principal.empty()) {