Trait SingletonSubscriber 

pub trait SingletonSubscriber<P, L = InsertOrder>:
    Send
    + Sync
    + 'static
where
    P: Serialize + DeserializeOwned + Send + Sync + 'static + Unpin,
    L: Lane,
{
    type Batch: Default + Send + 'static;

    const SUBSCRIPTION: StreamSelection = StreamSelection::All;

    // Provided methods
    fn handle_persistent<'inv>(
        &self,
        ctx: EventCtx<'inv, Self::Batch>,
        event: &EventDelivery<P, L>,
    ) -> impl Future<Output = Result<Handled<'inv>, Box<dyn Error + Send + Sync>>> + Send { ... }
    fn handle_undecodable(
        &self,
        error: &UndecodableDelivery<L>,
    ) -> impl Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send { ... }
    fn flush(
        &self,
        op: &mut FlushOp<'_, L>,
        items: Self::Batch,
    ) -> impl Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send { ... }
    fn handle_ephemeral(
        &self,
        event: &Arc<EphemeralOutboxEvent<P>>,
    ) -> impl Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send { ... }
}

Handles the events of one outbox listener job.

Exactly one instance exists per type, created at registration and always on. That presence is a contract, not an implementation detail: it is what licenses the ephemeral subscription below, since ephemeral events cannot be replayed and only an always-present consumer may hear them. It is also why a singleton subscriber has no way to pause — no pause_until, no staged chain. For a single-instance flow that does need to pause or stage, see KeyedSubscriber with one static key.

handle_persistent receives an EventCtx and must resolve it into a Handled token, deciding the transactional fate of every event:

• skip — not my event; costs no transaction at all. The checkpoint advances lazily.
• collect_with / the collect sugar — contribute an item to the pending batch’s Batch accumulator; a pure memory write. The runner applies the whole accumulator via flush exactly once per batch landing, inside the transaction that commits the checkpoint — N per-event statements become one batched flush. The batch lands when the ready persistent backlog is drained (it is never held open waiting on the network), when max_batch_size is reached, when a later event consumes, or on shutdown. Requires the work to tolerate whole-batch replay after a mid-batch failure.
• consume then commit — my event is its own atomic unit: the pending batch (items + checkpoint) lands before my work starts, and my op commits at return. Use for causally significant or risky work.

Ephemeral events are delivered on their own stream and are handled between batches: they never interrupt a pending batch, and nothing is pending while handle_ephemeral runs. Most handlers only consume one of the two streams — declare it via SUBSCRIPTION and the other stream is never even subscribed.

L is the delivery lane, which decides what a position is — for the event, the flush and the checkpoint. It defaults to InsertOrder; impl SingletonSubscriber<P, CommitOrder> for X is the same code reading CommitSequences. A subscription cannot later move to the other lane — registration refuses it.

Provided Associated Constants

const SUBSCRIPTION: StreamSelection = StreamSelection::All

Which delivery streams this handler’s job subscribes to. Defaults to All.

Declaring a single-stream mode is a contract, not a filter: the other stream is never subscribed, so its handler method is never called — overriding handle_ephemeral on a PersistentOnly handler (or handle_persistent on an EphemeralOnly one) is dead code.

Required Associated Types

type Batch: Default + Send + 'static

Accumulator for events resolved via collect_with — Vec<T> for append-style batching, HashMap<K, V> for keyed coalescing folds, or any other Default container. Handlers that never collect use ().

Provided Methods

fn handle_persistent<'inv>( &self, ctx: EventCtx<'inv, Self::Batch>, event: &EventDelivery<P, L>, ) -> impl Future<Output = Result<Handled<'inv>, Box<dyn Error + Send + Sync>>> + Send

The event, plus its position on L. It derefs to the event, and a handler retaining it past the call clones only an Arc via inner.

fn handle_undecodable( &self, error: &UndecodableDelivery<L>, ) -> impl Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send

Handle a persistent event whose stored payload could not be decoded into P — delivered as the persistent stream’s Err arm and never as an ordinary event, so it cannot reach handle_persistent. It arrives at the position an ordinary event would have had, and derefs to the UndecodableEventError carrying the raw payload and serde error.

The runner lands the pending batch before invoking this — like handle_ephemeral, nothing is pending while it runs and no batch transaction spans the await, and work completed up to the event is durable regardless of the outcome.

The default fails with the error as-is: a payload the consumer cannot decode is a runtime bug (schema drift, a producer ahead of this consumer, foreign rows in the table) that must surface loudly, not silently pass by. On Err the job fails with its checkpoint parked at the sequence before the event — every retry re-reads the event from the database, so deploying a consumer that understands the payload resumes the pipeline automatically, in order, with nothing skipped.

Override and return Ok(()) only when this handler genuinely wants to move past payloads it cannot decode — an explicit, auditable decision (consider recording error.failure somewhere durable first). The checkpoint then advances over the event exactly as for a skip. Any work done here must tolerate replay (the event is redelivered if the acknowledgement’s checkpoint was not yet persisted at a crash).

fn flush( &self, op: &mut FlushOp<'_, L>, items: Self::Batch, ) -> impl Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send

Apply everything collected since the last flush. Called at most once per batch landing, inside the batch transaction, before the checkpoint commits — items and pointer land atomically.

op is the batch op behind a restricted FlushOp view: execute statements or register commit hooks on it for work that must share the checkpoint’s fate; ignore it when flushing to a foreign database (then make the writes idempotent — the checkpoint only advances after Ok, and a failure replays and re-collects the whole batch). Its position is where the batch lands on L.

fn handle_ephemeral( &self, event: &Arc<EphemeralOutboxEvent<P>>, ) -> impl Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send

Handed the shared Arc for the same reason as handle_persistent, though an ephemeral event carries no sequence and belongs to no batch, so there is rarely anything to retain.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors