Trait PipelineCallback 

pub trait PipelineCallback:
    Send
    + 'static
    + Send {
    // Required methods
    fn execute_cycle(
        &mut self,
        source_batches: &FxHashMap<Arc<str>, Vec<RecordBatch>>,
        watermark: i64,
    ) -> impl Future<Output = Result<CycleOutcome, CycleError>> + Send;
    fn complete_pending_vnode_transition(
        &mut self,
    ) -> impl Future<Output = Result<bool, CycleError>> + Send;
    fn drain_checkpoint_edges_until(
        &mut self,
        deadline: Instant,
    ) -> impl Future<Output = Result<(), CycleError>> + Send;
    fn write_to_sinks(
        &mut self,
        results: &FxHashMap<Arc<str>, Vec<RecordBatch>>,
        deadline: Option<Instant>,
    ) -> impl Future<Output = Result<(), CycleError>> + Send;
    fn extract_watermark(
        &mut self,
        source_name: &str,
        batch: &RecordBatch,
        admission_floor: i64,
    ) -> Result<(), CycleError>;
    fn reconcile_source_input_channels(
        &mut self,
        source_name: &str,
        input_channels: Option<Arc<[Vec<u8>]>>,
    ) -> Result<(), CycleError>;
    fn filter_late_rows(
        &self,
        source_name: &str,
        batch: &RecordBatch,
    ) -> Result<Option<RecordBatch>, CycleError>;
    fn current_watermark(&self) -> i64;
    fn abandon_checkpoint_attempt(
        &mut self,
        _attempt: CheckpointAttempt,
        _reason: &str,
        _flags: u64,
        _assignment_fence: Option<CheckpointAssignmentFence>,
    ) -> impl Future<Output = Result<(), String>> + Send;
    fn cancel_source_barrier_attempt(
        &mut self,
        _attempt: CheckpointAttempt,
        _reason: &str,
    ) -> impl Future<Output = Result<(), String>> + Send;
    fn service_checkpoint_control(
        &mut self,
        source_offsets: FxHashMap<String, SourceCheckpoint>,
    ) -> impl Future<Output = CheckpointControlOutcome> + Send;
    fn checkpoint_with_barrier(
        &mut self,
        source_checkpoints: FxHashMap<String, SourceCheckpoint>,
        attempt: CheckpointAttempt,
        attempt_started: Instant,
        deadline: Instant,
        flags: u64,
        assignment_fence: Option<CheckpointAssignmentFence>,
    ) -> impl Future<Output = BarrierOutcome> + Send;
    fn record_cycle(&self, events_ingested: u64, batches: u64, elapsed_ns: u64);
    fn apply_control(&mut self, msg: ControlMsg);

    // Provided methods
    fn prepare_source_intake(&mut self) -> Result<(), String> { ... }
    fn pin_source_frontiers_for_new_cycle(&mut self) -> Result<(), String> { ... }
    fn push_to_streams(
        &self,
        results: &FxHashMap<Arc<str>, Vec<RecordBatch>>,
    ) -> Result<(), CycleError> { ... }
    fn commit_subscription_output(&mut self) { ... }
    fn abort_subscription_output(&mut self) { ... }
    fn update_mv_stores(
        &self,
        results: &FxHashMap<Arc<str>, Vec<RecordBatch>>,
    ) -> Result<(), CycleError> { ... }
    fn is_leader(&self) -> bool { ... }
    fn is_recovering(&mut self) -> bool { ... }
    fn fault_on_cycle_error(&self) -> bool { ... }
    fn take_pipeline_halt(&mut self) -> Option<String> { ... }
    fn take_pipeline_fault(&mut self) -> Option<String> { ... }
    fn cancel_checkpoint_tails_for_recovery(&mut self) -> bool { ... }
    fn record_checkpoint_failure(&mut self, _checkpoint_id: u64, _reason: &str) { ... }
    fn record_checkpoint_continuation_fault(
        &mut self,
        _attempt: CheckpointAttempt,
        _reason: &str,
    ) { ... }
    fn record_checkpoint_admission_failure(&mut self, _reason: &str) { ... }
    fn settle_checkpoint_tail_tasks(
        &mut self,
    ) -> impl Future<Output = Result<(), String>> + Send { ... }
    fn reserve_checkpoint_attempt(
        &mut self,
        _deadline: Instant,
    ) -> impl Future<Output = Result<CheckpointAttempt, String>> + Send { ... }
    fn publish_checkpoint_prepare(
        &mut self,
        _attempt: CheckpointAttempt,
        _attempt_started: Instant,
        _deadline: Instant,
        _flags: u64,
        _assignment_fence: Option<CheckpointAssignmentFence>,
    ) -> impl Future<Output = Result<(), String>> + Send { ... }
    fn resolve_authoritative_follower_abort(
        &mut self,
        _attempt: CheckpointAttempt,
    ) -> Result<(), String> { ... }
    fn checkpoint_assignment_for_admission(
        &mut self,
        _deadline: Instant,
    ) -> impl Future<Output = CheckpointAssignmentAdmission> + Send { ... }
    fn checkpoint_control_wake(&self) -> Option<CheckpointControlWake> { ... }
    fn shuffle_work_wake(&self) -> Option<Arc<Notify>> { ... }
    fn tick_idle_watermark(&mut self) { ... }
    fn record_cycle_phases(
        &self,
        _execute_ns: u64,
        _output_store_ns: u64,
        _sink_enqueue_ns: u64,
    ) { ... }
    fn note_cycle_error(&self) { ... }
    fn is_backpressured(&self) -> bool { ... }
    fn external_output_pressure(&self) -> ExternalOutputPressure { ... }
    fn intake_paused(&self) -> bool { ... }
    fn has_deferred_input(&self) -> bool { ... }
    fn has_runnable_deferred_input(&self) -> bool { ... }
    fn reserve_subscription_cut(
        &mut self,
        _attempt: CheckpointAttempt,
    ) -> Result<(), String> { ... }
    fn abort_subscription_cut(&mut self, _attempt: CheckpointAttempt) { ... }
    fn publish_barrier(
        &mut self,
        _attempt: CheckpointAttempt,
    ) -> Result<(), String> { ... }
    fn invalidate_subscriptions(&self, _reason: &str) { ... }
    fn settle_sink_epoch_for_shutdown(
        &mut self,
    ) -> impl Future<Output = Result<(), String>> + Send { ... }
    fn close_sinks(&mut self) -> impl Future<Output = Result<(), String>> + Send { ... }
    fn set_barrier_injectors(&mut self, injectors: Vec<SourceBarrierControl>) { ... }
}

