Reference append-only, NDJSON-backed IOfflineQueue that persists payload, idempotencyKey, and attempts across process restarts with no third-party dependency. More...

#include <file_offline_queue.hpp>

Inheritance diagram for morph::offline::FileOfflineQueue:

Public Member Functions
	FileOfflineQueue (::morph::exec::IExecutor &owner, std::filesystem::path path, ::morph::core::FileIoOps ioOps={}, std::optional< std::size_t > maxDepth=std::nullopt)
	Opens (or creates) the queue log at `path`, replaying and compacting whatever is already there; the queue belongs to `owner`.
	FileOfflineQueue (const FileOfflineQueue &)=delete
FileOfflineQueue &	operator= (const FileOfflineQueue &)=delete
	FileOfflineQueue (FileOfflineQueue &&)=delete
FileOfflineQueue &	operator= (FileOfflineQueue &&)=delete
uint64_t	enqueue (std::string payload) override
	Appends `payload` with no idempotency key. On the owner.
uint64_t	enqueue (std::string payload, std::string idempotencyKey) override
	Appends `payload` carrying `idempotencyKey`. On the owner. A non-empty key already present on a pending item is deduplicated: the existing item's id is returned and nothing new is written.
std::vector< QueueItem >	drain () const override
	Returns all pending items in ascending-id (enqueue) order. On the owner.
std::size_t	size () const override
	Returns the number of pending items. On the owner.
std::optional< std::size_t >	maxDepth () const override
	Returns the configured maximum depth, or std::nullopt if unbounded. Callable anywhere: fixed at construction.
void	markDone (uint64_t itemId) override
	Tombstones `itemId`, on the owner. No-op if not found.
void	setAttempts (uint64_t itemId, Attempts attempts) override
	Persists an updated attempt count for `itemId`, on the owner. No-op if not found.
::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< 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::size_t >	size (::morph::exec::IExecutor &replyExec) const
	Counts the pending items on the queue's owner; delivered on `replyExec`.
Public Member Functions inherited from morph::offline::IOfflineQueue
::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
void	setIdempotencyKey (uint64_t itemId, std::string idempotencyKey) override
	Stamps an idempotency key onto an already-enqueued item, on the owner. No-op if `itemId` is not found. Reachable only if a caller invokes the base IOfflineQueue::enqueue(payload, key) default through an IOfflineQueue& — this class's own enqueue(payload, key) override above stamps the key inline instead.
::morph::async::Completion< uint64_t >	askEnqueue (::morph::exec::IExecutor &replyExec, std::string payload, std::string idempotencyKey) override
	Enqueues on the owner, answered on `replyExec`.
::morph::async::Completion< std::vector< QueueItem > >	askDrain (::morph::exec::IExecutor &replyExec) const override
	Snapshots on the owner, answered on `replyExec`.
::morph::async::Completion< std::size_t >	askSize (::morph::exec::IExecutor &replyExec) const override
	Counts on the owner, answered on `replyExec`.

Detailed Description

Reference append-only, NDJSON-backed IOfflineQueue that persists payload, idempotencyKey, and attempts across process restarts with no third-party dependency.

Each mutation (enqueue, markDone, setAttempts, setIdempotencyKey) appends one JSON line and immediately fflush+fsyncs it, so it is a committed transaction before the call returns. On construction, the file is replayed line by line (last-write-wins per id — a later "put" overwrites an earlier one, a "done" tombstones it) and then rewritten in compacted form (one "put" line per surviving item), which both bounds file growth and heals a torn trailing line left by a crash mid-write — the same tolerance FileActionLog::entries() (include/morph/journal/file_action_log.hpp) gives a malformed trailing line: it is logged and skipped, not thrown: a malformed line anywhere else is genuine corruption and is rethrown.

New ids resume from the highest id ever seen in the file (including tombstoned ones), so a fresh item can never collide with an old tombstone — unlike FileActionLog's process-local seq, which does not need this property because it never reuses/removes entries.

One owner: The file handle and the pending items belong to the executor given at construction (see IOfflineQueue, "One owner"): each mutation's write and fsync run there, one at a time. Not safe for multiple processes to open the same path concurrently — same restriction as FileActionLog.

Idempotency-key dedup: A keyed enqueue does a linear scan over the currently-pending items to look for a matching idempotencyKey — O(pending items) per call. That is fine at the queue depths this reference implementation targets; a host with high-volume keyed enqueues should prefer SqliteOfflineQueue, whose dedup is index-backed.

Definition at line 146 of file file_offline_queue.hpp.

Constructor & Destructor Documentation

◆ FileOfflineQueue()

morph::offline::FileOfflineQueue::FileOfflineQueue	(	::morph::exec::IExecutor &	owner,
		std::filesystem::path	path,
		::morph::core::FileIoOps	ioOps = {},
		std::optional< std::size_t >	maxDepth = std::nullopt )

inline

