spate_core/config/
feed.rs

1//! The `feed:` section: the work feed a feed-reading source reads from, and
2//! the keys every backend shares.
3
4use super::{ComponentConfig, redact};
5use serde::Deserialize;
6use std::num::{NonZeroU32, NonZeroUsize};
7use std::time::Duration;
8
9/// The pipeline's `feed:` section, handed to the source through
10/// [`Source::configure_feed`](crate::source::Source::configure_feed).
11///
12/// One key names the backend, with the same single-key shape as `source:`;
13/// its body holds the backend's own keys and is decoded by the backend.
14/// `units_held` applies whatever the backend; `attempt_cap` applies to the
15/// in-process and Kafka feeds, and a backend that cannot honor it refuses
16/// it; a source that re-enqueues units caps them at it. `location` names a job location for a source that reads one, and
17/// `settling_window` is the settling window of a source that ends a job
18/// with a location; the in-process feed refuses both.
19///
20/// ```yaml
21/// feed:
22///   memory: {}
23///   units_held: 32
24/// ```
25///
26/// Construct one in code with [`FeedSection::new`] and set the optional
27/// fields. The struct is `#[non_exhaustive]` so new keys can be added
28/// without breaking callers.
29#[derive(Clone, PartialEq)]
30#[non_exhaustive]
31pub struct FeedSection {
32    backend: ComponentConfig,
33    /// Units the source holds and reads at once. `None` leaves the source's
34    /// own default.
35    pub units_held: Option<NonZeroUsize>,
36    /// The delivery on which a failed unit moves to its feed's dead-letter
37    /// destination, on the in-process and Kafka feeds. The Kafka feed also
38    /// moves a unit whose recorded deliveries have reached it, without
39    /// delivering it again. `None` leaves the backend's default. A backend
40    /// that cannot honor the key refuses it. A source that re-enqueues units
41    /// caps their re-enqueues at it.
42    pub attempt_cap: Option<NonZeroU32>,
43    /// A job location, an object-store prefix, for a source that reads
44    /// one; that source defines what the prefix holds. `None` when the
45    /// section names none.
46    /// The in-process feed refuses one.
47    pub location: Option<String>,
48    /// The settling window of a job with a `location`, for a source that ends
49    /// one through an [`EndCheck`](crate::feed::EndCheck): how long the feed
50    /// must read empty before the job ends. The source's page says what else
51    /// it bounds. `None` leaves
52    /// [`DEFAULT_SETTLING_WINDOW`](crate::feed::DEFAULT_SETTLING_WINDOW). The
53    /// in-process feed refuses one.
54    pub settling_window: Option<Duration>,
55}
56
57// Hand-written: a location URL can carry credentials in its userinfo or
58// query. The destructure lists every field so a new one cannot reach `Debug`
59// unredacted.
60impl std::fmt::Debug for FeedSection {
61    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
62        let FeedSection {
63            backend,
64            units_held,
65            attempt_cap,
66            location,
67            settling_window,
68        } = self;
69        f.debug_struct("FeedSection")
70            .field("backend", backend)
71            .field("units_held", units_held)
72            .field("attempt_cap", attempt_cap)
73            .field("location", &location.as_deref().map(redact::url))
74            .field("settling_window", settling_window)
75            .finish()
76    }
77}
78
79impl FeedSection {
80    /// A section naming `backend`, with every other key unset. Errors from
81    /// the backend's body carry paths such as `feed.memory.<key>`.
82    ///
83    /// ```
84    /// use spate_core::config::{ComponentConfig, FeedSection, YamlValue};
85    /// use std::num::NonZeroUsize;
86    ///
87    /// let mut feed = FeedSection::new(ComponentConfig::new(
88    ///     "memory",
89    ///     YamlValue::Mapping(Default::default()),
90    /// ));
91    /// feed.units_held = NonZeroUsize::new(32);
92    ///
93    /// assert_eq!(feed.backend().type_tag(), "memory");
94    /// ```
95    #[must_use]
96    pub fn new(mut backend: ComponentConfig) -> FeedSection {
97        backend.set_section("feed");
98        FeedSection {
99            backend,
100            units_held: None,
101            attempt_cap: None,
102            location: None,
103            settling_window: None,
104        }
105    }
106
107    /// The backend component (`<backend>: { ... }`). Its
108    /// [`type_tag`](ComponentConfig::type_tag) is the key a feed's
109    /// [`backend`](crate::feed::Feed::backend) must return.
110    #[must_use]
111    pub fn backend(&self) -> &ComponentConfig {
112        &self.backend
113    }
114
115    /// The section's `settling_window`, or
116    /// [`DEFAULT_SETTLING_WINDOW`](crate::feed::DEFAULT_SETTLING_WINDOW) when
117    /// it sets none.
118    #[must_use]
119    pub fn settling_window_or_default(&self) -> Duration {
120        self.settling_window
121            .unwrap_or(crate::feed::DEFAULT_SETTLING_WINDOW)
122    }
123}
124
125impl<'de> Deserialize<'de> for FeedSection {
126    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
127    where
128        D: serde::Deserializer<'de>,
129    {
130        use serde::de::Error as _;
131        let mut keys = serde_yaml::Mapping::deserialize(deserializer)?;
132        let units_held = keys
133            .remove("units_held")
134            .map(|v| {
135                Option::<NonZeroUsize>::deserialize(v)
136                    .map_err(|e| D::Error::custom(format!("units_held: {e}")))
137            })
138            .transpose()?
139            .flatten();
140        let attempt_cap = keys
141            .remove("attempt_cap")
142            .map(|v| {
143                Option::<NonZeroU32>::deserialize(v)
144                    .map_err(|e| D::Error::custom(format!("attempt_cap: {e}")))
145            })
146            .transpose()?
147            .flatten();
148        let location = keys
149            .remove("location")
150            .map(|v| {
151                Option::<String>::deserialize(v)
152                    .map_err(|e| D::Error::custom(format!("location: {e}")))
153            })
154            .transpose()?
155            .flatten();
156        let settling_window = keys
157            .remove("settling_window")
158            .map(|v| {
159                humantime_serde::Serde::<Option<Duration>>::deserialize(v)
160                    .map(humantime_serde::Serde::into_inner)
161                    .map_err(|e| D::Error::custom(format!("settling_window: {e}")))
162            })
163            .transpose()?
164            .flatten();
165        // Every key left names a backend, so a misspelled common key shows up
166        // here as a second backend.
167        if keys.len() != 1 {
168            let found: Vec<String> = keys
169                .keys()
170                .map(|k| match k.as_str() {
171                    Some(name) => format!("`{name}`"),
172                    None => "a non-string key".to_owned(),
173                })
174                .collect();
175            return Err(D::Error::custom(if found.is_empty() {
176                "`feed:` names no backend; add one, e.g. `memory: {}`".to_owned()
177            } else {
178                format!(
179                    "`feed:` must name exactly one backend beside `units_held`, \
180                     `attempt_cap`, `location` and `settling_window`; found {}",
181                    found.join(", ")
182                )
183            }));
184        }
185        let mut backend = ComponentConfig::deserialize(serde_yaml::Value::Mapping(keys))
186            .map_err(D::Error::custom)?;
187        // The backend has no chain terminal, so a `chunk:` peeled from its body
188        // would otherwise vanish without an error.
189        if backend.has_chunk() {
190            return Err(D::Error::custom(format!(
191                "{}: `chunk:` is only valid on a sink section",
192                backend.type_tag()
193            )));
194        }
195        backend.set_section("feed");
196        Ok(FeedSection {
197            backend,
198            units_held,
199            attempt_cap,
200            location,
201            settling_window,
202        })
203    }
204}
205
206#[cfg(test)]
207mod tests {
208    use super::*;
209
210    fn parse(yaml: &str) -> Result<FeedSection, serde_yaml::Error> {
211        serde_yaml::from_str(yaml)
212    }
213
214    #[derive(Debug, Deserialize)]
215    #[serde(deny_unknown_fields)]
216    struct Body {
217        x: u32,
218    }
219
220    /// Pins that the backend is the one key left beside the common keys, and
221    /// that every common key is read.
222    #[test]
223    fn one_backend_sits_beside_the_common_keys() {
224        let s = parse(
225            "units_held: 12\nattempt_cap: 3\nlocation: s3://ops/jobs/a/\n\
226             settling_window: 2m 30s\nmemory: { x: 7 }\n",
227        )
228        .unwrap();
229        assert_eq!(s.backend().type_tag(), "memory");
230        assert_eq!(s.units_held, NonZeroUsize::new(12));
231        assert_eq!(s.attempt_cap, NonZeroU32::new(3));
232        assert_eq!(s.location.as_deref(), Some("s3://ops/jobs/a/"));
233        assert_eq!(s.settling_window, Some(Duration::from_secs(150)));
234        let body: Body = s.backend().deserialize_into().unwrap();
235        assert_eq!(body.x, 7);
236
237        let bare = parse("memory: {}\nlocation: null\nsettling_window: null\n").unwrap();
238        assert_eq!(bare.units_held, None);
239        assert_eq!(bare.attempt_cap, None);
240        assert_eq!(bare.location, None);
241        assert_eq!(bare.settling_window, None);
242
243        let built = FeedSection::new(ComponentConfig::new(
244            "memory",
245            serde_yaml::Value::Mapping(Default::default()),
246        ));
247        assert_eq!(built.location, None);
248        assert_eq!(built.settling_window, None);
249    }
250
251    /// Pins the error for a section with no backend key.
252    #[test]
253    fn a_section_with_no_backend_is_refused() {
254        let err = parse("units_held: 4\n").unwrap_err().to_string();
255        assert!(err.contains("names no backend"), "{err}");
256    }
257
258    /// Pins that two backend keys are refused, naming both.
259    #[test]
260    fn two_backends_are_refused_naming_both() {
261        let err = parse("memory: {}\nother: {}\n").unwrap_err().to_string();
262        assert!(err.contains("exactly one backend"), "{err}");
263        assert!(err.contains("`memory`") && err.contains("`other`"), "{err}");
264    }
265
266    /// Pins that a misspelled common key reads as a second backend, so it is
267    /// refused by name and never ignored.
268    #[test]
269    fn a_misspelled_common_key_is_refused_by_name() {
270        let err = parse("memory: {}\nunit_held: 4\n").unwrap_err().to_string();
271        assert!(err.contains("`unit_held`"), "{err}");
272
273        let err = parse("memory: {}\nlocaton: s3://ops/a/\n")
274            .unwrap_err()
275            .to_string();
276        assert!(err.contains("`locaton`"), "{err}");
277        assert!(err.contains("`location`"), "{err}");
278
279        let err = parse("memory: {}\nsettling_windw: 5m\n")
280            .unwrap_err()
281            .to_string();
282        assert!(err.contains("`settling_windw`"), "{err}");
283        assert!(err.contains("`settling_window`"), "{err}");
284    }
285
286    /// Pins that a `settling_window` that is not a duration is refused under
287    /// its own key.
288    #[test]
289    fn a_settling_window_that_is_not_a_duration_is_refused() {
290        for value in ["[1]", "lots", "5"] {
291            let err = parse(&format!("memory: {{}}\nsettling_window: {value}\n"))
292                .unwrap_err()
293                .to_string();
294            assert!(err.starts_with("settling_window: "), "{value}: {err}");
295        }
296    }
297
298    /// Pins that a section with no `settling_window` reads as five minutes,
299    /// and one that sets it reads as its value.
300    #[test]
301    fn an_absent_settling_window_reads_as_five_minutes() {
302        let s = parse("sqs: {}\n").unwrap();
303        assert_eq!(s.settling_window_or_default(), Duration::from_secs(300));
304        let s = parse("sqs: {}\nsettling_window: 90s\n").unwrap();
305        assert_eq!(s.settling_window_or_default(), Duration::from_secs(90));
306    }
307
308    fn load(feed: &str) -> Result<crate::config::PipelineConfig, String> {
309        let yaml = format!(
310            "pipeline: {{ name: demo }}\nsource: {{ memory: {{}} }}\nsink: {{ memory: {{}} }}\n\
311             feed: {feed}\n"
312        );
313        crate::config::PipelineConfig::from_str(&yaml).map_err(|e| e.to_string())
314    }
315
316    /// Pins that loading a file refuses a zero `settling_window`.
317    #[test]
318    fn a_zero_settling_window_is_refused() {
319        let err = load("{ sqs: {}, settling_window: 0s }").unwrap_err();
320        assert!(err.contains("`feed.settling_window`"), "{err}");
321    }
322
323    /// Pins the window's floor at one read interval: 20 s loads, 19 s is
324    /// refused.
325    #[test]
326    fn a_settling_window_below_the_floor_is_refused() {
327        let cfg = load("{ sqs: {}, settling_window: 20s }").expect("the floor loads");
328        assert_eq!(
329            cfg.feed.unwrap().settling_window,
330            Some(Duration::from_secs(20))
331        );
332        let err = load("{ sqs: {}, settling_window: 19s 999ms }").unwrap_err();
333        assert!(
334            err.contains("`feed.settling_window`") && err.contains("20s"),
335            "{err}"
336        );
337    }
338
339    /// Pins the window's ceiling at a day: `24h` loads, a second more and a
340    /// value that would overflow a deadline are refused.
341    #[test]
342    fn a_settling_window_above_a_day_is_refused() {
343        load("{ sqs: {}, settling_window: 24h }").expect("a day loads");
344        for value in ["24h 1s", "1000000years"] {
345            let err = load(&format!("{{ sqs: {{}}, settling_window: {value} }}")).unwrap_err();
346            assert!(
347                err.contains("`feed.settling_window`") && err.contains("1day"),
348                "{value}: {err}"
349            );
350        }
351    }
352
353    /// Pins that loading a file refuses a `settling_window` beside the
354    /// in-process feed, naming the rule.
355    #[test]
356    fn a_memory_feed_with_a_settling_window_is_refused() {
357        let err = load("{ memory: {}, settling_window: 1m }").unwrap_err();
358        assert!(
359            err.contains("`feed.settling_window`") && err.contains("`memory`"),
360            "{err}"
361        );
362    }
363
364    /// Pins that a `location` that is not a string is refused under its own
365    /// key.
366    #[test]
367    fn a_location_that_is_not_a_string_is_refused() {
368        let err = parse("memory: {}\nlocation: [1]\n")
369            .unwrap_err()
370            .to_string();
371        assert!(err.starts_with("location: "), "{err}");
372    }
373
374    /// Pins that `Debug` shows the location without its userinfo or query.
375    #[test]
376    fn a_section_debug_redacts_the_location() {
377        let s = parse("memory: {}\nlocation: s3://u:secret@ops/jobs/a?x=1\nsettling_window: 1m\n")
378            .unwrap();
379        let shown = format!("{s:?}");
380        assert!(shown.contains("ops/jobs/a"), "{shown}");
381        assert!(shown.contains("settling_window: Some(60s)"), "{shown}");
382        assert!(!shown.contains("secret"), "{shown}");
383        assert!(!shown.contains("x=1"), "{shown}");
384    }
385
386    /// Pins that loading a file refuses a `location` beside the in-process
387    /// feed, naming the rule.
388    #[test]
389    fn a_memory_feed_with_a_location_is_refused_at_load() {
390        let yaml = "pipeline: { name: demo }\nsource: { memory: {} }\nsink: { memory: {} }\n\
391                    feed: { memory: {}, location: s3://ops/jobs/a/ }\n";
392        let err = crate::config::PipelineConfig::from_str(yaml)
393            .unwrap_err()
394            .to_string();
395        assert!(
396            err.contains("`feed.location`") && err.contains("`memory`"),
397            "{err}"
398        );
399    }
400
401    /// Pins that loading a file keeps a `location` beside a backend other
402    /// than the in-process feed.
403    #[test]
404    fn a_location_beside_a_durable_backend_loads() {
405        let yaml = "pipeline: { name: demo }\nsource: { memory: {} }\nsink: { memory: {} }\n\
406                    feed: { sqs: {}, location: s3://ops/jobs/a/ }\n";
407        let cfg = crate::config::PipelineConfig::from_str(yaml).expect("loads");
408        let feed = cfg.feed.expect("feed section");
409        assert_eq!(feed.location.as_deref(), Some("s3://ops/jobs/a/"));
410    }
411
412    /// Pins that `units_held: 0` is refused.
413    #[test]
414    fn zero_units_held_is_refused() {
415        let err = parse("memory: {}\nunits_held: 0\n")
416            .unwrap_err()
417            .to_string();
418        assert!(err.starts_with("units_held: "), "{err}");
419    }
420
421    /// Pins that a `chunk:` under the backend is refused.
422    #[test]
423    fn chunk_under_the_backend_is_refused() {
424        let err = parse("memory: { chunk: { target_bytes: 1MiB } }\n")
425            .unwrap_err()
426            .to_string();
427        assert!(err.contains("chunk"), "{err}");
428    }
429
430    /// Pins that a `chunk:` in a body built in code is refused, as on the YAML path.
431    #[test]
432    fn chunk_under_a_built_backend_is_refused() {
433        let base = "pipeline: { name: demo }\nsource: { memory: {} }\nsink: { memory: {} }\n";
434        let mut cfg = crate::config::PipelineConfig::from_str(base).unwrap();
435        let body = serde_yaml::from_str("chunk: { target_bytes: 1MiB }").unwrap();
436        cfg.feed = Some(FeedSection::new(ComponentConfig::new("memory", body)));
437        let err = cfg.validate().unwrap_err().to_string();
438        assert!(err.contains("not `feed`"), "{err}");
439    }
440
441    /// Pins that an error in the backend's body carries its path from the
442    /// root, `feed.<backend>.<key>`, on the YAML path and from `new`.
443    #[test]
444    fn backend_errors_are_rooted_at_feed() {
445        let s = parse("memory: { x: lots }\n").unwrap();
446        let err = s
447            .backend()
448            .deserialize_into::<Body>()
449            .unwrap_err()
450            .to_string();
451        assert!(err.starts_with("feed.memory.x"), "{err}");
452
453        let body = serde_yaml::from_str("x: lots").unwrap();
454        let built = FeedSection::new(ComponentConfig::new("memory", body));
455        let err = built
456            .backend()
457            .deserialize_into::<Body>()
458            .unwrap_err()
459            .to_string();
460        assert!(err.starts_with("feed.memory.x"), "{err}");
461    }
462}
spate_core/config/feed.rs

spate_core/config/
feed.rs
