surreal-sync-runtime 0.6.0

Shared runtime: apply pipeline, init, SurrealDB config, and transform loading for surreal-sync
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

//! Apply options: batching, in-flight window, timeout, failure policy.

use std::time::Duration;

/// What to do when a batch fails transform or sink apply.
///
/// - [`Fail`](Self::Fail) (default): stop; do not `advance_watermark` the failed
///   batch or any later position. Operator restarts resume from the last
///   successful watermark advance.
/// - [`Skip`](Self::Skip): log, do **not** write the failed batch, but still
///   call [`SourceDriver::note_sunk_events`](crate::pipeline::SourceDriver::note_sunk_events)
///   and advance past it. This can lose data by explicit operator choice.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum FailurePolicy {
    /// Stop on failure; leave checkpoint unchanged for the failed batch.
    #[default]
    Fail,
    /// Drop the failed batch (no sink write) and advance the watermark past it.
    Skip,
}

/// Options for [`crate::pipeline::run_change_feed`], [`crate::pipeline::write_rows`], and
/// [`crate::pipeline::ApplyContext`].
///
/// `max_in_flight` is the apply **window size** (default 1): concurrent
/// transforms plus batches awaiting/in ordered sink. W=1 and W=16 share the
/// same runtime — reads may overlap transforms and ordered writes; source
/// watermark advance stays after sink success.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApplyOpts {
    /// Maximum batches in the apply window (transforming + awaiting/in sink).
    /// Default: 1.
    pub max_in_flight: usize,
    /// Accumulate this many changes before starting a transform batch.
    /// Default: 1000.
    pub batch_size: usize,
    /// Flush a partial batch after this idle wait.
    /// Default: 500ms.
    pub batch_max_wait: Duration,
    /// Per-batch transform timeout.
    /// Default: 60s.
    pub timeout: Duration,
    /// Transform / sink failure handling.
    /// Default: [`FailurePolicy::Fail`].
    pub failure_policy: FailurePolicy,
}

impl Default for ApplyOpts {
    fn default() -> Self {
        Self {
            max_in_flight: 1,
            batch_size: 1000,
            batch_max_wait: Duration::from_millis(500),
            timeout: Duration::from_secs(60),
            failure_policy: FailurePolicy::Fail,
        }
    }
}

impl ApplyOpts {
    /// Apply options for an identity (no-config) sync path.
    ///
    /// Pins `max_in_flight = 1` and `batch_size = 1` so omit-`--transforms-config`
    /// keeps per-event CDC cadence (no multi-change buffering, no overlapping
    /// transform window). Full-sync / ad-hoc helpers still go through the same
    /// [`crate::pipeline::ApplyContext`] path; there is no separate oneshot bypass.
    pub fn identity() -> Self {
        Self::default().with_batch_size(1).with_max_in_flight(1)
    }

    /// Builder: set in-flight window size (clamped to at least 1).
    pub fn with_max_in_flight(mut self, n: usize) -> Self {
        self.max_in_flight = n.max(1);
        self
    }

    /// Builder: set batch size (clamped to at least 1).
    pub fn with_batch_size(mut self, n: usize) -> Self {
        self.batch_size = n.max(1);
        self
    }

    /// Builder: set max wait before flushing a partial batch.
    pub fn with_batch_max_wait(mut self, d: Duration) -> Self {
        self.batch_max_wait = d;
        self
    }

    /// Builder: set per-batch transform timeout.
    pub fn with_timeout(mut self, d: Duration) -> Self {
        self.timeout = d;
        self
    }

    /// Builder: set failure policy.
    pub fn with_failure_policy(mut self, policy: FailurePolicy) -> Self {
        self.failure_policy = policy;
        self
    }
}