fix(cubestore): stop reporting remote files nothing references

rust/cubestore/cubestore/src/app_metrics.rs

@@ -148,6 +148,10 @@ pub static REMOTE_FS_FILES_TO_REMOVE: Gauge = metrics::gauge("cs.remote_fs.files
 pub static REMOTE_FS_FILES_SIZE_TO_REMOVE: Gauge =
     metrics::gauge("cs.remote_fs.files_to_remove.size");
 
+/// Startup warmup files that were gone by the time the pass reached them, which says how far
+/// behind its snapshot a pass ran rather than that anything is wrong.
+pub static WARMUP_MISSING: Counter = metrics::counter("cs.warmup.missing");
+
 /// Cache Store Cache
 pub static CACHESTORE_TTL_PERSIST: Counter = metrics::counter("cs.cachestore.ttl.persist");
 pub static CACHESTORE_TTL_BUFFER: Gauge = metrics::gauge("cs.cachestore.ttl.buffer");

rust/cubestore/cubestore/src/cluster/mod.rs

@@ -19,7 +19,7 @@ use crate::cluster::worker_services::{
     WorkerProcessing,
 };
 
-use crate::ack_error;
+use crate::app_metrics;
 use crate::cluster::message::NetworkMessage;
 use crate::cluster::rate_limiter::{ProcessRateLimiter, TaskType, TraceIndex};
 use crate::cluster::transport::{ClusterTransport, MetaStoreTransport, WorkerConnection};
@@ -2118,13 +2118,11 @@ impl ClusterImpl {
                     log::debug!("Startup warmup cancelled");
                     return;
                 }
-                // TODO: propagate 'not found' and log in debug mode. Compaction might remove files,
-                //       so they are not errors most of the time.
-                ack_error!(
-                    self.remote_fs
-                        .download_file(file, p.get_row().file_size())
-                        .await
-                );
+                let result = self
+                    .remote_fs
+                    .download_file(file.clone(), p.get_row().file_size())
+                    .await;
+                self.report_warmup_download(result, &file);
             }
             for c in chunks {
                 if self.stop_token.is_cancelled() {
@@ -2134,21 +2132,33 @@ impl ClusterImpl {
                 if c.get_row().in_memory() {
                     continue;
                 }
+                let file = chunk_file_name(c.get_id(), c.get_row().suffix());
                 let result = self
                     .remote_fs
-                    .download_file(
-                        chunk_file_name(c.get_id(), c.get_row().suffix()),
-                        c.get_row().file_size(),
-                    )
+                    .download_file(file.clone(), c.get_row().file_size())
                     .await;
-                // TODO: propagate 'not found' and log in debug mode. Compaction might remove files,
-                //       so they are not errors most of the time.
-                ack_error!(result);
+                self.report_warmup_download(result, &file);
             }
         }
         log::debug!("Startup warmup finished");
         return;
     }
+
+    /// A file that is gone by the time the pass reaches it is the normal case: the walk takes as
+    /// long as it takes and compaction removes what it replaces meanwhile, so its absence says how
+    /// far behind the snapshot the pass ran rather than that anything is wrong. A file that is
+    /// really missing is reported by the query that needs it, which is also the one that can tell,
+    /// since by then the metastore either still names the file or does not.
+    fn report_warmup_download(&self, result: Result<String, CubeError>, remote_path: &str) {
+        match result {
+            Ok(_) => {}
+            Err(e) if e.is_file_not_found() => {
+                app_metrics::WARMUP_MISSING.increment();
+                log::debug!("Skipping warmup of {}: {}", remote_path, e.message);
+            }
+            Err(e) => log::error!("Warmup of {} failed: {:?}", remote_path, e),
+        }
+    }
 }
 
 struct LoopbackConnection {

rust/cubestore/cubestore/src/lib.rs

@@ -15,6 +15,7 @@ use datafusion::cube_ext::catch_unwind::PanicError;
 use datafusion::parquet::errors::ParquetError;
 use flexbuffers::{DeserializationError, ReaderError};
 use log::SetLoggerError;
+use serde::Serializer;
 use serde_derive::{Deserialize, Serialize};
 use sqlparser::parser::ParserError;
 use std::any::Any;
@@ -62,13 +63,41 @@ pub struct CubeError {
 
 impl std::error::Error for CubeError {}
 
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[derive(Debug, Clone, Deserialize, PartialEq)]
 pub enum CubeErrorCauseType {
     User,
     Internal,
     CorruptData,
     WrongConnection,
     Panic,
+    /// A remote object is absent rather than temporarily unreachable.
+    FileNotFound,
+}
+
+impl CubeErrorCauseType {
+    /// How the cause looks to everything outside the process that produced it: peers reading a
+    /// serialized [CubeError] and [crate::scheduler] matching the job error strings that
+    /// [CubeError::to_string] produced. [CubeErrorCauseType::FileNotFound] is a refinement of
+    /// [CubeErrorCauseType::CorruptData] that only the local process acts on, so it takes the same
+    /// name and index and stays readable by a node or a stored job row of any version.
+    fn external_repr(&self) -> (u32, &'static str) {
+        match self {
+            CubeErrorCauseType::User => (0, "User"),
+            CubeErrorCauseType::Internal => (1, "Internal"),
+            CubeErrorCauseType::CorruptData | CubeErrorCauseType::FileNotFound => {
+                (2, "CorruptData")
+            }
+            CubeErrorCauseType::WrongConnection => (3, "WrongConnection"),
+            CubeErrorCauseType::Panic => (4, "Panic"),
+        }
+    }
+}
+
+impl serde::Serialize for CubeErrorCauseType {
+    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        let (index, name) = self.external_repr();
+        serializer.serialize_unit_variant("CubeErrorCauseType", index, name)
+    }
 }
 
 impl CubeError {
@@ -135,9 +164,24 @@ impl CubeError {
         }
     }
 
+    pub fn file_not_found(message: String) -> CubeError {
+        CubeError {
+            message,
+            backtrace: String::new(),
+            cause: CubeErrorCauseType::FileNotFound,
+        }
+    }
+
     pub fn is_corrupt_data(&self) -> bool {
         match self.cause {
-            CubeErrorCauseType::CorruptData => true,
+            CubeErrorCauseType::CorruptData | CubeErrorCauseType::FileNotFound => true,
+            _ => false,
+        }
+    }
+
+    pub fn is_file_not_found(&self) -> bool {
+        match self.cause {
+            CubeErrorCauseType::FileNotFound => true,
             _ => false,
         }
     }
@@ -180,7 +224,11 @@ impl fmt::Display for CubeError {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         match self.cause {
             CubeErrorCauseType::User => f.write_fmt(format_args!("{}", self.message)),
-            _ => f.write_fmt(format_args!("{:?}: {}", self.cause, self.message)),
+            _ => f.write_fmt(format_args!(
+                "{}: {}",
+                self.cause.external_repr().1,
+                self.message
+            )),
         }
     }
 }
@@ -519,3 +567,53 @@ impl Into<ArrowError> for CubeError {
         ArrowError::ExternalError(Box::new(self))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde::{Deserialize as _, Serialize as _};
+
+    fn to_flexbuffer(e: &CubeError) -> Vec<u8> {
+        let mut ser = flexbuffers::FlexbufferSerializer::new();
+        e.serialize(&mut ser).unwrap();
+        ser.take_buffer()
+    }
+
+    #[test]
+    fn file_not_found_travels_as_corrupt_data() {
+        let not_found = CubeError::file_not_found("f.parquet is gone".to_string());
+        let corrupt_data = CubeError::corrupt_data("f.parquet is gone".to_string());
+
+        // A node running any other version has to be able to read the result.
+        assert_eq!(to_flexbuffer(&not_found), to_flexbuffer(&corrupt_data));
+
+        let buffer = to_flexbuffer(&not_found);
+        let reader = flexbuffers::Reader::get_root(buffer.as_slice()).unwrap();
+        assert_eq!(CubeError::deserialize(reader).unwrap(), corrupt_data);
+    }
+
+    #[test]
+    fn file_not_found_reads_as_corrupt_data() {
+        // `scheduler` classifies a failed job by looking for "CorruptData" in the error string a
+        // worker stored via `to_string`, so the rendering has to stay the same.
+        let not_found = CubeError::file_not_found("f.parquet is gone".to_string());
+        assert_eq!(
+            not_found.to_string(),
+            CubeError::corrupt_data("f.parquet is gone".to_string()).to_string()
+        );
+        assert!(not_found.to_string().contains("CorruptData"));
+    }
+
+    #[test]
+    fn file_not_found_is_corrupt_data_locally() {
+        let not_found = CubeError::file_not_found("f.parquet is gone".to_string());
+        assert!(not_found.is_file_not_found());
+        assert!(not_found.is_corrupt_data());
+
+        let corrupt_data = CubeError::corrupt_data("f.parquet is gone".to_string());
+        assert!(!corrupt_data.is_file_not_found());
+        assert!(corrupt_data.is_corrupt_data());
+
+        assert!(!CubeError::internal("nope".to_string()).is_file_not_found());
+    }
+}

