Function run_source_runtime_with 

pub async fn run_source_runtime_with<D, S, T>(
    driver: &mut D,
    sink: &S,
    transformer: Arc<T>,
    apply_opts: &ApplyOpts,
    runtime_opts: &SourceRuntimeOpts,
) -> Result<RuntimeExit>
where
    D: SourceDriver,
    S: SurrealSink,
    T: BatchTransformer + 'static,

Like run_source_runtime but accepts any BatchTransformer behind Arc.

Pipeline overlap model (identity and non-identity share this path):

• Reads continue while prior batches transform or sink.
• Transforms overlap up to max_in_flight window occupancy (transforming + awaiting/in-flight ordered sink).
• Writes stay ordered; the fill loop does not await sink before the next poll/transform start. Source advance_watermark runs only after that batch’s sink succeeds.