Callback trait for the coordinator to interact with the rest of the DB. Trait exists for test seam; production impl is ConnectorPipelineCallback.

Required Methods

fn execute_cycle( &mut self, source_batches: &FxHashMap<Arc<str>, Vec<RecordBatch>>, watermark: i64, ) -> impl Future<Output = Result<CycleOutcome, CycleError>> + Send

Execute a SQL cycle over the accumulated source batches. Err is a whole-cycle failure (all domains, or a backpressure halt); per-domain faults surface in CycleOutcome so healthy domains still commit.

fn complete_pending_vnode_transition( &mut self, ) -> impl Future<Output = Result<bool, CycleError>> + Send

Complete assignment-scoped vnode lifecycle work without admitting source input.

Recovery can close intake before a predecessor transition has run. Implementations return true only when pending work completed; false means idle or waiting for its exact assignment certificate.

fn drain_checkpoint_edges_until( &mut self, deadline: Instant, ) -> impl Future<Output = Result<(), CycleError>> + Send

Drain every graph input that belongs to the frozen checkpoint cut.

Implementations must deliver each drain pass’s outputs before returning. The returned future is one owner transaction: its caller must await an explicit result or destroy the complete callback/coordinator generation. It must not drop the future while retaining that generation. Operators may temporarily own graph input across an await, and a completed graph pass may have partially published materialized-view, stream, or sink output.

An implementation may cancel a nested operation for its absolute deadline or process lease only when it regains control and records or returns a disposition that prevents checkpoint capture from passing incomplete publication. A future caller that needs an outer timeout, select!, or task abort must first add a coordinator-owned attempt transaction covering graph/output publication, source-barrier ownership, and attempt cleanup. The current implementation checks its absolute deadline between complete graph passes and inside sink-publication awaits whose outcomes are consumed before capture.

fn write_to_sinks( &mut self, results: &FxHashMap<Arc<str>, Vec<RecordBatch>>, deadline: Option<Instant>, ) -> impl Future<Output = Result<(), CycleError>> + Send

Write cycle results to sinks, bounded by deadline when this is a checkpoint drain.

fn extract_watermark( &mut self, source_name: &str, batch: &RecordBatch, admission_floor: i64, ) -> Result<(), CycleError>

Extract watermark from a batch for a given source.

Errors

Returns an error when the source batch has invalid event-time metadata.

fn reconcile_source_input_channels( &mut self, source_name: &str, input_channels: Option<Arc<[Vec<u8>]>>, ) -> Result<(), CycleError>

Install the exact input-channel inventory carried by a source cursor.

Errors

Returns an error when the inventory is invalid or conflicts with recovered progress.

fn filter_late_rows( &self, source_name: &str, batch: &RecordBatch, ) -> Result<Option<RecordBatch>, CycleError>