rust/cubestore/cubestore/src/remotefs/gcs.rs

@@ -118,6 +118,22 @@ impl GCSRemoteFs {
 
 di_service!(GCSRemoteFs, [RemoteFs, ExtendedRemoteFs]);
 
+/// GCS answers 404 for an object that is already gone where S3 answers 204. Absent is the state a
+/// caller of `delete_file` asked for and none of them can act on the difference, so it is reported
+/// as done rather than as a failure they have to log.
+///
+/// A missing bucket carries the same reason, and the client crate keeps the message and the code
+/// private, so the two cannot be told apart here. A bucket that is not there fails every upload and
+/// download loudly, which is the signal to go by.
+fn is_absent_object(e: &cloud_storage::Error) -> bool {
+    match e {
+        cloud_storage::Error::Google(response) => {
+            response.errors_has_reason(&cloud_storage::Reason::NotFound)
+        }
+        _ => false,
+    }
+}
+
 #[async_trait]
 impl RemoteFs for GCSRemoteFs {
     async fn temp_upload_path(&self, remote_path: String) -> Result<String, CubeError> {
@@ -247,8 +263,13 @@ impl RemoteFs for GCSRemoteFs {
         );
         let time = SystemTime::now();
         debug!("Deleting {}", remote_path);
-        Object::delete(self.bucket.as_str(), self.gcs_path(&remote_path).as_str()).await?;
-        info!("Deleting {} ({:?})", remote_path, time.elapsed()?);
+        match Object::delete(self.bucket.as_str(), self.gcs_path(&remote_path).as_str()).await {
+            Ok(()) => info!("Deleting {} ({:?})", remote_path, time.elapsed()?),
+            Err(e) if is_absent_object(&e) => {
+                debug!("File {} is already absent in remote fs", remote_path);
+            }
+            Err(e) => return Err(e.into()),
+        }
 
         let _guard = acquire_lock("delete file", self.delete_mut.lock()).await?;
         let local = self.dir.as_path().join(remote_path);
@@ -353,3 +374,34 @@ impl GCSRemoteFs {
         )
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The payload a bucket really answers with, so a change in how the crate maps reasons shows
+    /// up here instead of as a returning flood of errors from the scheduler.
+    const NOT_FOUND_RESPONSE: &str = r#"{"error":{"errors":[{"domain":"global",
+        "reason":"notFound","message":"No such object: bucket/1829139-ihr4wfxi.parquet"}],
+        "code":404,"message":"No such object: bucket/1829139-ihr4wfxi.parquet"}}"#;
+
+    const FORBIDDEN_RESPONSE: &str = r#"{"error":{"errors":[{"domain":"global",
+        "reason":"forbidden","message":"Access denied"}],"code":403,"message":"Access denied"}}"#;
+
+    fn google_error(payload: &str) -> cloud_storage::Error {
+        cloud_storage::Error::Google(serde_json::from_str(payload).unwrap())
+    }
+
+    #[test]
+    fn absent_object_is_recognized() {
+        assert!(is_absent_object(&google_error(NOT_FOUND_RESPONSE)));
+    }
+
+    #[test]
+    fn other_failures_are_not_absent_objects() {
+        assert!(!is_absent_object(&google_error(FORBIDDEN_RESPONSE)));
+        assert!(!is_absent_object(&cloud_storage::Error::Other(
+            "connection reset".to_string()
+        )));
+    }
+}

rust/cubestore/cubestore/src/remotefs/queue.rs

@@ -288,12 +288,23 @@ impl RemoteFs for QueueRemoteFs {
                             return Ok(f);
                         }
                         Err(err) => {
-                            //Check if file doesn't exists in remoteFs
-                            if self.remote_fs.list(file.clone()).await?.is_empty() {
-                                return Err(CubeError::corrupt_data(format!(
-                                    "File {} doesn't exist in remote file system",
-                                    file
-                                )));
+                            // A listing is the only way to tell an absent object from a remote
+                            // that can't be reached at all: object stores answer 404 for both a
+                            // missing key and a missing bucket. A listing that fails itself leaves
+                            // the download error to speak for both, rather than replacing it.
+                            match self.remote_fs.list(file.clone()).await {
+                                Ok(listing) if listing.is_empty() => {
+                                    return Err(CubeError::file_not_found(format!(
+                                        "File {} doesn't exist in remote file system",
+                                        file
+                                    )));
+                                }
+                                Ok(_) => {}
+                                Err(list_err) => log::warn!(
+                                    "Could not check whether {} exists in remote fs: {}",
+                                    file,
+                                    list_err
+                                ),
                             }
                             return Err(err);
                         }