sui_indexer_alt_framework/
config.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::collections::BTreeSet;
5use std::sync::Mutex;
6
7use serde::Deserialize;
8use serde::Serialize;
9use tracing::warn;
10
11use sui_futures::stream::ConcurrencyConfig as RuntimeConcurrencyConfig;
12
13/// Serde-friendly concurrency configuration used by both the ingestion and processor stages.
14///
15/// Use `{ kind = "fixed", value = 10 }` for constant concurrency, or
16/// `{ kind = "adaptive", initial = 5, min = 1, max = 20 }` for adaptive concurrency that adjusts
17/// based on downstream channel fill fraction.
18///
19/// Adaptive mode accepts optional tuning overrides:
20/// - `dead_band`: `[low, high]` fill fraction thresholds (default: [0.6, 0.85])
21#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
22#[serde(tag = "kind", rename_all = "lowercase")]
23pub enum ConcurrencyConfig {
24    Fixed {
25        value: usize,
26    },
27    Adaptive {
28        initial: usize,
29        min: usize,
30        max: usize,
31        #[serde(default, skip_serializing_if = "Option::is_none")]
32        dead_band: Option<[f64; 2]>,
33    },
34}
35
36impl ConcurrencyConfig {
37    pub fn initial(&self) -> usize {
38        match self {
39            Self::Fixed { value } => *value,
40            Self::Adaptive { initial, .. } => *initial,
41        }
42    }
43
44    pub fn min(&self) -> usize {
45        let v = match self {
46            Self::Fixed { value } => *value,
47            Self::Adaptive { min, .. } => *min,
48        };
49        assert!(v >= 1, "min concurrency must be >= 1");
50        v
51    }
52
53    pub fn max(&self) -> usize {
54        match self {
55            Self::Fixed { value } => *value,
56            Self::Adaptive { max, .. } => *max,
57        }
58    }
59
60    pub fn is_adaptive(&self) -> bool {
61        matches!(self, Self::Adaptive { .. })
62    }
63}
64
65impl From<ConcurrencyConfig> for RuntimeConcurrencyConfig {
66    fn from(config: ConcurrencyConfig) -> Self {
67        match config {
68            ConcurrencyConfig::Fixed { value } => Self::fixed(value),
69            ConcurrencyConfig::Adaptive {
70                initial,
71                min,
72                max,
73                dead_band,
74            } => {
75                let mut c = Self::adaptive(initial, min, max);
76                if let Some([low, high]) = dead_band {
77                    c = c.with_dead_band(low, high);
78                }
79                c
80            }
81        }
82    }
83}
84
85/// Log a warning that the config key `key` is set but no longer has any effect. Each key is
86/// reported at most once per process, however many pipelines set it.
87pub fn warn_ignored_config_key(key: &'static str) {
88    static WARNED: Mutex<BTreeSet<&'static str>> = Mutex::new(BTreeSet::new());
89    let first_time = WARNED
90        .lock()
91        .unwrap_or_else(|poisoned| poisoned.into_inner())
92        .insert(key);
93    if first_time {
94        warn!(key, "Config key is no longer used and is ignored");
95    }
96}
sui_indexer_alt_framework/config.rs

sui_indexer_alt_framework/
config.rs
