Interface for durable storage of actions that could not be delivered. More...

#include <offline_queue.hpp>

Inheritance diagram for morph::offline::IOfflineQueue:

Public Member Functions
virtual uint64_t	enqueue (std::string payload)=0
	Appends `payload` to the queue with no idempotency key.
virtual uint64_t	enqueue (std::string payload, std::string idempotencyKey)
	Appends `payload` carrying the cross-subsystem `idempotencyKey`.
virtual std::vector< QueueItem >	drain () const =0
	Returns all pending items in enqueue order without removing them.
virtual void	markDone (uint64_t itemId)=0
	Removes the item identified by `itemId`.
virtual std::size_t	size () const
	Returns the number of pending items without removing them.
virtual std::optional< std::size_t >	maxDepth () const
	Returns the configured maximum depth, or std::nullopt if unbounded.
virtual void	setAttempts (uint64_t itemId, Attempts attempts)
	Persists an updated attempt count for an item. Default: no-op.
::morph::async::Completion< uint64_t >	enqueue (::morph::exec::IExecutor &replyExec, std::string payload, std::string idempotencyKey={})
	Appends `payload` carrying `idempotencyKey` on the queue's owner; the new item's id is delivered on `replyExec`.
::morph::async::Completion< std::vector< QueueItem > >	drain (::morph::exec::IExecutor &replyExec) const
	Snapshots the pending items on the queue's owner; delivered on `replyExec`.
::morph::async::Completion< std::size_t >	size (::morph::exec::IExecutor &replyExec) const
	Counts the pending items on the queue's owner; delivered on `replyExec`.

Protected Member Functions
virtual ::morph::async::Completion< uint64_t >	askEnqueue (::morph::exec::IExecutor &replyExec, std::string payload, std::string idempotencyKey)
	What enqueue(replyExec, payload, key) asks. The default enqueues where it is called, for an implementation with no owner; one with an owner enqueues there.
virtual ::morph::async::Completion< std::vector< QueueItem > >	askDrain (::morph::exec::IExecutor &replyExec) const
	What drain(replyExec) asks. The default drains where it is called; one with an owner drains there.
virtual ::morph::async::Completion< std::size_t >	askSize (::morph::exec::IExecutor &replyExec) const
	What size(replyExec) asks. The default counts where it is called; one with an owner counts there.
virtual void	setIdempotencyKey (uint64_t itemId, std::string idempotencyKey)
	Stamps an idempotency key onto an already-enqueued item.

Detailed Description

Interface for durable storage of actions that could not be delivered.

Items accumulate while the system is offline and are replayed by SyncWorker on reconnect. The interface is intentionally minimal so that implementations can range from in-memory (InMemoryOfflineQueue) to SQLite or file-backed stores.

One owner: Every queue morph ships belongs to the executor it is given at construction. Its synchronous verbs answer only there; its writes (markDone, setAttempts) run there, posted from anywhere else; a caller that is not on the owner asks through the completion overloads (enqueue(replyExec, …), drain(replyExec), size(replyExec)). A SyncWorker draining the queue runs on the queue's owner. See docs/spec/offline/offline.md, "One owner".

Definition at line 177 of file offline_queue.hpp.

Member Function Documentation

◆ enqueue() [1/3]

virtual uint64_t morph::offline::IOfflineQueue::enqueue ( std::string payload )

nodiscardpure virtual

Appends payload to the queue with no idempotency key.

Parameters

payload Serialised action to persist.

Returns: A stable id that can be passed to markDone().

Implemented in morph::offline::FileOfflineQueue, morph::offline::InMemoryOfflineQueue, and morph::offline::SqliteOfflineQueue.

◆ enqueue() [2/3]

virtual uint64_t morph::offline::IOfflineQueue::enqueue	(	std::string	payload,
		std::string	idempotencyKey )

inlinenodiscardvirtual

Appends payload carrying the cross-subsystem idempotencyKey.

The key is stored on the resulting QueueItem::idempotencyKey so a replay consumer can dedup this op against ones the journal (or a prior replay) already applied (see QueueItem::idempotencyKey and docs/spec/offline/offline.md).

Uniqueness is a floor, not a prohibition: The queue never interprets the key, and this interface does not require uniqueness enforcement: a conforming implementation may store the same non-empty key twice, so a replay consumer must dedup on the key regardless. Enqueue-time dedup is an allowed strengthening, not a contract violation. Of the implementations morph ships, InMemoryOfflineQueue never dedups, while FileOfflineQueue (linear scan) and SqliteOfflineQueue (partial unique index) both do.

What a dedup hit does, in every implementation that dedups

It applies only to a non-empty key already carried by a pending item. An empty key is never a dedup token — two empty-key enqueues always produce two distinct items, in every implementation.
The call succeeds and returns the existing item's id rather than a fresh one, so the return value is not a reliable signal that anything was stored.
It is first-write-wins with silent payload loss: payload is discarded, the pending item keeps the payload it already had, and no error is raised and nothing is reported to the caller. A caller that re-enqueues a corrected payload under an unchanged key therefore loses the correction — mint a new key when the payload changes.
markDone() releases the key: once the pending item is gone, the same key enqueues normally again.
It survives a restart in a durable queue — re-enqueuing a key that a pending persisted item still carries is a hit after a reopen.

