faucet_source_dynamodb/
config.rs

1//! Configuration for the DynamoDB source. No I/O here.
2
3use faucet_common_dynamodb::{DynamoDbCredentials, RetryPolicy, ValueFormat};
4use faucet_core::FaucetError;
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use std::collections::BTreeMap;
9
10/// What the source reads.
11#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
12#[serde(rename_all = "snake_case")]
13pub enum ReadMode {
14    /// Full-table `Scan`, optionally parallel (`segments`).
15    #[default]
16    Scan,
17    /// `Query` by `key_condition_expression` (optionally on `index_name`).
18    Query,
19    /// DynamoDB Streams change data capture.
20    Streams,
21}
22
23impl ReadMode {
24    /// Lowercase name, as written in configs.
25    pub fn as_str(self) -> &'static str {
26        match self {
27            Self::Scan => "scan",
28            Self::Query => "query",
29            Self::Streams => "streams",
30        }
31    }
32}
33
34/// Where a stream shard starts when no bookmark exists for it.
35#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
36#[serde(rename_all = "snake_case")]
37pub enum StreamStart {
38    /// Oldest record still retained (up to 24 hours).
39    #[default]
40    TrimHorizon,
41    /// Only changes made after the consumer starts.
42    Latest,
43}
44
45/// What to do when the bookmark points before the Streams trim horizon (a
46/// resume after more than 24 hours, a re-enabled stream, a shard that expired
47/// before it was read).
48#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
49#[serde(rename_all = "snake_case")]
50pub enum OnGap {
51    /// Fail the run, naming the shards whose changes are no longer available.
52    #[default]
53    Fail,
54    /// Re-read the whole table (emitted as `op: "r"` envelopes), then resume
55    /// the stream from the trim horizon. The run is a complete read: afterwards
56    /// the destination rows of items it did not see (deleted while their stream
57    /// records were trimmed) are deleted, so the sink must support cleanup, with
58    /// `write_mode: upsert` and a `key`. `resnapshot_deletes: false` skips that
59    /// delete (and the requirement).
60    Resnapshot,
61}
62
63/// Configuration for [`DynamoDbSource`](crate::DynamoDbSource).
64#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
65#[serde(deny_unknown_fields)]
66pub struct DynamoDbSourceConfig {
67    /// Table name.
68    pub table_name: String,
69    /// AWS region. `None` uses the SDK default chain (env, profile, IMDS).
70    #[serde(default, skip_serializing_if = "Option::is_none")]
71    pub region: Option<String>,
72    /// Custom endpoint URL (DynamoDB Local, LocalStack, VPC endpoints).
73    #[serde(default, skip_serializing_if = "Option::is_none")]
74    #[schemars(extend("x-faucet-resource" = "url"))]
75    pub endpoint_url: Option<String>,
76    /// AWS credentials. Defaults to the SDK default provider chain.
77    #[serde(default)]
78    pub credentials: DynamoDbCredentials,
79
80    /// `scan` (default), `query`, or `streams`.
81    #[serde(default)]
82    pub mode: ReadMode,
83
84    /// Parallel scan segments (`TotalSegments`), 1–1000000. Scan only.
85    #[serde(default = "default_segments")]
86    pub segments: u32,
87    /// Segments read concurrently. Default 8.
88    #[serde(default = "default_max_concurrency")]
89    pub max_concurrency: usize,
90    /// Secondary index to scan or query.
91    #[serde(default, skip_serializing_if = "Option::is_none")]
92    pub index_name: Option<String>,
93    /// `ProjectionExpression` — the attributes to return.
94    #[serde(default, skip_serializing_if = "Option::is_none")]
95    pub projection: Option<String>,
96    /// `FilterExpression` applied server-side after the read.
97    #[serde(default, skip_serializing_if = "Option::is_none")]
98    #[schemars(extend("x-faucet-code" = true))]
99    pub filter_expression: Option<String>,
100    /// `KeyConditionExpression`. Required for `query`, rejected otherwise.
101    #[serde(default, skip_serializing_if = "Option::is_none")]
102    #[schemars(extend("x-faucet-code" = true))]
103    pub key_condition_expression: Option<String>,
104    /// `ExpressionAttributeNames` (`#name` → attribute).
105    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
106    pub expression_attribute_names: BTreeMap<String, String>,
107    /// `ExpressionAttributeValues` (`:value` → plain JSON value).
108    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
109    pub expression_attribute_values: BTreeMap<String, Value>,
110    /// Strongly consistent reads (table and local secondary indexes only).
111    #[serde(default)]
112    pub consistent_read: bool,
113    /// Query order: ascending sort key when `true` (default).
114    #[serde(default = "default_true")]
115    pub scan_index_forward: bool,
116    /// `Limit` per Scan/Query request (items evaluated per call).
117    #[serde(default, skip_serializing_if = "Option::is_none")]
118    pub page_limit: Option<u32>,
119
120    /// Stream ARN. Defaults to the table's `LatestStreamArn`. Streams only.
121    #[serde(default, skip_serializing_if = "Option::is_none")]
122    pub stream_arn: Option<String>,
123    /// Start for shards with no bookmark and no parent that was read.
124    #[serde(default)]
125    pub start_position: StreamStart,
126    /// Reaction to a bookmark older than the trim horizon.
127    #[serde(default)]
128    pub on_gap: OnGap,
129    /// `on_gap: resnapshot`: after the re-read, delete the destination rows of
130    /// items it did not see (default `true`; the sink must support cleanup).
131    /// `false` keeps the resnapshot for any sink but leaves the destination
132    /// rows of items deleted while their stream records were trimmed.
133    #[serde(default = "default_true")]
134    pub resnapshot_deletes: bool,
135    /// `on_gap: resnapshot`: the most distinct keys the re-read tracks for its
136    /// delete (default 1,000,000; memory grows with it). A table with more
137    /// items fails that run once with nothing deleted; its position is saved,
138    /// so the next run resumes streaming.
139    #[serde(default = "default_resnapshot_max_keys")]
140    pub resnapshot_max_keys: usize,
141    /// Base wait between `GetRecords` calls per shard, in ms (floored at 200).
142    #[serde(default = "default_poll_interval_ms")]
143    pub poll_interval_ms: u64,
144    /// `GetRecords` `Limit` per request (1–1000). Default 1000.
145    #[serde(default = "default_records_per_request")]
146    pub records_per_request: u32,
147    /// Shards read concurrently. Default 4.
148    #[serde(default = "default_shard_concurrency")]
149    pub shard_concurrency: usize,
150    /// Streams: stop after this many seconds without a new record. At least
151    /// one of `idle_termination_secs` / `max_messages` is required in streams
152    /// mode.
153    #[serde(default, skip_serializing_if = "Option::is_none")]
154    pub idle_termination_secs: Option<u64>,
155    /// Streams: stop after this many change records.
156    #[serde(default, skip_serializing_if = "Option::is_none")]
157    pub max_messages: Option<usize>,
158
159    /// Throttle / transient-failure retry policy.
160    #[serde(default)]
161    pub retry: RetryPolicy,
162
163    /// Records per emitted page. `0` is the "no batching" sentinel. Default
164    /// 1000.
165    #[serde(default = "default_batch_size")]
166    pub batch_size: usize,
167
168    /// How attribute values appear in records: `plain` JSON (default; sets
169    /// become arrays, binaries base64 strings, imprecise numbers decimal
170    /// strings) or `typed` DynamoDB attribute JSON (`{"S": "a"}`,
171    /// `{"SS": ["x"]}`, `{"B": "<base64>"}`) that a `value_format: typed`
172    /// DynamoDB sink writes back losslessly. Applies to scan/query items and
173    /// to Streams `key` / `before` / `after` images.
174    #[serde(default)]
175    pub value_format: ValueFormat,
176}
177
178fn default_segments() -> u32 {
179    1
180}
181fn default_max_concurrency() -> usize {
182    8
183}
184fn default_true() -> bool {
185    true
186}
187fn default_poll_interval_ms() -> u64 {
188    1000
189}
190fn default_records_per_request() -> u32 {
191    1000
192}
193fn default_shard_concurrency() -> usize {
194    4
195}
196fn default_batch_size() -> usize {
197    faucet_core::DEFAULT_BATCH_SIZE
198}
199fn default_resnapshot_max_keys() -> usize {
200    faucet_core::DEFAULT_COMPLETE_READ_MAX_KEYS
201}
202
203/// DynamoDB's `TotalSegments` ceiling.
204pub const MAX_SEGMENTS: u32 = 1_000_000;
205
206impl DynamoDbSourceConfig {
207    /// Scan config with defaults for everything but the table name.
208    pub fn new(table_name: impl Into<String>) -> Self {
209        Self {
210            table_name: table_name.into(),
211            region: None,
212            endpoint_url: None,
213            credentials: DynamoDbCredentials::default(),
214            mode: ReadMode::default(),
215            segments: default_segments(),
216            max_concurrency: default_max_concurrency(),
217            index_name: None,
218            projection: None,
219            filter_expression: None,
220            key_condition_expression: None,
221            expression_attribute_names: BTreeMap::new(),
222            expression_attribute_values: BTreeMap::new(),
223            consistent_read: false,
224            scan_index_forward: true,
225            page_limit: None,
226            stream_arn: None,
227            start_position: StreamStart::default(),
228            on_gap: OnGap::default(),
229            resnapshot_deletes: true,
230            resnapshot_max_keys: default_resnapshot_max_keys(),
231            poll_interval_ms: default_poll_interval_ms(),
232            records_per_request: default_records_per_request(),
233            shard_concurrency: default_shard_concurrency(),
234            idle_termination_secs: None,
235            max_messages: None,
236            retry: RetryPolicy::default(),
237            batch_size: default_batch_size(),
238            value_format: ValueFormat::default(),
239        }
240    }
241
242    /// Fail-fast validation, called from `DynamoDbSource::new`.
243    pub fn validate(&self) -> Result<(), FaucetError> {
244        let err = |m: String| Err(FaucetError::Config(format!("dynamodb source: {m}")));
245        if self.table_name.trim().is_empty() {
246            return err("table_name must not be empty".into());
247        }
248        faucet_core::validate_batch_size(self.batch_size)?;
249        if self.segments == 0 || self.segments > MAX_SEGMENTS {
250            return err(format!(
251                "segments must be 1..={MAX_SEGMENTS} (got {})",
252                self.segments
253            ));
254        }
255        if self.segments > 1 && self.mode != ReadMode::Scan {
256            return err("segments > 1 is only valid in scan mode".into());
257        }
258        if self.max_concurrency == 0 {
259            return err("max_concurrency must be at least 1".into());
260        }
261        if self.resnapshot_max_keys == 0 {
262            return err("resnapshot_max_keys must be at least 1".into());
263        }
264        if self.page_limit == Some(0) {
265            return err("page_limit must be at least 1".into());
266        }
267        match (self.mode, &self.key_condition_expression) {
268            (ReadMode::Query, None) => {
269                return err("mode: query requires key_condition_expression".into());
270            }
271            (ReadMode::Scan | ReadMode::Streams, Some(_)) => {
272                return err(format!(
273                    "key_condition_expression is only valid in query mode (mode: {})",
274                    self.mode.as_str()
275                ));
276            }
277            _ => {}
278        }
279        if let Some(bad) = self
280            .expression_attribute_names
281            .keys()
282            .find(|k| !k.starts_with('#'))
283        {
284            return err(format!(
285                "expression_attribute_names key '{bad}' must start with '#'"
286            ));
287        }
288        if let Some(bad) = self
289            .expression_attribute_values
290            .keys()
291            .find(|k| !k.starts_with(':'))
292        {
293            return err(format!(
294                "expression_attribute_values key '{bad}' must start with ':'"
295            ));
296        }
297        if self.mode == ReadMode::Streams {
298            if self.records_per_request == 0 || self.records_per_request > 1000 {
299                return err(format!(
300                    "records_per_request must be 1..=1000 (got {})",
301                    self.records_per_request
302                ));
303            }
304            if self.shard_concurrency == 0 {
305                return err("shard_concurrency must be at least 1".into());
306            }
307            if self.idle_termination_secs.is_none() && self.max_messages.is_none() {
308                return err(
309                    "mode: streams needs at least one of idle_termination_secs / max_messages \
310                     so a run can terminate"
311                        .into(),
312                );
313            }
314            if self.projection.is_some()
315                || self.filter_expression.is_some()
316                || self.index_name.is_some()
317            {
318                return err(
319                    "projection / filter_expression / index_name do not apply to mode: streams"
320                        .into(),
321                );
322            }
323        } else if self.stream_arn.is_some() {
324            return err("stream_arn is only valid in streams mode".into());
325        }
326        if self.idle_termination_secs == Some(0) {
327            return err("idle_termination_secs must be at least 1".into());
328        }
329        if self.max_messages == Some(0) {
330            return err("max_messages must be at least 1".into());
331        }
332        Ok(())
333    }
334
335    /// Effective per-shard poll interval (floored at 200 ms).
336    pub fn poll_interval(&self) -> std::time::Duration {
337        std::time::Duration::from_millis(self.poll_interval_ms.max(200))
338    }
339}
340
341#[cfg(test)]
342mod tests {
343    use super::*;
344    use serde_json::json;
345
346    fn streams() -> DynamoDbSourceConfig {
347        let mut c = DynamoDbSourceConfig::new("t");
348        c.mode = ReadMode::Streams;
349        c.idle_termination_secs = Some(5);
350        c
351    }
352
353    #[test]
354    fn defaults_are_sensible() {
355        let c = DynamoDbSourceConfig::new("t");
356        assert_eq!(c.mode, ReadMode::Scan);
357        assert_eq!(c.segments, 1);
358        assert_eq!(c.start_position, StreamStart::TrimHorizon);
359        assert_eq!(c.on_gap, OnGap::Fail);
360        assert!(c.scan_index_forward);
361        c.validate().unwrap();
362        assert_eq!(ReadMode::Query.as_str(), "query");
363    }
364
365    #[test]
366    fn validation_bounds() {
367        let bad = |f: &dyn Fn(&mut DynamoDbSourceConfig), needle: &str| {
368            let mut c = DynamoDbSourceConfig::new("t");
369            f(&mut c);
370            let e = c.validate().unwrap_err().to_string();
371            assert!(e.contains(needle), "{e} !~ {needle}");
372        };
373        bad(&|c| c.table_name = " ".into(), "table_name");
374        bad(&|c| c.batch_size = faucet_core::MAX_BATCH_SIZE + 1, "batch");
375        bad(&|c| c.segments = 0, "segments");
376        bad(&|c| c.segments = MAX_SEGMENTS + 1, "segments");
377        bad(
378            &|c| {
379                c.segments = 4;
380                c.mode = ReadMode::Query;
381                c.key_condition_expression = Some("pk = :p".into());
382            },
383            "only valid in scan",
384        );
385        bad(&|c| c.max_concurrency = 0, "max_concurrency");
386        bad(&|c| c.page_limit = Some(0), "page_limit");
387        bad(&|c| c.mode = ReadMode::Query, "requires key_condition");
388        bad(
389            &|c| c.key_condition_expression = Some("x".into()),
390            "only valid in query",
391        );
392        bad(
393            &|c| {
394                c.expression_attribute_names.insert("n".into(), "x".into());
395            },
396            "must start with '#'",
397        );
398        bad(
399            &|c| {
400                c.expression_attribute_values.insert("v".into(), json!(1));
401            },
402            "must start with ':'",
403        );
404        bad(&|c| c.stream_arn = Some("arn".into()), "stream_arn");
405        bad(&|c| c.idle_termination_secs = Some(0), "idle_termination");
406        bad(&|c| c.max_messages = Some(0), "max_messages");
407    }
408
409    #[test]
410    fn streams_validation() {
411        streams().validate().unwrap();
412        let mut c = streams();
413        c.idle_termination_secs = None;
414        assert!(c.validate().unwrap_err().to_string().contains("terminate"));
415        c.max_messages = Some(3);
416        c.validate().unwrap();
417        let mut c = streams();
418        c.records_per_request = 0;
419        assert!(c.validate().is_err());
420        c.records_per_request = 1001;
421        assert!(c.validate().is_err());
422        let mut c = streams();
423        c.shard_concurrency = 0;
424        assert!(c.validate().is_err());
425        let mut c = streams();
426        c.projection = Some("a".into());
427        assert!(
428            c.validate()
429                .unwrap_err()
430                .to_string()
431                .contains("do not apply")
432        );
433    }
434
435    #[test]
436    fn poll_interval_floor() {
437        let mut c = streams();
438        c.poll_interval_ms = 10;
439        assert_eq!(c.poll_interval().as_millis(), 200);
440    }
441
442    #[test]
443    fn full_config_parses_from_yaml() {
444        let yaml = r##"
445table_name: orders
446region: us-east-1
447endpoint_url: http://127.0.0.1:8000
448credentials: { type: access_key, config: { access_key_id: a, secret_access_key: b } }
449mode: query
450index_name: by_customer
451projection: "id, #s"
452filter_expression: "#s = :open"
453key_condition_expression: "customer = :c"
454expression_attribute_names: { "#s": status }
455expression_attribute_values: { ":c": "cust-1", ":open": "OPEN" }
456consistent_read: false
457scan_index_forward: false
458page_limit: 100
459retry: { max_retries: 3 }
460batch_size: 500
461"##;
462        let c: DynamoDbSourceConfig = yaml_serde::from_str(yaml).unwrap();
463        c.validate().unwrap();
464        assert_eq!(c.mode, ReadMode::Query);
465        assert!(!c.scan_index_forward);
466        assert_eq!(c.retry.max_retries, 3);
467
468        let yaml = "table_name: t\nmode: streams\non_gap: resnapshot\nstart_position: latest\nmax_messages: 10\n";
469        let c: DynamoDbSourceConfig = yaml_serde::from_str(yaml).unwrap();
470        c.validate().unwrap();
471        assert_eq!(c.on_gap, OnGap::Resnapshot);
472        assert_eq!(c.start_position, StreamStart::Latest);
473        assert!(yaml_serde::from_str::<DynamoDbSourceConfig>("table_name: t\nbogus: 1\n").is_err());
474    }
475}
faucet_source_dynamodb/config.rs

faucet_source_dynamodb/
config.rs
