orxtra v0.13.0 /dispatch.src.orxtra.dispatch._dispatch_worker
Updated
On this page

DispatchWorker is the durable at-least-once dispatch loop: polls events past a cursor, matches persistent subscriptions, executes or accumulator-buffers actions, and records completions.

#dispatch.src.orxtra.dispatch._dispatch_worker

#dispatch.src.orxtra.dispatch._dispatch_worker

Dispatcher worker: durable event-processing loop.

Polls the events table via a DispatchBackend, matches events against persistent subscriptions, executes actions through the ActionExecutor protocol, and records per-event-action completion records for at-least-once delivery.

The worker lives in dispatch (orchestration layer) and depends only on protocols -- never on the concrete ServicesActionExecutor. Services wires the concrete implementation; CLI registers the command.

#DispatchWorker

Durable dispatcher worker with at-least-once delivery.

Constructor takes:

  • backend: a DispatchBackend with cursor/completion/poll methods

(PgDispatchBackend or InMemoryDispatchBackend)

  • action_executor: the ActionExecutor protocol (for WorkflowAction)
  • flush_scheduler: the FlushScheduler protocol
  • pool: an asyncpg Pool (for LISTEN/NOTIFY)
  • cursor_name: identifies this worker's cursor position
  • events_channel: the PG NOTIFY channel name
  • poll_interval: fallback poll interval in seconds
  • batch_size: max events per poll

The main loop:

  1. Read cursor position (or start from beginning)
  2. LISTEN on events_channel as a wake hint
  3. Poll for unprocessed events since cursor
  4. For each event: match subscriptions, check completion records,

execute action, record completion, advance cursor

  1. Wait for NOTIFY or poll interval, repeat

#run

python
async def run(self) -> None

Main loop: poll, match, execute, advance, wait.

#stop

python
async def stop(self) -> None

Signal the main loop to exit cleanly.

#_wait_for_wake_or_stop

python
async def _wait_for_wake_or_stop(self) -> None

Wait until either wake_event or stop_event is set.

#_setup_listen

python
async def _setup_listen(self) -> None

Acquire a dedicated connection and LISTEN on the events channel.

#_teardown_listen

python
async def _teardown_listen(self) -> None

Remove the listener and release the connection.

#_on_notify

python
def _on_notify(self, _connection: asyncpg.Connection[Any] | asyncpg.pool.PoolConnectionProxy[Any], _pid: int, _channel: str, _payload: object) -> None

NOTIFY callback: set the wake event to trigger an immediate poll.

#_poll_and_process

python
async def _poll_and_process(self) -> int

Poll for new events and process them. Returns count processed.

#_maybe_sweep_expired_inbox

python
async def _maybe_sweep_expired_inbox(self) -> None

Expire pending inbox items with past deadlines, at most once per interval.

The min-interval guard ensures this does not fire per-batch under load (the loop continues immediately when events are available).

#_execute_action_immediate

python
async def _execute_action_immediate(self, sub_action: SubscriptionAction, event_id: UUID, event_type: str, principal_id: UUID | None, data: dict[str, Any] | None, owner_principal_id: UUID) -> None

Execute a single action immediately (no accumulator).

#_buffer_or_flush

python
async def _buffer_or_flush(self, sub_action: SubscriptionAction, event_id: UUID, owner_principal_id: UUID) -> None

Buffer event for accumulator; flush inline if count threshold reached.

  • Buffer the event via the backend.
  • If count threshold is set and pending >= threshold, claim and execute.
  • If time threshold is set, schedule a deferred flush via FlushScheduler.
  • Whichever fires first wins (standard accumulator contract).

#_make_flush_callback

python
def _make_flush_callback(self, sub_action: SubscriptionAction, owner_principal_id: UUID) -> Any

Create a zero-arg async callback for FlushScheduler.

#_flush_action

python
async def _flush_action(self, sub_action: SubscriptionAction, owner_principal_id: UUID) -> None

Claim a batch from the accumulator buffer and execute the action.

#_make_event_fire_callback

python
def _make_event_fire_callback(self, owner_principal_id: UUID) -> Any

Create a callback for EventAction dispatch.

A derived event re-fired by an EventAction is attributed to the OWNING SUBSCRIPTION's principal: the subscription whose action triggered the re-fire is the actor behind the new event. Every event the worker processes flows through a subscription, so an owner is always in scope.

Search