Log suggest truncate message delays

quickwit/quickwit-actors/src/actor_handle.rs

@@ -266,16 +266,16 @@ impl<A: Actor> ActorHandle<A> {
         let observable_state_or_timeout =
             scheduler_client.timeout(crate::OBSERVE_TIMEOUT, rx).await;
         match observable_state_or_timeout {
-            Ok(Ok(state)) => {
+            Some(Ok(state)) => {
                 let obs_type = ObservationType::Alive;
                 Observation { obs_type, state }
             }
-            Ok(Err(_)) => {
+            Some(Err(_)) => {
                 let state = self.last_observation().clone();
                 let obs_type = ObservationType::PostMortem;
                 Observation { obs_type, state }
             }
-            Err(_) => {
+            None => {
                 let state = self.last_observation().clone();
                 let obs_type = if self.actor_context.state().is_exit() {
                     ObservationType::PostMortem

quickwit/quickwit-actors/src/actor_state.rs

@@ -77,7 +77,7 @@ impl AtomicState {
     pub(crate) fn pause(&self) {
         let _ = self
             .0
-            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |state| {
+            .try_update(Ordering::SeqCst, Ordering::SeqCst, |state| {
                 if ActorState::from(state).is_running() {
                     return Some(ActorState::Paused as u32);
                 }

quickwit/quickwit-actors/src/scheduler.rs

@@ -143,17 +143,14 @@ impl SchedulerClient {
         let _ = oneshot_rx.await;
     }
 
-    pub async fn timeout<O>(
-        &self,
-        duration: Duration,
-        fut: impl Future<Output = O>,
-    ) -> Result<O, ()> {
+    /// Returns `None` if `duration` elapses before `fut` completes.
+    pub async fn timeout<O>(&self, duration: Duration, fut: impl Future<Output = O>) -> Option<O> {
         tokio::select! {
             _ = self.sleep(duration) => {
-                Err(())
+                None
             },
             future_output = fut => {
-                Ok(future_output)
+                Some(future_output)
             }
         }
     }

quickwit/quickwit-cli/src/lib.rs

@@ -406,7 +406,7 @@ pub mod busy_detector {
 
     fn emit_debug(delta: u64, now: u64) {
         if NEXT_DEBUG_TIMESTAMP
-            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |next_debug| {
+            .try_update(Ordering::Relaxed, Ordering::Relaxed, |next_debug| {
                 if next_debug < now {
                     Some(now + DEBUG_SUPPRESSION_MICROS)
                 } else {

quickwit/quickwit-codegen/example/src/lib.rs

@@ -34,6 +34,7 @@ pub use crate::hello::*;
 pub type HelloResult<T> = Result<T, HelloError>;
 
 #[derive(Debug, Clone)]
+#[allow(dead_code)]
 struct Counter<S> {
     counter: Arc<AtomicUsize>,
     inner: S,

quickwit/quickwit-common/src/rate_limited_tracing.rs

@@ -100,7 +100,7 @@ pub fn should_log<F: Fn() -> Instant>(
     let mut update_time = false;
 
     let update_res =
-        count_atomic.fetch_update(Ordering::Release, Ordering::Acquire, |current_count| {
+        count_atomic.try_update(Ordering::Release, Ordering::Acquire, |current_count| {
             let mut current_count: LogSiteMetadata = current_count.into();
             if generation == current_count.generation {
                 // we can update generation&time, so we can definitely log

quickwit/quickwit-directories/src/caching_directory.rs

@@ -67,7 +67,7 @@ impl fmt::Debug for CachingFileHandle {
         write!(
             f,
             "CachingFileHandle(path={:?}, underlying={:?})",
-            &self.path,
+            self.path,
             self.underlying_filehandle.as_ref()
         )
     }

quickwit/quickwit-directories/src/debug_proxy_directory.rs

@@ -180,7 +180,7 @@ impl<D: Directory> FileHandle for DebugProxyFileHandle<D> {
 
 impl<D: Directory> fmt::Debug for DebugProxyFileHandle<D> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "DebugProxyFileHandle({:?})", &self.underlying)
+        write!(f, "DebugProxyFileHandle({:?})", self.underlying)
     }
 }
 

quickwit/quickwit-directories/src/hot_directory.rs

@@ -408,7 +408,7 @@ impl FileHandle for FileSliceWithCache {
 
 impl fmt::Debug for FileSliceWithCache {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "FileSliceWithCache({:?})", &self.underlying)
+        write!(f, "FileSliceWithCache({:?})", self.underlying)
     }
 }
 

quickwit/quickwit-directories/src/storage_directory.rs

@@ -42,7 +42,7 @@ impl fmt::Debug for StorageDirectoryFileHandle {
         write!(
             f,
             "StorageDirectoryFileHandle({:?}, dir={:?})",
-            &self.path, self.storage_directory
+            self.path, self.storage_directory
         )
     }
 }

quickwit/quickwit-doc-mapper/src/doc_mapper/field_mapping_entry.rs

@@ -805,7 +805,7 @@ impl TryFrom<FieldMappingEntryForSerialization> for FieldMappingEntry {
             QuickwitFieldType::parse_type_id(&value.type_id).ok_or_else(|| {
                 format!(
                     "field `{}` has an unknown type: `{}`",
-                    &value.name, &value.type_id
+                    value.name, value.type_id
                 )
             })?;
         let mapping_type = deserialize_mapping_type(

quickwit/quickwit-index-management/src/garbage_collection.rs

@@ -28,11 +28,11 @@ use quickwit_metastore::{
 };
 use quickwit_metrics::Counter;
 use quickwit_proto::metastore::{
-    DeleteSplitsRequest, ListSplitsRequest, MarkSplitsForDeletionRequest, MetastoreError,
-    MetastoreService, MetastoreServiceClient,
+    DeleteSplitsRequest, ListSplitsRequest, MarkSplitsForDeletionRequest, MetastoreService,
+    MetastoreServiceClient,
 };
 use quickwit_proto::types::{IndexUid, SplitId};
-use quickwit_storage::{BulkDeleteError, Storage};
+use quickwit_storage::Storage;
 use thiserror::Error;
 use time::OffsetDateTime;
 use tracing::{error, instrument};
@@ -66,9 +66,7 @@ impl RecordGcMetrics for Option<GcMetrics> {
 #[error("failed to delete splits from storage and/or metastore")]
 pub struct DeleteSplitsError {
     successes: Vec<SplitInfo>,
-    storage_error: Option<BulkDeleteError>,
     storage_failures: Vec<SplitInfo>,
-    metastore_error: Option<MetastoreError>,
     metastore_failures: Vec<SplitInfo>,
 }
 
@@ -442,7 +440,6 @@ pub async fn delete_splits_from_storage_and_metastore(
         progress.record_progress();
     }
     let mut successes = Vec::with_capacity(split_infos.len());
-    let mut storage_error: Option<BulkDeleteError> = None;
     let mut storage_failures = Vec::new();
 
     match delete_result {
@@ -463,11 +460,11 @@ pub async fn delete_splits_from_storage_and_metastore(
                 .collect::<Vec<_>>();
             error!(
                 error=?bulk_delete_error.error,
+                failures=?PrettySample::new(&bulk_delete_error.failures, 5),
                 index_id=index_uid.index_id,
                 "failed to delete split file(s) {:?} from storage",
                 PrettySample::new(&failed_split_paths, 5),
             );
-            storage_error = Some(bulk_delete_error);
         }
     };
     if !successes.is_empty() {
@@ -491,9 +488,7 @@ pub async fn delete_splits_from_storage_and_metastore(
             );
             let delete_splits_error = DeleteSplitsError {
                 successes: Vec::new(),
-                storage_error,
                 storage_failures,
-                metastore_error: Some(metastore_error),
                 metastore_failures: successes,
             };
             return Err(delete_splits_error);
@@ -502,9 +497,7 @@ pub async fn delete_splits_from_storage_and_metastore(
     if !storage_failures.is_empty() {
         let delete_splits_error = DeleteSplitsError {
             successes,
-            storage_error,
             storage_failures,
-            metastore_error: None,
             metastore_failures: Vec::new(),
         };
         return Err(delete_splits_error);
@@ -513,8 +506,8 @@ pub async fn delete_splits_from_storage_and_metastore(
 }
 
 #[cfg(test)]
-#[allow(clippy::result_large_err)] // BulkDeleteError is large; acceptable in mock closures
 mod tests {
+    use std::collections::BTreeMap;
     use std::time::Duration;
 
     use itertools::Itertools;
@@ -525,7 +518,7 @@ mod tests {
         SplitState, StageSplitsRequestExt, metastore_for_test,
     };
     use quickwit_proto::metastore::{
-        CreateIndexRequest, EntityKind, MockMetastoreService, StageSplitsRequest,
+        CreateIndexRequest, EntityKind, MetastoreError, MockMetastoreService, StageSplitsRequest,
     };
     use quickwit_proto::types::IndexUid;
     use quickwit_storage::{
@@ -870,7 +863,7 @@ mod tests {
                     code: Some("AccessDenied".to_string()),
                     ..Default::default()
                 };
-                let failures = HashMap::from_iter([(split_path, delete_failure)]);
+                let failures = BTreeMap::from_iter([(split_path, delete_failure)]);
                 let bulk_delete_error = BulkDeleteError {
                     successes,
                     failures,
@@ -961,7 +954,7 @@ mod tests {
                     code: Some("AccessDenied".to_string()),
                     ..Default::default()
                 };
-                let failures = HashMap::from_iter([(split_path, delete_failure)]);
+                let failures = BTreeMap::from_iter([(split_path, delete_failure)]);
                 let bulk_delete_error = BulkDeleteError {
                     successes,
                     failures,

quickwit/quickwit-indexing/src/actors/publisher.rs

@@ -138,12 +138,9 @@ async fn suggest_truncate(
     if let Some(source_mailbox) = source_mailbox_opt.as_ref()
         && let Some(checkpoint) = checkpoint_delta_opt
     {
-        let _ = ctx
-            .send_message(
-                source_mailbox,
-                SuggestTruncate(checkpoint.source_delta.get_source_checkpoint()),
-            )
-            .await;
+        let source_checkpoint = checkpoint.source_delta.get_source_checkpoint();
+        let message = SuggestTruncate::new(source_checkpoint);
+        let _ = ctx.send_message(source_mailbox, message).await;
     }
 }
 
@@ -403,7 +400,7 @@ mod tests {
         let suggest_truncate_checkpoints: Vec<SourceCheckpoint> = source_inbox
             .drain_for_test_typed::<SuggestTruncate>()
             .into_iter()
-            .map(|msg| msg.0)
+            .map(|msg| msg.checkpoint)
             .collect();
 
         assert_eq!(suggest_truncate_checkpoints.len(), 1);
@@ -481,7 +478,7 @@ mod tests {
         let suggest_truncate_checkpoints: Vec<SourceCheckpoint> = source_inbox
             .drain_for_test_typed::<SuggestTruncate>()
             .into_iter()
-            .map(|msg| msg.0)
+            .map(|msg| msg.checkpoint)
             .collect();
 
         assert_eq!(suggest_truncate_checkpoints.len(), 1);

quickwit/quickwit-indexing/src/actors/uploader.rs

@@ -148,7 +148,7 @@ impl SplitsUpdateSender {
                 {
                     bail!(
                         "failed to send upload split `{:?}`. the publisher is probably dead",
-                        &publisher_message
+                        publisher_message
                     );
                 }
             }

quickwit/quickwit-indexing/src/source/file_source.rs

@@ -311,14 +311,14 @@ mod tests {
         let batch2 = indexer_msgs[1].downcast_ref::<RawDocBatch>().unwrap();
         let command = indexer_msgs[2].downcast_ref::<Command>().unwrap();
         assert_eq!(
-            format!("{:?}", &batch1.checkpoint_delta),
+            format!("{:?}", batch1.checkpoint_delta),
             format!(
                 "∆({}:{})",
                 uri, "(00000000000000000000..00000000000005242895]"
             )
         );
         assert_eq!(
-            format!("{:?}", &batch2.checkpoint_delta),
+            format!("{:?}", batch2.checkpoint_delta),
             format!(
                 "∆({}:{})",
                 uri, "(00000000000005242895..~00000000000005397105]"

quickwit/quickwit-indexing/src/source/ingest/mod.rs

@@ -1630,8 +1630,6 @@ mod tests {
 
         fetch_message_tx
             .send(Err(FetchStreamError {
-                index_uid: IndexUid::for_test("test-index", 0),
-                source_id: "test-source".into(),
                 shard_id: ShardId::from(1),
                 ingest_error: IngestV2Error::Internal("test-error".to_string()),
             }))

quickwit/quickwit-indexing/src/source/mod.rs

@@ -74,7 +74,7 @@ mod void_source;
 use std::collections::BTreeSet;
 use std::path::PathBuf;
 use std::sync::LazyLock;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use async_trait::async_trait;
 use bytes::Bytes;
@@ -96,7 +96,9 @@ use quickwit_common::metrics::{
     IN_FLIGHT_KINESIS_SOURCE, IN_FLIGHT_OTHER_SOURCE, IN_FLIGHT_PUBSUB_SOURCE,
     IN_FLIGHT_PULSAR_SOURCE,
 };
+use quickwit_common::pretty::PrettyDisplay;
 use quickwit_common::pubsub::EventBroker;
+use quickwit_common::rate_limited_warn;
 use quickwit_common::runtimes::RuntimeType;
 use quickwit_config::{
     FileSourceNotification, FileSourceParams, IndexingSettings, SourceConfig, SourceParams,
@@ -511,8 +513,23 @@ pub async fn check_source_connectivity(
     }
 }
 
+/// Suggest truncate messages processed later than this threshold trigger a warning.
+const SUGGEST_TRUNCATE_DELAY_WARN_THRESHOLD: Duration = Duration::from_millis(100);
+
 #[derive(Debug)]
-pub struct SuggestTruncate(pub SourceCheckpoint);
+pub struct SuggestTruncate {
+    pub checkpoint: SourceCheckpoint,
+    pub created_at: Instant,
+}
+
+impl SuggestTruncate {
+    pub fn new(checkpoint: SourceCheckpoint) -> Self {
+        Self {
+            checkpoint,
+            created_at: Instant::now(),
+        }
+    }
+}
 
 #[async_trait]
 impl Handler<SuggestTruncate> for SourceActor {
@@ -523,7 +540,21 @@ impl Handler<SuggestTruncate> for SourceActor {
         suggest_truncate: SuggestTruncate,
         ctx: &SourceContext,
     ) -> Result<(), ActorExitStatus> {
-        let SuggestTruncate(checkpoint) = suggest_truncate;
+        let SuggestTruncate {
+            checkpoint,
+            created_at,
+        } = suggest_truncate;
+        let delay = created_at.elapsed();
+
+        if delay > SUGGEST_TRUNCATE_DELAY_WARN_THRESHOLD {
+            rate_limited_warn!(
+                limit_per_min = 6,
+                source = self.source.name(),
+                delay_ms = delay.as_millis() as u64,
+                "suggest truncate message processed with delay of {}",
+                delay.pretty_display()
+            );
+        }
 
         if let Err(error) = self.source.suggest_truncate(checkpoint, ctx).await {
             // Failing to process suggest truncate does not

quickwit/quickwit-indexing/src/source/pulsar_source.rs

@@ -609,7 +609,7 @@ mod pulsar_broker_tests {
         checkpoint
             .try_apply_delta(checkpoints!(partition => truncate_to))
             .expect("Create checkpoint");
-        let truncate = SuggestTruncate(checkpoint);
+        let truncate = SuggestTruncate::new(checkpoint);
         source_handle
             .mailbox()
             .send_message(truncate)