offline: bound IOfflineQueue depth with a reject-newest overflow policy

docs/spec/core/observability.md

@@ -25,7 +25,7 @@ if the seam did not exist. `RemoteServer` additionally exposes a
 
 ## Metrics
 
-`Metric` is a closed `enum class` of eight observation kinds:
+`Metric` is a closed `enum class` of nine observation kinds:
 
 | Enumerator | Kind | Meaning |
 |---|---|---|
@@ -37,6 +37,7 @@ if the seam did not exist. `RemoteServer` additionally exposes a
 | `queueDepth` | gauge | `IOfflineQueue` pending items at the start of a `SyncWorker::run()`. |
 | `reconnectAttempts` | counter | `ReconnectCoordinator::onOnline`'s `tryReconnect` attempts. |
 | `reconnectOutcome` | counter | One per `onOnline()` call, tagged by outcome. |
+| `queueOverflow` | counter | An `IOfflineQueue::enqueue()` call rejected because the queue was at its configured `maxDepth()` (see `docs/spec/offline/offline.md`). |
 
 Each observation is a `MetricEvent{metric, value, tags}`; `tags` is a
 `std::span<const std::pair<std::string_view, std::string_view>>` carrying
@@ -112,6 +113,7 @@ a deployment's transport (e.g. `QtWebSocketServer`) is expected to expose
 | `LocalBackend::execute`'s strand task | `executeLatencyMs`, `executeInFlight`, `executeErrors` | `beginSpan`/`endSpan` around `localOp` |
 | `SyncWorker::run()` | `queueDepth` (once, at drain) | — |
 | `ReconnectCoordinator::onOnline()` | `reconnectAttempts` (per attempt), `reconnectOutcome` (once, tagged `outcome`) | — |
+| `InMemoryOfflineQueue::enqueue`, `FileOfflineQueue::enqueue`, `SqliteOfflineQueue::enqueue` | `queueOverflow` (once per rejected call, at the queue's configured `maxDepth()`) | — |
 
 `registerCount`/`deregisterCount` count every call, not just successful ones —
 an unauthorized or malformed `register` still increments it, so the counter
@@ -178,7 +180,7 @@ the `~StrandExecutor` note below).
 
 ### `Metric`
 