Filter late rows while preserving any validated hidden source metadata.

Errors

Returns an error when hidden metadata is malformed or the batch cannot be filtered.

fn current_watermark(&self) -> i64

Current pipeline watermark.

fn abandon_checkpoint_attempt( &mut self, _attempt: CheckpointAttempt, _reason: &str, _flags: u64, _assignment_fence: Option<CheckpointAssignmentFence>, ) -> impl Future<Output = Result<(), String>> + Send

Abandon a reserved attempt that cannot reach the durable checkpoint tail. Transactional sinks must roll back that epoch before the next attempt begins.

fn cancel_source_barrier_attempt( &mut self, _attempt: CheckpointAttempt, _reason: &str, ) -> impl Future<Output = Result<(), String>> + Send

Cancel an exact follower source-barrier attempt before capture.

An attempt admitted with follower ownership releases its exact local reservation and publishes a negative barrier acknowledgement, even if this process changes role while the attempt is active. Originator-owned attempts use abandon_checkpoint_attempt.

fn service_checkpoint_control( &mut self, source_offsets: FxHashMap<String, SourceCheckpoint>, ) -> impl Future<Output = CheckpointControlOutcome> + Send

Service a cluster follower announcement observed from the leader.

Periodic, manual, and shutdown admission belongs exclusively to the streaming coordinator. This control seam must never originate a local checkpoint.

fn checkpoint_with_barrier( &mut self, source_checkpoints: FxHashMap<String, SourceCheckpoint>, attempt: CheckpointAttempt, attempt_started: Instant, deadline: Instant, flags: u64, assignment_fence: Option<CheckpointAssignmentFence>, ) -> impl Future<Output = BarrierOutcome> + Send

Called when all sources have aligned on a barrier.

fn record_cycle(&self, events_ingested: u64, batches: u64, elapsed_ns: u64)

Record cycle metrics.

fn apply_control(&mut self, msg: ControlMsg)

Apply a DDL control message (add/drop stream) to the running pipeline.

Provided Methods

fn prepare_source_intake(&mut self) -> Result<(), String>

Install any newly published recovery cut before the coordinator removes another source message from its FIFO. The default is a no-op outside a clustered source-handoff runtime.

Errors

Returns an error if a pending recovery cut cannot be installed.

fn pin_source_frontiers_for_new_cycle(&mut self) -> Result<(), String>

Pin decision-bound source frontiers before a new source-admission cycle begins.

The coordinator does not call this while graph-owned deferred input is replaying, so one snapshot covers source filtering, initial execution, and every retained replay pass.

Errors

Returns an error when the frontier snapshot cannot be installed.

fn push_to_streams( &self, results: &FxHashMap<Arc<str>, Vec<RecordBatch>>, ) -> Result<(), CycleError>

Push cycle results to stream subscriptions.

Errors

Returns an error if subscription delivery rejects the cycle output.

fn commit_subscription_output(&mut self)

Publish aggregate output bookkeeping after every cycle consumer accepted the output.

fn abort_subscription_output(&mut self)

Restore retryable aggregate bookkeeping after cycle publication fails.

fn update_mv_stores( &self, results: &FxHashMap<Arc<str>, Vec<RecordBatch>>, ) -> Result<(), CycleError>

Update materialized view stores with cycle results.

Errors

Returns an error if materialized state cannot apply the cycle output.

fn is_leader(&self) -> bool

true if this node is the cluster leader, or in single-node mode.

fn is_recovering(&mut self) -> bool

true while a coordinated restart is in flight; the checkpoint admission gate holds. Default false.

fn fault_on_cycle_error(&self) -> bool

true if a fatal cycle error should fault for recovery rather than drop-and-continue (exactly-once, or coordinated recovery). Default false (at-least-once drops).

fn take_pipeline_halt(&mut self) -> Option<String>

Take a pending deterministic pipeline halt. A halt is permanent for the current deployment and therefore takes precedence over consistency faults that would otherwise trigger recovery of the same poison input.

fn take_pipeline_fault(&mut self) -> Option<String>

Take a pending consistency fault from checkpointing or a poisoned sink epoch. The coordinator stops intake so recovery can replay from the last committed cut.

fn cancel_checkpoint_tails_for_recovery(&mut self) -> bool

Cancel checkpoint tails only after coordinated recovery owns the lifecycle fence. Returns whether that fence was observed, so shutdown need not repeat cancellation.

fn record_checkpoint_failure(&mut self, _checkpoint_id: u64, _reason: &str)

Record a checkpoint failure observed by the coordinator. Exactly-once implementations fault for recovery; weaker guarantees may retain retry-on-next-interval behaviour.

fn record_checkpoint_continuation_fault( &mut self, _attempt: CheckpointAttempt, _reason: &str, )

