morph::offline::SyncWorker Class Reference

Replays queued actions from an IOfflineQueue on reconnect. More...

#include <sync_worker.hpp>

Public Types
using	ReplayFunction = std::function<bool(const std::string& payload)>
	Callable that attempts to replay a single queued item, reporting only success or failure.
using	DetailedReplayFunction = std::function<ReplayOutcome(const std::string& payload)>
	Callable that attempts to replay a single queued item and reports what it achieved.
using	DeadLetterSink = std::function<void(const QueueItem& poisoned)>
	Callable invoked when an item exhausts its retry budget, just before it is removed from the queue.

Public Member Functions
	SyncWorker (::morph::exec::IExecutor &owner, IOfflineQueue &queue, ReplayFunction replay, DeadLetterSink deadLetterSink=nullptr)
	Constructs a worker that drains queue using replay, on owner.
	SyncWorker (::morph::exec::IExecutor &owner, IOfflineQueue &queue, DetailedReplayFunction replay, DeadLetterSink deadLetterSink=nullptr)
	Constructs a worker that drains queue using replay, which distinguishes an undelivered replay from a rejected one.
::morph::async::Completion< SyncResult >	run (::morph::exec::IExecutor &replyExec)
	Drains the queue, replaying each item via the replay function, on the owner.
void	stop ()
	Signals an in-progress drain to stop after the current item.

Detailed Description

Replays queued actions from an IOfflineQueue on reconnect.

SyncWorker drains the queue and calls a caller-supplied replay function for each item. The framework has no knowledge of what "replay" means — that is entirely the caller's domain. Retry semantics are baked in with hard-coded sensible defaults; there are intentionally no setters.

Retry & dead-letter (built-in defaults)