These guarantees are pinned for every shipped implementation by tests/offline_queue_conformance.hpp, which is told which policy each implementation has and asserts it.

The default implementation delegates to enqueue(std::move(payload)) and then stamps the key via setIdempotencyKey, so existing IOfflineQueue implementations keep working without overriding it.

Non-atomicity of this default (implementors, read this): This default makes two separate virtual calls (enqueue(payload), then setIdempotencyKey(itemId, key)), and nothing ties them together. In an implementation reachable from more than one thread, a markDone(itemId) landing between them could erase the item before the key is stamped (silently dropping the key rather than throwing), and a drain() in that same window can observe the item with an empty idempotency key. A subclass that stores keys and cares about atomicity must override the two-argument enqueue and stamp the key in the same step that inserts the item, exactly as every implementation morph ships already does (FileOfflineQueue, InMemoryOfflineQueue, SqliteOfflineQueue all override it inline) — this base default exists only so a subclass that has not been updated yet keeps compiling and working for the non-concurrent case, not as an atomicity guarantee.

Parameters

payload	Serialised action to persist.
idempotencyKey	Stable dedup token for the logical op; may be empty.

Returns: A stable id that can be passed to markDone().

Reimplemented in morph::offline::FileOfflineQueue, morph::offline::InMemoryOfflineQueue, and morph::offline::SqliteOfflineQueue.

Definition at line 247 of file offline_queue.hpp.

◆ drain() [1/2]

virtual std::vector< QueueItem > morph::offline::IOfflineQueue::drain ( ) const

nodiscardpure virtual

Returns all pending items in enqueue order without removing them.

Items remain in the queue until markDone() is called. It is safe to call drain() multiple times — items survive a crash between drain() and the corresponding markDone() call.

Enqueue order is the implementation's to keep; the id does not imply it: "Enqueue order" is a property this method requires and that QueueItem::id does not supply. Every implementation morph ships mints ids that happen to increase with insertion — an in-memory counter, a file offset, SQLite's INTEGER PRIMARY KEY AUTOINCREMENT — so each of them satisfies this method by presenting rows ordered by id. That is a property of those three stores, not of the type: a uint64_t neither promises monotonicity nor rules out an id minted from a GUID, a content hash, or a sharded sequence, and a store that reuses the id of a removed row does not order correctly either. An implementation over such a store must carry its own insertion sequence and order on that; ordering by the id would compile, survive a casual smoke test, and replay a user's actions out of order. tests/offline_queue_conformance.hpp asserts the ordering, so an implementation that gets this wrong fails there.

Returns: Snapshot of all pending items.

Implemented in morph::offline::FileOfflineQueue, morph::offline::InMemoryOfflineQueue, and morph::offline::SqliteOfflineQueue.

◆ markDone()

virtual void morph::offline::IOfflineQueue::markDone ( uint64_t itemId )

pure virtual

Removes the item identified by itemId.

No-op if itemId is not found.

Parameters

itemId Id returned by the corresponding enqueue() call.

Implemented in morph::offline::FileOfflineQueue, morph::offline::InMemoryOfflineQueue, and morph::offline::SqliteOfflineQueue.

◆ size() [1/2]

virtual std::size_t morph::offline::IOfflineQueue::size ( ) const

inlinenodiscardvirtual

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.

Returns: Current pending item count.

Reimplemented in morph::offline::FileOfflineQueue, morph::offline::InMemoryOfflineQueue, and morph::offline::SqliteOfflineQueue.

Definition at line 289 of file offline_queue.hpp.

◆ maxDepth()

virtual std::optional< std::size_t > morph::offline::IOfflineQueue::maxDepth ( ) const

inlinenodiscardvirtual

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.

Returns: The capacity enqueue() enforces, or std::nullopt if none.

Reimplemented in morph::offline::FileOfflineQueue, morph::offline::InMemoryOfflineQueue, and morph::offline::SqliteOfflineQueue.

Definition at line 296 of file offline_queue.hpp.

◆ setAttempts()

virtual void morph::offline::IOfflineQueue::setAttempts	(	uint64_t	itemId,
		Attempts	attempts )

inlinevirtual

Persists an updated attempt count for an item. Default: no-op.

