From 076448c3085002cc9e48d6a98e0b4c1212045f0e Mon Sep 17 00:00:00 2001 From: Yaraslau Tamashevich Date: Mon, 17 Aug 2026 14:32:32 +0300 Subject: [PATCH 1/4] offline: bound IOfflineQueue depth with a reject-newest overflow policy (morph#112) IOfflineQueue previously had no depth bound or overflow policy -- an unbounded producer (a host stuck offline for a long time) could grow InMemoryOfflineQueue/FileOfflineQueue/SqliteOfflineQueue without limit. Adds an opt-in, per-instance maxDepth (std::optional, default std::nullopt/unbounded) as the last constructor parameter on all three shipped implementations, enforced with a reject-newest policy: once a bounded queue holds maxDepth() items, enqueue() throws the new OfflineQueueFullError (a std::runtime_error carrying maxDepth and currentSize) instead of evicting an older item or invoking an app-defined eviction callback. Evict-oldest would silently destroy data the caller believes durable and break replay ordering with no error raised anywhere; IOfflineQueue's existing virtual interface is already the seam for a host that wants different eviction semantics (subclass it directly), so no second overlapping policy mechanism was added. Each enqueue() rejection emits the new Metric::queueOverflow counter (observability.hpp) with the rejection-time size, immediately before throwing. Per-implementation enforcement: - InMemoryOfflineQueue: capacity checked under the existing lock, before push_back. No idempotency-key dedup exists here, so there is no dedup-vs-capacity ordering question. - FileOfflineQueue: capacity checked after the existing keyed-dedup scan, before appendPut -- a dedup hit (re-enqueue of an already- pending key) always succeeds even on a full queue, since it inserts nothing new. - SqliteOfflineQueue: capacity checked (SELECT COUNT(*), via a new countLocked() helper) before both INSERT paths, including the keyed ON CONFLICT ... DO NOTHING path. This means a call that would have resolved to a dedup hit can still be rejected if the queue is full at that moment -- a deliberate, documented conservatism rather than a second round-trip to special-case it. drain() moves from non-const to const across IOfflineQueue and all three implementations, since none of their bodies mutate instance state beyond taking a lock; this lets the new size() default (drain().size()) be const too, and every shipped implementation overrides size() with a direct O(1)/indexed count instead. SqliteOfflineQueue needed `mutable sqlite3* _db` (plus its existing mutex made mutable) since SQLite's C API takes a non-const handle throughout and prepare()/bindText()/bindInt64()/stepOrThrow() all needed to become callable from const context -- the idiomatic fix for a logically-const method wrapping a C API with no const overloads. Two test-only IOfflineQueue subclasses (MinimalQueue in test_offline_queue.cpp, NonDurableQueue in test_sync_worker.cpp) updated their drain() overrides to const to match. docs/spec/offline/offline.md's "Offline queue" section documents size()/maxDepth(), OfflineQueueFullError, the reject-newest policy and its rationale (with a per-implementation enforcement-order note), the queueOverflow metric, and that maxDepth is a per-construction parameter (not persisted on disk -- a reopened FileOfflineQueue/ SqliteOfflineQueue must pass the same value again). Verified every existing enqueue()/constructor call site under include/, examples/, and tests/ still compiles unchanged (the new maxDepth parameter defaults to std::nullopt everywhere) -- confirmed via a full default build (examples/concepts included). 18 new tests added across test_offline_queue.cpp (7), test_file_offline_queue.cpp (5), test_sqlite_offline_queue.cpp (5), and test_sync_worker.cpp (1), all passing. Full suite: 1049 -> 1067 test cases (morph_tests, MSVC/Ninja Debug), all passing, no regressions. morph_offline_sqlite_tests: 5 -> 10 test cases; all 5 new overflow tests pass. 2 pre-existing, unrelated failures on that binary (Windows file-lock on an unscoped SqliteOfflineQueue local still holding its WAL file open when the test tries to remove() it) reproduce identically on unmodified master. Co-Authored-By: Claude Sonnet 5 --- docs/spec/offline/offline.md | 62 +++++++++++ include/morph/core/observability.hpp | 3 + include/morph/offline/file_offline_queue.hpp | 35 +++++- include/morph/offline/offline_queue.hpp | 74 +++++++++++- .../morph/offline/sqlite_offline_queue.hpp | 70 ++++++++++-- .../test_sqlite_offline_queue.cpp | 105 ++++++++++++++++++ tests/test_file_offline_queue.cpp | 91 +++++++++++++++ tests/test_offline_queue.cpp | 83 +++++++++++++- tests/test_sync_worker.cpp | 24 +++- 9 files changed, 529 insertions(+), 18 deletions(-) diff --git a/docs/spec/offline/offline.md b/docs/spec/offline/offline.md index 88969b06..ee5e7566 100644 --- a/docs/spec/offline/offline.md +++ b/docs/spec/offline/offline.md @@ -175,8 +175,70 @@ while offline; `SyncWorker` drains and replays them on reconnect. | `drain` | `std::vector 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 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 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 diff --git a/include/morph/core/observability.hpp b/include/morph/core/observability.hpp index e45f5db8..0d931a05 100644 --- a/include/morph/core/observability.hpp +++ b/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`. diff --git a/include/morph/offline/file_offline_queue.hpp b/include/morph/offline/file_offline_queue.hpp index dfdce716..0a60668d 100644 --- a/include/morph/offline/file_offline_queue.hpp +++ b/include/morph/offline/file_offline_queue.hpp @@ -9,6 +9,7 @@ #include #include #include +#include #include #include #include @@ -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 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(_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 drain() override { + std::vector drain() const override { std::scoped_lock const lock{_mtx}; std::vector 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 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 _items; uint64_t _nextId{0}; + std::optional _maxDepth; }; } // namespace morph::offline diff --git a/include/morph/offline/offline_queue.hpp b/include/morph/offline/offline_queue.hpp index 6069e5f8..1cfe3fdb 100644 --- a/include/morph/offline/offline_queue.hpp +++ b/include/morph/offline/offline_queue.hpp @@ -5,9 +5,13 @@ #include #include #include +#include +#include #include #include +#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 maxDepth The configured capacity that was reached. + /// @param currentSize Number of pending items at the time of rejection + /// (equal to maxDepth for a well-behaved implementation). + OfflineQueueFullError(std::size_t maxDepth, std::size_t currentSize) + : std::runtime_error("IOfflineQueue: enqueue rejected, queue is at capacity (" + + std::to_string(currentSize) + "/" + std::to_string(maxDepth) + ")"), + maxDepth{maxDepth}, currentSize{currentSize} {} + + /// @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,7 +131,7 @@ 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 drain() = 0; + virtual std::vector drain() const = 0; /// @brief Removes the item identified by @p itemId. /// @@ -110,6 +139,21 @@ struct IOfflineQueue { /// @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 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 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(_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 drain() override { + std::vector drain() const override { std::scoped_lock const lock{_mtx}; return std::vector{_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 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 _items; uint64_t _nextId{0}; + std::optional _maxDepth; }; } // namespace morph::offline diff --git a/include/morph/offline/sqlite_offline_queue.hpp b/include/morph/offline/sqlite_offline_queue.hpp index b20e6a7b..417d8d15 100644 --- a/include/morph/offline/sqlite_offline_queue.hpp +++ b/include/morph/offline/sqlite_offline_queue.hpp @@ -7,10 +7,12 @@ #include #include #include +#include #include #include #include +#include "../core/observability.hpp" #include "offline_queue.hpp" namespace morph::offline { @@ -94,10 +96,16 @@ class SqliteOfflineQueue : public IOfflineQueue { /// @brief Opens (or creates) the queue database at @p path, creating the /// schema if it does not already exist. - /// @param path SQLite database file. + /// @param path SQLite database file. + /// @param maxDepth Maximum number of pending rows `enqueue()` will admit + /// before throwing `OfflineQueueFullError`; `std::nullopt` (the + /// default) means unbounded. Not persisted in the database itself + /// — a per-construction parameter, so a reopen must pass it again + /// to keep the same cap enforced. /// @throws SqliteOfflineQueueError if the database cannot be opened or /// the schema cannot be created. - explicit SqliteOfflineQueue(std::filesystem::path path) : _path{std::move(path)} { + explicit SqliteOfflineQueue(std::filesystem::path path, std::optional maxDepth = std::nullopt) + : _path{std::move(path)}, _maxDepth{maxDepth} { if (sqlite3_open(_path.string().c_str(), &_db) != SQLITE_OK) { std::string msg = "SqliteOfflineQueue: failed to open " + _path.string() + ": " + (_db != nullptr ? sqlite3_errmsg(_db) : "unknown error"); @@ -136,8 +144,10 @@ class SqliteOfflineQueue : public IOfflineQueue { /// @brief Inserts @p payload with an empty idempotency key. /// @param payload Serialised action to persist. /// @return The new row's id (`SELECT last_insert_rowid()`). + /// @throws OfflineQueueFullError if the queue is already at `maxDepth()`. uint64_t enqueue(std::string payload) override { std::scoped_lock const lock{_mtx}; + checkCapacityLocked(); detail::StatementGuard guard{ prepare("INSERT INTO morph_offline_queue (payload, idempotency_key, attempts, enqueued_at) " "VALUES (?, '', 0, ?);")}; @@ -153,9 +163,16 @@ class SqliteOfflineQueue : public IOfflineQueue { /// @param payload Serialised action to persist. /// @param idempotencyKey Stable dedup token; empty means "no dedup". /// @return The new row's id, or the existing row's id on a dedup hit. + /// @throws OfflineQueueFullError if the queue is already at `maxDepth()`. + /// Checked before the insert is attempted, so a call that would + /// have resolved to a dedup hit (inserting nothing) can also be + /// rejected when the queue happens to be full at the same time — + /// a documented, accepted conservatism rather than an extra + /// round trip to special-case it. uint64_t enqueue(std::string payload, std::string idempotencyKey) override { std::scoped_lock const lock{_mtx}; if (idempotencyKey.empty()) { + checkCapacityLocked(); detail::StatementGuard guard{ prepare("INSERT INTO morph_offline_queue (payload, idempotency_key, attempts, enqueued_at) " "VALUES (?, '', 0, ?);")}; @@ -165,6 +182,7 @@ class SqliteOfflineQueue : public IOfflineQueue { return static_cast(sqlite3_last_insert_rowid(_db)); } + checkCapacityLocked(); detail::StatementGuard insertGuard{ prepare("INSERT INTO morph_offline_queue (payload, idempotency_key, attempts, enqueued_at) " "VALUES (?, ?, 0, ?) " @@ -196,7 +214,7 @@ class SqliteOfflineQueue : public IOfflineQueue { /// @brief Returns all pending rows in ascending-id (enqueue) order. /// @return Snapshot of all pending items; the table is unchanged. - std::vector drain() override { + std::vector drain() const override { std::scoped_lock const lock{_mtx}; detail::StatementGuard guard{ prepare("SELECT id, payload, idempotency_key, attempts FROM morph_offline_queue ORDER BY id;")}; @@ -245,6 +263,17 @@ class SqliteOfflineQueue : public IOfflineQueue { stepOrThrow(guard.get(), "setAttempts"); } + /// @brief Returns the number of pending rows. Thread-safe. + /// @return Current pending item count (`COUNT(*)` against the table). + std::size_t size() const override { + std::scoped_lock const lock{_mtx}; + return countLocked(); + } + + /// @brief Returns the configured maximum depth, or `std::nullopt` if unbounded. + /// @return The capacity `enqueue()` enforces, or `std::nullopt` if none. + std::optional maxDepth() const override { return _maxDepth; } + protected: /// @brief Stamps an idempotency key onto an already-inserted row. No-op if /// @p itemId is absent. Reachable only if a caller invokes the base @@ -271,7 +300,7 @@ class SqliteOfflineQueue : public IOfflineQueue { } } - sqlite3_stmt* prepare(const char* sql) { + sqlite3_stmt* prepare(const char* sql) const { sqlite3_stmt* stmt = nullptr; if (sqlite3_prepare_v2(_db, sql, -1, &stmt, nullptr) != SQLITE_OK) { throw SqliteOfflineQueueError{std::string{"SqliteOfflineQueue: prepare failed: "} + sqlite3_errmsg(_db)}; @@ -279,19 +308,19 @@ class SqliteOfflineQueue : public IOfflineQueue { return stmt; } - void bindText(sqlite3_stmt* stmt, int index, const std::string& value) { + void bindText(sqlite3_stmt* stmt, int index, const std::string& value) const { if (sqlite3_bind_text(stmt, index, value.c_str(), -1, detail::kSqliteTransient) != SQLITE_OK) { throw SqliteOfflineQueueError{std::string{"SqliteOfflineQueue: bind failed: "} + sqlite3_errmsg(_db)}; } } - void bindInt64(sqlite3_stmt* stmt, int index, std::int64_t value) { + void bindInt64(sqlite3_stmt* stmt, int index, std::int64_t value) const { if (sqlite3_bind_int64(stmt, index, value) != SQLITE_OK) { throw SqliteOfflineQueueError{std::string{"SqliteOfflineQueue: bind failed: "} + sqlite3_errmsg(_db)}; } } - void stepOrThrow(sqlite3_stmt* stmt, const char* what) { + void stepOrThrow(sqlite3_stmt* stmt, const char* what) const { // A busy/error code is treated the same as reaching the end -- a // production consumer wanting to distinguish SQLITE_BUSY should retry // instead, but a single in-process mutex around the whole connection @@ -313,9 +342,32 @@ class SqliteOfflineQueue : public IOfflineQueue { .count(); } + /// @brief Returns the current row count. Caller must hold `_mtx`. + /// @return `COUNT(*)` against `morph_offline_queue`. + std::size_t countLocked() const { + detail::StatementGuard guard{prepare("SELECT COUNT(*) FROM morph_offline_queue;")}; + sqlite3_step(guard.get()); + return static_cast(sqlite3_column_int64(guard.get(), 0)); + } + + /// @brief Throws `OfflineQueueFullError` if the queue is already at + /// `maxDepth()`. Caller must hold `_mtx`. No-op if unbounded. + void checkCapacityLocked() const { + if (!_maxDepth) { + return; + } + std::size_t const current = countLocked(); + if (current >= *_maxDepth) { + ::morph::observe::detail::emitMetric(::morph::observe::Metric::queueOverflow, + static_cast(current)); + throw OfflineQueueFullError{*_maxDepth, current}; + } + } + std::filesystem::path _path; - sqlite3* _db = nullptr; - std::mutex _mtx; + mutable sqlite3* _db = nullptr; + mutable std::mutex _mtx; + std::optional _maxDepth; }; } // namespace morph::offline diff --git a/tests/offline_sqlite/test_sqlite_offline_queue.cpp b/tests/offline_sqlite/test_sqlite_offline_queue.cpp index 78758a0e..470e13c9 100644 --- a/tests/offline_sqlite/test_sqlite_offline_queue.cpp +++ b/tests/offline_sqlite/test_sqlite_offline_queue.cpp @@ -4,8 +4,12 @@ #include #include #include +#include #include #include +#include + +#include #include #include @@ -151,3 +155,104 @@ TEST_CASE("morph::offline::SqliteOfflineQueue + SyncWorker: poison item dead-let } removeDbFiles(dbPath); } + +// ── Coverage: maxDepth / overflow policy (morph#112) ─────────────────────── + +TEST_CASE("morph::offline::SqliteOfflineQueue: enqueue at maxDepth throws OfflineQueueFullError", "[sqlite][overflow]") { + auto dbPath = tempDbPath(); + removeDbFiles(dbPath); + { + morph::offline::SqliteOfflineQueue queue{dbPath, 2}; + queue.enqueue("a"); + queue.enqueue("b"); + REQUIRE_THROWS_AS(queue.enqueue("c"), morph::offline::OfflineQueueFullError); + REQUIRE(queue.drain().size() == 2); + } + removeDbFiles(dbPath); +} + +TEST_CASE("morph::offline::SqliteOfflineQueue: maxDepth survives destroying and reopening over the same file", + "[sqlite][overflow]") { + auto dbPath = tempDbPath(); + removeDbFiles(dbPath); + { + morph::offline::SqliteOfflineQueue queue{dbPath, 1}; + queue.enqueue("a"); + REQUIRE_THROWS_AS(queue.enqueue("b"), morph::offline::OfflineQueueFullError); + } + { + // Reopened with the same maxDepth argument -- still enforced. maxDepth + // is a per-construction parameter, not persisted in the database itself. + morph::offline::SqliteOfflineQueue queue{dbPath, 1}; + REQUIRE(queue.drain().size() == 1); + REQUIRE_THROWS_AS(queue.enqueue("b"), morph::offline::OfflineQueueFullError); + } + removeDbFiles(dbPath); +} + +TEST_CASE("morph::offline::SqliteOfflineQueue: size() matches COUNT(*) against the table", "[sqlite][overflow]") { + auto dbPath = tempDbPath(); + removeDbFiles(dbPath); + { + morph::offline::SqliteOfflineQueue queue{dbPath}; + REQUIRE(queue.size() == 0); + queue.enqueue("a"); + queue.enqueue("b"); + queue.enqueue("c"); + REQUIRE(queue.size() == 3); + + // Cross-check via a raw query against the same database file. + sqlite3* raw = nullptr; + REQUIRE(sqlite3_open(dbPath.string().c_str(), &raw) == SQLITE_OK); + sqlite3_stmt* stmt = nullptr; + REQUIRE(sqlite3_prepare_v2(raw, "SELECT COUNT(*) FROM morph_offline_queue;", -1, &stmt, nullptr) == + SQLITE_OK); + REQUIRE(sqlite3_step(stmt) == SQLITE_ROW); + auto const rawCount = static_cast(sqlite3_column_int64(stmt, 0)); + sqlite3_finalize(stmt); + sqlite3_close(raw); + + REQUIRE(rawCount == queue.size()); + } + removeDbFiles(dbPath); +} + +TEST_CASE("morph::offline::SqliteOfflineQueue: a dedup hit at capacity is rejected (documented conservatism)", + "[sqlite][overflow]") { + auto dbPath = tempDbPath(); + removeDbFiles(dbPath); + { + morph::offline::SqliteOfflineQueue queue{dbPath, 1}; + queue.enqueue("first-payload", "op-1"); + // The keyed path checks capacity BEFORE attempting the insert, so a + // call that would otherwise resolve to a dedup hit (inserting + // nothing) is still rejected once the queue is full -- documented + // conservatism, not a bug. + REQUIRE_THROWS_AS(queue.enqueue("second-payload", "op-1"), morph::offline::OfflineQueueFullError); + REQUIRE(queue.drain().size() == 1); + } + removeDbFiles(dbPath); +} + +TEST_CASE("morph::offline::SqliteOfflineQueue: enqueue at maxDepth emits queueOverflow metric", + "[sqlite][overflow][observability]") { + morph::observe::ScopedObserveOverride guard; + auto dbPath = tempDbPath(); + removeDbFiles(dbPath); + { + morph::offline::SqliteOfflineQueue queue{dbPath, 1}; + queue.enqueue("a"); + + std::vector samples; + morph::observe::setMetricSink([&](const morph::observe::MetricEvent& evt) { + if (evt.metric == morph::observe::Metric::queueOverflow) { + samples.push_back(evt.value); + } + }); + + REQUIRE_THROWS_AS(queue.enqueue("b"), morph::offline::OfflineQueueFullError); + REQUIRE(samples.size() == 1); + REQUIRE(samples[0] == 1.0); + } + removeDbFiles(dbPath); +} diff --git a/tests/test_file_offline_queue.cpp b/tests/test_file_offline_queue.cpp index e101e3a7..43a74443 100644 --- a/tests/test_file_offline_queue.cpp +++ b/tests/test_file_offline_queue.cpp @@ -6,7 +6,9 @@ #include #include #include +#include #include +#include #include #include @@ -507,3 +509,92 @@ TEST_CASE( REQUIRE_THROWS_AS(morph::offline::FileOfflineQueue(path, ioOps), std::runtime_error); std::filesystem::remove(path); } + +// ── Coverage: maxDepth / overflow policy (morph#112) ─────────────────────── + +TEST_CASE("morph::offline::FileOfflineQueue: enqueue at maxDepth throws OfflineQueueFullError", + "[file_queue][overflow]") { + auto path = tempQueuePath(); + std::filesystem::remove(path); + { + morph::offline::FileOfflineQueue queue{path, morph::core::FileIoOps{}, 2}; + queue.enqueue("a"); + queue.enqueue("b"); + REQUIRE_THROWS_AS(queue.enqueue("c"), morph::offline::OfflineQueueFullError); + REQUIRE(queue.drain().size() == 2); + } + std::filesystem::remove(path); +} + +TEST_CASE("morph::offline::FileOfflineQueue: maxDepth survives destroying and reopening over the same file", + "[file_queue][overflow]") { + auto path = tempQueuePath(); + std::filesystem::remove(path); + { + morph::offline::FileOfflineQueue queue{path, morph::core::FileIoOps{}, 1}; + queue.enqueue("a"); + REQUIRE_THROWS_AS(queue.enqueue("b"), morph::offline::OfflineQueueFullError); + } + { + // Reopened with the same maxDepth argument -- still enforced. maxDepth + // is a per-construction parameter, not persisted in the file itself. + morph::offline::FileOfflineQueue queue{path, morph::core::FileIoOps{}, 1}; + REQUIRE(queue.drain().size() == 1); + REQUIRE_THROWS_AS(queue.enqueue("b"), morph::offline::OfflineQueueFullError); + } + std::filesystem::remove(path); +} + +TEST_CASE("morph::offline::FileOfflineQueue: a dedup hit on a full queue does not throw", "[file_queue][overflow]") { + auto path = tempQueuePath(); + std::filesystem::remove(path); + { + morph::offline::FileOfflineQueue queue{path, morph::core::FileIoOps{}, 1}; + auto id1 = queue.enqueue("first-payload", "op-1"); + // The dedup scan runs before the capacity check, so a repeat of the + // same idempotencyKey on a full queue returns the existing id instead + // of throwing. + auto id2 = queue.enqueue("second-payload", "op-1"); + REQUIRE(id1 == id2); + REQUIRE(queue.drain().size() == 1); + } + std::filesystem::remove(path); +} + +TEST_CASE("morph::offline::FileOfflineQueue: size() reflects live pending count", "[file_queue][overflow]") { + auto path = tempQueuePath(); + std::filesystem::remove(path); + { + morph::offline::FileOfflineQueue queue{path}; + REQUIRE(queue.size() == 0); + auto id1 = queue.enqueue("a"); + queue.enqueue("b"); + REQUIRE(queue.size() == 2); + queue.markDone(id1); + REQUIRE(queue.size() == 1); + } + std::filesystem::remove(path); +} + +TEST_CASE("morph::offline::FileOfflineQueue: enqueue at maxDepth emits queueOverflow metric", + "[file_queue][overflow][observability]") { + morph::observe::ScopedObserveOverride guard; + auto path = tempQueuePath(); + std::filesystem::remove(path); + { + morph::offline::FileOfflineQueue queue{path, morph::core::FileIoOps{}, 1}; + queue.enqueue("a"); + + std::vector samples; + morph::observe::setMetricSink([&](const morph::observe::MetricEvent& evt) { + if (evt.metric == morph::observe::Metric::queueOverflow) { + samples.push_back(evt.value); + } + }); + + REQUIRE_THROWS_AS(queue.enqueue("b"), morph::offline::OfflineQueueFullError); + REQUIRE(samples.size() == 1); + REQUIRE(samples[0] == 1.0); + } + std::filesystem::remove(path); +} diff --git a/tests/test_offline_queue.cpp b/tests/test_offline_queue.cpp index b11e33fa..e87d216e 100644 --- a/tests/test_offline_queue.cpp +++ b/tests/test_offline_queue.cpp @@ -1,7 +1,9 @@ // SPDX-License-Identifier: Apache-2.0 #include +#include #include +#include #include #include #include @@ -101,7 +103,7 @@ struct MinimalQueue : morph::offline::IOfflineQueue { items.push_back(morph::offline::QueueItem{.id = itemId, .payload = std::move(payload), .idempotencyKey = {}}); return itemId; } - std::vector drain() override { return items; } + std::vector drain() const override { return items; } void markDone(uint64_t) override {} std::vector items; @@ -192,3 +194,82 @@ TEST_CASE("morph::offline::IOfflineQueue: default setAttempts is a no-op", "[que base.setAttempts(id, 7); REQUIRE(queue.items[0].attempts == 0); } + +// ── Coverage: maxDepth / overflow policy (morph#112) ─────────────────────── + +TEST_CASE("morph::offline::InMemoryOfflineQueue: enqueue below maxDepth succeeds", "[queue][overflow]") { + morph::offline::InMemoryOfflineQueue queue{3}; + REQUIRE_NOTHROW(queue.enqueue("a")); + REQUIRE_NOTHROW(queue.enqueue("b")); + REQUIRE(queue.size() == 2); + REQUIRE(queue.maxDepth() == std::optional{3}); +} + +TEST_CASE("morph::offline::InMemoryOfflineQueue: enqueue at maxDepth throws OfflineQueueFullError", + "[queue][overflow]") { + morph::offline::InMemoryOfflineQueue queue{2}; + queue.enqueue("a"); + queue.enqueue("b"); + REQUIRE_THROWS_AS(queue.enqueue("c"), morph::offline::OfflineQueueFullError); + REQUIRE(queue.size() == 2); // the rejected item must not grow the queue +} + +TEST_CASE("morph::offline::InMemoryOfflineQueue: markDone frees capacity for a subsequent enqueue", + "[queue][overflow]") { + morph::offline::InMemoryOfflineQueue queue{1}; + auto id = queue.enqueue("a"); + REQUIRE_THROWS_AS(queue.enqueue("b"), morph::offline::OfflineQueueFullError); + + queue.markDone(id); + + REQUIRE_NOTHROW(queue.enqueue("b")); + REQUIRE(queue.size() == 1); +} + +TEST_CASE("morph::offline::InMemoryOfflineQueue: default constructor is unbounded", "[queue][overflow]") { + morph::offline::InMemoryOfflineQueue queue; + REQUIRE(queue.maxDepth() == std::nullopt); + for (int i = 0; i < 10000; ++i) { + REQUIRE_NOTHROW(queue.enqueue("item" + std::to_string(i))); + } + REQUIRE(queue.size() == 10000); +} + +TEST_CASE("morph::offline::IOfflineQueue: default size() delegates to drain().size()", "[queue][overflow]") { + MinimalQueue queue; + morph::offline::IOfflineQueue& base = queue; + base.enqueue("a"); + base.enqueue("b"); + base.enqueue("c"); + // MinimalQueue does not override size(), so this resolves to + // IOfflineQueue's default, which calls drain().size(). + REQUIRE(base.size() == 3); + REQUIRE(base.maxDepth() == std::nullopt); +} + +TEST_CASE("morph::offline::OfflineQueueFullError: carries maxDepth and currentSize", "[queue][overflow]") { + morph::offline::OfflineQueueFullError const error{5, 5}; + REQUIRE(error.maxDepth == 5); + REQUIRE(error.currentSize == 5); + std::string const what = error.what(); + REQUIRE(what.find("5") != std::string::npos); +} + +TEST_CASE("morph::offline::InMemoryOfflineQueue: enqueue at maxDepth emits queueOverflow metric", + "[queue][overflow][observability]") { + morph::observe::ScopedObserveOverride guard; + morph::offline::InMemoryOfflineQueue queue{1}; + queue.enqueue("a"); + + std::vector samples; + morph::observe::setMetricSink([&](const morph::observe::MetricEvent& evt) { + if (evt.metric == morph::observe::Metric::queueOverflow) { + samples.push_back(evt.value); + } + }); + + REQUIRE_THROWS_AS(queue.enqueue("b"), morph::offline::OfflineQueueFullError); + + REQUIRE(samples.size() == 1); + REQUIRE(samples[0] == 1.0); +} diff --git a/tests/test_sync_worker.cpp b/tests/test_sync_worker.cpp index b5904d77..e75007b0 100644 --- a/tests/test_sync_worker.cpp +++ b/tests/test_sync_worker.cpp @@ -305,7 +305,7 @@ struct NonDurableQueue : morph::offline::IOfflineQueue { items.push_back(morph::offline::QueueItem{.id = itemId, .payload = std::move(payload), .idempotencyKey = {}}); return itemId; } - std::vector drain() override { return items; } + std::vector drain() const override { return items; } void markDone(uint64_t itemId) override { std::erase_if(items, [itemId](const morph::offline::QueueItem& item) { return item.id == itemId; }); } @@ -363,3 +363,25 @@ TEST_CASE("morph::offline::SyncWorker: run() emits queueDepth with the pending c REQUIRE(samples.size() == 1); REQUIRE(samples[0] == 3.0); } + +TEST_CASE("morph::offline::SyncWorker: run() over a queue at maxDepth still drains and replays normally", + "[sync][overflow]") { + // maxDepth bounds enqueue(); it has no bearing on drain()/replay -- a + // full queue still drains and replays every pending item exactly as an + // unbounded one would. + morph::offline::InMemoryOfflineQueue queue{3}; + queue.enqueue("a"); + queue.enqueue("b"); + queue.enqueue("c"); + REQUIRE_THROWS_AS(queue.enqueue("d"), morph::offline::OfflineQueueFullError); + + morph::offline::SyncWorker worker{queue, [](const std::string&) { return true; }}; + auto result = worker.run(); + + REQUIRE(result.successful == 3); + REQUIRE(result.failed == 0); + REQUIRE(queue.drain().empty()); + + // Capacity freed up by the successful replay -- enqueue succeeds again. + REQUIRE_NOTHROW(queue.enqueue("e")); +} From 81c34de4a0bef36b07b8b5d50a60957daacde590 Mon Sep 17 00:00:00 2001 From: Yaraslau Tamashevich Date: Mon, 17 Aug 2026 14:53:18 +0300 Subject: [PATCH 2/4] fix CI: OfflineQueueFullError's constructor params shadow its own fields Rename the constructor parameters (maxDepth/currentSize) to maxDepthValue/currentSizeValue -- clang's -Wshadow-field-in-constructor (enabled under -Weverything -Werror on the Linux clang-tidy-diff and Windows clangcl-release CI legs) correctly flagged the original parameter names as shadowing the identically-named public fields they initialize. MSVC (this worktree's local build toolchain) doesn't enable this warning, so it wasn't caught before pushing. No functional change; member field names and the exception's public API are unchanged. Co-Authored-By: Claude Sonnet 5 --- include/morph/offline/offline_queue.hpp | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/include/morph/offline/offline_queue.hpp b/include/morph/offline/offline_queue.hpp index 1cfe3fdb..b0f34319 100644 --- a/include/morph/offline/offline_queue.hpp +++ b/include/morph/offline/offline_queue.hpp @@ -71,13 +71,13 @@ struct QueueItem { /// `IOfflineQueue` directly rather than layering a second policy mechanism on /// top of this one. struct OfflineQueueFullError : std::runtime_error { - /// @param maxDepth The configured capacity that was reached. - /// @param currentSize Number of pending items at the time of rejection - /// (equal to maxDepth for a well-behaved implementation). - OfflineQueueFullError(std::size_t maxDepth, std::size_t currentSize) + /// @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(currentSize) + "/" + std::to_string(maxDepth) + ")"), - maxDepth{maxDepth}, currentSize{currentSize} {} + std::to_string(currentSizeValue) + "/" + std::to_string(maxDepthValue) + ")"), + maxDepth{maxDepthValue}, currentSize{currentSizeValue} {} /// @brief The configured capacity that was reached. std::size_t maxDepth; From 70fdc909d4f6c36fbbac6d5f26e37bd37f330fe5 Mon Sep 17 00:00:00 2001 From: Yaraslau Tamashevich Date: Mon, 17 Aug 2026 15:15:50 +0300 Subject: [PATCH 3/4] fix CI: document queueOverflow in docs/spec/core/observability.md The Header <-> spec sync check flagged this PR's include/morph/core/ observability.hpp change (the new Metric::queueOverflow enumerator) with no matching docs/spec/core/** update -- docs/spec/offline/ offline.md was updated (correctly, from the offline-queue side) but the Metric enum's own spec, which explicitly enumerated "eight observation kinds" in two places, was not. Adds queueOverflow to both the enumerator table and the call-site table (fired by all three IOfflineQueue::enqueue implementations), and updates both enumerator-count references from eight to nine. Co-Authored-By: Claude Sonnet 5 --- docs/spec/core/observability.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/docs/spec/core/observability.md b/docs/spec/core/observability.md index 3fac4bea..6544ff34 100644 --- a/docs/spec/core/observability.md +++ b/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>` 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` From 3f53dedadb14a9cc22574741e0c0504167cddc2f Mon Sep 17 00:00:00 2001 From: Yaraslau Tamashevich Date: Mon, 17 Aug 2026 15:44:11 +0300 Subject: [PATCH 4/4] fix CI: cover FileOfflineQueue::maxDepth() -- the one real codecov/patch miss Confirmed via llvm-cov's own per-line HTML report (ground truth, not codecov's report which has an unrelated pre-existing per-instantiation line-record issue tracked in morph#92): FileOfflineQueue::maxDepth() was the only genuinely uncovered line in this diff. Its InMemoryOfflineQueue and SqliteOfflineQueue counterparts both have a direct maxDepth() assertion in their own test files; FileOfflineQueue's test file added the throw/reopen/dedup/size()/metric coverage but never called maxDepth() itself. Adds the missing assertion to the existing "enqueue at maxDepth throws" test (mirrors test_offline_queue.cpp's pattern) plus a dedicated "maxDepth() is std::nullopt when unbounded" case, mirroring what InMemoryOfflineQueue's own test file already covers. Co-Authored-By: Claude Sonnet 5 --- tests/test_file_offline_queue.cpp | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/test_file_offline_queue.cpp b/tests/test_file_offline_queue.cpp index 43a74443..b11cd258 100644 --- a/tests/test_file_offline_queue.cpp +++ b/tests/test_file_offline_queue.cpp @@ -518,6 +518,7 @@ TEST_CASE("morph::offline::FileOfflineQueue: enqueue at maxDepth throws OfflineQ std::filesystem::remove(path); { morph::offline::FileOfflineQueue queue{path, morph::core::FileIoOps{}, 2}; + REQUIRE(queue.maxDepth() == std::optional{2}); queue.enqueue("a"); queue.enqueue("b"); REQUIRE_THROWS_AS(queue.enqueue("c"), morph::offline::OfflineQueueFullError); @@ -526,6 +527,17 @@ TEST_CASE("morph::offline::FileOfflineQueue: enqueue at maxDepth throws OfflineQ std::filesystem::remove(path); } +TEST_CASE("morph::offline::FileOfflineQueue: maxDepth() is std::nullopt when unbounded", + "[file_queue][overflow]") { + auto path = tempQueuePath(); + std::filesystem::remove(path); + { + morph::offline::FileOfflineQueue queue{path}; + REQUIRE(queue.maxDepth() == std::nullopt); + } + std::filesystem::remove(path); +} + TEST_CASE("morph::offline::FileOfflineQueue: maxDepth survives destroying and reopening over the same file", "[file_queue][overflow]") { auto path = tempQueuePath();