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

sui_indexer_alt/
config.rs