• Each item is retried up to 5 cumulative attempts. The count is seeded from the larger of the drained QueueItem::attempts (durable, if the queue persists it) and this worker's own in-memory count, so a queue that overrides IOfflineQueue::setAttempts() makes the budget survive a process restart; a queue that does not (the default no-op) keeps the count purely in-memory, exactly as before.
• Only a delivered failure spends an attempt. A replay reporting ReplayOutcome::Undelivered leaves the item queued with its count untouched, in memory and on disk alike, so a run of reconnect flaps cannot walk queued work to the dead-letter sink. The boolean ReplayFunction cannot express that outcome; a caller that can tell the two apart should use DetailedReplayFunction.
• After every charged failure, the new count is written back through IOfflineQueue::setAttempts() (a no-op unless the queue overrides it) so a persisting queue's next drain() — this run or after a restart — sees the updated value.
• Items that fail their 5th cumulative attempt are dropped from the queue. If a DeadLetterSink is set, it is invoked with the exhausted item instead of the default log line; if unset, the item is logged at morph::log::LogLevel::error (the payload appears in the log line) — today's behavior, unchanged.
• Items that succeed reset their attempt counter implicitly (they're removed).
• There is no public knob on the retry cap — if your application needs different retry math, wrap or replace SyncWorker. The framework's promise is "obvious, safe defaults that the GUI never has to think about."

Replay contract

DetailedReplayFunction (three outcomes — prefer it):

• ReplayOutcome::Succeeded → removed from the queue.
• ReplayOutcome::Rejected → delivered and refused; attempt counter incremented; left in queue if under the cap, otherwise dropped (dead-lettered).
• ReplayOutcome::Undelivered → never reached the peer; nothing is charged and the item stays queued.

ReplayFunction (two outcomes — unchanged for existing callers):

• Return true → same as Succeeded.
• Return false → same as Rejected. It is deliberately not read as Undelivered: every caller written against the boolean form means "this attempt failed", and silently treating that as undelivered would retry a poison payload forever.

Either form:

• Throw → treated as Rejected. A throw reports failure but not delivery, and an unknown failure has to be charged for the same reason.

One owner

A worker belongs to the executor it is given, which must run one task at a time: the coordinator's offline strand (ReconnectCoordinator::strand()), or any other strand. Its queue drain, attempt counts and dead-lettering run only in that owner's tasks. run() is callable from any thread: on the owner it drains at once; elsewhere it posts the drain to the owner, so two runs never overlap. Its answer is delivered on the executor the caller names, so a caller attaches to it where it already runs. stop() is the one cross-thread signal, an atomic.

The drain calls the queue's synchronous verbs, which answer only on the queue's own owner (see IOfflineQueue, "One owner"): give the queue the worker's owner, or the executor that strand runs over.

Definition at line 139 of file sync_worker.hpp.

Member Typedef Documentation

ReplayFunction

using morph::offline::SyncWorker::ReplayFunction = std::function<bool(const std::string& payload)>

Callable that attempts to replay a single queued item, reporting only success or failure.

Retained unchanged for existing callers, and false keeps meaning exactly what it always did — ReplayOutcome::Rejected, i.e. one attempt spent. A caller that can tell an undelivered replay from a refused one should use DetailedReplayFunction instead; this overload cannot express the difference and so cannot avoid the dead-lettering it causes.

Definition at line 149 of file sync_worker.hpp.

DetailedReplayFunction

using morph::offline::SyncWorker::DetailedReplayFunction = std::function<ReplayOutcome(const std::string& payload)>

Callable that attempts to replay a single queued item and reports what it achieved.

The three-outcome counterpart of ReplayFunction: it is the only form that can report ReplayOutcome::Undelivered and so keep a reconnect flap from spending the item's retry budget.

Definition at line 157 of file sync_worker.hpp.

DeadLetterSink

using morph::offline::SyncWorker::DeadLetterSink = std::function<void(const QueueItem& poisoned)>

Callable invoked when an item exhausts its retry budget, just before it is removed from the queue.

Receives the exhausted item (payload, idempotencyKey, and its final cumulative attempt count). If unset, SyncWorker keeps the default behavior: log at morph::log::LogLevel::error and drop. A throwing sink is caught and logged — the item is still removed.

Definition at line 166 of file sync_worker.hpp.

Constructor & Destructor Documentation

SyncWorker() [1/2]

morph::offline::SyncWorker::SyncWorker ( ::morph::exec::IExecutor & owner, IOfflineQueue & queue, ReplayFunction replay, DeadLetterSink deadLetterSink = nullptr )

inline

Constructs a worker that drains queue using replay, on owner.

Parameters

owner	The executor every drain runs on; must run one task at a time (a strand). Borrowed: it must outlive this worker and keep running tasks until it is destroyed.
queue	Queue to drain on each run() call, belonging to owner (or to the executor it runs over). Borrowed, not owned: it must outlive this worker.
replay	Function called for each pending item. Stored and invoked for this worker's whole lifetime, so anything the callable refers to must outlive it.
deadLetterSink	Optional hook invoked with the exhausted item instead of the default log-and-drop path when an item exhausts its retry budget. Default: unset. Retained on the same terms as replay.

Definition at line 183 of file sync_worker.hpp.

SyncWorker() [2/2]

morph::offline::SyncWorker::SyncWorker ( ::morph::exec::IExecutor & owner, IOfflineQueue & queue, DetailedReplayFunction replay, DeadLetterSink deadLetterSink = nullptr )

inline

Constructs a worker that drains queue using replay, which distinguishes an undelivered replay from a rejected one.

Identical to the overload above except for the replay callable's return type. Only ReplayOutcome::Rejected spends an attempt, so a reconnect flap that never reached the peer no longer walks queued work towards the dead-letter sink.

The two overloads are unambiguous: ReplayOutcome is a scoped enum, so neither return type implicitly converts to the other.

Parameters

owner	The executor every drain runs on; must run one task at a time (a strand). Borrowed: it must outlive this worker and keep running tasks until it is destroyed.
queue	Queue to drain on each run() call, belonging to owner (or to the executor it runs over). Borrowed, not owned: it must outlive this worker.
replay	Function called for each pending item. Stored and invoked for this worker's whole lifetime, so anything the callable refers to must outlive it.
deadLetterSink	Optional hook invoked with the exhausted item instead of the default log-and-drop path when an item exhausts its retry budget. Default: unset. Retained on the same terms as replay.

Definition at line 215 of file sync_worker.hpp.

Member Function Documentation

run()

::morph::async::Completion< SyncResult > morph::offline::SyncWorker::run ( ::morph::exec::IExecutor & replyExec )

inline

Drains the queue, replaying each item via the replay function, on the owner.

On the owner the drain runs before this returns; from any other thread it is posted there, after any drain posted before it. It processes every pending item, or stops early once stop() is signalled.

If stop() was called before the drain starts, the drain does nothing, resets the stop flag and answers an empty result.

A posted drain refers to this worker, so the worker must outlive every drain it has posted: destroy it once every run() has settled, or after its owner is closed.

Parameters

replyExec

Executor the answer is delivered on and the caller attaches its callbacks on: the caller's own. Borrowed: it must outlive the returned Completion.

Returns

A Completion settled on the owner with the counts of successful / failed / dead-lettered replays, delivered on replyExec.

Definition at line 240 of file sync_worker.hpp.

stop()

void morph::offline::SyncWorker::stop ( )

inline

Signals an in-progress drain to stop after the current item.

Callable from any thread. The drain clears the flag at its start, so stopping is one-shot — but a stop() that lands during a drain leaves the flag set on return, so the next drain takes its early-out and drains nothing; work resumes only on the drain after that.

Definition at line 257 of file sync_worker.hpp.

---

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

• morph/offline/sync_worker.hpp