sui_indexer_alt/
config.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::num::NonZeroUsize;
5
6use serde::Deserialize;
7use serde::Serialize;
8use sui_indexer_alt_framework::config::ConcurrencyConfig;
9use sui_indexer_alt_framework::config::warn_ignored_config_key;
10use sui_indexer_alt_framework::ingestion::IngestionConfig;
11use sui_indexer_alt_framework::pipeline;
12use sui_indexer_alt_framework::pipeline::CommitterConfig;
13use sui_indexer_alt_framework::pipeline::concurrent::ConcurrentConfig;
14use sui_indexer_alt_framework::pipeline::concurrent::PrunerConfig;
15use sui_indexer_alt_framework::pipeline::sequential::SequentialConfig;
16use tracing::warn;
17
18/// Trait for merging configuration structs together.
19pub trait Merge: Sized {
20    fn merge(self, other: Self) -> anyhow::Result<Self>;
21}
22
23#[derive(Clone, Default, Debug, Deserialize, Serialize)]
24#[serde(default, rename_all = "kebab-case", deny_unknown_fields)]
25pub struct IndexerConfig {
26    /// How checkpoints are read by the indexer.
27    pub ingestion: IngestionLayer,
28
29    /// Default configuration for committers that is shared by all pipelines. Pipelines can
30    /// override individual settings in their own configuration sections.
31    pub committer: CommitterLayer,
32
33    /// Default configuration for pruners that is shared by all concurrent pipelines. Pipelines can
34    /// override individual settings in their own configuration sections. Concurrent pipelines
35    /// still need to specify a pruner configuration (although it can be empty) to indicate that
36    /// they want to enable pruning, but when they do, any missing values will be filled in by this
37    /// config.
38    pub pruner: PrunerLayer,
39
40    /// Per-pipeline configurations.
41    pub pipeline: PipelineLayer,
42}
43
44// Configuration layers apply overrides over a base configuration. When reading configs from a
45// file, we read them into layer types, and then apply those layers onto an existing configuration
46// (such as the default configuration) to `finish()` them.
47//
48// Treating configs as layers allows us to support configuration merging, where multiple
49// configuration files can be combined into one final configuration. Having a separate type for
50// reading configs also allows us to detect and warn against unrecognised fields.
51
52#[derive(Clone, Default, Debug, Deserialize, Serialize)]
53#[serde(default, rename_all = "kebab-case", deny_unknown_fields)]
54pub struct IngestionLayer {
55    pub ingest_concurrency: Option<ConcurrencyConfig>,
56    pub retry_interval_ms: Option<u64>,
57    pub streaming_backoff_initial_batch_size: Option<NonZeroUsize>,
58    pub streaming_backoff_max_batch_size: Option<usize>,
59    pub streaming_connection_timeout_ms: Option<u64>,
60    pub streaming_statement_timeout_ms: Option<u64>,
61    pub min_cohort_boundary: Option<u64>,
62
63    /// Deprecated: accepted (and ignored) so old configs don't fail to parse. Replaced by
64    /// per-pipeline `ingestion.subscriber-channel-size`.
65    pub checkpoint_buffer_size: Option<usize>,
66}
67
68#[derive(Clone, Default, Debug, Deserialize, Serialize)]
69#[serde(default, rename_all = "kebab-case", deny_unknown_fields)]
70pub struct SequentialLayer {
71    pub committer: Option<CommitterLayer>,
72    pub ingestion: Option<PipelineIngestionLayer>,
73    pub fanout: Option<ConcurrencyConfig>,
74    /// Deprecated and ignored: batch size is controlled by the committer's `min-batch-rows` and
75    /// `max-batch-delay-ms`.
76    pub min_eager_rows: Option<usize>,
77    pub max_pending_rows: Option<usize>,
78    pub max_batch_checkpoints: Option<usize>,
79    pub processor_channel_size: Option<usize>,
80    pub pipeline_depth: Option<usize>,
81}
82
83#[derive(Clone, Default, Debug, Deserialize, Serialize)]
84#[serde(default, rename_all = "kebab-case", deny_unknown_fields)]
85pub struct ConcurrentLayer {
86    pub committer: Option<CommitterLayer>,
87    pub ingestion: Option<PipelineIngestionLayer>,
88    pub pruner: Option<PrunerLayer>,
89    pub fanout: Option<ConcurrencyConfig>,
90    /// Deprecated and ignored: batch size is controlled by the committer's `min-batch-rows` and
91    /// `max-batch-delay-ms`.
92    pub min_eager_rows: Option<usize>,
93    pub max_pending_rows: Option<usize>,
94    pub max_watermark_updates: Option<usize>,
95    pub processor_channel_size: Option<usize>,
96    pub collector_channel_size: Option<usize>,
97    pub committer_channel_size: Option<usize>,
98}
99
100#[derive(Clone, Default, Debug, Deserialize, Serialize)]
101#[serde(default, rename_all = "kebab-case", deny_unknown_fields)]
102pub struct PipelineIngestionLayer {
103    pub subscriber_channel_size: Option<usize>,
104}
105
106#[derive(Clone, Default, Debug, Deserialize, Serialize)]
107#[serde(default, rename_all = "kebab-case", deny_unknown_fields)]
108pub struct CommitterLayer {
109    pub write_concurrency: Option<usize>,
110    pub watermark_interval_ms: Option<u64>,
111
112    /// Minimum number of pending rows the collector waits for before cutting a batch. `0` cuts a
113    /// batch as soon as the committer has room for it.
114    pub min_batch_rows: Option<usize>,
115
116    /// Longest time, in milliseconds, the oldest pending row waits for `min-batch-rows` to
117    /// accumulate before a batch is cut anyway.
118    pub max_batch_delay_ms: Option<u64>,
119
120    /// Deprecated and ignored: batches are cut as soon as the committer has room; see
121    /// `min-batch-rows` and `max-batch-delay-ms`.
122    pub collect_interval_ms: Option<u64>,
123}
124
125#[derive(Clone, Default, Debug, Deserialize, Serialize)]
126#[serde(default, rename_all = "kebab-case", deny_unknown_fields)]
127pub struct PrunerLayer {
128    pub interval_ms: Option<u64>,
129    pub delay_ms: Option<u64>,
130    pub retention: Option<u64>,
131    pub max_chunk_size: Option<u64>,
132    pub prune_concurrency: Option<u64>,
133}
134
135#[derive(Clone, Default, Debug, Deserialize, Serialize)]
136#[serde(default, rename_all = "snake_case", deny_unknown_fields)]
137pub struct PipelineLayer {
138    // Sequential pipelines
139    pub sum_displays: Option<SequentialLayer>,
140
141    // All concurrent pipelines
142    pub cp_bloom_blocks: Option<ConcurrentLayer>,
143    pub cp_blooms: Option<ConcurrentLayer>,
144    pub cp_digests: Option<ConcurrentLayer>,
145    pub cp_sequence_numbers: Option<ConcurrentLayer>,
146    pub ev_emit_mod: Option<ConcurrentLayer>,
147    pub ev_struct_inst: Option<ConcurrentLayer>,
148    pub kv_checkpoints: Option<ConcurrentLayer>,
149    pub kv_epoch_ends: Option<ConcurrentLayer>,
150    pub kv_epoch_starts: Option<ConcurrentLayer>,
151    pub kv_feature_flags: Option<ConcurrentLayer>,
152    pub kv_objects: Option<ConcurrentLayer>,
153    pub kv_packages: Option<ConcurrentLayer>,
154    pub kv_protocol_configs: Option<ConcurrentLayer>,
155    pub kv_transactions: Option<ConcurrentLayer>,
156    pub obj_versions: Option<ConcurrentLayer>,
157    pub tx_affected_addresses: Option<ConcurrentLayer>,
158    pub tx_affected_objects: Option<ConcurrentLayer>,
159    pub tx_balance_changes: Option<ConcurrentLayer>,
160    pub tx_calls: Option<ConcurrentLayer>,
161    pub tx_digests: Option<ConcurrentLayer>,
162    pub tx_kinds: Option<ConcurrentLayer>,
163}
164
165impl IndexerConfig {
166    /// Generate an example configuration, suitable for demonstrating the fields available to
167    /// configure.
168    pub fn example() -> Self {
169        let mut example: Self = Default::default();
170
171        example.ingestion = IngestionConfig::default().into();
172        example.committer = CommitterConfig::default().into();
173        example.pruner = PrunerConfig::default().into();
174        example.pipeline = PipelineLayer::example();
175
176        example
177    }
178
179    /// Generate a configuration suitable for testing. This is the same as the example
180    /// configuration, but with reduced concurrency and faster polling intervals so tests spend
181    /// less time waiting.
182    pub fn for_test() -> Self {
183        Self::example()
184            .merge(IndexerConfig {
185                ingestion: IngestionLayer {
186                    retry_interval_ms: Some(10),
187                    ingest_concurrency: Some(ConcurrencyConfig::Fixed { value: 1 }),
188                    ..Default::default()
189                },
190                committer: CommitterLayer {
191                    watermark_interval_ms: Some(50),
192                    write_concurrency: Some(1),
193                    ..Default::default()
194                },
195                pruner: PrunerLayer {
196                    interval_ms: Some(50),
197                    delay_ms: Some(0),
198                    ..Default::default()
199                },
200                ..Default::default()
201            })
202            .expect("Merge failed for test configuration")
203    }
204}
205
206impl IngestionLayer {
207    pub fn finish(self, base: IngestionConfig) -> anyhow::Result<IngestionConfig> {
208        if self.checkpoint_buffer_size.is_some() {
209            warn!(
210                "Config field `checkpoint-buffer-size` is deprecated and ignored. Remove it from \
211                 your config; set `subscriber-channel-size` under each pipeline's `ingestion` \
212                 section if you need to override the default."
213            );
214        }
215
216        Ok(IngestionConfig {
217            ingest_concurrency: self.ingest_concurrency.unwrap_or(base.ingest_concurrency),
218            retry_interval_ms: self.retry_interval_ms.unwrap_or(base.retry_interval_ms),
219            streaming_backoff_initial_batch_size: self
220                .streaming_backoff_initial_batch_size
221                .unwrap_or(base.streaming_backoff_initial_batch_size),
222            streaming_backoff_max_batch_size: self
223                .streaming_backoff_max_batch_size
224                .unwrap_or(base.streaming_backoff_max_batch_size),
225            streaming_connection_timeout_ms: self
226                .streaming_connection_timeout_ms
227                .unwrap_or(base.streaming_connection_timeout_ms),
228            streaming_statement_timeout_ms: self
229                .streaming_statement_timeout_ms
230                .unwrap_or(base.streaming_statement_timeout_ms),
231            min_cohort_boundary: self.min_cohort_boundary.unwrap_or(base.min_cohort_boundary),
232        })
233    }
234}
235
236impl SequentialLayer {
237    pub fn finish(self, base: SequentialConfig) -> anyhow::Result<SequentialConfig> {
238        if self.min_eager_rows.is_some() {
239            warn_ignored_config_key("min-eager-rows");
240        }
241
242        Ok(SequentialConfig {
243            committer: if let Some(committer) = self.committer {
244                committer.finish(base.committer)?
245            } else {
246                base.committer
247            },
248            ingestion: if let Some(ingestion) = self.ingestion {
249                ingestion.finish(base.ingestion)
250            } else {
251                base.ingestion
252            },
253            fanout: self.fanout.or(base.fanout),
254            max_pending_rows: self.max_pending_rows.or(base.max_pending_rows),
255            max_batch_checkpoints: self.max_batch_checkpoints.or(base.max_batch_checkpoints),
256            processor_channel_size: self.processor_channel_size.or(base.processor_channel_size),
257            pipeline_depth: self.pipeline_depth.or(base.pipeline_depth),
258        })
259    }
260}
261
262impl ConcurrentLayer {
263    /// Unlike other parameters, `pruner` will appear in the finished configuration only if they
264    /// appear in the layer *and* in the base.
265    pub fn finish(self, base: ConcurrentConfig) -> anyhow::Result<ConcurrentConfig> {
266        if self.min_eager_rows.is_some() {
267            warn_ignored_config_key("min-eager-rows");
268        }
269
270        Ok(ConcurrentConfig {
271            committer: if let Some(committer) = self.committer {
272                committer.finish(base.committer)?
273            } else {
274                base.committer
275            },
276            ingestion: if let Some(ingestion) = self.ingestion {
277                ingestion.finish(base.ingestion)
278            } else {
279                base.ingestion
280            },
281            pruner: match (self.pruner, base.pruner) {
282                (None, _) | (_, None) => None,
283                (Some(pruner), Some(base)) => Some(pruner.finish(base)?),
284            },
285            fanout: self.fanout.or(base.fanout),
286            max_pending_rows: self.max_pending_rows.or(base.max_pending_rows),
287            max_watermark_updates: self.max_watermark_updates.or(base.max_watermark_updates),
288            processor_channel_size: self.processor_channel_size.or(base.processor_channel_size),
289            collector_channel_size: self.collector_channel_size.or(base.collector_channel_size),
290            committer_channel_size: self.committer_channel_size.or(base.committer_channel_size),
291        })
292    }
293}
294
295impl PipelineIngestionLayer {
296    pub fn finish(self, base: pipeline::IngestionConfig) -> pipeline::IngestionConfig {
297        pipeline::IngestionConfig {
298            subscriber_channel_size: self
299                .subscriber_channel_size
300                .or(base.subscriber_channel_size),
301        }
302    }
303}
304
305impl CommitterLayer {
306    pub fn finish(self, base: CommitterConfig) -> anyhow::Result<CommitterConfig> {
307        if self.collect_interval_ms.is_some() {
308            warn_ignored_config_key("collect-interval-ms");
309        }
310
311        Ok(CommitterConfig {
312            write_concurrency: self.write_concurrency.unwrap_or(base.write_concurrency),
313            watermark_interval_ms: self
314                .watermark_interval_ms
315                .unwrap_or(base.watermark_interval_ms),
316            watermark_interval_jitter_ms: 0,
317            min_batch_rows: self.min_batch_rows.unwrap_or(base.min_batch_rows),
318            max_batch_delay_ms: self.max_batch_delay_ms.unwrap_or(base.max_batch_delay_ms),
319        })
320    }
321}
322
323impl PrunerLayer {
324    pub fn finish(self, base: PrunerConfig) -> anyhow::Result<PrunerConfig> {
325        Ok(PrunerConfig {
326            interval_ms: self.interval_ms.unwrap_or(base.interval_ms),
327            delay_ms: self.delay_ms.unwrap_or(base.delay_ms),
328            retention: self.retention.unwrap_or(base.retention),
329            max_chunk_size: self.max_chunk_size.unwrap_or(base.max_chunk_size),
330            prune_concurrency: self.prune_concurrency.unwrap_or(base.prune_concurrency),
331        })
332    }
333}
334
335impl PipelineLayer {
336    /// Generate an example configuration, suitable for demonstrating the fields available to
337    /// configure.
338    pub fn example() -> Self {
339        PipelineLayer {
340            cp_blooms: Some(Default::default()),
341            cp_bloom_blocks: Some(Default::default()),
342            cp_digests: Some(Default::default()),
343            sum_displays: Some(Default::default()),
344            cp_sequence_numbers: Some(Default::default()),
345            ev_emit_mod: Some(Default::default()),
346            ev_struct_inst: Some(Default::default()),
347            kv_checkpoints: Some(Default::default()),
348            kv_epoch_ends: Some(Default::default()),
349            kv_epoch_starts: Some(Default::default()),
350            kv_feature_flags: Some(Default::default()),
351            kv_objects: Some(Default::default()),
352            kv_packages: Some(Default::default()),
353            kv_protocol_configs: Some(Default::default()),
354            kv_transactions: Some(Default::default()),
355            obj_versions: Some(Default::default()),
356            tx_affected_addresses: Some(Default::default()),
357            tx_affected_objects: Some(Default::default()),
358            tx_balance_changes: Some(Default::default()),
359            tx_calls: Some(Default::default()),
360            tx_digests: Some(Default::default()),
361            tx_kinds: Some(Default::default()),
362        }
363    }
364}
365
366impl Merge for IndexerConfig {
367    fn merge(self, other: IndexerConfig) -> anyhow::Result<IndexerConfig> {
368        Ok(IndexerConfig {
369            ingestion: self.ingestion.merge(other.ingestion)?,
370            committer: self.committer.merge(other.committer)?,
371            pruner: self.pruner.merge(other.pruner)?,
372            pipeline: self.pipeline.merge(other.pipeline)?,
373        })
374    }
375}
376
377impl Merge for IngestionLayer {
378    fn merge(self, other: IngestionLayer) -> anyhow::Result<IngestionLayer> {
379        Ok(IngestionLayer {
380            ingest_concurrency: other.ingest_concurrency.or(self.ingest_concurrency),
381            retry_interval_ms: other.retry_interval_ms.or(self.retry_interval_ms),
382            streaming_backoff_initial_batch_size: other
383                .streaming_backoff_initial_batch_size
384                .or(self.streaming_backoff_initial_batch_size),
385            streaming_backoff_max_batch_size: other
386                .streaming_backoff_max_batch_size
387                .or(self.streaming_backoff_max_batch_size),
388            streaming_connection_timeout_ms: other
389                .streaming_connection_timeout_ms
390                .or(self.streaming_connection_timeout_ms),
391            streaming_statement_timeout_ms: other
392                .streaming_statement_timeout_ms
393                .or(self.streaming_statement_timeout_ms),
394            min_cohort_boundary: other.min_cohort_boundary.or(self.min_cohort_boundary),
395            checkpoint_buffer_size: other.checkpoint_buffer_size.or(self.checkpoint_buffer_size),
396        })
397    }
398}
399
400impl Merge for SequentialLayer {
401    fn merge(self, other: SequentialLayer) -> anyhow::Result<SequentialLayer> {
402        Ok(SequentialLayer {
403            committer: self.committer.merge(other.committer)?,
404            ingestion: self.ingestion.merge(other.ingestion)?,
405            fanout: other.fanout.or(self.fanout),
406            min_eager_rows: other.min_eager_rows.or(self.min_eager_rows),
407            max_pending_rows: other.max_pending_rows.or(self.max_pending_rows),
408            max_batch_checkpoints: other.max_batch_checkpoints.or(self.max_batch_checkpoints),
409            processor_channel_size: other.processor_channel_size.or(self.processor_channel_size),
410            pipeline_depth: other.pipeline_depth.or(self.pipeline_depth),
411        })
412    }
413}
414
415impl Merge for ConcurrentLayer {
416    fn merge(self, other: ConcurrentLayer) -> anyhow::Result<ConcurrentLayer> {
417        Ok(ConcurrentLayer {
418            committer: self.committer.merge(other.committer)?,
419            ingestion: self.ingestion.merge(other.ingestion)?,
420            pruner: self.pruner.merge(other.pruner)?,
421            fanout: other.fanout.or(self.fanout),
422            min_eager_rows: other.min_eager_rows.or(self.min_eager_rows),
423            max_pending_rows: other.max_pending_rows.or(self.max_pending_rows),
424            max_watermark_updates: other.max_watermark_updates.or(self.max_watermark_updates),
425            processor_channel_size: other.processor_channel_size.or(self.processor_channel_size),
426            collector_channel_size: other.collector_channel_size.or(self.collector_channel_size),
427            committer_channel_size: other.committer_channel_size.or(self.committer_channel_size),
428        })
429    }
430}
431
432impl Merge for PipelineIngestionLayer {
433    fn merge(self, other: PipelineIngestionLayer) -> anyhow::Result<PipelineIngestionLayer> {
434        Ok(PipelineIngestionLayer {
435            subscriber_channel_size: other
436                .subscriber_channel_size
437                .or(self.subscriber_channel_size),
438        })
439    }
440}
441
442impl Merge for CommitterLayer {
443    fn merge(self, other: CommitterLayer) -> anyhow::Result<CommitterLayer> {
444        Ok(CommitterLayer {
445            write_concurrency: other.write_concurrency.or(self.write_concurrency),
446            watermark_interval_ms: other.watermark_interval_ms.or(self.watermark_interval_ms),
447            min_batch_rows: other.min_batch_rows.or(self.min_batch_rows),
448            max_batch_delay_ms: other.max_batch_delay_ms.or(self.max_batch_delay_ms),
449            collect_interval_ms: other.collect_interval_ms.or(self.collect_interval_ms),
450        })
451    }
452}
453
454impl Merge for PrunerLayer {
455    /// Last write takes precedence for all fields except the `retention`, which takes the max of
456    /// all available values.
457    fn merge(self, other: PrunerLayer) -> anyhow::Result<PrunerLayer> {
458        Ok(PrunerLayer {
459            interval_ms: other.interval_ms.or(self.interval_ms),
460            delay_ms: other.delay_ms.or(self.delay_ms),
461            retention: match (other.retention, self.retention) {
462                (Some(a), Some(b)) => Some(a.max(b)),
463                (Some(a), _) | (_, Some(a)) => Some(a),
464                (None, None) => None,
465            },
466            max_chunk_size: other.max_chunk_size.or(self.max_chunk_size),
467            prune_concurrency: other.prune_concurrency.or(self.prune_concurrency),
468        })
469    }
470}
471
472impl Merge for PipelineLayer {
473    fn merge(self, other: PipelineLayer) -> anyhow::Result<PipelineLayer> {
474        Ok(PipelineLayer {
475            cp_blooms: self.cp_blooms.merge(other.cp_blooms)?,
476            cp_bloom_blocks: self.cp_bloom_blocks.merge(other.cp_bloom_blocks)?,
477            cp_digests: self.cp_digests.merge(other.cp_digests)?,
478            sum_displays: self.sum_displays.merge(other.sum_displays)?,
479            cp_sequence_numbers: self.cp_sequence_numbers.merge(other.cp_sequence_numbers)?,
480            ev_emit_mod: self.ev_emit_mod.merge(other.ev_emit_mod)?,
481            ev_struct_inst: self.ev_struct_inst.merge(other.ev_struct_inst)?,
482            kv_checkpoints: self.kv_checkpoints.merge(other.kv_checkpoints)?,
483            kv_epoch_ends: self.kv_epoch_ends.merge(other.kv_epoch_ends)?,
484            kv_epoch_starts: self.kv_epoch_starts.merge(other.kv_epoch_starts)?,
485            kv_feature_flags: self.kv_feature_flags.merge(other.kv_feature_flags)?,
486            kv_objects: self.kv_objects.merge(other.kv_objects)?,
487            kv_packages: self.kv_packages.merge(other.kv_packages)?,
488            kv_protocol_configs: self.kv_protocol_configs.merge(other.kv_protocol_configs)?,
489            kv_transactions: self.kv_transactions.merge(other.kv_transactions)?,
490            obj_versions: self.obj_versions.merge(other.obj_versions)?,
491            tx_affected_addresses: self
492                .tx_affected_addresses
493                .merge(other.tx_affected_addresses)?,
494            tx_affected_objects: self.tx_affected_objects.merge(other.tx_affected_objects)?,
495            tx_balance_changes: self.tx_balance_changes.merge(other.tx_balance_changes)?,
496            tx_calls: self.tx_calls.merge(other.tx_calls)?,
497            tx_digests: self.tx_digests.merge(other.tx_digests)?,
498            tx_kinds: self.tx_kinds.merge(other.tx_kinds)?,
499        })
500    }
501}
502
503impl<T: Merge> Merge for Option<T> {
504    fn merge(self, other: Option<T>) -> anyhow::Result<Option<T>> {
505        Ok(match (self, other) {
506            (Some(a), Some(b)) => Some(a.merge(b)?),
507            (Some(a), _) | (_, Some(a)) => Some(a),
508            (None, None) => None,
509        })
510    }
511}
512
513impl From<IngestionConfig> for IngestionLayer {
514    fn from(config: IngestionConfig) -> Self {
515        Self {
516            ingest_concurrency: Some(config.ingest_concurrency),
517            retry_interval_ms: Some(config.retry_interval_ms),
518            streaming_backoff_initial_batch_size: Some(config.streaming_backoff_initial_batch_size),
519            streaming_backoff_max_batch_size: Some(config.streaming_backoff_max_batch_size),
520            streaming_connection_timeout_ms: Some(config.streaming_connection_timeout_ms),
521            streaming_statement_timeout_ms: Some(config.streaming_statement_timeout_ms),
522            min_cohort_boundary: Some(config.min_cohort_boundary),
523            checkpoint_buffer_size: None,
524        }
525    }
526}
527
528impl From<SequentialConfig> for SequentialLayer {
529    fn from(config: SequentialConfig) -> Self {
530        Self {
531            committer: Some(config.committer.into()),
532            ingestion: Some(config.ingestion.into()),
533            fanout: config.fanout,
534            min_eager_rows: None,
535            max_pending_rows: config.max_pending_rows,
536            max_batch_checkpoints: config.max_batch_checkpoints,
537            processor_channel_size: config.processor_channel_size,
538            pipeline_depth: config.pipeline_depth,
539        }
540    }
541}
542
543impl From<ConcurrentConfig> for ConcurrentLayer {
544    fn from(config: ConcurrentConfig) -> Self {
545        Self {
546            committer: Some(config.committer.into()),
547            ingestion: Some(config.ingestion.into()),
548            pruner: config.pruner.map(Into::into),
549            fanout: config.fanout,
550            min_eager_rows: None,
551            max_pending_rows: config.max_pending_rows,
552            max_watermark_updates: config.max_watermark_updates,
553            processor_channel_size: config.processor_channel_size,
554            collector_channel_size: config.collector_channel_size,
555            committer_channel_size: config.committer_channel_size,
556        }
557    }
558}
559
560impl From<pipeline::IngestionConfig> for PipelineIngestionLayer {
561    fn from(config: pipeline::IngestionConfig) -> Self {
562        Self {
563            subscriber_channel_size: config.subscriber_channel_size,
564        }
565    }
566}
567
568impl From<CommitterConfig> for CommitterLayer {
569    fn from(config: CommitterConfig) -> Self {
570        Self {
571            write_concurrency: Some(config.write_concurrency),
572            watermark_interval_ms: Some(config.watermark_interval_ms),
573            min_batch_rows: Some(config.min_batch_rows),
574            max_batch_delay_ms: Some(config.max_batch_delay_ms),
575            collect_interval_ms: None,
576        }
577    }
578}
579
580impl From<PrunerConfig> for PrunerLayer {
581    fn from(config: PrunerConfig) -> Self {
582        Self {
583            interval_ms: Some(config.interval_ms),
584            delay_ms: Some(config.delay_ms),
585            retention: Some(config.retention),
586            max_chunk_size: Some(config.max_chunk_size),
587            prune_concurrency: Some(config.prune_concurrency),
588        }
589    }
590}
591
592#[cfg(test)]
593mod tests {
594    use super::*;
595
596    macro_rules! assert_matches {
597        ($value:expr, $pattern:pat $(,)?) => {
598            let value = $value;
599            assert!(
600                matches!(value, $pattern),
601                "Did not match pattern:\nexpected: {}\nactual: {value:#?}",
602                stringify!($pattern)
603            );
604        };
605    }
606
607    #[test]
608    fn merge_recursive() {
609        let this = PipelineLayer {
610            sum_displays: Some(SequentialLayer {
611                committer: Some(CommitterLayer {
612                    write_concurrency: Some(10),
613                    min_batch_rows: Some(1000),
614                    ..Default::default()
615                }),
616                max_pending_rows: Some(100),
617                ..Default::default()
618            }),
619            ev_emit_mod: Some(ConcurrentLayer {
620                committer: Some(CommitterLayer {
621                    write_concurrency: Some(5),
622                    min_batch_rows: Some(500),
623                    ..Default::default()
624                }),
625                ..Default::default()
626            }),
627            ..Default::default()
628        };
629
630        let that = PipelineLayer {
631            sum_displays: Some(SequentialLayer {
632                committer: Some(CommitterLayer {
633                    write_concurrency: Some(5),
634                    watermark_interval_ms: Some(500),
635                    ..Default::default()
636                }),
637                max_pending_rows: Some(200),
638                ..Default::default()
639            }),
640            ev_emit_mod: None,
641            ..Default::default()
642        };
643
644        let this_then_that = this.clone().merge(that.clone()).unwrap();
645        let that_then_this = that.clone().merge(this.clone()).unwrap();
646
647        assert_matches!(
648            this_then_that,
649            PipelineLayer {
650                sum_displays: Some(SequentialLayer {
651                    committer: Some(CommitterLayer {
652                        write_concurrency: Some(5),
653                        min_batch_rows: Some(1000),
654                        watermark_interval_ms: Some(500),
655                        ..
656                    }),
657                    max_pending_rows: Some(200),
658                    ..
659                }),
660                ev_emit_mod: Some(ConcurrentLayer {
661                    committer: Some(CommitterLayer {
662                        write_concurrency: Some(5),
663                        min_batch_rows: Some(500),
664                        watermark_interval_ms: None,
665                        ..
666                    }),
667                    pruner: None,
668                    ..
669                }),
670                ..
671            },
672        );
673
674        assert_matches!(
675            that_then_this,
676            PipelineLayer {
677                sum_displays: Some(SequentialLayer {
678                    committer: Some(CommitterLayer {
679                        write_concurrency: Some(10),
680                        min_batch_rows: Some(1000),
681                        watermark_interval_ms: Some(500),
682                        ..
683                    }),
684                    max_pending_rows: Some(100),
685                    ..
686                }),
687                ev_emit_mod: Some(ConcurrentLayer {
688                    committer: Some(CommitterLayer {
689                        write_concurrency: Some(5),
690                        min_batch_rows: Some(500),
691                        watermark_interval_ms: None,
692                        ..
693                    }),
694                    pruner: None,
695                    ..
696                }),
697                ..
698            },
699        );
700    }
701
702    #[test]
703    fn merge_pruner() {
704        let this = PrunerLayer {
705            interval_ms: None,
706            delay_ms: Some(100),
707            retention: Some(200),
708            max_chunk_size: Some(300),
709            prune_concurrency: Some(1),
710        };
711
712        let that = PrunerLayer {
713            interval_ms: Some(400),
714            delay_ms: None,
715            retention: Some(500),
716            max_chunk_size: Some(600),
717            prune_concurrency: Some(2),
718        };
719
720        let this_then_that = this.clone().merge(that.clone()).unwrap();
721        let that_then_this = that.clone().merge(this.clone()).unwrap();
722
723        assert_matches!(
724            this_then_that,
725            PrunerLayer {
726                interval_ms: Some(400),
727                delay_ms: Some(100),
728                retention: Some(500),
729                max_chunk_size: Some(600),
730                prune_concurrency: Some(2),
731            },
732        );
733
734        assert_matches!(
735            that_then_this,
736            PrunerLayer {
737                interval_ms: Some(400),
738                delay_ms: Some(100),
739                retention: Some(500),
740                max_chunk_size: Some(300),
741                prune_concurrency: Some(1),
742            },
743        );
744    }
745
746    #[test]
747    fn finish_concurrent_unpruned_override() {
748        let layer = ConcurrentLayer {
749            committer: None,
750            pruner: None,
751            ..Default::default()
752        };
753
754        let base = ConcurrentConfig {
755            committer: CommitterConfig {
756                write_concurrency: 5,
757                max_batch_delay_ms: 50,
758                watermark_interval_ms: 500,
759                ..Default::default()
760            },
761            pruner: Some(PrunerConfig::default()),
762            ..Default::default()
763        };
764
765        assert_matches!(
766            layer.finish(base).unwrap(),
767            ConcurrentConfig {
768                committer: CommitterConfig {
769                    write_concurrency: 5,
770                    max_batch_delay_ms: 50,
771                    watermark_interval_ms: 500,
772                    ..
773                },
774                pruner: None,
775                ..
776            },
777        );
778    }
779
780    #[test]
781    fn finish_concurrent_no_pruner() {
782        let layer = ConcurrentLayer {
783            committer: None,
784            pruner: None,
785            ..Default::default()
786        };
787
788        let base = ConcurrentConfig {
789            committer: CommitterConfig {
790                write_concurrency: 5,
791                max_batch_delay_ms: 50,
792                watermark_interval_ms: 500,
793                ..Default::default()
794            },
795            pruner: None,
796            ..Default::default()
797        };
798
799        assert_matches!(
800            layer.finish(base).unwrap(),
801            ConcurrentConfig {
802                committer: CommitterConfig {
803                    write_concurrency: 5,
804                    max_batch_delay_ms: 50,
805                    watermark_interval_ms: 500,
806                    ..
807                },
808                pruner: None,
809                ..
810            },
811        );
812    }
813
814    #[test]
815    fn finish_concurrent_pruner() {
816        let layer = ConcurrentLayer {
817            committer: None,
818            pruner: Some(PrunerLayer {
819                interval_ms: Some(1000),
820                ..Default::default()
821            }),
822            ..Default::default()
823        };
824
825        let base = ConcurrentConfig {
826            committer: CommitterConfig {
827                write_concurrency: 5,
828                max_batch_delay_ms: 50,
829                watermark_interval_ms: 500,
830                ..Default::default()
831            },
832            pruner: Some(PrunerConfig {
833                interval_ms: 100,
834                delay_ms: 200,
835                retention: 300,
836                max_chunk_size: 400,
837                prune_concurrency: 1,
838            }),
839            ..Default::default()
840        };
841
842        assert_matches!(
843            layer.finish(base).unwrap(),
844            ConcurrentConfig {
845                committer: CommitterConfig {
846                    write_concurrency: 5,
847                    max_batch_delay_ms: 50,
848                    watermark_interval_ms: 500,
849                    ..
850                },
851                pruner: Some(PrunerConfig {
852                    interval_ms: 1000,
853                    delay_ms: 200,
854                    retention: 300,
855                    max_chunk_size: 400,
856                    prune_concurrency: 1,
857                }),
858                ..
859            },
860        );
861    }
862
863    #[test]
864    fn detect_unrecognized_fields() {
865        let err = toml::from_str::<IndexerConfig>(
866            r#"
867            i_dont_exist = "foo"
868            "#,
869        )
870        .unwrap_err();
871
872        assert!(
873            err.to_string().contains("i_dont_exist"),
874            "Unexpected error: {err}"
875        );
876    }
877
878    #[test]
879    fn deprecated_checkpoint_buffer_size_parses() {
880        let config: IndexerConfig = toml::from_str(
881            r#"
882            [ingestion]
883            checkpoint-buffer-size = 5000
884            "#,
885        )
886        .expect("deprecated `checkpoint-buffer-size` must deserialize without error");
887
888        assert_eq!(config.ingestion.checkpoint_buffer_size, Some(5000));
889    }
890
891    #[test]
892    fn deprecated_batching_keys_parse_and_are_ignored() {
893        let config: IndexerConfig = toml::from_str(
894            r#"
895            [committer]
896            collect-interval-ms = 500
897
898            [pipeline.ev_emit_mod]
899            min-eager-rows = 100
900            "#,
901        )
902        .expect("deprecated batching keys must deserialize without error");
903
904        assert_eq!(config.committer.collect_interval_ms, Some(500));
905        let ev_emit_mod = config.pipeline.ev_emit_mod.expect("ev_emit_mod layer");
906        assert_eq!(ev_emit_mod.min_eager_rows, Some(100));
907
908        let committer = config.committer.finish(CommitterConfig::default()).unwrap();
909        assert_eq!(
910            committer.min_batch_rows,
911            CommitterConfig::default().min_batch_rows
912        );
913        assert_eq!(
914            committer.max_batch_delay_ms,
915            CommitterConfig::default().max_batch_delay_ms
916        );
917    }
918}
sui_indexer_alt/config.rs

sui_indexer_alt/
config.rs
