chore: correct tokio::spawn usage

There was code of the form `tokio::spawn(async move { foo().await }`

While this works without issues, it's redundant, as `tokio::spawn`
needs a future as the argument. Therefore the new code now
has only `tokio::spawn(foo())`.

+

perform `cargo fmt`

src/common/infra/cluster/etcd.rs

@@ -76,7 +76,7 @@ async fn register() -> Result<()> {
     let locker = dist_lock::lock("/nodes/register", cfg.limit.node_heartbeat_ttl as u64).await?;
 
     // 2. watch node list
-    tokio::task::spawn(async move { super::watch_node_list().await });
+    tokio::task::spawn(super::watch_node_list());
 
     // 3. get node list
     let node_list = match super::list_nodes().await {

src/common/infra/config.rs

@@ -179,7 +179,7 @@ mod tests {
     #[tokio::test]
     async fn test_update_cache() {
         let (tx, rx) = mpsc::channel(100);
-        tokio::spawn(async move { update_cache(rx).await });
+        tokio::spawn(update_cache(rx));
         tx.send(infra::db::nats::NatsEvent::Connected)
             .await
             .unwrap();

src/config/src/meta/dashboards/v3/mod.rs

@@ -222,7 +222,7 @@ pub struct PanelConfig {
 pub struct DrillDown {
     #[serde(skip_serializing_if = "Option::is_none")]
     name: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none", rename="type")]
+    #[serde(skip_serializing_if = "Option::is_none", rename = "type")]
     type_field: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     target_blank: Option<bool>,

src/config/src/meta/dashboards/v5/mod.rs

@@ -318,7 +318,7 @@ pub struct ColorCfg {
 #[derive(Debug, Clone, PartialEq, Hash, Serialize, Deserialize, ToSchema)]
 #[serde(rename_all = "camelCase")]
 pub struct Mapping {
-    #[serde(skip_serializing_if = "Option::is_none", rename="type")]
+    #[serde(skip_serializing_if = "Option::is_none", rename = "type")]
     typee: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     value: Option<String>,
@@ -328,7 +328,7 @@ pub struct Mapping {
     to: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pattern: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none", rename="match")]
+    #[serde(skip_serializing_if = "Option::is_none", rename = "match")]
     matchh: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     color: Option<String>,
@@ -341,7 +341,7 @@ pub struct Mapping {
 pub struct DrillDown {
     #[serde(skip_serializing_if = "Option::is_none")]
     name: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none", rename="type")]
+    #[serde(skip_serializing_if = "Option::is_none", rename = "type")]
     type_field: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     target_blank: Option<bool>,
@@ -356,7 +356,7 @@ pub struct DrillDown {
 pub struct MarkLine {
     #[serde(skip_serializing_if = "Option::is_none")]
     name: Option<String>,
-    #[serde(skip_serializing_if = "Option::is_none", rename="type")]
+    #[serde(skip_serializing_if = "Option::is_none", rename = "type")]
     typee: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     value: Option<String>,

src/config/src/meta/self_reporting/error.rs

@@ -48,7 +48,10 @@ pub struct PipelineError {
     pub pipeline_name: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub error: Option<String>,
-    #[serde(skip_serializing_if = "HashMap::is_empty", serialize_with = "serialize_values_only")]
+    #[serde(
+        skip_serializing_if = "HashMap::is_empty",
+        serialize_with = "serialize_values_only"
+    )]
     pub node_errors: HashMap<String, NodeErrors>,
 }
 

src/infra/src/db/etcd.rs

@@ -52,7 +52,7 @@ pub async fn init() {
         return;
     }
     // enable keep alive for auth token
-    tokio::task::spawn(async move { keep_alive_connection().await });
+    tokio::task::spawn(keep_alive_connection());
 }
 
 pub struct Etcd {

src/infra/src/table/migration/m20250107_160900_delete_bad_dashboards.rs

@@ -687,7 +687,7 @@ mod v3 {
     pub struct DrillDown {
         #[serde(skip_serializing_if = "Option::is_none")]
         name: Option<String>,
-        #[serde(skip_serializing_if = "Option::is_none", rename="type")]
+        #[serde(skip_serializing_if = "Option::is_none", rename = "type")]
         type_field: Option<String>,
         #[serde(skip_serializing_if = "Option::is_none")]
         target_blank: Option<bool>,
@@ -1026,7 +1026,7 @@ mod v4 {
     pub struct DrillDown {
         #[serde(skip_serializing_if = "Option::is_none")]
         name: Option<String>,
-        #[serde(skip_serializing_if = "Option::is_none", rename="type")]
+        #[serde(skip_serializing_if = "Option::is_none", rename = "type")]
         type_field: Option<String>,
         #[serde(skip_serializing_if = "Option::is_none")]
         target_blank: Option<bool>,
@@ -1041,7 +1041,7 @@ mod v4 {
     pub struct MarkLine {
         #[serde(skip_serializing_if = "Option::is_none")]
         name: Option<String>,
-        #[serde(skip_serializing_if = "Option::is_none", rename="type")]
+        #[serde(skip_serializing_if = "Option::is_none", rename = "type")]
         typee: Option<String>,
         #[serde(skip_serializing_if = "Option::is_none")]
         value: Option<String>,
@@ -1430,7 +1430,7 @@ mod v5 {
     #[derive(Deserialize)]
     #[serde(rename_all = "camelCase")]
     pub struct Mapping {
-        #[serde(skip_serializing_if = "Option::is_none", rename="type")]
+        #[serde(skip_serializing_if = "Option::is_none", rename = "type")]
         typee: Option<String>,
         #[serde(skip_serializing_if = "Option::is_none")]
         value: Option<String>,
@@ -1440,7 +1440,7 @@ mod v5 {
         to: Option<String>,
         #[serde(skip_serializing_if = "Option::is_none")]
         pattern: Option<String>,
-        #[serde(skip_serializing_if = "Option::is_none", rename="match")]
+        #[serde(skip_serializing_if = "Option::is_none", rename = "match")]
         matchh: Option<String>,
         #[serde(skip_serializing_if = "Option::is_none")]
         color: Option<String>,
@@ -1453,7 +1453,7 @@ mod v5 {
     pub struct DrillDown {
         #[serde(skip_serializing_if = "Option::is_none")]
         name: Option<String>,
-        #[serde(skip_serializing_if = "Option::is_none", rename="type")]
+        #[serde(skip_serializing_if = "Option::is_none", rename = "type")]
         type_field: Option<String>,
         #[serde(skip_serializing_if = "Option::is_none")]
         target_blank: Option<bool>,
@@ -1468,7 +1468,7 @@ mod v5 {
     pub struct MarkLine {
         #[serde(skip_serializing_if = "Option::is_none")]
         name: Option<String>,
-        #[serde(skip_serializing_if = "Option::is_none", rename="type")]
+        #[serde(skip_serializing_if = "Option::is_none", rename = "type")]
         typee: Option<String>,
         #[serde(skip_serializing_if = "Option::is_none")]
         value: Option<String>,

src/job/alert_manager.rs

@@ -60,14 +60,14 @@ pub async fn run() -> Result<(), anyhow::Error> {
         .await?;
     }
 
-    tokio::task::spawn(async move { run_schedule_jobs().await });
-    tokio::task::spawn(async move { watch_timeout_jobs().await });
+    tokio::task::spawn(run_schedule_jobs());
+    tokio::task::spawn(watch_timeout_jobs());
     for i in 0..cfg.limit.search_job_workers {
-        tokio::task::spawn(async move { run_search_jobs(i).await });
+        tokio::task::spawn(run_search_jobs(i));
     }
-    tokio::task::spawn(async move { run_check_running_search_jobs().await });
-    tokio::task::spawn(async move { run_delete_jobs_by_retention().await });
-    tokio::task::spawn(async move { run_delete_jobs().await });
+    tokio::task::spawn(run_check_running_search_jobs());
+    tokio::task::spawn(run_delete_jobs_by_retention());
+    tokio::task::spawn(run_delete_jobs());
 
     Ok(())
 }

src/job/compactor.rs

@@ -46,19 +46,19 @@ pub async fn run() -> Result<(), anyhow::Error> {
         log::error!("[COMPACTOR::JOB] start merge job scheduler error: {e}");
     }
 
-    tokio::task::spawn(async move { run_generate_job().await });
-    tokio::task::spawn(async move { run_generate_old_data_job().await });
+    tokio::task::spawn(run_generate_job());
+    tokio::task::spawn(run_generate_old_data_job());
     #[cfg(feature = "enterprise")]
-    tokio::task::spawn(async move { run_generate_downsampling_job().await });
-    tokio::task::spawn(async move { run_merge(scheduler.tx()).await });
-    tokio::task::spawn(async move { run_retention().await });
-    tokio::task::spawn(async move { run_delay_deletion().await });
-    tokio::task::spawn(async move { run_sync_to_db().await });
+    tokio::task::spawn(run_generate_downsampling_job());
+    tokio::task::spawn(run_merge(scheduler.tx()));
+    tokio::task::spawn(run_retention());
+    tokio::task::spawn(run_delay_deletion());
+    tokio::task::spawn(run_sync_to_db());
     #[cfg(feature = "enterprise")]
-    tokio::task::spawn(async move { run_downsampling_sync_to_db().await });
-    tokio::task::spawn(async move { run_check_running_jobs().await });
-    tokio::task::spawn(async move { run_clean_done_jobs().await });
-    tokio::task::spawn(async move { run_compactor_pending_jobs_metric().await });
+    tokio::task::spawn(run_downsampling_sync_to_db());
+    tokio::task::spawn(run_check_running_jobs());
+    tokio::task::spawn(run_clean_done_jobs());
+    tokio::task::spawn(run_compactor_pending_jobs_metric());
 
     Ok(())
 }

src/job/files/mod.rs

@@ -31,9 +31,9 @@ pub async fn run() -> Result<(), anyhow::Error> {
     // load pending delete files to memory cache
     crate::service::db::file_list::local::load_pending_delete().await?;
 
-    tokio::task::spawn(async move { parquet::run().await });
-    tokio::task::spawn(async move { broadcast::run().await });
-    tokio::task::spawn(async move { clean_empty_dirs().await });
+    tokio::task::spawn(parquet::run());
+    tokio::task::spawn(broadcast::run());
+    tokio::task::spawn(clean_empty_dirs());
 
     Ok(())
 }

src/job/flatten_compactor.rs

@@ -58,7 +58,7 @@ pub async fn run() -> Result<(), anyhow::Error> {
         });
     }
 
-    tokio::task::spawn(async move { run_generate(tx).await });
+    tokio::task::spawn(run_generate(tx));
     Ok(())
 }
 

src/job/mod.rs

@@ -97,7 +97,7 @@ pub async fn init() -> Result<(), anyhow::Error> {
         && (LOCAL_NODE.is_ingester() || LOCAL_NODE.is_querier() || LOCAL_NODE.is_alert_manager())
     {
         // Try to download the mmdb files, if its not disabled.
-        tokio::task::spawn(async move { mmdb_downloader::run().await });
+        tokio::task::spawn(mmdb_downloader::run());
     }
 
     db::user::cache().await.expect("user cache failed");
@@ -111,9 +111,9 @@ pub async fn init() -> Result<(), anyhow::Error> {
         .expect("organization settings cache sync failed");
 
     // watch org users
-    tokio::task::spawn(async move { db::user::watch().await });
-    tokio::task::spawn(async move { db::org_users::watch().await });
-    tokio::task::spawn(async move { db::organization::watch().await });
+    tokio::task::spawn(db::user::watch());
+    tokio::task::spawn(db::org_users::watch());
+    tokio::task::spawn(db::organization::watch());
 
     // check version
     db::metas::version::set()
@@ -125,13 +125,13 @@ pub async fn init() -> Result<(), anyhow::Error> {
 
     // Auth auditing should be done by router also
     #[cfg(feature = "enterprise")]
-    tokio::task::spawn(async move { self_reporting::run_audit_publish().await });
+    tokio::task::spawn(self_reporting::run_audit_publish());
     #[cfg(feature = "cloud")]
-    tokio::task::spawn(async move { self_reporting::cloud_events::flush_cloud_events().await });
+    tokio::task::spawn(self_reporting::cloud_events::flush_cloud_events());
 
     #[cfg(feature = "enterprise")]
     {
-        tokio::task::spawn(async move { db::ofga::watch().await });
+        tokio::task::spawn(db::ofga::watch());
         db::ofga::cache().await.expect("ofga model cache failed");
         o2_openfga::authorizer::authz::init_open_fga().await;
         // RBAC model
@@ -142,7 +142,7 @@ pub async fn init() -> Result<(), anyhow::Error> {
         }
     }
 
-    tokio::task::spawn(async move { promql_self_consume::run().await });
+    tokio::task::spawn(promql_self_consume::run());
     // Router doesn't need to initialize job
     if LOCAL_NODE.is_router() && LOCAL_NODE.is_single_role() {
         return Ok(());
@@ -150,27 +150,27 @@ pub async fn init() -> Result<(), anyhow::Error> {
 
     // telemetry run
     if cfg.common.telemetry_enabled && LOCAL_NODE.is_querier() {
-        tokio::task::spawn(async move { telemetry::run().await });
+        tokio::task::spawn(telemetry::run());
     }
 
-    tokio::task::spawn(async move { self_reporting::run().await });
+    tokio::task::spawn(self_reporting::run());
 
     // cache short_urls
-    tokio::task::spawn(async move { db::short_url::watch().await });
+    tokio::task::spawn(db::short_url::watch());
     db::short_url::cache()
         .await
         .expect("short url cache failed");
 
     // initialize metadata watcher
-    tokio::task::spawn(async move { db::schema::watch().await });
-    tokio::task::spawn(async move { db::functions::watch().await });
-    tokio::task::spawn(async move { db::compact::retention::watch().await });
-    tokio::task::spawn(async move { db::metrics::watch_prom_cluster_leader().await });
-    tokio::task::spawn(async move { db::alerts::templates::watch().await });
-    tokio::task::spawn(async move { db::alerts::destinations::watch().await });
-    tokio::task::spawn(async move { db::alerts::realtime_triggers::watch().await });
-    tokio::task::spawn(async move { db::alerts::alert::watch().await });
-    tokio::task::spawn(async move { db::organization::org_settings_watch().await });
+    tokio::task::spawn(db::schema::watch());
+    tokio::task::spawn(db::functions::watch());
+    tokio::task::spawn(db::compact::retention::watch());
+    tokio::task::spawn(db::metrics::watch_prom_cluster_leader());
+    tokio::task::spawn(db::alerts::templates::watch());
+    tokio::task::spawn(db::alerts::destinations::watch());
+    tokio::task::spawn(db::alerts::realtime_triggers::watch());
+    tokio::task::spawn(db::alerts::alert::watch());
+    tokio::task::spawn(db::organization::org_settings_watch());
 
     // pipeline not used on compactors
     if LOCAL_NODE.is_ingester() || LOCAL_NODE.is_querier() || LOCAL_NODE.is_alert_manager() {
@@ -179,10 +179,10 @@ pub async fn init() -> Result<(), anyhow::Error> {
 
     #[cfg(feature = "enterprise")]
     if LOCAL_NODE.is_ingester() || LOCAL_NODE.is_querier() {
-        tokio::task::spawn(async move { db::session::watch().await });
+        tokio::task::spawn(db::session::watch());
     }
     if LOCAL_NODE.is_ingester() || LOCAL_NODE.is_querier() || LOCAL_NODE.is_alert_manager() {
-        tokio::task::spawn(async move { db::enrichment_table::watch().await });
+        tokio::task::spawn(db::enrichment_table::watch());
     }
 
     tokio::task::yield_now().await;
@@ -235,31 +235,29 @@ pub async fn init() -> Result<(), anyhow::Error> {
         }
     }
 
-    tokio::task::spawn(async move { files::run().await });
-    tokio::task::spawn(async move { stats::run().await });
-    tokio::task::spawn(async move { compactor::run().await });
-    tokio::task::spawn(async move { flatten_compactor::run().await });
-    tokio::task::spawn(async move { metrics::run().await });
-    tokio::task::spawn(async move { promql::run().await });
-    tokio::task::spawn(async move { alert_manager::run().await });
-    tokio::task::spawn(async move { file_downloader::run().await });
+    tokio::task::spawn(files::run());
+    tokio::task::spawn(stats::run());
+    tokio::task::spawn(compactor::run());
+    tokio::task::spawn(flatten_compactor::run());
+    tokio::task::spawn(metrics::run());
+    tokio::task::spawn(promql::run());
+    tokio::task::spawn(alert_manager::run());
+    tokio::task::spawn(file_downloader::run());
 
     if LOCAL_NODE.is_compactor() {
-        tokio::task::spawn(async move { file_list_dump::run().await });
+        tokio::task::spawn(file_list_dump::run());
     }
 
     // load metrics disk cache
-    tokio::task::spawn(async move { crate::service::promql::search::init().await });
+    tokio::task::spawn(crate::service::promql::search::init());
     // start pipeline data retention
     #[cfg(feature = "enterprise")]
-    tokio::task::spawn(
-        async move { o2_enterprise::enterprise::pipeline::pipeline_job::run().await },
-    );
+    tokio::task::spawn(o2_enterprise::enterprise::pipeline::pipeline_job::run());
 
     #[cfg(feature = "enterprise")]
-    tokio::task::spawn(async move { cipher::run().await });
+    tokio::task::spawn(cipher::run());
     #[cfg(feature = "enterprise")]
-    tokio::task::spawn(async move { db::keys::watch().await });
+    tokio::task::spawn(db::keys::watch());
 
     // additional for cloud
     #[cfg(feature = "cloud")]
@@ -279,8 +277,8 @@ pub async fn init() -> Result<(), anyhow::Error> {
     log::info!("Job initialization complete");
 
     // Syslog server start
-    tokio::task::spawn(async move { db::syslog::watch().await });
-    tokio::task::spawn(async move { db::syslog::watch_syslog_settings().await });
+    tokio::task::spawn(db::syslog::watch());
+    tokio::task::spawn(db::syslog::watch_syslog_settings());
 
     let start_syslog = *SYSLOG_ENABLED.read();
     if start_syslog {

src/job/stats.rs

@@ -19,9 +19,9 @@ use tokio::time;
 use crate::service::{compact::stats::update_stats_from_file_list, db};
 
 pub async fn run() -> Result<(), anyhow::Error> {
-    tokio::task::spawn(async move { update_node_memory_usage().await });
-    tokio::task::spawn(async move { file_list_update_stats().await });
-    tokio::task::spawn(async move { cache_stream_stats().await });
+    tokio::task::spawn(update_node_memory_usage());
+    tokio::task::spawn(file_list_update_stats());
+    tokio::task::spawn(cache_stream_stats());
     Ok(())
 }
 

src/main.rs

@@ -359,7 +359,7 @@ async fn main() -> Result<(), anyhow::Error> {
         .expect("Deferred jobs failed to init");
 
     if cfg.log.events_enabled {
-        tokio::task::spawn(async move { zo_logger::send_logs().await });
+        tokio::task::spawn(zo_logger::send_logs());
     }
     if cfg.common.telemetry_enabled {
         tokio::task::spawn(async move {
@@ -753,9 +753,7 @@ async fn init_http_server() -> Result<(), anyhow::Error> {
         .disable_signals()
         .run();
     let handle = server.handle();
-    tokio::task::spawn(async move {
-        graceful_shutdown(handle).await;
-    });
+    tokio::task::spawn(graceful_shutdown(handle));
     server.await?;
     Ok(())
 }
@@ -861,9 +859,7 @@ async fn init_http_server_without_tracing() -> Result<(), anyhow::Error> {
         .disable_signals()
         .run();
     let handle = server.handle();
-    tokio::task::spawn(async move {
-        graceful_shutdown(handle).await;
-    });
+    tokio::task::spawn(graceful_shutdown(handle));
     server.await?;
     Ok(())
 }
@@ -1104,9 +1100,7 @@ async fn init_script_server() -> Result<(), anyhow::Error> {
         .disable_signals()
         .run();
     let handle = server.handle();
-    tokio::task::spawn(async move {
-        graceful_shutdown(handle).await;
-    });
+    tokio::task::spawn(graceful_shutdown(handle));
     server.await?;
 
     log::info!("HTTP server stopped");

src/report_server/src/server.rs

@@ -40,9 +40,7 @@ pub async fn spawn_server() -> Result<(), anyhow::Error> {
     .run();
 
     let handle = server.handle();
-    tokio::task::spawn(async move {
-        graceful_shutdown(handle).await;
-    });
+    tokio::task::spawn(graceful_shutdown(handle));
     server.await?;
     log::info!("Report server stopped");
     Ok(())

src/service/alerts/scheduler/worker.rs

@@ -140,7 +140,6 @@ impl SchedulerWorker {
         if let Err(e) = handler_job.await {
             return Err(anyhow::anyhow!("Error in handler: {}", e));
         }
-
         Ok(())
     }
 }

src/service/compact/flatten.rs

@@ -62,8 +62,7 @@ pub async fn run_generate(worker_tx: mpsc::Sender<FileKey>) -> Result<(), anyhow
                     infra::schema::get_settings(&org_id, &stream_name, stream_type)
                         .await
                         .unwrap_or_default();
-                let defined_schema_fields =
-                    stream_setting.defined_schema_fields;
+                let defined_schema_fields = stream_setting.defined_schema_fields;
                 if defined_schema_fields.is_empty() {
                     continue;
                 }

src/service/metadata/distinct_values.rs

@@ -109,7 +109,7 @@ impl Default for DistinctValues {
 
 impl DistinctValues {
     pub fn new() -> Self {
-        tokio::task::spawn(async move { run_flush().await });
+        tokio::task::spawn(run_flush());
         Self {
             channel: handle_channel(),
             shutdown: Arc::new(AtomicBool::new(false)),

src/service/pipeline/batch_execution.rs

@@ -220,7 +220,7 @@ impl ExecutablePipeline {
         }
 
         // Spawn tasks for each node
-        let mut node_tasks = Vec::new();
+        let mut node_tasks = Vec::with_capacity(self.sorted_nodes.len());
         for (idx, node_id) in self.sorted_nodes.iter().enumerate() {
             let pl_id_cp = self.id.to_string();
             let org_id_cp = org_id.to_string();
@@ -239,22 +239,19 @@ impl ExecutablePipeline {
 
             // WARN: Do not change. Processing node can only be done in a task, as the internals of
             // remote wal writer depends on the task id.
-            let task = tokio::spawn(async move {
-                process_node(
-                    pl_id_cp,
-                    idx,
-                    org_id_cp,
-                    node,
-                    node_receiver,
-                    child_senders,
-                    vrl_runtime,
-                    result_sender_cp,
-                    error_sender_cp,
-                    pipeline_name,
-                    stream_name,
-                )
-                .await
-            });
+            let task = tokio::spawn(process_node(
+                pl_id_cp,
+                idx,
+                org_id_cp,
+                node,
+                node_receiver,
+                child_senders,
+                vrl_runtime,
+                result_sender_cp,
+                error_sender_cp,
+                pipeline_name,
+                stream_name,
+            ));
             node_tasks.push(task);
         }
 

src/service/self_reporting/queues.rs

@@ -56,9 +56,12 @@ fn initialize_usage_queue() -> ReportingQueue {
 
     for thread_id in 0..cfg.limit.usage_reporting_thread_num {
         let msg_receiver = msg_receiver.clone();
-        tokio::task::spawn(async move {
-            self_reporting_ingest_job(thread_id, msg_receiver, batch_size, timeout).await
-        });
+        tokio::task::spawn(self_reporting_ingest_job(
+            thread_id,
+            msg_receiver,
+            batch_size,
+            timeout,
+        ));
     }
 
     ReportingQueue::new(msg_sender)
@@ -81,9 +84,12 @@ fn initialize_error_queue() -> ReportingQueue {
 
     for thread_id in 0..cfg.limit.usage_reporting_thread_num {
         let msg_receiver = msg_receiver.clone();
-        tokio::task::spawn(async move {
-            self_reporting_ingest_job(thread_id, msg_receiver, batch_size, timeout).await
-        });
+        tokio::task::spawn(self_reporting_ingest_job(
+            thread_id,
+            msg_receiver,
+            batch_size,
+            timeout,
+        ));
     }
 
     ReportingQueue::new(msg_sender)