surreal_sync_runtime/pipeline/apply/
feed.rs

1//! Change feed trait: thin adapter over [`crate::pipeline::SourceDriver`] for tests and simple row feeds.
2//!
3//! Prefer [`crate::pipeline::SourceDriver`] + [`crate::pipeline::run_source_runtime`] for production ports.
4//! [`ChangeFeed`] / [`crate::pipeline::run_change_feed`] remain for tests and simple row-CDC sources
5//! that do not need schema refresh, ad-hoc snapshot, or other control hooks.
6
7use crate::pipeline::apply::event::{ApplyEvent, PositionedEvent};
8use anyhow::Result;
9use surreal_sync_core::Change;
10
11/// A CDC / incremental row event plus the source position to advance after sink success.
12///
13/// Prefer [`PositionedEvent`] when the feed may also emit relation changes.
14#[derive(Debug, Clone)]
15pub struct PositionedChange<P> {
16    /// Universal change to transform and apply.
17    pub change: Change,
18    /// Source position associated with this change (checkpoint candidate).
19    pub position: P,
20}
21
22impl<P> PositionedChange<P> {
23    /// Construct a positioned change.
24    pub fn new(change: Change, position: P) -> Self {
25        Self { change, position }
26    }
27
28    /// Convert to the unified [`PositionedEvent`] form.
29    pub fn into_event(self) -> PositionedEvent<P> {
30        PositionedEvent::new(ApplyEvent::Change(self.change), self.position)
31    }
32}
33
34impl<P> From<PositionedChange<P>> for PositionedEvent<P> {
35    fn from(pc: PositionedChange<P>) -> Self {
36        pc.into_event()
37    }
38}
39
40/// Source-facing incremental **row** feed: poll events, advance watermark after durable sink apply.
41///
42/// Sources do **not** batch for transforms — the shared apply loop owns batching,
43/// the in-flight window, ordered sink apply, and contiguous sink-safe watermark.
44///
45/// For sources that also emit relation edges or need DDL / cancel / checkpoint
46/// hooks, implement [`crate::pipeline::SourceDriver`] instead (or wrap this feed with
47/// [`crate::pipeline::ChangeFeedDriver`]).
48#[async_trait::async_trait]
49pub trait ChangeFeed: Send {
50    /// Checkpoint / resume position type.
51    type Position: Clone + Send + Sync + 'static;
52
53    /// Next source events (may be empty on idle).
54    async fn poll_changes(&mut self) -> Result<Vec<PositionedChange<Self::Position>>>;
55
56    /// Mark `position` sink-safe. May be in-memory only, broker offset commit, or
57    /// both — durable store writes belong in
58    /// [`SourceDriver::persist_checkpoint`](crate::pipeline::SourceDriver::persist_checkpoint)
59    /// unless policy is [`CheckpointPolicy::AdvanceOnly`](crate::pipeline::CheckpointPolicy::AdvanceOnly).
60    async fn advance_watermark(&mut self, position: Self::Position) -> Result<()>;
61
62    /// Whether the feed will produce no more changes (EOF).
63    ///
64    /// Default: `false` (endless sources). Scripted / finite feeds override so
65    /// [`crate::pipeline::run_change_feed`] can exit after draining in-flight work.
66    fn is_finished(&self) -> bool {
67        false
68    }
69}
surreal_sync_runtime/pipeline/apply/feed.rs

surreal_sync_runtime/pipeline/apply/
feed.rs
