Struct ApplyContext 

pub struct ApplyContext<'a, S, T, P = ()> { /* private fields */ }

Library / custom-loop driver sharing the same ordered apply path as crate::pipeline::run_source_runtime.

Accepts row Change and RelationChange into one buffer / window. Identity and non-identity pipelines share one path: every batch is spawned onto the transform JoinSet (identity is an async no-op). Sink apply is ordered and may overlap with polling / transforming later batches; push_* / flush still return Some(position) only after that batch’s sink succeeds. They do not call SourceDriver::advance_watermark / ChangeFeed::advance_watermark.

Poisoning after FailurePolicy::Fail

After a batch fails under FailurePolicy::Fail, this context is poisoned: successors are discarded, and further push/flush return Err immediately. Do not reuse — create a new context and replay from the last successful sink-safe watermark.

Implementations

impl<'a, S, T, P> ApplyContext<'a, S, T, P>

where S: SurrealSink, T: BatchTransformer + 'static, P: Clone + Send + Sync + 'static,

pub fn new(sink: &'a S, transformer: Arc<T>, opts: &'a ApplyOpts) -> Self

Create an apply context bound to a sink, transformer, and options.

pub fn is_poisoned(&self) -> bool

Whether this context was poisoned by a FailurePolicy::Fail error.

pub fn buffer_len(&self) -> usize

Number of events waiting to form a batch.

pub fn in_flight_count(&self) -> usize

Number of batches currently transforming (JoinSet path).

pub fn sink_in_flight(&self) -> bool

Whether an ordered sink apply is in progress (slot reserved).

pub fn window_occupancy(&self) -> usize

Batches occupying the apply window: transforming + awaiting/in sink.

ApplyOpts::max_in_flight bounds this total so a slow sink back-pressures new transforms while still allowing overlap (reads / transforms / writes).

pub fn take_sunk_change_count(&mut self) -> u64

Take and reset the count of events sunk since the previous take.

pub fn completed_waiting_count(&self) -> usize

Number of transform results waiting for ordered sink apply.

pub fn has_unsunk_work(&self) -> bool

Whether any buffered, in-flight, completed-waiting, or sinking work remains.

Used by checkpoint policies that must not persist a read-ahead position while transform/apply still has unsunk work.

pub fn is_fully_drained(&self) -> bool

Whether the apply window is fully drained (no unsunk work).

pub async fn push_change( &mut self, change: Change, position: P, ) -> Result<Option<P>>

Push one row change; may start transforms and drain ordered sink.

Returns the last position successfully sunk (caller should advance_watermark).

pub async fn push_relation_change( &mut self, change: RelationChange, position: P, ) -> Result<Option<P>>

Push one relation change into the same window as row changes.

pub async fn push_event( &mut self, event: ApplyEvent, position: P, ) -> Result<Option<P>>

Push a unified positioned event.

pub async fn flush(&mut self) -> Result<Option<P>>

Flush remaining buffered events and wait for in-flight work.

pub async fn write_rows(&self, rows: Vec<Row>) -> Result<()>

Transform then sink rows (same as write_rows_with).

pub async fn write_relations(&self, relations: Vec<Relation>) -> Result<()>

Transform then sink relations (same as write_relations_with).