-`enum class Metric : std::uint8_t` — see [Metrics](#metrics) for the eight enumerators.
+`enum class Metric : std::uint8_t` — see [Metrics](#metrics) for the nine enumerators.
 
 ### `MetricEvent`
 

docs/spec/offline/offline.md

@@ -175,8 +175,70 @@ while offline; `SyncWorker` drains and replays them on reconnect.
 | `drain` | `std::vector<QueueItem> drain()` | Returns all pending items in enqueue order, without removing them. Safe to call multiple times — items survive between `drain()` and the corresponding `markDone()`. |
 | `markDone` | `void markDone(uint64_t itemId)` | Removes the item identified by `itemId`. No-op if not found. |
 | `setAttempts` | `void setAttempts(uint64_t itemId, uint32_t attempts)` | Persists an updated attempt count for an item. **Public** (unlike `setIdempotencyKey`) because `SyncWorker` calls it from outside the queue after every failed replay. Default no-op; `InMemoryOfflineQueue` overrides it to update the in-deque item. A queue that overrides it to store the count durably makes `SyncWorker`'s retry budget survive a process restart. |
+| `size` | `std::size_t size() const` | Number of pending items, without removing them. Default calls `drain().size()` — correct but O(n) and allocates a full snapshot to answer a size query; every shipped implementation overrides it with a direct count. |
+| `maxDepth` | `std::optional<std::size_t> maxDepth() const` | The capacity `enqueue()` enforces, or `std::nullopt` if unbounded. Default: `std::nullopt` — preserves current behavior for any `IOfflineQueue` subclass written before this method existed. |
 | `setIdempotencyKey` (protected) | `void setIdempotencyKey(uint64_t itemId, std::string key)` | Hook the default two-arg `enqueue` uses to stamp the key onto an already-enqueued item. Default no-op; `InMemoryOfflineQueue` records the key directly instead. |
 
+`drain()` is `const` — it takes a snapshot and mutates nothing, so `size()`'s
+default can call it (and so can an application) without needing a non-`const`
+reference to the queue.
+
+#### Depth bound and overflow policy
+
+`IOfflineQueue` has no depth bound by default — `maxDepth()` returns
+`std::nullopt` and `enqueue()` never rejects an item on capacity grounds
+unless a concrete queue is constructed with an explicit bound. Every shipped
+implementation (`InMemoryOfflineQueue`, `FileOfflineQueue`,
+`SqliteOfflineQueue`) takes an `std::optional<std::size_t> maxDepth =
+std::nullopt` as the **last** constructor parameter; passing a value turns on
+enforcement for that instance.
+
+**Policy: reject-newest.** Once a bounded queue holds `maxDepth()` items, a
+further `enqueue()` throws `OfflineQueueFullError` instead of admitting the
+new item — the queue never silently evicts an older item or invokes an
+app-defined eviction callback:
+
+```cpp
+/// Thrown by enqueue() when the queue is at its configured maxDepth().
+struct OfflineQueueFullError : std::runtime_error {
+    OfflineQueueFullError(std::size_t maxDepth, std::size_t currentSize);
+    std::size_t maxDepth;     // the configured capacity that was reached
+    std::size_t currentSize;  // pending items at the time of rejection
+};
+```
+
+`maxDepth`/`currentSize` are equal for a well-behaved implementation; both are
+carried on the exception so a caller can log or branch on the numbers without
+re-querying the queue. Immediately before throwing, each implementation emits
+the `queueOverflow` counter metric with the rejection-time size as its value
+(see [observability.md](../core/observability.md)).
+
+Per-implementation notes:
+
+- **`InMemoryOfflineQueue`** checks capacity under its existing lock, before
+  the deque `push_back`. It has no idempotency-key dedup at all, so there is
+  no dedup-hit-vs-capacity ordering question here.
+- **`FileOfflineQueue`** runs its existing keyed-dedup scan *first*; the
+  capacity check sits after it, before `appendPut`. A dedup hit (a re-enqueue
+  of an already-pending key) therefore always succeeds and returns the
+  existing id, even on a full queue — it inserts nothing new, so there is
+  nothing to reject.
+- **`SqliteOfflineQueue`** checks capacity (`SELECT COUNT(*)`) before
+  attempting either INSERT path (empty-key and keyed). For the keyed path,
+  this means the check runs *before* the `INSERT ... ON CONFLICT ... DO
+  NOTHING` can resolve to a dedup hit — a re-enqueue of an already-queued key
+  can be rejected if the queue happens to be full at that moment, even though
+  it would have inserted nothing. This is a deliberate, documented
+  conservatism: avoiding it would require a second round trip (insert
+  speculatively, then check whether it was actually a no-op conflict) purely
+  to special-case a narrow situation (re-enqueuing an already-queued
+  idempotency key while the queue is simultaneously full).
+
+`maxDepth` is a per-construction parameter, not persisted in the file or
+database — a host that reopens `FileOfflineQueue`/`SqliteOfflineQueue` over
+the same path must pass the same `maxDepth` argument again to keep the same
+cap enforced; nothing on disk remembers it.
+
 ### `InMemoryOfflineQueue`
 
 Thread-safe in-memory implementation of `IOfflineQueue`. Items live in a

include/morph/core/observability.hpp

@@ -30,6 +30,9 @@ enum class Metric : std::uint8_t {
     reconnectAttempts,
     /// @brief Counter, tagged by outcome: `ReconnectCoordinator::onOnline` results.
     reconnectOutcome,
+    /// @brief Counter: an `IOfflineQueue::enqueue()` call rejected because the
+    ///        queue was at its configured `maxDepth()`.
+    queueOverflow,
 };
 
 /// @brief One metric observation delivered to the installed `MetricSink`.

include/morph/offline/file_offline_queue.hpp

@@ -9,6 +9,7 @@
 #include <glaze/glaze.hpp>
 #include <map>
 #include <mutex>
+#include <optional>
 #include <stdexcept>
 #include <string>
 #include <string_view>
@@ -17,6 +18,7 @@
 
 #include "../core/file_io_ops.hpp"
 #include "../core/logger.hpp"
+#include "../core/observability.hpp"
 #include "offline_queue.hpp"
 
 #ifdef _WIN32
@@ -140,11 +142,17 @@ class FileOfflineQueue : public IOfflineQueue {
     ///        syscalls. Test-only seam — see `morph::core::FileIoOps`'s own
     ///        docs — for forcing the failure branches that need a real
     ///        OS-level I/O error to reach.
+    /// @param maxDepth Maximum number of pending items `enqueue()` will admit
+    ///        before throwing `OfflineQueueFullError`; `std::nullopt` (the
+    ///        default) means unbounded. Not persisted in the file itself — a
+    ///        per-construction parameter, so a reopen must pass it again to
+    ///        keep the same cap enforced.
     /// @throws FileOfflineQueueError if @p path exists but contains a
     ///         malformed non-trailing line.
     /// @throws std::runtime_error if @p path cannot be opened/rewritten.
-    explicit FileOfflineQueue(std::filesystem::path path, ::morph::core::FileIoOps ioOps = {})
-        : _path{std::move(path)}, _io{std::move(ioOps)} {
+    explicit FileOfflineQueue(std::filesystem::path path, ::morph::core::FileIoOps ioOps = {},
+                              std::optional<std::size_t> maxDepth = std::nullopt)
+        : _path{std::move(path)}, _io{std::move(ioOps)}, _maxDepth{maxDepth} {
         load();
         compact();
         _file = _io.fopen(_path.string(), "a");
@@ -177,6 +185,8 @@ class FileOfflineQueue : public IOfflineQueue {
     /// @param payload        Serialised action to persist.
     /// @param idempotencyKey Stable dedup token; empty means "no dedup".
     /// @return The new item's id, or the existing item's id on a dedup hit.
+    /// @throws OfflineQueueFullError if the queue is already at `maxDepth()`
+    ///         (a dedup hit above bypasses this check and always succeeds).
     uint64_t enqueue(std::string payload, std::string idempotencyKey) override {
         std::scoped_lock const lock{_mtx};
         if (!idempotencyKey.empty()) {
@@ -186,6 +196,11 @@ class FileOfflineQueue : public IOfflineQueue {
                 }
             }
         }
+        if (_maxDepth && _items.size() >= *_maxDepth) {
+            ::morph::observe::detail::emitMetric(::morph::observe::Metric::queueOverflow,
+                                                 static_cast<double>(_items.size()));
+            throw OfflineQueueFullError{*_maxDepth, _items.size()};
+        }
         uint64_t const itemId = ++_nextId;
         QueueItem item{.id = itemId, .payload = std::move(payload), .idempotencyKey = std::move(idempotencyKey)};
         appendPut(item);
@@ -195,7 +210,7 @@ class FileOfflineQueue : public IOfflineQueue {
 
     /// @brief Returns all pending items in ascending-id (enqueue) order.
     /// @return Snapshot of all pending items; the file itself is unchanged.
-    std::vector<QueueItem> drain() override {
+    std::vector<QueueItem> drain() const override {
         std::scoped_lock const lock{_mtx};
         std::vector<QueueItem> out;
         out.reserve(_items.size());
@@ -205,6 +220,17 @@ class FileOfflineQueue : public IOfflineQueue {
         return out;
     }
 
+    /// @brief Returns the number of pending items. Thread-safe.
+    /// @return Current pending item count.
+    std::size_t size() const override {
+        std::scoped_lock const lock{_mtx};
+        return _items.size();
+    }
+
+    /// @brief Returns the configured maximum depth, or `std::nullopt` if unbounded.
+    /// @return The capacity `enqueue()` enforces, or `std::nullopt` if none.
+    std::optional<std::size_t> maxDepth() const override { return _maxDepth; }
+
     /// @brief Tombstones @p itemId. No-op if not found.
     /// @param itemId Id returned by the corresponding `enqueue()` call.
     void markDone(uint64_t itemId) override {
@@ -384,9 +410,10 @@ class FileOfflineQueue : public IOfflineQueue {
     std::filesystem::path _path;
     ::morph::core::FileIoOps _io;
     std::FILE* _file = nullptr;
-    std::mutex _mtx;
+    mutable std::mutex _mtx;
     std::map<uint64_t, QueueItem> _items;
     uint64_t _nextId{0};
+    std::optional<std::size_t> _maxDepth;
 };
 
 }  // namespace morph::offline

include/morph/offline/offline_queue.hpp

@@ -5,9 +5,13 @@
 #include <cstdint>
 #include <deque>
 #include <mutex>
+#include <optional>
+#include <stdexcept>
 #include <string>
 #include <vector>
 
+#include "../core/observability.hpp"
+
 namespace morph::offline {
 
 /// @brief An item stored in the offline queue.
@@ -56,6 +60,31 @@ struct QueueItem {
     uint32_t attempts{0};
 };
 
+/// @brief Thrown by `enqueue()` when the queue is at its configured `maxDepth()`.
+///
+/// `IOfflineQueue` enforces a reject-newest overflow policy: once the queue
+/// holds `maxDepth()` items, a further `enqueue()` throws instead of silently
+/// evicting an older item or invoking an app-defined callback. Evicting the
+/// oldest item would destroy data the caller believes durable and break
+/// replay ordering with no error raised anywhere; a host that genuinely wants
+/// different eviction semantics already has the seam for it — subclass
+/// `IOfflineQueue` directly rather than layering a second policy mechanism on
+/// top of this one.
+struct OfflineQueueFullError : std::runtime_error {
+    /// @param maxDepthValue    The configured capacity that was reached.
+    /// @param currentSizeValue Number of pending items at the time of rejection
+    ///                         (equal to maxDepth for a well-behaved implementation).
+    OfflineQueueFullError(std::size_t maxDepthValue, std::size_t currentSizeValue)
+        : std::runtime_error("IOfflineQueue: enqueue rejected, queue is at capacity (" +
+                              std::to_string(currentSizeValue) + "/" + std::to_string(maxDepthValue) + ")"),
+          maxDepth{maxDepthValue}, currentSize{currentSizeValue} {}
+
+    /// @brief The configured capacity that was reached.
+    std::size_t maxDepth;
+    /// @brief Number of pending items at the time of rejection.
+    std::size_t currentSize;
+};
+
 // ── Interface ─────────────────────────────────────────────────────────────────
 
 /// @brief Interface for durable storage of actions that could not be delivered.
@@ -102,14 +131,29 @@ struct IOfflineQueue {
     /// call `drain()` multiple times — items survive a crash between `drain()`
     /// and the corresponding `markDone()` call.
     /// @return Snapshot of all pending items.
-    virtual std::vector<QueueItem> drain() = 0;
+    virtual std::vector<QueueItem> drain() const = 0;
 
     /// @brief Removes the item identified by @p itemId.
     ///
     /// No-op if @p itemId is not found.
     /// @param itemId Id returned by the corresponding `enqueue()` call.
     virtual void markDone(uint64_t itemId) = 0;
 
+    /// @brief Returns the number of pending items without removing them.
+    ///
+    /// Default implementation calls `drain().size()` — correct but O(n) and
+    /// allocates a full snapshot vector to answer a size query. Override for
+    /// an O(1) or index-backed answer.
+    /// @return Current pending item count.
+    virtual std::size_t size() const { return drain().size(); }
+
+    /// @brief Returns the configured maximum depth, or `std::nullopt` if unbounded.
+    ///
+    /// Default: `std::nullopt` (unbounded) — preserves current behavior for any
+    /// `IOfflineQueue` subclass that predates this method.
+    /// @return The capacity `enqueue()` enforces, or `std::nullopt` if none.
+    virtual std::optional<std::size_t> maxDepth() const { return std::nullopt; }
+
     /// @brief Persists an updated attempt count for an item. Default: no-op.
     ///
     /// A durable queue overrides this to store the count so the retry budget
@@ -147,6 +191,12 @@ class InMemoryOfflineQueue : public IOfflineQueue {
 public:
     using IOfflineQueue::enqueue;  // keep the two-arg overload visible
 
+    /// @brief Constructs an in-memory queue, optionally bounded.
+    /// @param maxDepth Maximum number of pending items `enqueue()` will admit
+    ///        before throwing `OfflineQueueFullError`; `std::nullopt` (the
+    ///        default) means unbounded.
+    explicit InMemoryOfflineQueue(std::optional<std::size_t> maxDepth = std::nullopt) : _maxDepth{maxDepth} {}
+
     /// @brief Appends @p payload and returns a monotonically increasing id.
     /// @param payload Serialised action to store.
     /// @return Unique id for this item.
@@ -157,8 +207,14 @@ class InMemoryOfflineQueue : public IOfflineQueue {
     /// @param payload        Serialised action to store.
     /// @param idempotencyKey Stable dedup token; stored verbatim on the item.
     /// @return Unique id for this item.
+    /// @throws OfflineQueueFullError if the queue is already at `maxDepth()`.
     uint64_t enqueue(std::string payload, std::string idempotencyKey) override {
         std::scoped_lock const lock{_mtx};
+        if (_maxDepth && _items.size() >= *_maxDepth) {
+            ::morph::observe::detail::emitMetric(::morph::observe::Metric::queueOverflow,
+                                                 static_cast<double>(_items.size()));
+            throw OfflineQueueFullError{*_maxDepth, _items.size()};
+        }
         uint64_t const itemId = ++_nextId;
         _items.push_back(
             QueueItem{.id = itemId, .payload = std::move(payload), .idempotencyKey = std::move(idempotencyKey)});
@@ -167,11 +223,22 @@ class InMemoryOfflineQueue : public IOfflineQueue {
 
     /// @brief Returns a snapshot of all pending items. Thread-safe.
     /// @return Copy of all items in insertion order.
-    std::vector<QueueItem> drain() override {
+    std::vector<QueueItem> drain() const override {
         std::scoped_lock const lock{_mtx};
         return std::vector<QueueItem>{_items.begin(), _items.end()};
     }
 
+    /// @brief Returns the number of pending items. Thread-safe.
+    /// @return Current pending item count.
+    std::size_t size() const override {
+        std::scoped_lock const lock{_mtx};
+        return _items.size();
+    }
+
+    /// @brief Returns the configured maximum depth, or `std::nullopt` if unbounded.
+    /// @return The capacity `enqueue()` enforces, or `std::nullopt` if none.
+    std::optional<std::size_t> maxDepth() const override { return _maxDepth; }
+
     /// @brief Removes the item with @p itemId from the queue. Thread-safe.
     ///
     /// No-op if @p itemId is not found.
@@ -202,9 +269,10 @@ class InMemoryOfflineQueue : public IOfflineQueue {
     }
 
 private:
-    std::mutex _mtx;
+    mutable std::mutex _mtx;
     std::deque<QueueItem> _items;
     uint64_t _nextId{0};
+    std::optional<std::size_t> _maxDepth;
 };
 
 }  // namespace morph::offline