perf(scanner): cut per-object allocations in the scan hot path (#6274)

This commit is contained in:
houseme
2026-08-19 23:33:49 +08:00
committed by GitHub
parent 0126f359e3
commit 81332718e6
4 changed files with 149 additions and 53 deletions
+57 -3
View File
@@ -26,6 +26,8 @@ use http::HeaderMap;
use rustfs_config::server_config::{Config as ServerConfig, get_global_server_config as config_get_global_server_config}; use rustfs_config::server_config::{Config as ServerConfig, get_global_server_config as config_get_global_server_config};
use std::path::PathBuf; use std::path::PathBuf;
use std::sync::Arc; use std::sync::Arc;
use std::sync::RwLock;
use std::time::{Duration, Instant};
use storage_api::owner::{ use storage_api::owner::{
ECSTORE_BUCKET_META_PREFIX, ECSTORE_RUSTFS_META_BUCKET, ECSTORE_STORAGE_FORMAT_FILE, ECSTORE_STORAGECLASS_RRS, ECSTORE_BUCKET_META_PREFIX, ECSTORE_RUSTFS_META_BUCKET, ECSTORE_STORAGE_FORMAT_FILE, ECSTORE_STORAGECLASS_RRS,
ECSTORE_STORAGECLASS_STANDARD, ECSTORE_TRANSITION_COMPLETE, EcstoreBucketTargetSys, EcstoreBucketVersioningSys, EcstoreDisk, ECSTORE_STORAGECLASS_STANDARD, ECSTORE_TRANSITION_COMPLETE, EcstoreBucketTargetSys, EcstoreBucketVersioningSys, EcstoreDisk,
@@ -33,7 +35,7 @@ use storage_api::owner::{
EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreLcEventSrc, EcstoreLifecycle, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreLcEventSrc, EcstoreLifecycle,
EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts, EcstoreReplicationConfigurationExt, EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts, EcstoreReplicationConfigurationExt,
EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore,
EcstoreTierConfig, EcstoreVersioningApi, HTTPPreconditions, HTTPRangeSpec, ObjectIO, ObjectOperations, ObjectToDelete, EcstoreVersioningApi, HTTPPreconditions, HTTPRangeSpec, ObjectIO, ObjectOperations, ObjectToDelete,
ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule,
ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config,
ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache, ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache,
@@ -363,8 +365,46 @@ pub(crate) fn resolve_scanner_server_config() -> Option<ServerConfig> {
config_get_global_server_config() config_get_global_server_config()
} }
pub(crate) async fn list_runtime_tiers() -> Vec<EcstoreTierConfig> { /// How long the scanner caches the runtime tier-name list before re-reading
ecstore_get_global_tier_config_mgr().read().await.list_tiers() /// the tier configuration manager.
const TIER_NAME_CACHE_TTL: Duration = Duration::from_secs(30);
/// Process-wide TTL cache of runtime tier names.
///
/// The scan hot path only needs tier *names* to seed `SizeSummary::tier_stats`
/// per object, but every `list_tiers()` call clones each full `TierConfig`
/// (endpoints, credentials, prefixes) from the global manager. Caching just
/// the names keeps the per-object cost at an `Arc` clone.
///
/// Staleness bounds: a newly added tier starts showing up in scans at most
/// `TIER_NAME_CACHE_TTL` later; a removed tier can leave an all-zero
/// `TierStats` seed behind for one cache generation, which merges harmlessly
/// by key in per-object accounting and disappears on the next refresh.
static TIER_NAME_CACHE: RwLock<Option<(Instant, Arc<[String]>)>> = RwLock::new(None);
/// Tier names currently registered in the tier configuration, cached for
/// `TIER_NAME_CACHE_TTL`.
pub(crate) async fn runtime_tier_names() -> Arc<[String]> {
{
let cached = TIER_NAME_CACHE.read().unwrap_or_else(|err| err.into_inner()).clone();
if let Some((refreshed_at, names)) = cached
&& refreshed_at.elapsed() < TIER_NAME_CACHE_TTL
{
return names;
}
}
let tiers = ecstore_get_global_tier_config_mgr().read().await.list_tiers();
let names: Arc<[String]> = tiers.iter().map(|tier| tier.name.clone()).collect::<Vec<_>>().into();
*TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = Some((Instant::now(), Arc::clone(&names)));
names
}
/// Test-only cache reset; the production cache has no invalidation hook
/// because the TTL is its only refresh path.
#[cfg(test)]
fn reset_tier_name_cache_for_test() {
*TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = None;
} }
pub(crate) async fn enqueue_runtime_free_version(oi: ScannerObjectInfo) { pub(crate) async fn enqueue_runtime_free_version(oi: ScannerObjectInfo) {
@@ -561,6 +601,20 @@ mod tests {
use super::*; use super::*;
use serial_test::serial; use serial_test::serial;
#[tokio::test]
#[serial]
async fn runtime_tier_names_serves_cached_arc_within_ttl() {
reset_tier_name_cache_for_test();
// The tier config manager is unconfigured in unit tests, so the
// first call populates the cache from an empty tier list...
let first = runtime_tier_names().await;
assert!(first.is_empty());
// ...and a second call within the TTL must return the cached Arc
// (pointer-equal) without re-reading the manager.
let second = runtime_tier_names().await;
assert!(Arc::ptr_eq(&first, &second));
}
#[test] #[test]
#[serial] #[serial]
fn foreground_read_guard_tracks_stream_lifetime() { fn foreground_read_guard_tracks_stream_lifetime() {
+13 -20
View File
@@ -45,7 +45,7 @@ use rustfs_common::metrics::{
use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit, trace_subscriber_count}; use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit, trace_subscriber_count};
use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams}; use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration}; use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, VersioningConfiguration};
use time::OffsetDateTime; use time::OffsetDateTime;
use tokio::select; use tokio::select;
use tokio::sync::mpsc; use tokio::sync::mpsc;
@@ -53,10 +53,10 @@ use tokio_util::sync::CancellationToken;
use tracing::{debug, error, warn}; use tracing::{debug, error, warn};
use crate::{ use crate::{
BucketVersioningSys, Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts, Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts, ReplicationConfig,
ReplicationConfig, ReplicationHealObject, ReplicationQueueAdmission, ReplicationStatusType, STORAGE_FORMAT_FILE, ReplicationHealObject, ReplicationQueueAdmission, ReplicationStatusType, STORAGE_FORMAT_FILE, ScannerDiskExt as _,
ScannerDiskExt as _, ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule,
apply_transition_rule, enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object, enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object,
path2_bucket_object_with_base_path, queue_replication_heal, scanner_is_erasure, path2_bucket_object_with_base_path, queue_replication_heal, scanner_is_erasure,
scanner_replication_config_for_lifecycle_eval, scanner_replication_config_for_lifecycle_eval,
}; };
@@ -934,6 +934,7 @@ impl ScannerItem {
&mut self, &mut self,
object_infos: Vec<ObjectInfo>, object_infos: Vec<ObjectInfo>,
lock_retention: Option<Arc<ObjectLockConfiguration>>, lock_retention: Option<Arc<ObjectLockConfiguration>>,
versioning_config: VersioningConfiguration,
size_summary: &mut SizeSummary, size_summary: &mut SizeSummary,
) { ) {
if object_infos.is_empty() { if object_infos.is_empty() {
@@ -958,21 +959,8 @@ impl ScannerItem {
"Scanner lifecycle evaluation started" "Scanner lifecycle evaluation started"
); );
let versioning_config = match BucketVersioningSys::get(&self.bucket).await { // `versioning_config` is resolved once per object by the caller
Ok(versioning_config) => versioning_config, // (`get_size`) and handed in; only `prefix_enabled` is consulted here.
Err(_) => {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
state = "versioning_lookup_failed_defaulting",
"Scanner lifecycle action falling back to default bucket versioning"
);
Default::default()
}
};
let Some(lifecycle) = self.lifecycle.as_ref() else { let Some(lifecycle) = self.lifecycle.as_ref() else {
let mut cumulative_size = 0; let mut cumulative_size = 0;
@@ -1402,6 +1390,11 @@ impl ScannerItem {
fn alert_excessive_versions(&self, remaining_versions: usize, cumulative_size: i64) { fn alert_excessive_versions(&self, remaining_versions: usize, cumulative_size: i64) {
ensure_scanner_alert_metrics_registered(); ensure_scanner_alert_metrics_registered();
let (too_many_versions, too_large_versions) = should_alert_excessive_versions(remaining_versions, cumulative_size); let (too_many_versions, too_large_versions) = should_alert_excessive_versions(remaining_versions, cumulative_size);
// Threshold check first so healthy objects never pay for the
// object-path allocation below.
if !too_many_versions && !too_large_versions {
return;
}
let object_path = self.object_path(); let object_path = self.object_path();
if too_many_versions { if too_many_versions {
global_metrics().record_scanner_source_executed(ScannerWorkSource::Alerts, 1); global_metrics().record_scanner_source_executed(ScannerWorkSource::Alerts, 1);
+70 -20
View File
@@ -32,7 +32,9 @@ use rustfs_data_usage::{BucketTargetUsageInfo, BucketUsageInfo};
use rustfs_filemeta::FileMeta; use rustfs_filemeta::FileMeta;
use rustfs_lock::{LockError, NamespaceLockGuard}; use rustfs_lock::{LockError, NamespaceLockGuard};
use rustfs_utils::path::path_join_buf; use rustfs_utils::path::path_join_buf;
use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration}; use s3s::dto::{
BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration, VersioningConfiguration,
};
use sha2::{Digest as _, Sha256}; use sha2::{Digest as _, Sha256};
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use std::future::Future; use std::future::Future;
@@ -55,7 +57,7 @@ use crate::{
BucketTargetSys, BucketVersioningSys, Disk, DiskError, ECStore, EcstoreError as Error, EcstoreResult as Result, BucketTargetSys, BucketVersioningSys, Disk, DiskError, ECStore, EcstoreError as Error, EcstoreResult as Result,
RUSTFS_META_BUCKET, ReplicationConfig, STORAGE_FORMAT_FILE, ScannerDiskExt as _, ScannerLifecycleConfigExt as _, RUSTFS_META_BUCKET, ReplicationConfig, STORAGE_FORMAT_FILE, ScannerDiskExt as _, ScannerLifecycleConfigExt as _,
ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, enqueue_runtime_free_version, ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, enqueue_runtime_free_version,
get_lifecycle_config, get_object_lock_config, get_replication_config, list_runtime_tiers, storageclass, get_lifecycle_config, get_object_lock_config, get_replication_config, runtime_tier_names, storageclass,
}; };
pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file"; pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file";
@@ -63,6 +65,11 @@ pub(crate) const SCANNER_METADATA_CORRUPT_ERROR: &str = "scanner metadata corrup
pub(crate) const SCANNER_METADATA_TRANSIENT_ERROR: &str = "scanner metadata transient"; pub(crate) const SCANNER_METADATA_TRANSIENT_ERROR: &str = "scanner metadata transient";
const LOG_COMPONENT_SCANNER: &str = "scanner"; const LOG_COMPONENT_SCANNER: &str = "scanner";
const LOG_SUBSYSTEM_IO: &str = "io"; const LOG_SUBSYSTEM_IO: &str = "io";
// Mirrors `scanner_folder.rs` so the versioning-lookup fallback warn keeps its
// historical `rustfs::scanner::folder` lifecycle event identity after the
// lookup moved into `get_size`.
const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle";
const EVENT_SCANNER_LIFECYCLE_ACTION: &str = "scanner_lifecycle_action";
const EVENT_SCANNER_DISK_BUCKET_STATE: &str = "scanner_disk_bucket_state"; const EVENT_SCANNER_DISK_BUCKET_STATE: &str = "scanner_disk_bucket_state";
const EVENT_SCANNER_DATA_USAGE_STREAM: &str = "scanner_data_usage_stream"; const EVENT_SCANNER_DATA_USAGE_STREAM: &str = "scanner_data_usage_stream";
const EVENT_SCANNER_CACHE_PERSIST_STATE: &str = "scanner_cache_persist_state"; const EVENT_SCANNER_CACHE_PERSIST_STATE: &str = "scanner_cache_persist_state";
@@ -3822,6 +3829,24 @@ impl ScannerIOCache for SetDisks {
} }
} }
/// Seed [`SizeSummary::tier_stats`] from the cached tier-name list.
///
/// Preserves the original seeding semantics: with no tiers configured the map
/// stays completely empty (STANDARD/RRS are not seeded either); otherwise the
/// standard storage classes are seeded alongside every configured tier so
/// per-object accounting always finds its tier key.
fn tier_stats_template(tier_names: &[String]) -> HashMap<String, TierStats> {
let mut tier_stats = HashMap::with_capacity(tier_names.len() + 2);
for tier_name in tier_names {
tier_stats.insert(tier_name.clone(), TierStats::default());
}
if !tier_stats.is_empty() {
tier_stats.insert(storageclass::STANDARD.to_string(), TierStats::default());
tier_stats.insert(storageclass::RRS.to_string(), TierStats::default());
}
tier_stats
}
#[async_trait::async_trait] #[async_trait::async_trait]
impl ScannerIODisk for Disk { impl ScannerIODisk for Disk {
async fn get_size(&self, mut item: ScannerItem) -> Result<SizeSummary> { async fn get_size(&self, mut item: ScannerItem) -> Result<SizeSummary> {
@@ -3861,10 +3886,26 @@ impl ScannerIODisk for Disk {
} }
}; };
let versioned = BucketVersioningSys::get(&item.bucket) // Single versioning lookup per object, shared with `apply_actions`
.await // (which used to query it a second time). On failure keep the
.map(|v| v.versioned(&item.object_path())) // historical fallback: default configuration (versioned = false) plus
.unwrap_or(false); // the warn that `apply_actions` used to emit.
let versioning_config = match BucketVersioningSys::get(&item.bucket).await {
Ok(versioning_config) => versioning_config,
Err(_) => {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %item.bucket,
state = "versioning_lookup_failed_defaulting",
"Scanner lifecycle action falling back to default bucket versioning"
);
VersioningConfiguration::default()
}
};
let versioned = versioning_config.versioned(&item.object_path());
let object_infos = fivs let object_infos = fivs
.versions .versions
@@ -3879,19 +3920,10 @@ impl ScannerIODisk for Disk {
let mut size_summary = SizeSummary::default(); let mut size_summary = SizeSummary::default();
let tiers = list_runtime_tiers().await; // Tier names come from the process-wide TTL cache; seeding from them
// replaces the per-object clone of every full TierConfig.
for tier in tiers.iter() { let tier_names = runtime_tier_names().await;
size_summary.tier_stats.insert(tier.name.clone(), TierStats::default()); size_summary.tier_stats = tier_stats_template(&tier_names);
}
if !size_summary.tier_stats.is_empty() {
size_summary
.tier_stats
.insert(storageclass::STANDARD.to_string(), TierStats::default());
size_summary
.tier_stats
.insert(storageclass::RRS.to_string(), TierStats::default());
}
let lock_config = object_lock_config_for_scanner_item(&item).await; let lock_config = object_lock_config_for_scanner_item(&item).await;
@@ -3901,7 +3933,8 @@ impl ScannerIODisk for Disk {
// `object_infos`. // `object_infos`.
global_metrics().record_scanner_versions_scanned(object_infos.len() as u64); global_metrics().record_scanner_versions_scanned(object_infos.len() as u64);
item.apply_actions(object_infos, lock_config, &mut size_summary).await; item.apply_actions(object_infos, lock_config, versioning_config, &mut size_summary)
.await;
if !free_version_infos.is_empty() { if !free_version_infos.is_empty() {
for oi in free_version_infos { for oi in free_version_infos {
@@ -4968,6 +5001,23 @@ mod tests {
assert!(is_xl_meta_path("/data/bucket/object/xl.meta")); assert!(is_xl_meta_path("/data/bucket/object/xl.meta"));
} }
#[test]
fn tier_stats_template_seeds_tiers_and_standard_classes() {
let template = tier_stats_template(&["WARM".to_string(), "COLD".to_string()]);
assert_eq!(template.len(), 4);
for tier in ["WARM", "COLD", storageclass::STANDARD, storageclass::RRS] {
assert_eq!(template.get(tier), Some(&TierStats::default()), "missing seed for tier {tier}");
}
}
#[test]
fn tier_stats_template_stays_empty_without_tiers() {
let template = tier_stats_template(&[]);
assert!(template.is_empty());
}
#[tokio::test] #[tokio::test]
async fn get_size_treats_missing_metadata_as_skip_file() { async fn get_size_treats_missing_metadata_as_skip_file() {
let temp_dir = std::env::temp_dir().join(format!("rustfs-scanner-missing-meta-{}", Uuid::new_v4())); let temp_dir = std::env::temp_dir().join(format!("rustfs-scanner-missing-meta-{}", Uuid::new_v4()));
+9 -10
View File
@@ -99,7 +99,6 @@ pub(crate) use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks;
pub(crate) use rustfs_ecstore::api::storage::ECStore as EcstoreStore; pub(crate) use rustfs_ecstore::api::storage::ECStore as EcstoreStore;
#[cfg(test)] #[cfg(test)]
pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx; pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx;
pub(crate) use rustfs_ecstore::api::tier::tier_config::TierConfig as EcstoreTierConfig;
use rustfs_storage_api as storage_contracts; use rustfs_storage_api as storage_contracts;
pub(crate) mod owner { pub(crate) mod owner {
@@ -114,15 +113,15 @@ pub(crate) mod owner {
EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreEventArgs, EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreEventArgs,
EcstoreLcEventSrc, EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts, EcstoreLcEventSrc, EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts,
EcstoreReplicationConfigurationExt, EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreReplicationConfigurationExt, EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard,
EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreTierConfig, EcstoreVersioningApi, EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreVersioningApi, ScannerReplicationHealObject,
ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ecstore_apply_transition_rule,
ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config,
ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache,
ecstore_invalidate_admin_data_usage_snapshot_cache, ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_is_erasure_sd,
ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info,
ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, ecstore_read_config,
ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config,
ecstore_save_config, ecstore_send_event, scanner_replication_config_for_lifecycle_eval, ecstore_send_event, scanner_replication_config_for_lifecycle_eval,
}; };
#[cfg(test)] #[cfg(test)]