Skip to main content

common/
lib.rs

1//! Common utilities and types for RCP file operation tools
2//!
3//! This crate provides shared functionality used across all RCP tools (`rcp`, `rrm`, `rlink`, `rcmp`).
4//! It includes core operations (copy, remove, link, compare), progress reporting, metadata preservation, and runtime configuration.
5//!
6//! # Core Modules
7//!
8//! - [`mod@copy`] - File copying operations with metadata preservation and error handling
9//! - [`mod@rm`] - File removal operations
10//! - [`mod@link`] - Hard-linking operations
11//! - [`mod@cmp`] - File comparison operations (metadata-based)
12//! - [`mod@preserve`] - Metadata preservation settings and operations
13//! - [`mod@progress`] - Progress tracking and reporting
14//! - [`mod@filecmp`] - File metadata comparison utilities
15//! - [`mod@remote_tracing`] - Remote tracing support for distributed operations
16//!
17//! # Key Types
18//!
19//! ## `RcpdType`
20//!
21//! Identifies the role of a remote copy daemon:
22//! - `Source` - reads files from source host
23//! - `Destination` - writes files to destination host
24//!
25//! ## `ProgressType`
26//!
27//! Controls progress reporting display:
28//! - `Auto` - automatically choose based on terminal type
29//! - `ProgressBar` - animated progress bar (for interactive terminals)
30//! - `TextUpdates` - periodic text updates (for logging/non-interactive)
31//!
32//! # Progress Reporting
33//!
34//! The crate provides a global progress tracking system accessible via [`get_progress()`].
35//! Progress can be displayed in different formats depending on the execution context.
36//!
37//! Progress output goes to stderr, while logs go to stdout, allowing users to redirect logs to a file while still viewing interactive progress.
38//!
39//! # Runtime Configuration
40//!
41//! The [`run`] function provides a unified entry point for all RCP tools with support for:
42//! - Progress tracking and reporting
43//! - Logging configuration (quiet/verbose modes)
44//! - Resource limits (max workers, open files, throttling)
45//! - Tokio runtime setup
46//! - Remote tracing integration
47//!
48//! # Examples
49//!
50//! ## Basic Copy Operation
51//!
52//! ```rust,no_run
53//! use std::path::Path;
54//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
55//! let src = Path::new("/source");
56//! let dst = Path::new("/destination");
57//!
58//! let settings = common::copy::Settings {
59//!     dereference: false,
60//!     fail_early: false,
61//!     overwrite: false,
62//!     overwrite_compare: Default::default(),
63//!     overwrite_filter: None,
64//!     ignore_existing: false,
65//!     chunk_size: 0,
66//!     skip_specials: false,
67//!     remote_copy_buffer_size: 0,
68//!     filter: None,
69//!     dry_run: None,
70//!     delete: None,
71//! };
72//! let preserve = common::preserve::preserve_none();
73//!
74//! let summary = common::copy(src, dst, &settings, &preserve).await?;
75//! println!("Copied {} files", summary.files_copied);
76//! # Ok(())
77//! # }
78//! ```
79//!
80//! ## Metadata Comparison
81//!
82//! ```rust,no_run
83//! use std::path::Path;
84//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
85//! let src = Path::new("/path1");
86//! let dst = Path::new("/path2");
87//!
88//! // output differences to stdout (use false for quiet mode)
89//! let log = common::cmp::LogWriter::new(None, true, common::cmp::OutputFormat::default()).await?;
90//! let settings = common::cmp::Settings {
91//!     fail_early: false,
92//!     exit_early: false,
93//!     expand_missing: false,
94//!     compare: Default::default(),
95//!     filter: None,
96//! };
97//!
98//! let summary = common::cmp(src, dst, &log, &settings).await?;
99//! println!("Comparison complete: {}", summary);
100//! # Ok(())
101//! # }
102//! ```
103
104use anyhow::Context;
105use std::io::IsTerminal;
106use tracing::instrument;
107
108mod auto_meta;
109pub mod chmod;
110pub mod cli;
111pub mod cmp;
112pub mod config;
113pub mod copy;
114pub mod copy_data;
115pub mod delete;
116pub mod dry_run;
117pub mod error;
118pub mod error_collector;
119pub mod filegen;
120pub mod filter;
121pub mod histogram_logger;
122pub mod histogram_panel;
123pub mod link;
124pub mod observability;
125pub mod preserve;
126pub mod remote_tracing;
127pub mod rm;
128mod runtime_setup;
129pub mod safedir;
130mod settings_parse;
131pub mod version;
132
133pub mod filecmp;
134pub mod progress;
135mod testutils;
136pub mod toctou_check;
137pub mod walk;
138pub mod walk_driver;
139
140pub use config::{
141    AutoMetaThrottleConfig, DryRunMode, DryRunWarnings, OutputConfig, RuntimeConfig,
142    ThrottleConfig, TracingConfig,
143};
144// Re-export `Side` from the congestion crate so downstream binaries
145// (rcp, rrm, …) and integration tests can pass `common::Side::Source` /
146// `common::Side::Destination` to `walk::next_entry_probed` and friends
147// without taking a direct dependency on `congestion`.
148pub use congestion::{MetadataOp, Side};
149pub use progress::{RcpdProgressPrinter, SerializableProgress};
150// Re-export the runtime-stat / trace-filename helpers that moved into
151// `runtime_setup` so downstream binaries keep reaching them as
152// `common::collect_runtime_stats`, etc.
153pub use runtime_setup::{
154    collect_runtime_stats, generate_debug_log_filename, generate_trace_filename,
155};
156pub use settings_parse::{
157    parse_compare_settings, parse_metadata_cmp_settings, parse_preserve_settings,
158    validate_update_compare_vs_preserve,
159};
160
161// Define RcpdType in common since remote depends on common
162#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, enum_map::Enum)]
163pub enum RcpdType {
164    Source,
165    Destination,
166}
167
168impl std::fmt::Display for RcpdType {
169    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
170        match self {
171            RcpdType::Source => write!(f, "source"),
172            RcpdType::Destination => write!(f, "destination"),
173        }
174    }
175}
176
177// Type alias for progress snapshots
178pub type ProgressSnapshot<T> = enum_map::EnumMap<RcpdType, T>;
179
180/// runtime statistics collected from a process (CPU time, memory usage)
181#[derive(Clone, Debug, Default, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
182pub struct RuntimeStats {
183    /// user-mode CPU time in milliseconds
184    pub cpu_time_user_ms: u64,
185    /// kernel-mode CPU time in milliseconds
186    pub cpu_time_kernel_ms: u64,
187    /// peak resident set size in bytes
188    pub peak_rss_bytes: u64,
189}
190
191/// runtime stats collected from remote rcpd processes for display at the end of a remote copy
192#[derive(Debug, Default)]
193pub struct RemoteRuntimeStats {
194    pub source_host: String,
195    pub source_stats: RuntimeStats,
196    pub dest_host: String,
197    pub dest_stats: RuntimeStats,
198}
199
200/// checks if a host string refers to the local machine.
201/// returns true for `localhost`, `127.0.0.1`, `::1`, `[::1]`, or the actual hostname
202#[must_use]
203pub fn is_localhost(host: &str) -> bool {
204    if host == "localhost" || host == "127.0.0.1" || host == "::1" || host == "[::1]" {
205        return true;
206    }
207    // check against actual hostname using gethostname
208    let mut buf = [0u8; 256];
209    // Safety: gethostname writes to buf and returns 0 on success
210    let result = unsafe { libc::gethostname(buf.as_mut_ptr() as *mut libc::c_char, buf.len()) };
211    if result == 0
212        && let Ok(hostname_cstr) = std::ffi::CStr::from_bytes_until_nul(&buf)
213        && let Ok(hostname) = hostname_cstr.to_str()
214        && host == hostname
215    {
216        return true;
217    }
218    false
219}
220
221pub(crate) static PROGRESS: std::sync::LazyLock<progress::Progress> =
222    std::sync::LazyLock::new(progress::Progress::new);
223pub(crate) static PBAR: std::sync::LazyLock<indicatif::ProgressBar> =
224    std::sync::LazyLock::new(indicatif::ProgressBar::new_spinner);
225pub(crate) static REMOTE_RUNTIME_STATS: std::sync::LazyLock<
226    std::sync::Mutex<Option<RemoteRuntimeStats>>,
227> = std::sync::LazyLock::new(|| std::sync::Mutex::new(None));
228static HISTOGRAM_LOGGER_CANCEL: std::sync::Mutex<Option<tokio::sync::watch::Sender<bool>>> =
229    std::sync::Mutex::new(None);
230static HISTOGRAM_LOGGER_HANDLE: std::sync::Mutex<Option<tokio::task::JoinHandle<()>>> =
231    std::sync::Mutex::new(None);
232
233pub(crate) fn store_logger_cancel(tx: tokio::sync::watch::Sender<bool>) {
234    *HISTOGRAM_LOGGER_CANCEL
235        .lock()
236        .expect("histogram logger cancel mutex poisoned") = Some(tx);
237}
238
239pub(crate) fn store_logger_handle(handle: tokio::task::JoinHandle<()>) {
240    *HISTOGRAM_LOGGER_HANDLE
241        .lock()
242        .expect("histogram logger handle mutex poisoned") = Some(handle);
243}
244
245fn take_logger_handle() -> Option<tokio::task::JoinHandle<()>> {
246    HISTOGRAM_LOGGER_HANDLE
247        .lock()
248        .expect("histogram logger handle mutex poisoned")
249        .take()
250}
251
252fn signal_logger_cancel() {
253    if let Some(tx) = HISTOGRAM_LOGGER_CANCEL
254        .lock()
255        .expect("histogram logger cancel mutex poisoned")
256        .take()
257        && let Err(err) = tx.send(true)
258    {
259        tracing::debug!("histogram-logger cancel send failed (already gone): {err:#}");
260    }
261}
262
263#[must_use]
264pub fn get_progress() -> &'static progress::Progress {
265    &PROGRESS
266}
267
268/// stores remote runtime stats for display at the end of a remote copy operation
269pub fn set_remote_runtime_stats(stats: RemoteRuntimeStats) {
270    *REMOTE_RUNTIME_STATS.lock().unwrap() = Some(stats);
271}
272
273struct ProgressTracker {
274    lock_cvar: std::sync::Arc<(std::sync::Mutex<bool>, std::sync::Condvar)>,
275    pbar_thread: Option<std::thread::JoinHandle<()>>,
276}
277
278#[derive(Copy, Clone, Debug, Default, clap::ValueEnum)]
279pub enum ProgressType {
280    #[default]
281    #[value(name = "auto", alias = "Auto")]
282    Auto,
283    #[value(name = "ProgressBar", alias = "progress-bar")]
284    ProgressBar,
285    #[value(name = "TextUpdates", alias = "text-updates")]
286    TextUpdates,
287}
288
289pub enum GeneralProgressType {
290    User {
291        progress_type: ProgressType,
292        kind: progress::LocalProgressKind,
293    },
294    Remote(tokio::sync::mpsc::UnboundedSender<remote_tracing::TracingMessage>),
295    RemoteMaster {
296        progress_type: ProgressType,
297        get_progress_snapshot:
298            Box<dyn Fn() -> ProgressSnapshot<SerializableProgress> + Send + 'static>,
299    },
300}
301
302impl std::fmt::Debug for GeneralProgressType {
303    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
304        match self {
305            GeneralProgressType::User {
306                progress_type,
307                kind,
308            } => write!(f, "User(progress_type: {progress_type:?}, kind: {kind:?})"),
309            GeneralProgressType::Remote(_) => write!(f, "Remote(<sender>)"),
310            GeneralProgressType::RemoteMaster { progress_type, .. } => {
311                write!(
312                    f,
313                    "RemoteMaster(progress_type: {progress_type:?}, <function>)"
314                )
315            }
316        }
317    }
318}
319
320#[derive(Debug)]
321pub struct ProgressSettings {
322    pub progress_type: GeneralProgressType,
323    pub progress_delay: Option<String>,
324}
325
326fn progress_bar(
327    lock: &std::sync::Mutex<bool>,
328    cvar: &std::sync::Condvar,
329    delay_opt: &Option<std::time::Duration>,
330    kind: progress::LocalProgressKind,
331) {
332    let delay = delay_opt.unwrap_or(std::time::Duration::from_millis(200));
333    PBAR.set_style(
334        indicatif::ProgressStyle::with_template("{spinner:.cyan} {msg}")
335            .unwrap()
336            .tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]),
337    );
338    let mut prog_printer = progress::make_local_printer(kind, &PROGRESS);
339    let mut is_done = lock.lock().unwrap();
340    loop {
341        PBAR.set_position(PBAR.position() + 1); // do we need to update?
342        let mut msg = prog_printer.print().unwrap();
343        msg.push_str(&observability::render_lines());
344        msg.push_str(&render_panel_from_registry());
345        PBAR.set_message(msg);
346        let result = cvar.wait_timeout(is_done, delay).unwrap();
347        is_done = result.0;
348        if *is_done {
349            break;
350        }
351    }
352    PBAR.finish_and_clear();
353}
354
355fn get_datetime_prefix() -> String {
356    chrono::Local::now()
357        .format("%Y-%m-%dT%H:%M:%S%.3f%:z")
358        .to_string()
359}
360
361fn text_updates(
362    lock: &std::sync::Mutex<bool>,
363    cvar: &std::sync::Condvar,
364    delay_opt: &Option<std::time::Duration>,
365    kind: progress::LocalProgressKind,
366) {
367    let delay = delay_opt.unwrap_or(std::time::Duration::from_secs(10));
368    let mut prog_printer = progress::make_local_printer(kind, &PROGRESS);
369    let mut is_done = lock.lock().unwrap();
370    loop {
371        eprintln!("=======================");
372        eprintln!(
373            "{}\n--{}{}{}",
374            get_datetime_prefix(),
375            prog_printer.print().unwrap(),
376            observability::render_lines(),
377            render_panel_from_registry(),
378        );
379        let result = cvar.wait_timeout(is_done, delay).unwrap();
380        is_done = result.0;
381        if *is_done {
382            break;
383        }
384    }
385}
386
387fn rcpd_updates(
388    lock: &std::sync::Mutex<bool>,
389    cvar: &std::sync::Condvar,
390    delay_opt: &Option<std::time::Duration>,
391    sender: tokio::sync::mpsc::UnboundedSender<remote_tracing::TracingMessage>,
392) {
393    tracing::debug!("Starting rcpd progress updates");
394    let delay = delay_opt.unwrap_or(std::time::Duration::from_millis(200));
395    let mut is_done = lock.lock().unwrap();
396    loop {
397        if remote_tracing::send_progress_update(&sender, &PROGRESS).is_err() {
398            // channel closed, receiver is done
399            tracing::debug!("Progress update channel closed, stopping progress updates");
400            break;
401        }
402        let result = cvar.wait_timeout(is_done, delay).unwrap();
403        is_done = result.0;
404        if *is_done {
405            break;
406        }
407    }
408}
409
410fn remote_master_updates<F>(
411    lock: &std::sync::Mutex<bool>,
412    cvar: &std::sync::Condvar,
413    delay_opt: &Option<std::time::Duration>,
414    get_progress_snapshot: F,
415    progress_type: ProgressType,
416) where
417    F: Fn() -> ProgressSnapshot<SerializableProgress> + Send + 'static,
418{
419    let interactive = match progress_type {
420        ProgressType::Auto => std::io::stderr().is_terminal(),
421        ProgressType::ProgressBar => true,
422        ProgressType::TextUpdates => false,
423    };
424    if interactive {
425        PBAR.set_style(
426            indicatif::ProgressStyle::with_template("{spinner:.cyan} {msg}")
427                .unwrap()
428                .tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]),
429        );
430        let delay = delay_opt.unwrap_or(std::time::Duration::from_millis(200));
431        let mut printer = RcpdProgressPrinter::new();
432        let mut is_done = lock.lock().unwrap();
433        loop {
434            let progress_map = get_progress_snapshot();
435            let source_progress = &progress_map[RcpdType::Source];
436            let destination_progress = &progress_map[RcpdType::Destination];
437            PBAR.set_position(PBAR.position() + 1); // do we need to update?
438            let mut msg = printer
439                .print(source_progress, destination_progress)
440                .unwrap();
441            msg.push_str(&render_panel_from_registry());
442            PBAR.set_message(msg);
443            let result = cvar.wait_timeout(is_done, delay).unwrap();
444            is_done = result.0;
445            if *is_done {
446                break;
447            }
448        }
449        PBAR.finish_and_clear();
450    } else {
451        let delay = delay_opt.unwrap_or(std::time::Duration::from_secs(10));
452        let mut printer = RcpdProgressPrinter::new();
453        let mut is_done = lock.lock().unwrap();
454        loop {
455            let progress_map = get_progress_snapshot();
456            let source_progress = &progress_map[RcpdType::Source];
457            let destination_progress = &progress_map[RcpdType::Destination];
458            eprintln!("=======================");
459            eprintln!(
460                "{}\n--{}{}",
461                get_datetime_prefix(),
462                printer
463                    .print(source_progress, destination_progress)
464                    .unwrap(),
465                render_panel_from_registry(),
466            );
467            let result = cvar.wait_timeout(is_done, delay).unwrap();
468            is_done = result.0;
469            if *is_done {
470                break;
471            }
472        }
473    }
474}
475
476impl ProgressTracker {
477    pub fn new(progress_type: GeneralProgressType, delay_opt: Option<std::time::Duration>) -> Self {
478        let lock_cvar =
479            std::sync::Arc::new((std::sync::Mutex::new(false), std::sync::Condvar::new()));
480        let lock_cvar_clone = lock_cvar.clone();
481        let pbar_thread = std::thread::spawn(move || {
482            let (lock, cvar) = &*lock_cvar_clone;
483            match progress_type {
484                GeneralProgressType::Remote(sender) => {
485                    rcpd_updates(lock, cvar, &delay_opt, sender);
486                }
487                GeneralProgressType::RemoteMaster {
488                    progress_type,
489                    get_progress_snapshot,
490                } => {
491                    remote_master_updates(
492                        lock,
493                        cvar,
494                        &delay_opt,
495                        get_progress_snapshot,
496                        progress_type,
497                    );
498                }
499                GeneralProgressType::User {
500                    progress_type,
501                    kind,
502                } => {
503                    let interactive = match progress_type {
504                        ProgressType::Auto => std::io::stderr().is_terminal(),
505                        ProgressType::ProgressBar => true,
506                        ProgressType::TextUpdates => false,
507                    };
508                    if interactive {
509                        progress_bar(lock, cvar, &delay_opt, kind);
510                    } else {
511                        text_updates(lock, cvar, &delay_opt, kind);
512                    }
513                }
514            }
515        });
516        Self {
517            lock_cvar,
518            pbar_thread: Some(pbar_thread),
519        }
520    }
521}
522
523impl Drop for ProgressTracker {
524    fn drop(&mut self) {
525        let (lock, cvar) = &*self.lock_cvar;
526        let mut is_done = lock.lock().unwrap();
527        *is_done = true;
528        cvar.notify_one();
529        drop(is_done);
530        if let Some(pbar_thread) = self.pbar_thread.take() {
531            pbar_thread.join().unwrap();
532        }
533    }
534}
535
536pub async fn cmp(
537    src: &std::path::Path,
538    dst: &std::path::Path,
539    log: &cmp::LogWriter,
540    settings: &cmp::Settings,
541) -> Result<cmp::Summary, anyhow::Error> {
542    cmp::cmp(&PROGRESS, src, dst, log, settings).await
543}
544
545pub async fn copy(
546    src: &std::path::Path,
547    dst: &std::path::Path,
548    settings: &copy::Settings,
549    preserve: &preserve::Settings,
550) -> Result<copy::Summary, copy::Error> {
551    copy::copy(&PROGRESS, src, dst, settings, preserve, false).await
552}
553
554pub async fn rm(path: &std::path::Path, settings: &rm::Settings) -> Result<rm::Summary, rm::Error> {
555    rm::rm(&PROGRESS, path, settings).await
556}
557
558pub async fn chmod(
559    path: &std::path::Path,
560    settings: &chmod::Settings,
561) -> Result<chmod::Summary, chmod::Error> {
562    chmod::chmod(&PROGRESS, path, settings).await
563}
564
565pub async fn link(
566    src: &std::path::Path,
567    dst: &std::path::Path,
568    update: &Option<std::path::PathBuf>,
569    settings: &link::Settings,
570) -> Result<link::Summary, link::Error> {
571    let cwd = std::env::current_dir()
572        .with_context(|| "failed to get current working directory")
573        .map_err(|err| link::Error::new(err, link::Summary::default()))?;
574    link::link(&PROGRESS, &cwd, src, dst, update, settings, false).await
575}
576
577fn render_panel_from_registry() -> String {
578    let entries = observability::registered_histograms();
579    if entries.is_empty() {
580        return String::new();
581    }
582    let snapshots: Vec<hdrhistogram::Histogram<u64>> = entries
583        .iter()
584        .map(|e| (*e.snapshot_rx.borrow()).clone())
585        .collect();
586    let units: Vec<histogram_panel::PanelUnit> = entries
587        .iter()
588        .zip(snapshots.iter())
589        .map(|(e, snap)| histogram_panel::PanelUnit {
590            label: e.label,
591            histogram: snap,
592            interval: e.interval,
593        })
594        .collect();
595    histogram_panel::render_histogram_panel(&units)
596}
597
598#[instrument(skip(func))] // "func" is not Debug printable
599pub fn run<Fut, Summary, Error>(
600    progress: Option<ProgressSettings>,
601    output: OutputConfig,
602    runtime_config: RuntimeConfig,
603    throttle_config: ThrottleConfig,
604    tracing_config: TracingConfig,
605    func: impl FnOnce() -> Fut,
606) -> Option<Summary>
607// we return an Option rather than a Result to indicate that callers of this function should NOT print the error
608where
609    Summary: std::fmt::Display,
610    Error: std::fmt::Display + std::fmt::Debug,
611    Fut: std::future::Future<Output = Result<Summary, Error>>,
612{
613    // force initialization of PROGRESS to set start_time at the beginning of the run
614    // (for remote master operations, PROGRESS is otherwise only accessed at the end in
615    // print_runtime_stats(), leading to near-zero walltime)
616    let _ = get_progress();
617    if let Err(e) = throttle_config.validate() {
618        eprintln!("Configuration error: {e}");
619        return None;
620    }
621    let OutputConfig {
622        quiet,
623        verbose,
624        print_summary,
625        suppress_runtime_stats,
626    } = output;
627    // tracing guards must outlive the runtime so chrome/flame traces flush
628    // extract trace_identifier before install_tracing_subscriber consumes tracing_config
629    let trace_identifier = tracing_config.trace_identifier.clone();
630    if let Err(e) =
631        runtime_setup::validate_histogram_log_target(&throttle_config, &trace_identifier)
632    {
633        eprintln!("Configuration error: {e}");
634        return None;
635    }
636    let _tracing_guards = runtime_setup::install_tracing_subscriber(quiet, verbose, tracing_config);
637    let res = {
638        let runtime = runtime_setup::build_tokio_runtime(&runtime_config, &throttle_config);
639        runtime_setup::spawn_throttle_replenishers(&runtime, &throttle_config, &trace_identifier);
640        let res = {
641            let _progress_tracker = progress.map(|settings| {
642                tracing::debug!("Requesting progress updates {settings:?}");
643                let delay = settings.progress_delay.map(|delay_str| {
644                    humantime::parse_duration(&delay_str)
645                        .expect("Couldn't parse duration out of --progress-delay")
646                });
647                ProgressTracker::new(settings.progress_type, delay)
648            });
649            runtime.block_on(func())
650        };
651        match &res {
652            Ok(summary) => {
653                if print_summary || verbose > 0 {
654                    println!("{summary}");
655                }
656            }
657            Err(err) => {
658                if !quiet {
659                    println!("{err:?}");
660                }
661            }
662        }
663        if (print_summary || verbose > 0)
664            && !suppress_runtime_stats
665            && let Err(err) = runtime_setup::print_runtime_stats()
666        {
667            println!("Failed to print runtime stats: {err:?}");
668        }
669        // Signal the histogram logger to exit cleanly so its final
670        // snapshot is written before the runtime drops and aborts it.
671        // No-op when histograms are disabled.
672        signal_logger_cancel();
673        if let Some(handle) = take_logger_handle() {
674            // Bound the wait so a stuck logger can't hang shutdown — 1s is
675            // generous: the logger only does a snapshot+flush on cancel.
676            let _ = runtime.block_on(async {
677                tokio::time::timeout(std::time::Duration::from_secs(1), handle).await
678            });
679        }
680        res
681        // runtime drops here, cancelling all spawned tasks (control
682        // loop, adapter, replenishers) and releasing their permits.
683    };
684    // Clear process-wide state so a second `run()` in the same process
685    // starts on a clean slate. Without this, a later run inherits the
686    // previous auto-meta sample sink and ops-in-flight cap, and its
687    // probes acquire against stale limits even when auto_meta is off.
688    reset_process_throttle_state();
689    res.ok()
690}
691
692/// Reset process-wide throttle + congestion state to its pre-`run()`
693/// defaults. Called by [`run`] on exit so callers that invoke it more
694/// than once in a single process (library users, integration tests
695/// outside this crate) aren't affected by the previous invocation's
696/// decisions.
697fn reset_process_throttle_state() {
698    congestion::clear_sample_sink();
699    observability::clear();
700    for &side in &throttle::Side::ALL {
701        for &op in &throttle::MetadataOp::ALL {
702            throttle::set_max_ops_in_flight(throttle::Resource::meta(side, op), 0);
703        }
704    }
705    throttle::disable_ops_throttle();
706    // Without these resets, a second run() in the same process inherits
707    // the previous run's open-files cap and iops-throttle even when the
708    // caller passes 0 ("no limit"): `set_max_open_files` / `init_iops_tokens`
709    // are skipped on 0, leaving the prior `setup(N)` in force. setup(0)
710    // disables the semaphore, so the next run sees a clean slate and can
711    // either re-init with a fresh value or stay disabled.
712    throttle::set_max_open_files(0);
713    throttle::init_iops_tokens(0);
714}