Struct ActivityEventPublisher 

pub struct ActivityEventPublisher { /* private fields */ }

The durable transcript sequencer + live fan-out for one deployment.

Cloneable: the broadcast sender and the store handle are shared, so every clone sequences into the same O keyspace and fans out to the same live subscribers. The store is the single source of store_seq monotonicity; the broadcast is best-effort live tail only.

Implementations

impl ActivityEventPublisher

pub fn new( store: Arc<dyn ObservabilityStore>, capacity: NonZeroUsize, batch: TranscriptBatchPolicy, ) -> Self

Build a publisher over store with a live broadcast of capacity and the drain’s batch flush policy.

capacity is the bounded live-tail buffer; a subscriber that lags beyond it receives one typed TranscriptStreamLagged then re-resumes from the durable tail. It must be non-zero (validated by the caller’s config).

batch is a REQUIRED argument, not an option with a fallback: the flush policy is the operator’s ruling on store cost against transcript latency and this type refuses to invent one (see TranscriptBatchPolicy).

pub async fn publish( &self, event: &ActivityEvent, ) -> Result<Option<u64>, StoreError>

Sequence + persist + fan out one event.

Ephemeral events are fanned out live with store_seq: None and are NEVER persisted. Non-ephemeral events are appended to the O keyspace under the commit-allocated store_seq (via the read-head -> append(expected_seq) -> on-conflict-re-read-head-and-retry loop), then fanned out carrying that store_seq. Returns the assigned store_seq for a persisted event, or None for an ephemeral one.

A send with no live subscribers is not an error (the calm no-dashboard case); the durable append is the primary artifact.

Errors

A StoreError from the durable append (after exhausting the retry budget on pathological contention, or any non-conflict backend error).

pub async fn publish_all( &self, events: &[ActivityEvent], ) -> Result<Vec<Option<u64>>, StoreError>

Sequence + persist + fan out a batch of events as FEW durable commits as the batch allows: one commit per distinct (workflow, run, activity, attempt) stream present in events.

Returns one entry per input event, in input order: the assigned store_seq for a persisted event, or None for an ephemeral one and for one past the stream’s retention cap — the same per-event answer Self::publish gives, because publish IS this method over a slice of one.

Events keep their arrival order within a stream, and a stream’s whole share of the batch is committed at one expected_seq under the same read-head -> append -> on-conflict-re-read-head-and-retry loop the single-event path used, so store_seq monotonicity and the single-writer law are untouched: a SequenceConflict leaves the WHOLE batch unwritten (the store contract is all-or-nothing) and the batch is retried at the observed head.

Errors

A StoreError from a durable append. Streams are independent, so a failure on one stream does not abandon the others: every stream in the batch is attempted and the FIRST error is returned once they have all been tried. (A caller with a single-stream batch — every drain call in this server — sees exactly the single-event behaviour.)

pub async fn replay_from( &self, key: &ActivityStreamKey, from_seq: u64, ) -> Result<Vec<ActivityRecord>, StoreError>

Read the durable O tail for key with store_seq >= from_seq.

The priming read a resuming transcript client replays before splicing onto the live stream. from_seq = 0 replays the whole persisted transcript.

Errors

A StoreError from the durable read.

pub async fn list_streams( &self, workflow_id: &WorkflowId, run_id: &RunId, ) -> Result<Vec<ActivityStreamSummary>, StoreError>

Enumerate the retained transcript streams of ONE RUN of workflow_id from the durable O keyspace (empty for a run with none — old runs simply have no retained transcript).

The run is required, never an optional filter: a workflow-wide enumeration over a continue-as-new chain would list several generations’ streams under coordinates that collide pairwise, and the caller could not tell them apart.

Errors

A StoreError from the durable enumeration.

pub fn subscribe( &self, key: ActivityStreamKey, after_seq: Option<u64>, ) -> BoxStream<'static, Result<ActivityEvent, TranscriptStreamLagged>>

Subscribe to the live transcript tail for key, suppressing every event for a DIFFERENT stream and every persisted event already covered by the resume cursor.

The broadcast is deployment-wide (one channel), so this filters to key’s (workflow, run, activity, attempt) stream — an event from a sibling continue-as-new generation of the same workflow fails the key comparison and is suppressed, exactly like a different attempt’s.

after_seq dedups the splice seam exactly like the cluster channel: attach this receiver BEFORE reading the priming Self::replay_from tail, so an event that races the priming read is retained by the receiver and applied after it (deduped on store_seq).

The cursor is an Option because store_seq is 0-based (the first event is store_seq == 0): after_seq = None is a FRESH subscriber that has applied nothing and must see every event including store_seq == 0; after_seq = Some(n) has already applied through store_seq == n, so events with store_seq <= n are suppressed at the seam. Ephemeral events (which carry store_seq: None) for key are ALWAYS forwarded live — they have no sequence to dedup and are never replayed.

Trait Implementations

impl Clone for ActivityEventPublisher

fn clone(&self) -> ActivityEventPublisher

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for ActivityEventPublisher

fn fmt(&self, formatter: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.