core: amortise LocalBackend's pending-completion compaction (fixes #528)

CHANGELOG.md

@@ -11,6 +11,27 @@ API surface).
 
 ### Changed
 
+- **`LocalBackend::execute` no longer rescans the pending-completion list on
+  every dispatch.** `trackPending` used to `std::erase_if` the whole `_pending`
+  vector before each append, so admitting one call with *n* already in flight
+  cost *n* atomic `weak_ptr::expired()` loads under `_pendingMtx`, and a burst of
+  *n* cost O(n²) — before any model work started. Measured against one parked
+  model (clang 22.1.8, `-O2`, 8-core Linux), timing only the `execute()` calls:
+  32,000 queued executes spent **362 ms** in admission alone, at **24.1 µs** per
+  admission over the last tenth; 16,000 spent 77.8 ms. The sweep is now
+  amortised — it runs only when the list reaches a threshold re-armed at twice
+  the surviving entry count after each sweep — which brings the same case to
+  **7.6 ms** total and **0.31 µs** per admission, flat in *n*, and within noise of
+  the 7.6 ms measured with the sweep deleted outright.
+
+  The list is now bounded at twice the live count rather than exactly it, which
+  is the whole price; a new `LocalBackend::trackedPendingCount()` makes that
+  observable. `cancelPending` is unaffected and still fails every live
+  completion on a backend swap or `~Bridge`: it never saw dead entries in the
+  first place, because `weak.lock()` has always skipped them. See
+  `docs/spec/core/backend.md`, "The pending list and its amortised compaction",
+  and morph#528.
+
 - **`Quantity::equation()` writes out at most 100 derivation steps by default,
   and takes the limit as an argument.** A derivation has no bound — a
   data-driven `total = total + row` loop records one step per iteration — and

docs/spec/core/backend.md

@@ -517,7 +517,8 @@ there, rather than once per backend.
   `beginSpan`/`endSpan` around `localOp` — see [observability.md](observability.md).
   Both `registerModel` and `deregisterModel` emit `registerCount`/`deregisterCount`.
 - `cancelPending` — snapshots the pending list under the pending mutex, delivers
