etl_core/source/
mod.rs

1//! Source abstraction: a control plane ([`Source`]) and a data plane
2//! ([`SourceLane`]).
3//!
4//! A source is poll-based — no `futures::Stream`. The control plane surfaces
5//! lane assignment and revocation as events and owns commits and
6//! pause/resume; each lane is a pollable unit pinned to one pipeline thread
7//! (for Kafka: a partition queue), yielding payloads that **borrow** the
8//! source's buffers for the duration of one `push_batch` call. See
9//! `docs/DESIGN.md` (§ Source abstraction, § Frozen v1 contracts).
10
11mod barrier;
12
13pub use barrier::DrainBarrier;
14
15use crate::checkpoint::AckIssuer;
16use crate::checkpoint::AckRef;
17use crate::error::SourceError;
18use crate::record::{PartitionId, RawPayload};
19use std::time::Duration;
20
21/// Identifier of one source lane within an assignment (dense,
22/// source-assigned; stable until the lane is revoked).
23#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
24pub struct LaneId(pub u32);
25
26/// One poll's worth of borrowed payloads. Streaming — payloads are handed
27/// out one at a time, and every payload shares the batch lifetime `'buf`.
28/// The batch carries exactly one [`AckRef`], issued by the lane through its
29/// [`AckIssuer`]; records derived from these payloads clone it.
30pub trait PayloadBatch<'buf> {
31    /// The next payload, or `None` when the batch is exhausted.
32    fn next_payload(&mut self) -> Option<RawPayload<'buf>>;
33
34    /// The acknowledgement handle covering every payload in this batch.
35    fn ack(&self) -> &AckRef;
36}
37
38/// Data-plane pollable unit of a source, owned by one pipeline thread.
39///
40/// Contract: payloads yielded by [`SourceLane::poll`] are valid only until
41/// the returned batch is dropped, which happens before the next `poll` call
42/// on the same lane — records must be consumed or encoded within that
43/// window (the operator chain guarantees this by construction).
44pub trait SourceLane: Send {
45    /// The borrowed batch type (a GAT so payloads can borrow lane buffers).
46    type Batch<'a>: PayloadBatch<'a>
47    where
48        Self: 'a;
49
50    /// This lane's identity within the current assignment.
51    fn id(&self) -> LaneId;
52
53    /// The source partition this lane reads. Used for checkpoint issuing
54    /// and shard routing fallback.
55    fn partition(&self) -> PartitionId;
56
57    /// Poll up to `max_records` payloads, waiting at most `timeout`.
58    /// `Ok(None)` means nothing arrived — the driver treats it as idle.
59    /// Implementations must not busy-spin when idle: block up to `timeout`.
60    fn poll(
61        &mut self,
62        max_records: usize,
63        timeout: Duration,
64    ) -> Result<Option<Self::Batch<'_>>, SourceError>;
65}
66
67/// Control-plane event returned by [`Source::poll_events`].
68#[derive(Debug)]
69#[non_exhaustive]
70pub enum SourceEvent<L> {
71    /// New lanes were assigned; the runtime distributes them across
72    /// pipeline threads. The source bumps its assignment epoch first.
73    LanesAssigned(Vec<L>),
74    /// Lanes are being revoked. The runtime trips the [`DrainBarrier`] for
75    /// the owning threads, which stop the lanes, flush in-flight records,
76    /// and arrive; the source completes the revocation (final synchronous
77    /// commit included) only after [`DrainBarrier::wait`] returns.
78    LanesRevoked {
79        /// Which lanes to stop.
80        lanes: Vec<LaneId>,
81        /// Barrier the owning pipeline threads arrive at once drained.
82        barrier: DrainBarrier,
83    },
84    /// Nothing happened within the timeout.
85    Idle,
86}
87
88/// Everything a source receives at [`Source::open`].
89#[derive(Debug)]
90#[non_exhaustive]
91pub struct SourceCtx {
92    /// Issuer for batch acknowledgement handles. Sources clone it into
93    /// every lane they construct; each lane issues one [`AckRef`] per poll
94    /// batch (`issue(partition, last_offset)`).
95    pub issuer: AckIssuer,
96}
97
98impl SourceCtx {
99    /// Context wrapping the checkpointer's issuer.
100    #[must_use]
101    pub fn new(issuer: AckIssuer) -> Self {
102        SourceCtx { issuer }
103    }
104}
105
106/// Control plane of a source. Driven by the runtime's controller from a
107/// single thread; lanes run on pipeline threads.
108pub trait Source: Send {
109    /// The lane type this source produces.
110    type Lane: SourceLane;
111
112    /// Connect and prepare. Called once before any other method.
113    fn open(&mut self, ctx: SourceCtx) -> Result<(), SourceError>;
114
115    /// Service control-plane work (rebalance callbacks, statistics) and
116    /// return the next event, waiting at most `timeout`. Must be called
117    /// regularly regardless of backpressure state.
118    fn poll_events(&mut self, timeout: Duration) -> Result<SourceEvent<Self::Lane>, SourceError>;
119
120    /// Store per-partition committable positions (each is the offset one
121    /// past the last acknowledged record). Positions are durable per the
122    /// source's own policy (e.g. interval auto-commit of stored offsets).
123    fn commit(&mut self, watermarks: &[(PartitionId, i64)]) -> Result<(), SourceError>;
124
125    /// Synchronously flush stored positions (shutdown, revocation).
126    fn flush_commits(&mut self) -> Result<(), SourceError> {
127        Ok(())
128    }
129
130    /// Stop fetching for `lanes` (backpressure). Optional capability:
131    /// sources that cannot pause rely on bounded-queue pushback alone.
132    fn pause(&mut self, lanes: &[LaneId]) -> Result<(), SourceError> {
133        let _ = lanes;
134        Ok(())
135    }
136
137    /// Resume fetching for `lanes`.
138    fn resume(&mut self, lanes: &[LaneId]) -> Result<(), SourceError> {
139        let _ = lanes;
140        Ok(())
141    }
142}
etl_core/source/mod.rs

etl_core/source/
mod.rs
