Reference SQLite-backed IOfflineQueue: persists payload, idempotencyKey, and attempts across process restarts. More...

#include <sqlite_offline_queue.hpp>

Inheritance diagram for morph::offline::SqliteOfflineQueue:

Public Types
enum class	Synchronous : std::uint8_t { normal , full }
	How hard each committed statement is pushed onto the platter. More...

Public Member Functions
int	synchronousLevel () const noexcept
	SQLite's numeric PRAGMA synchronous level for this connection, as read back at construction.
const std::string &	journalMode () const noexcept
	The journal_mode this database actually ended up in.
	SqliteOfflineQueue (::morph::exec::IExecutor &owner, std::filesystem::path path, std::optional< std::size_t > maxDepth=std::nullopt, ::morph::core::FileIoOps ioOps={}, Synchronous synchronous=Synchronous::normal, std::chrono::milliseconds busyTimeout=std::chrono::milliseconds{kBusyTimeoutMillis}, ::core::platform::WallClockRef wallClock=::core::platform::defaultSystemWallClock())
	Opens (or creates) the queue database at `path`, creating the schema if it does not already exist; the queue belongs to `owner`.
	SqliteOfflineQueue (const SqliteOfflineQueue &)=delete
SqliteOfflineQueue &	operator= (const SqliteOfflineQueue &)=delete
	SqliteOfflineQueue (SqliteOfflineQueue &&)=delete
SqliteOfflineQueue &	operator= (SqliteOfflineQueue &&)=delete
uint64_t	enqueue (std::string payload) override
	Inserts `payload` with an empty idempotency key. On the owner.
uint64_t	enqueue (std::string payload, std::string idempotencyKey) override
	Inserts `payload` carrying `idempotencyKey` in one write. On the owner. A non-empty key already present on a row is deduplicated: the existing row's id is returned and nothing new is inserted.
std::vector< QueueItem >	drain () const override
	Returns all pending rows in ascending-id (enqueue) order. On the owner.
void	markDone (uint64_t itemId) override
	Deletes the row identified by `itemId`, on the owner. No-op if absent.
void	setAttempts (uint64_t itemId, Attempts attempts) override
	Persists an updated attempt count for `itemId`, on the owner. No-op if absent.
std::size_t	size () const override
	Returns the number of pending rows. 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.
