surreal_sync_runtime/pipeline/apply/
opts.rs

1//! Apply options: batching, in-flight window, timeout, failure policy.
2
3use std::time::Duration;
4
5/// What to do when a batch fails transform or sink apply.
6///
7/// - [`Fail`](Self::Fail) (default): stop; do not `advance_watermark` the failed
8///   batch or any later position. Operator restarts resume from the last
9///   successful watermark advance.
10/// - [`Skip`](Self::Skip): log, do **not** write the failed batch, but still
11///   call [`SourceDriver::note_sunk_events`](crate::pipeline::SourceDriver::note_sunk_events)
12///   and advance past it. This can lose data by explicit operator choice.
13#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
14pub enum FailurePolicy {
15    /// Stop on failure; leave checkpoint unchanged for the failed batch.
16    #[default]
17    Fail,
18    /// Drop the failed batch (no sink write) and advance the watermark past it.
19    Skip,
20}
21
22/// Options for [`crate::pipeline::run_change_feed`], [`crate::pipeline::write_rows`], and
23/// [`crate::pipeline::ApplyContext`].
24///
25/// `max_in_flight` is the apply **window size** (default 1): concurrent
26/// transforms plus batches awaiting/in ordered sink. W=1 and W=16 share the
27/// same runtime — reads may overlap transforms and ordered writes; source
28/// watermark advance stays after sink success.
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub struct ApplyOpts {
31    /// Maximum batches in the apply window (transforming + awaiting/in sink).
32    /// Default: 1.
33    pub max_in_flight: usize,
34    /// Accumulate this many changes before starting a transform batch.
35    /// Default: 1000.
36    pub batch_size: usize,
37    /// Flush a partial batch after this idle wait.
38    /// Default: 500ms.
39    pub batch_max_wait: Duration,
40    /// Per-batch transform timeout.
41    /// Default: 60s.
42    pub timeout: Duration,
43    /// Transform / sink failure handling.
44    /// Default: [`FailurePolicy::Fail`].
45    pub failure_policy: FailurePolicy,
46}
47
48impl Default for ApplyOpts {
49    fn default() -> Self {
50        Self {
51            max_in_flight: 1,
52            batch_size: 1000,
53            batch_max_wait: Duration::from_millis(500),
54            timeout: Duration::from_secs(60),
55            failure_policy: FailurePolicy::Fail,
56        }
57    }
58}
59
60impl ApplyOpts {
61    /// Apply options for an identity (no-config) sync path.
62    ///
63    /// Pins `max_in_flight = 1` and `batch_size = 1` so omit-`--transforms-config`
64    /// keeps per-event CDC cadence (no multi-change buffering, no overlapping
65    /// transform window). Full-sync / ad-hoc helpers still go through the same
66    /// [`crate::pipeline::ApplyContext`] path; there is no separate oneshot bypass.
67    pub fn identity() -> Self {
68        Self::default().with_batch_size(1).with_max_in_flight(1)
69    }
70
71    /// Builder: set in-flight window size (clamped to at least 1).
72    pub fn with_max_in_flight(mut self, n: usize) -> Self {
73        self.max_in_flight = n.max(1);
74        self
75    }
76
77    /// Builder: set batch size (clamped to at least 1).
78    pub fn with_batch_size(mut self, n: usize) -> Self {
79        self.batch_size = n.max(1);
80        self
81    }
82
83    /// Builder: set max wait before flushing a partial batch.
84    pub fn with_batch_max_wait(mut self, d: Duration) -> Self {
85        self.batch_max_wait = d;
86        self
87    }
88
89    /// Builder: set per-batch transform timeout.
90    pub fn with_timeout(mut self, d: Duration) -> Self {
91        self.timeout = d;
92        self
93    }
94
95    /// Builder: set failure policy.
96    pub fn with_failure_policy(mut self, policy: FailurePolicy) -> Self {
97        self.failure_policy = policy;
98        self
99    }
100}
surreal_sync_runtime/pipeline/apply/opts.rs

surreal_sync_runtime/pipeline/apply/
opts.rs
