Added IO throttling for the merger pipeline.

quickwit/Cargo.toml

@@ -31,6 +31,7 @@ anyhow = "1"
 arc-swap = "1.4"
 assert-json-diff = "2"
 assert_cmd = "2"
+async-speed-limit = "0.4"
 async-trait = "0.1"
 atty = "0.2"
 azure_core = "0.5.0"
@@ -90,6 +91,7 @@ openssl-probe = "0.1.4"
 opentelemetry = { version = "0.18", features = ["rt-tokio"] }
 opentelemetry-jaeger = { version = "0.17", features = ["rt-tokio"] }
 opentelemetry-otlp = "0.11.0"
+pin-project-lite = "0.2.9"
 pnet = { version = "0.31.0", features = ["std"] }
 predicates = "2"
 prometheus = { version = "0.13", features = ["process"] }

quickwit/quickwit-actors/src/actor.rs

@@ -32,13 +32,14 @@ use tokio::sync::{oneshot, watch};
 use tracing::{debug, error};
 
 use crate::actor_state::{ActorState, AtomicState};
-use crate::progress::{Progress, ProtectedZoneGuard};
 use crate::registry::ActorRegistry;
 use crate::scheduler::{Callback, ScheduleEvent, Scheduler};
 use crate::spawn_builder::SpawnBuilder;
 #[cfg(any(test, feature = "testsuite"))]
 use crate::Universe;
-use crate::{AskError, Command, KillSwitch, Mailbox, QueueCapacity, SendError};
+use crate::{
+    AskError, Command, KillSwitch, Mailbox, Progress, ProtectedZoneGuard, QueueCapacity, SendError,
+};
 
 /// The actor exit status represents the outcome of the execution of an actor,
 /// after the end of the execution.

quickwit/quickwit-actors/src/lib.rs

@@ -36,10 +36,8 @@ mod actor_state;
 pub mod channel_with_priority;
 mod command;
 mod envelope;
-mod kill_switch;
 mod mailbox;
 mod observation;
-mod progress;
 mod registry;
 mod scheduler;
 mod spawn_builder;
@@ -52,9 +50,8 @@ mod universe;
 pub use actor::{Actor, ActorExitStatus, Handler};
 pub use actor_handle::{ActorHandle, Health, Supervisable};
 pub use command::Command;
-pub use kill_switch::KillSwitch;
 pub use observation::{Observation, ObservationType};
-pub use progress::{Progress, ProtectedZoneGuard};
+use quickwit_common::{KillSwitch, Progress, ProtectedZoneGuard};
 pub(crate) use scheduler::Scheduler;
 use thiserror::Error;
 pub use universe::Universe;

quickwit/quickwit-common/Cargo.toml

@@ -11,12 +11,14 @@ documentation = "https://quickwit.io/docs/"
 
 [dependencies]
 anyhow = { workspace = true }
+async-speed-limit = {workspace=true}
 colored = { workspace = true }
 env_logger = { workspace = true }
 home = { workspace = true }
 itertools = { workspace = true }
 num_cpus = { workspace = true }
 once_cell = { workspace = true }
+pin-project-lite = { workspace = true }
 pnet = { workspace = true }
 prometheus = { workspace = true }
 rand = { workspace = true }