Opens (or creates) the queue log at path, replaying and compacting whatever is already there; the queue belongs to owner.

Runs on the constructing thread, before anything else can reach the queue.

Parameters

owner	The executor every access runs on; must run one task at a time. Borrowed: it must outlive this queue and run what it posts.
path	NDJSON file to store queue state in.
ioOps	Injectable file-I/O primitives; defaults to the real 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.
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.

Exceptions

FileOfflineQueueError	if `path` exists but contains a malformed non-trailing line.
std::runtime_error	if `path` cannot be opened/rewritten.

Definition at line 171 of file file_offline_queue.hpp.

Member Function Documentation

◆ enqueue() [1/3]

uint64_t morph::offline::FileOfflineQueue::enqueue ( std::string payload )

inlinenodiscardoverridevirtual

Appends payload with no idempotency key. On the owner.

Parameters

payload Serialised action to persist.

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

Implements morph::offline::IOfflineQueue.

Definition at line 185 of file file_offline_queue.hpp.

◆ enqueue() [2/3]

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

inlinenodiscardoverridevirtual

Appends payload carrying idempotencyKey. On the owner. A non-empty key already present on a pending item is deduplicated: the existing item's id is returned and nothing new is written.

Parameters

payload	Serialised action to persist.
idempotencyKey	Stable dedup token; empty means "no dedup".

Returns: The new item's id, or the existing item's id on a dedup hit.

Exceptions

OfflineQueueFullError if the queue is already at maxDepth() (a dedup hit above bypasses this check and always succeeds).

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 196 of file file_offline_queue.hpp.

◆ drain() [1/2]

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

inlinenodiscardoverridevirtual

Returns all pending items in ascending-id (enqueue) order. On the owner.

Returns: Snapshot of all pending items; the file itself is unchanged.

Implements morph::offline::IOfflineQueue.

Definition at line 203 of file file_offline_queue.hpp.

◆ size() [1/2]

std::size_t morph::offline::FileOfflineQueue::size ( ) const

inlinenodiscardoverridevirtual

Returns the number of pending items. On the owner.

Returns: Current pending item count.

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 209 of file file_offline_queue.hpp.

◆ maxDepth()

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

inlinenodiscardoverridevirtual

Returns the configured maximum depth, or std::nullopt if unbounded. Callable anywhere: fixed at construction.

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

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 214 of file file_offline_queue.hpp.

◆ markDone()

void morph::offline::FileOfflineQueue::markDone ( uint64_t itemId )

inlineoverridevirtual

Tombstones itemId, on the owner. No-op if not found.

Called off the owner, the tombstone is posted there and this returns at once; a write that fails there is logged, and the item stays queued.

Parameters

itemId Id returned by the corresponding enqueue() call.

Implements morph::offline::IOfflineQueue.

Definition at line 221 of file file_offline_queue.hpp.

◆ setAttempts()

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

inlineoverridevirtual

Persists an updated attempt count for itemId, on the owner. No-op if not found.

Parameters

itemId	Id of the item whose count changed.
attempts	New cumulative attempt count to store.

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 229 of file file_offline_queue.hpp.

◆ setIdempotencyKey()

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

inlineoverrideprotectedvirtual

Stamps an idempotency key onto an already-enqueued item, on the owner. No-op if itemId is not found. Reachable only if a caller invokes the base IOfflineQueue::enqueue(payload, key) default through an IOfflineQueue& — this class's own enqueue(payload, key) override above stamps the key inline instead.

Parameters

itemId	Id of the item to stamp.
idempotencyKey	Key to store.

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 242 of file file_offline_queue.hpp.

◆ askEnqueue()

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

inlinenodiscardoverrideprotectedvirtual

Enqueues on the owner, answered on replyExec.

Parameters

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

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

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 254 of file file_offline_queue.hpp.

◆ askDrain()

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

inlinenodiscardoverrideprotectedvirtual

Snapshots on the owner, answered on replyExec.

Parameters

replyExec Where the answer is delivered.

Returns: The pending items.

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 267 of file file_offline_queue.hpp.

◆ askSize()

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

inlinenodiscardoverrideprotectedvirtual

Counts on the owner, answered on replyExec.

Parameters

replyExec Where the answer is delivered.

Returns: The pending item count.

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 276 of file 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.

◆ 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.

◆ 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.

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

morph/offline/file_offline_queue.hpp

Public Member Functions

Protected Member Functions

Detailed Description

Constructor & Destructor Documentation

◆ FileOfflineQueue()

Member Function Documentation

◆ enqueue() [1/3]

◆ enqueue() [2/3]

◆ drain() [1/2]

◆ size() [1/2]

◆ maxDepth()

◆ markDone()

◆ setAttempts()

◆ setIdempotencyKey()

◆ askEnqueue()

◆ askDrain()

◆ askSize()

◆ drain() [2/2]

◆ enqueue() [3/3]

◆ size() [2/2]