-  `exc` to every still-live state.
+  `exc` to every still-live state, and re-arms the compaction threshold. See
+  [The pending list and its amortised compaction](#the-pending-list-and-its-amortised-compaction).
 - `notifyBackendChanged` — under `_regMtx`, looks up only the models recorded in
   `_changeAware` (populated at registration from
   `IModelHolder::isBackendChangeAware()` — a compile-time answer per model type,
@@ -533,6 +534,58 @@ there, rather than once per backend.
 Each model instance gets its own strand so actions are serialised per-model
 without a global lock on the pool.
 
+### The pending list and its amortised compaction
+
+`_pending` is a `vector<weak_ptr<CompletionState<shared_ptr<void>>>>` guarded by
+`_pendingMtx`. It exists for exactly one reader — `cancelPending`, which swaps it
+out and fails everything still live on a backend swap or a `~Bridge`. Nothing
+else consults it, and nothing unlinks from it when a completion settles: an entry
+simply becomes a dead `weak_ptr` that `cancelPending`'s `weak.lock()` skips.
+
+Dead entries therefore have to be reclaimed by a sweep, and the question is only
+how often. Sweeping on **every** `execute`, as `trackPending` did before
+morph#528, makes admitting one call cost one atomic `weak_ptr::expired()` load
+per entry already in the list, under the mutex, before any work starts — so a
+burst of *n* costs O(n²). Measured against one parked model on an 8-core Linux
+box (clang 22.1.8, `-O2`), timing only the `execute()` calls themselves:
+
+| queued executes | total admission time | mean per admission | mean over the last 10% |
+|---|---|---|---|
+| 1 000 | 0.59 ms | 0.59 µs | 0.82 µs |
+| 4 000 | 5.01 ms | 1.25 µs | 2.19 µs |
+| 16 000 | 77.8 ms | 4.86 µs | 9.57 µs |
+| 32 000 | 362 ms | 11.3 µs | 24.1 µs |
+
+Total time quadruples per doubling of *n* and the per-admission cost doubles —
+the O(n²)/O(n) pair, not an artefact of some constant.
+
+The sweep is now **amortised**: `trackPending` sweeps only when `_pending.size()`
+reaches `_compactAt`, and each sweep re-arms `_compactAt` at twice the number of
+entries that survived it (floor 32, below which sweeping costs more than it
+reclaims). A sweep costs O(size) and at least `_compactAt / 2` appends must
+happen before the next one, so admission is amortised O(1) at any depth. On the
+same benchmark the 32 000-execute case drops from 362 ms to 7.6 ms — within noise
+of the 7.6 ms measured with the sweep deleted outright, so what is left is the
+`make_shared`, the registry lookup and the strand post, not the sweep.
+
+Two properties are the price and the guarantee:
+
+- **Memory.** The list is bounded at twice the live count plus the floor, rather
+  than at exactly the live count. `trackedPendingCount()` makes that observable;
+  the fixture in `tests/test_backend_extra.cpp` drives 3 072 settled-and-dropped
+  admissions past 48 parked ones and measures 112 entries left, against the 3 120
+  an uncompacted list would hold.
+- **`cancelPending` is unchanged.** It never saw dead entries in the first place
+  — `weak.lock()` has always skipped them — so carrying them for longer changes
+  nothing it observes. Every live state admitted across every sweep is still
+  reached, which is what the same fixture's second assertion checks. `cancelPending`
+  also resets `_compactAt` to the floor, since it has just emptied the list.
+
+What is **not** guarded by a test is the admission latency itself: a wall-clock
+assertion on a shared CI runner would be a flake rather than evidence, so the
+numbers above come from a benchmark and the test guards only the bound and the
+cancellation.
+
 ## `RemoteServer` — server-side message handler
 
 `RemoteServer` receives JSON envelopes (`morph::wire::Envelope`) from any
@@ -1953,7 +2006,8 @@ inside the class calls `close()` — no thread it joins can be waiting on it.
 | `deregisterModel(mid)` | Releases one attachment through `_instances` under `_regMtx`; erases from `_changeAware` when that destroys the instance. |
 | `notifyBackendChanged()` | Looks up the models recorded in `_changeAware` under `_regMtx`, then posts `onBackendChanged()` (the `IModelHolder` base virtual — no `dynamic_cast`) onto each such model's strand (outside the lock). Cost is O(change-aware models). |
 | `execute(mid, call, cbExec)` | Posts `call.localOp` on the model's strand with `ScopedContext`. Returns a `Completion`. |
-| `cancelPending(exc)` | Snapshots `_pending`, delivers `exc` to each live state. |
+| `cancelPending(exc)` | Snapshots `_pending`, delivers `exc` to each live state, and re-arms the compaction threshold. |
+| `trackedPendingCount()` | `[[nodiscard]] std::size_t trackedPendingCount() const` — size of `_pending` under `_pendingMtx`. **Not** the in-flight count: between sweeps the list also holds entries whose state is already destroyed. An upper bound on in-flight, and the observable that makes [the compaction policy](#the-pending-list-and-its-amortised-compaction)'s memory cost measurable. For in-flight *calls*, use `Bridge::pendingCalls()`. |
 
 ### `RemoteServer`
 
@@ -2097,6 +2151,7 @@ not a behavior change to the existing loopback-only default.
 | `handleInline` | Synchronous; caller-restricted to control messages | Safe to call from a worker-pool thread (e.g. from a `BridgeHandler` constructor). It is meant for `register`/`deregister` only; an `execute` envelope is rejected with an `err` reply, because `dispatchExecute` posts to the strand and would reply after `handleInline` returns (writing into an already-destroyed reply buffer). The rejection is now enforced by the code, matching the documented intent. |
 | `SimulatedRemoteBackend` factory ignored | Model construction delegated to `RemoteServer`'s `ModelRegistryFactory` | The factory closure lives on the client side; the server owns the actual instances. |
 | `cancelPending` snapshots | Weak-ptr snapshot under lock, then resolves outside | Avoids holding the lock while delivering exceptions to each state, preventing deadlock if a callback re-enters the backend. |
+| `_pending` compacted amortised, not intrusively | Sweep when `size() >= _compactAt`, re-arm at twice the survivors | The alternative considered in morph#528 was intrusive: give `CompletionState` a slot index and unlink on settle, making both registration and removal O(1) with no sweep at all. Rejected. It pushes a back-reference to the backend's table into a type shared by every backend, and puts a `_pendingMtx` acquisition on the settle path of every completion — turning a cost paid once per burst into contention paid by every strand thread on every result, on the exact path morph#579's value-handling contract just fixed. The amortised sweep buys the same O(1) admission for one `size_t` of state confined to `LocalBackend`, at the cost of a list bounded at 2× the live count instead of exactly it. |
 | `setReconnectHandler` | Default no-op | Only backends with a transport layer (e.g. `QtWebSocketBackend`) need to react to reconnects. `LocalBackend` and `SimulatedRemoteBackend` never invoke it. |
 | `setConnectHandler`/`setDisconnectHandler` on `IBackend`, not only `QtWebSocketBackend` | Same no-op-default pattern as `setReconnectHandler` | Connection state is a property of any transport-backed backend; a UI observing it shouldn't have to downcast to a concrete backend type. A purely local backend has no meaningful connection state, so the base-class hook is simply inert for it — no behavior change, matching the existing `setReconnectHandler` precedent exactly. |
 | `setDisconnectHandler` fires before reconnect scheduling | Ordering choice, not incidental | An instant successful reconnect must not look, from an observer's perspective, like nothing happened — the disconnected state must be visible even when the very next thing that happens is a fresh `connected`. |

include/morph/core/backend.hpp

@@ -1,9 +1,11 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #pragma once
+#include <algorithm>
 #include <array>
 #include <atomic>
 #include <chrono>
+#include <cstddef>
 #include <exception>
 #include <functional>
 #include <memory>
@@ -1345,6 +1347,7 @@ class LocalBackend : public detail::IBackend {
         {
             std::scoped_lock const lock{_pendingMtx};
             snapshot.swap(_pending);
+            _compactAt = kPendingCompactFloor;
         }
         for (auto& weak : snapshot) {
             if (auto state = weak.lock()) {
@@ -1353,6 +1356,20 @@ class LocalBackend : public detail::IBackend {
         }
     }
 
+    /// @brief Number of entries currently held in the pending list.
+    ///
+    /// **Not** the number of calls in flight: the list is compacted amortised
+    /// (see `trackPending`), so it also carries entries whose completion has
+    /// already been destroyed and which `cancelPending` would skip. It is an
+    /// upper bound on the in-flight count and a direct measure of what the
+    /// compaction policy costs in memory — which is what it exists to make
+    /// observable. For a count of in-flight *calls*, use `Bridge::pendingCalls()`.
+    /// @return Size of the pending list, live and dead entries alike.
+    [[nodiscard]] std::size_t trackedPendingCount() const {
+        std::scoped_lock const lock{_pendingMtx};
+        return _pending.size();
+    }
+
 private:
     /// @brief Builds a holder via @p factory, records it under a fresh id, and
     ///        returns that id. Caller holds `_regMtx`.
@@ -1389,9 +1406,34 @@ class LocalBackend : public detail::IBackend {
         return {mid, std::move(holder)};
     }
 
+    /// @brief Records @p state in the pending list, compacting it amortised-O(1).
+    ///
+    /// The list is append-only between compactions; dead entries are swept only
+    /// when its size reaches `_compactAt`, which each sweep re-arms at twice the
+    /// number of entries that survived it. Doubling the threshold off the live
+    /// count is what makes the sweep amortised: a sweep costs O(size), and at
+    /// least `_compactAt / 2` appends must happen before the next one, so the
+    /// per-append cost is O(1) however deep the queue gets. The list is
+    /// correspondingly bounded at twice the live count (plus the floor), which
+    /// is the whole price of dropping the per-dispatch scan.
+    ///
+    /// Before morph#528 this swept on *every* append, so admitting one call with
+    /// `n` in flight cost `n` atomic `weak_ptr::expired()` loads under
+    /// `_pendingMtx` and a burst of `n` cost O(n²) — measured at 362ms of pure
+    /// admission time for 32k queued executes against one slow model, against
+    /// 7.6ms without the sweep.
+    ///
+    /// Purely a cost change: `cancelPending` still sees every live state,
+    /// because a dead entry is one whose `CompletionState` is already gone and
+    /// which `cancelPending`'s `weak.lock()` has always skipped. Carrying dead
+    /// entries for longer changes nothing it observes.
+    /// @param state Completion state to track until it expires or is cancelled.
     void trackPending(const std::shared_ptr<::morph::async::detail::CompletionState<std::shared_ptr<void>>>& state) {
         std::scoped_lock const lock{_pendingMtx};
-        std::erase_if(_pending, [](const auto& weak) { return weak.expired(); });
+        if (_pending.size() >= _compactAt) {
+            std::erase_if(_pending, [](const auto& weak) { return weak.expired(); });
+            _compactAt = std::max(kPendingCompactFloor, _pending.size() * 2);
+        }
         _pending.emplace_back(state);
     }
 
@@ -1414,8 +1456,16 @@ class LocalBackend : public detail::IBackend {
     // directory is the state the two backends genuinely share.
     std::unordered_set<::morph::exec::detail::ModelId, ::morph::exec::detail::ModelIdHash> _changeAware;
     std::atomic<uint64_t> _nextId{0};
-    std::mutex _pendingMtx;
+    // Smallest size at which `trackPending` will sweep. Below it the sweep costs
+    // more than the handful of dead `weak_ptr`s it could reclaim, and a backend
+    // that only ever has a few calls in flight never sweeps at all.
+    static constexpr std::size_t kPendingCompactFloor = 32;
+    mutable std::mutex _pendingMtx;
     std::vector<std::weak_ptr<::morph::async::detail::CompletionState<std::shared_ptr<void>>>> _pending;
+    // Size at which `trackPending` next sweeps `_pending` for expired entries;
+    // re-armed at twice the surviving count after each sweep. Guarded by
+    // `_pendingMtx` along with `_pending` itself. See `trackPending` (morph#528).
+    std::size_t _compactAt = kPendingCompactFloor;
     // Concurrent in-flight executes, for the executeInFlight metric. A
     // shared_ptr (not a plain atomic member) so strand tasks hold their own
     // reference instead of capturing `this` — see execute()'s comment and the

scripts/branch_partial_allowlist.json

@@ -80,7 +80,7 @@
     },
     {
       "file": "include/morph/core/backend.hpp",
-      "line": 1228,
+      "line": 1230,
       "source": "if (const auto* inst = _instances.find(modelId)) {",
       "reason": "Unreachable by construction given the `_changeAware`/`_instances` invariant (core audit finding BK2). `_changeAware` is an index over the instance directory: an id enters it in `createHolder` (this file, when the holder answers `isBackendChangeAware()`) in the same `_regMtx`-held critical section that files the instance, and leaves it in `deregisterModel` only when `InstanceDirectory::release` reports the instance actually destroyed. `notifyBackendChanged()` (this function) holds the same `_regMtx` while walking `_changeAware` and looking each id up at this line, so every id it walks is still live -- the null arm cannot occur without a code change that breaks that subset invariant. Formerly keyed on `_models`, the map morph#523 replaced with the directory; the invariant and its reason are unchanged."
     },