::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`.

Static Public Attributes
static constexpr int	kBusyTimeoutMillis = 5000
	PRAGMA busy_timeout set at construction: how long a statement blocks on SQLITE_BUSY before giving up, in milliseconds.

Protected Member Functions
void	setIdempotencyKey (uint64_t itemId, std::string idempotencyKey) override
	Stamps an idempotency key onto an already-inserted row, on the owner. No-op if `itemId` is absent. 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 in the same INSERT 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
	Reads every pending row on the owner, answered on `replyExec`.
::morph::async::Completion< std::size_t >	askSize (::morph::exec::IExecutor &replyExec) const override
	Counts the rows on the owner, answered on `replyExec`.

Detailed Description

Reference SQLite-backed IOfflineQueue: persists payload, idempotencyKey, and attempts across process restarts.

Schema (one table, morph_offline_queue):

CREATE TABLE IF NOT EXISTS morph_offline_queue (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    payload         TEXT    NOT NULL,
    idempotency_key TEXT    NOT NULL DEFAULT '',
    attempts        INTEGER NOT NULL DEFAULT 0,
    enqueued_at     INTEGER NOT NULL
);
CREATE UNIQUE INDEX IF NOT EXISTS ix_queue_idem
    ON morph_offline_queue(idempotency_key) WHERE idempotency_key <> '';

id is AUTOINCREMENT, so ids are never reused and a re-opened queue re-presents each row under its stored id — stable across restarts. QueueItem::id remains queue-local (per docs/spec/offline/offline.md); cross-restart identity is carried by idempotencyKey, not id.

The partial unique index gives insert-time dedup for a non-empty idempotencyKey: a re-enqueue of the same key is a no-op that returns the existing row's id (INSERT ... ON CONFLICT ... DO NOTHING, then a lookup on a no-op conflict). Empty keys (the default) are exempt, so keyless items behave exactly as InMemoryOfflineQueue does — never deduplicated.

Crash safety: drain() never deletes, so a crash between drain() and markDone() loses nothing. Each write (enqueue, markDone, setAttempts, setIdempotencyKey) is its own committed statement; PRAGMA journal_mode=WAL and PRAGMA synchronous=FULL (both set once, at construction, and journal_mode read back and verified rather than trusted – some filesystems, e.g. NFS, silently fall back to delete mode) give the durability. PRAGMA busy_timeout is also set at construction: running every statement on one owner makes SQLITE_BUSY unreachable for a single instance, but the class documents no single-opener restriction. Construction also fsyncs the containing directory once, after sqlite3_open() and every schema-setup PRAGMA/CREATE succeed, closing the same fresh-file-creation directory-durability gap FileIoOps::syncPath closes for FileActionLog/FileOfflineQueue.

One owner: The connection belongs to the executor given at construction (see IOfflineQueue, "One owner"): every statement runs there, one at a time. The application's enqueue-on-failure path and SyncWorker's drain share the queue by sharing that owner.

Definition at line 115 of file sqlite_offline_queue.hpp.

Member Enumeration Documentation

◆ Synchronous

enum class morph::offline::SqliteOfflineQueue::Synchronous : std::uint8_t

strong

How hard each committed statement is pushed onto the platter.

Enumerator
normal	PRAGMA synchronous=NORMAL — the default, and SQLite's own recommendation under WAL. A crash or power loss can lose the most recent commits, which this queue's at-least-once delivery plus idempotencyKey dedup already absorb.
full	PRAGMA synchronous=FULL — fsync on every committed statement. Measured ~18x slower per mutation (~0.08ms to ~1.44ms on NVMe/btrfs, sqlite 3.53.4), and every mutation here is its own commit, so a 200-item SyncWorker drain goes from ~16ms to ~290ms — all of it on the queue's owner, where it holds up the producer's enqueue() too. Opt in when losing the last few commits is genuinely unacceptable.

Definition at line 129 of file sqlite_offline_queue.hpp.

Constructor & Destructor Documentation

◆ SqliteOfflineQueue()

morph::offline::SqliteOfflineQueue::SqliteOfflineQueue	(	::morph::exec::IExecutor &	owner,
		std::filesystem::path	path,
		std::optional< std::size_t >	maxDepth = std::nullopt,
		::morph::core::FileIoOps	ioOps = {},
		Synchronous	synchronous = Synchronous::normal,
		std::chrono::milliseconds	busyTimeout = std::chrono::milliseconds{kBusyTimeoutMillis},
		::core::platform::WallClockRef	wallClock = ::core::platform::defaultSystemWallClock() )

inlineexplicit

Opens (or creates) the queue database at path, creating the schema if it does not already exist; the queue belongs to owner.

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

Parameters

owner	The executor every statement runs on; must run one task at a time. Borrowed: it must outlive this queue and run what it posts.
path	SQLite database file.
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.
ioOps	Injectable directory-fsync primitive (FileIoOps::syncPath only – the rest of this class talks to SQLite directly, never through FileIoOps); defaults to the real syscall. Test-only seam, mirroring FileActionLog/FileOfflineQueue's own FileIoOps parameter, for forcing the directory-fsync failure branch below without needing a real OS-level failure.
synchronous	PRAGMA synchronous level; see Synchronous. Defaults to normal.
busyTimeout	How long a statement waits out a lock another connection holds before returning SQLITE_BUSY. Defaults to kBusyTimeoutMillis. One owner running every statement makes SQLITE_BUSY unreachable for a single instance, so the timeout only matters when something else has the database open — but when it does, the wait happens on the owner and holds up every other caller of this instance, a Qt GUI thread included. Pass 0 to restore fail-fast.
wallClock	Clock the enqueued_at column is stamped from; the system clock by default. Borrowed, so it must outlive the queue.

Exceptions

SqliteOfflineQueueError if the database cannot be opened, the schema cannot be created, or the containing directory cannot be fsynced after sqlite3_open() creates a brand-new file. A journal_mode that does not end up as wal is warned about, not thrown — see the constructor body.

Definition at line 199 of file sqlite_offline_queue.hpp.

Member Function Documentation

◆ synchronousLevel()

int morph::offline::SqliteOfflineQueue::synchronousLevel ( ) const

inlinenodiscardnoexcept

SQLite's numeric PRAGMA synchronous level for this connection, as read back at construction.

0 OFF, 1 NORMAL, 2 FULL, 3 EXTRA. Reported rather than inferred from the constructor argument: the argument is what was asked for, this is what SQLite confirmed.

Returns: The level in force on this queue's own connection.

Definition at line 152 of file sqlite_offline_queue.hpp.

◆ journalMode()

const std::string & morph::offline::SqliteOfflineQueue::journalMode ( ) const

inlinenodiscardnoexcept

The journal_mode this database actually ended up in.

Normally "wal". A filesystem without the shared-memory support WAL needs reports "delete"; :memory: and the temp spellings report "memory". Construction warns rather than throws on anything but "wal" (durability is carried by PRAGMA synchronous), so this is how a caller that cares can tell which mode it got.

Returns: The mode string as SQLite reported it at construction.

Definition at line 162 of file sqlite_offline_queue.hpp.

◆ enqueue() [1/3]

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

inlinenodiscardoverridevirtual

Inserts payload with an empty idempotency key. On the owner.

Parameters

payload Serialised action to persist.

Returns: The new row's id (SELECT last_insert_rowid()).

Exceptions

OfflineQueueFullError if the queue is already at maxDepth().

Implements morph::offline::IOfflineQueue.

Definition at line 218 of file sqlite_offline_queue.hpp.

◆ enqueue() [2/3]

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

inlinenodiscardoverridevirtual

Inserts payload carrying idempotencyKey in one write. On the owner. A non-empty key already present on a row is deduplicated: the existing row's id is returned and nothing new is inserted.

Parameters

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

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

Exceptions

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.

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 235 of file sqlite_offline_queue.hpp.

◆ drain() [1/2]

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

inlinenodiscardoverridevirtual

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

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

Implements morph::offline::IOfflineQueue.

Definition at line 241 of file sqlite_offline_queue.hpp.

◆ markDone()

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

inlineoverridevirtual

Deletes the row identified by itemId, on the owner. No-op if absent.

Called off the owner, the delete is posted there and this returns at once; a statement that fails there is logged, and the row stays queued.

Parameters

itemId Id returned by the corresponding enqueue() call.

Implements morph::offline::IOfflineQueue.

Definition at line 250 of file sqlite_offline_queue.hpp.

◆ setAttempts()

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

inlineoverridevirtual

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

Parameters

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

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 258 of file sqlite_offline_queue.hpp.

◆ size() [1/2]

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

inlinenodiscardoverridevirtual

Returns the number of pending rows. On the owner.

Returns: Current pending item count (COUNT(*) against the table).

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 265 of file sqlite_offline_queue.hpp.

◆ maxDepth()

std::optional< std::size_t > morph::offline::SqliteOfflineQueue::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 270 of file sqlite_offline_queue.hpp.

◆ setIdempotencyKey()

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

inlineoverrideprotectedvirtual

Stamps an idempotency key onto an already-inserted row, on the owner. No-op if itemId is absent. 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 in the same INSERT instead.

Parameters

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

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 281 of file sqlite_offline_queue.hpp.

◆ askEnqueue()

::morph::async::Completion< uint64_t > morph::offline::SqliteOfflineQueue::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 row's id, or the enqueue's error.

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 293 of file sqlite_offline_queue.hpp.

◆ askDrain()

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

inlinenodiscardoverrideprotectedvirtual

Reads every pending row on the owner, answered on replyExec.

Parameters

replyExec Where the answer is delivered.

Returns: The pending items, or the read's error.

Reimplemented from morph::offline::IOfflineQueue.

Definition at line 304 of file sqlite_offline_queue.hpp.

◆ askSize()

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

inlinenodiscardoverrideprotectedvirtual

Counts the rows 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 313 of file sqlite_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.

Member Data Documentation

◆ kBusyTimeoutMillis

int morph::offline::SqliteOfflineQueue::kBusyTimeoutMillis = 5000

staticconstexpr

PRAGMA busy_timeout set at construction: how long a statement blocks on SQLITE_BUSY before giving up, in milliseconds.

Definition at line 126 of file sqlite_offline_queue.hpp.

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

morph/offline/sqlite_offline_queue.hpp

Public Types

Public Member Functions

Static Public Attributes

Protected Member Functions

Detailed Description

Member Enumeration Documentation

◆ Synchronous

Constructor & Destructor Documentation

◆ SqliteOfflineQueue()

Member Function Documentation

◆ synchronousLevel()

◆ journalMode()

◆ enqueue() [1/3]

◆ enqueue() [2/3]

◆ drain() [1/2]

◆ markDone()

◆ setAttempts()

◆ size() [1/2]

◆ maxDepth()

◆ setIdempotencyKey()

◆ askEnqueue()

◆ askDrain()

◆ askSize()

◆ drain() [2/2]

◆ enqueue() [3/3]

◆ size() [2/2]

Member Data Documentation

◆ kBusyTimeoutMillis