A durable queue overrides this to store the count so the retry budget survives a restart — SyncWorker calls it after every failed replay, and reads the persisted value back through QueueItem::attempts on the next drain() (this run, or after a restart). InMemoryOfflineQueue overrides it to update the in-deque item; a queue that leaves it as the default no-op keeps the pre-existing process-local retry behavior (SyncWorker's own in-memory counter is then always authoritative, since QueueItem::attempts never advances).

Parameters

itemId	Id of the item whose attempt count changed.
attempts	New cumulative attempt count to persist. Typed, not a bare integer, so a transposed call does not compile — see Attempts for what that silently did before.

Reimplemented in morph::offline::FileOfflineQueue, morph::offline::InMemoryOfflineQueue, and morph::offline::SqliteOfflineQueue.

Definition at line 312 of file offline_queue.hpp.

◆ enqueue() [3/3]

::morph::async::Completion< uint64_t > morph::offline::IOfflineQueue::enqueue	(	::morph::exec::IExecutor &	replyExec,
		std::string	payload,
		std::string	idempotencyKey = {} )

inlinenodiscard

Appends payload carrying idempotencyKey on the queue's owner; the new item's id is delivered on replyExec.

The form of enqueue() for a caller that is not on the queue's owner. A full queue rejects the completion with OfflineQueueFullError.

Parameters

replyExec	Where the answer is delivered, and where the caller attaches to it. Borrowed: it must outlive the completion.
payload	Serialised action to persist.
idempotencyKey	Stable dedup token for the logical op; may be empty.

Returns: The id enqueue(payload, idempotencyKey) would have returned, or the error it would have thrown.

Definition at line 325 of file offline_queue.hpp.

◆ drain() [2/2]

::morph::async::Completion< std::vector< QueueItem > > morph::offline::IOfflineQueue::drain ( ::morph::exec::IExecutor & replyExec ) const

inlinenodiscard

Snapshots the pending items on the queue's owner; delivered on replyExec.

Parameters

replyExec Where the answer is delivered. Borrowed: it must outlive the completion.

Returns: What drain() would have returned.

Definition at line 335 of file offline_queue.hpp.

◆ size() [2/2]

::morph::async::Completion< std::size_t > morph::offline::IOfflineQueue::size ( ::morph::exec::IExecutor & replyExec ) const

inlinenodiscard

Counts the pending items on the queue's owner; delivered on replyExec.

Parameters

replyExec Where the answer is delivered. Borrowed: it must outlive the completion.

Returns: What size() would have returned.

Definition at line 345 of file offline_queue.hpp.

◆ askEnqueue()

virtual ::morph::async::Completion< uint64_t > morph::offline::IOfflineQueue::askEnqueue	(	::morph::exec::IExecutor &	replyExec,
		std::string	payload,
		std::string	idempotencyKey )

inlinenodiscardprotectedvirtual

What enqueue(replyExec, payload, key) asks. The default enqueues where it is called, for an implementation with no owner; one with an owner enqueues there.

Parameters

replyExec	Where the answer is delivered.
payload	Serialised action to persist.
idempotencyKey	Stable dedup token; may be empty.

Returns: The new item's id, or the enqueue's error.

Reimplemented in morph::offline::FileOfflineQueue, morph::offline::InMemoryOfflineQueue, and morph::offline::SqliteOfflineQueue.

Definition at line 358 of file offline_queue.hpp.

◆ askDrain()

virtual ::morph::async::Completion< std::vector< QueueItem > > morph::offline::IOfflineQueue::askDrain ( ::morph::exec::IExecutor & replyExec ) const

inlinenodiscardprotectedvirtual

What drain(replyExec) asks. The default drains where it is called; one with an owner drains there.

Parameters

replyExec Where the answer is delivered.

Returns: The pending items.

Reimplemented in morph::offline::FileOfflineQueue, morph::offline::InMemoryOfflineQueue, and morph::offline::SqliteOfflineQueue.

Definition at line 368 of file offline_queue.hpp.

◆ askSize()

virtual ::morph::async::Completion< std::size_t > morph::offline::IOfflineQueue::askSize ( ::morph::exec::IExecutor & replyExec ) const

inlinenodiscardprotectedvirtual

What size(replyExec) asks. The default counts where it is called; one with an owner counts there.

Parameters

replyExec Where the answer is delivered.

Returns: The pending item count.

Reimplemented in morph::offline::FileOfflineQueue, morph::offline::InMemoryOfflineQueue, and morph::offline::SqliteOfflineQueue.

Definition at line 377 of file offline_queue.hpp.

◆ setIdempotencyKey()

virtual void morph::offline::IOfflineQueue::setIdempotencyKey	(	uint64_t	itemId,
		std::string	idempotencyKey )

inlineprotectedvirtual

Stamps an idempotency key onto an already-enqueued item.

Called by the default enqueue(payload, key) overload so implementations that only override the single-argument enqueue still support keys. The default is a no-op (an implementation with no per-item key storage simply drops it); InMemoryOfflineQueue overrides it to record the key.

Parameters

itemId	Id of the item to stamp.
idempotencyKey	Key to store; ignored by the default no-op.

Reimplemented in morph::offline::FileOfflineQueue, and morph::offline::SqliteOfflineQueue.

Definition at line 389 of file offline_queue.hpp.

The documentation for this struct was generated from the following file:

morph/offline/offline_queue.hpp

Public Member Functions

Protected Member Functions

Detailed Description

Member Function Documentation

◆ enqueue() [1/3]

◆ enqueue() [2/3]

◆ drain() [1/2]

◆ markDone()

◆ size() [1/2]

◆ maxDepth()

◆ setAttempts()

◆ enqueue() [3/3]

◆ drain() [2/2]

◆ size() [2/2]

◆ askEnqueue()

◆ askDrain()

◆ askSize()

◆ setIdempotencyKey()