Record an invariant failure after the checkpoint itself became durable.

fn record_checkpoint_admission_failure(&mut self, _reason: &str)

Record a failure before an exact checkpoint attempt could be reserved.

fn settle_checkpoint_tail_tasks( &mut self, ) -> impl Future<Output = Result<(), String>> + Send

Join tracked asynchronous checkpoint tails before connector teardown.

fn reserve_checkpoint_attempt( &mut self, _deadline: Instant, ) -> impl Future<Output = Result<CheckpointAttempt, String>> + Send

Durably reserve the exact attempt before barriers are admitted to sources.

Implementations must never synthesize an in-memory checkpoint ID. A successful reservation may be abandoned, but its ID is permanently burned.

fn publish_checkpoint_prepare( &mut self, _attempt: CheckpointAttempt, _attempt_started: Instant, _deadline: Instant, _flags: u64, _assignment_fence: Option<CheckpointAssignmentFence>, ) -> impl Future<Output = Result<(), String>> + Send

Durably admit exact checkpoint artifacts, then publish the certified cluster Prepare, before any source or shuffle barrier is injected.

fn resolve_authoritative_follower_abort( &mut self, _attempt: CheckpointAttempt, ) -> Result<(), String>

Resolve exact local follower state after an authoritative pre-capture BarrierOutcome::Aborted. This operation must not publish control traffic or wait on the network because cluster authority has already terminated the attempt.

Errors

Returns an error if local follower state cannot be resolved exactly.

fn checkpoint_assignment_for_admission( &mut self, _deadline: Instant, ) -> impl Future<Output = CheckpointAssignmentAdmission> + Send

Capture the exact assignment certificate for a new attempt.

fn checkpoint_control_wake(&self) -> Option<CheckpointControlWake>

Wake the coordinator for leader-originated checkpoint control. None keeps local runtimes free of cluster polling.

fn shuffle_work_wake(&self) -> Option<Arc<Notify>>

Wake the coordinator when inbound data or deferred shuffle work becomes ready.

fn tick_idle_watermark(&mut self)

Demote sources idle past their timeout so a quiet input doesn’t pin the combined watermark.

fn record_cycle_phases( &self, _execute_ns: u64, _output_store_ns: u64, _sink_enqueue_ns: u64, )

Record the three timed phases of one successfully published normal cycle. Checkpoint graph drains use a separate execution path and must not call this hook.

fn note_cycle_error(&self)

Count a fatal cycle error that was dropped-and-continued (at-least-once only).

fn is_backpressured(&self) -> bool

true when internal buffers are near capacity.

fn external_output_pressure(&self) -> ExternalOutputPressure

Pressure from bounded checkpoint-committable output. A due checkpoint accelerates the next periodic cut without stopping compute. Commit backpressure keeps checkpoint completion and control traffic live but stops source and deferred-input work until the prepared cut resolves.

fn intake_paused(&self) -> bool

true while the runtime must not fold source or shuffle input into operator state. Cluster startup and coordinated recovery use this stronger fence; ordinary backpressure only pauses source polling.

fn has_deferred_input(&self) -> bool

true when deferred operators have pending input to drain.

fn has_runnable_deferred_input(&self) -> bool

true when retained work can run now without waiting for an external wake.

fn reserve_subscription_cut( &mut self, _attempt: CheckpointAttempt, ) -> Result<(), String>

Reserve each subscription log’s cursor at the aligned checkpoint cut.

Errors

Returns an error if the subscription cut cannot be reserved atomically.

fn abort_subscription_cut(&mut self, _attempt: CheckpointAttempt)

Discard an unresolved subscription cut after checkpoint failure.

fn publish_barrier(&mut self, _attempt: CheckpointAttempt) -> Result<(), String>

Resolve the exact cut for external SUBSCRIBE consumers after durable commit.

Errors

Returns an error if the committed cut cannot be published atomically.

fn invalidate_subscriptions(&self, _reason: &str)

Terminate provisional subscription delivery before shutdown or recovery replay.

fn settle_sink_epoch_for_shutdown( &mut self, ) -> impl Future<Output = Result<(), String>> + Send

Resolve durable ownership of any open checkpoint-committable sink epoch while its actor is still live. A failure must leave the actors open so lifecycle teardown can retry.

fn close_sinks(&mut self) -> impl Future<Output = Result<(), String>> + Send

Gracefully close sinks on shutdown (abort open transactions, flush) so a restart re-initialises cleanly. Every sink must be attempted; the result aggregates failures.

fn set_barrier_injectors(&mut self, injectors: Vec<SourceBarrierControl>)

Register the local source barrier injectors.

Dyn Compatibility

This trait is not dyn compatible.

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

Implementors