Cargo fmt

Cargo.toml

@@ -1,4 +1,3 @@
 [workspace]
-members = [
-    "server",
-]
+members = ["server"]
+resolver = "2"

server/src/analytics.rs

@@ -113,7 +113,9 @@ fn total_event_stats() -> (u64, u64, u64) {
     let mut total_json_bytes: u64 = 0;
 
     for stream in metadata::STREAM_INFO.list_streams() {
-        let Some(stats) = stats::get_current_stats(&stream, "json") else {continue;};
+        let Some(stats) = stats::get_current_stats(&stream, "json") else {
+            continue;
+        };
         total_events += stats.events;
         total_parquet_bytes += stats.storage;
         total_json_bytes += stats.ingestion;

server/src/event/format/json.rs

@@ -158,7 +158,9 @@ fn fields_mismatch(schema: &[Arc<Field>], body: &Value) -> bool {
         if val.is_null() {
             continue;
         }
-        let Some(field) = get_field(schema, name) else { return true };
+        let Some(field) = get_field(schema, name) else {
+            return true;
+        };
         if !valid_type(field.data_type(), val) {
             return true;
         }

server/src/query/filter_optimizer.rs

@@ -43,7 +43,9 @@ impl OptimizerRule for FilterOptimizerRule {
         config: &dyn datafusion::optimizer::OptimizerConfig,
     ) -> datafusion::error::Result<Option<datafusion::logical_expr::LogicalPlan>> {
         // if there are no patterns then the rule cannot be performed
-        let Some(filter_expr) = self.expr() else { return Ok(None); };
+        let Some(filter_expr) = self.expr() else {
+            return Ok(None);
+        };
 
         if let LogicalPlan::Filter(filter) = plan {
             if filter.predicate == filter_expr {
@@ -115,7 +117,9 @@ impl FilterOptimizerRule {
             Expr::Column(Column::from_name(&self.column)).like(lit(format!("%{}%", literal)))
         });
 
-        let Some(mut filter_expr) = patterns.next() else { return None };
+        let Some(mut filter_expr) = patterns.next() else {
+            return None;
+        };
         for expr in patterns {
             filter_expr = or(filter_expr, expr)
         }

server/src/storage/retention.rs

@@ -133,7 +133,9 @@ impl TryFrom<Vec<TaskView>> for Retention {
             if !duration.ends_with('d') {
                 return Err("missing 'd' suffix for duration value".to_string());
             }
-            let Ok(days) = duration[0..duration.len() - 1].parse() else { return Err("could not convert duration to an unsigned number".to_string());};
+            let Ok(days) = duration[0..duration.len() - 1].parse() else {
+                return Err("could not convert duration to an unsigned number".to_string());
+            };
 
             if set.contains(&task.action) {
                 return Err(format!(
@@ -184,8 +186,14 @@ mod action {
         log::info!("running retention task - delete");
         let retain_until = get_retain_until(Utc::now().date_naive(), days as u64);
 
-        let Ok(dates) = CONFIG.storage().get_object_store().list_dates(&stream_name)
-        .await else { return };
+        let Ok(dates) = CONFIG
+            .storage()
+            .get_object_store()
+            .list_dates(&stream_name)
+            .await
+        else {
+            return;
+        };
 
         let dates_to_delete = dates
             .into_iter()

server/src/storage/staging.rs

@@ -87,8 +87,9 @@ impl StorageDir {
     }
 
     pub fn arrow_files(&self) -> Vec<PathBuf> {
-        let Ok(dir) = self.data_path
-            .read_dir() else { return vec![] };
+        let Ok(dir) = self.data_path.read_dir() else {
+            return vec![];
+        };
 
         let paths: Vec<PathBuf> = dir
             .flatten()
@@ -143,8 +144,9 @@ impl StorageDir {
     }
 
     pub fn parquet_files(&self) -> Vec<PathBuf> {
-        let Ok(dir) = self.data_path
-            .read_dir() else { return vec![] };
+        let Ok(dir) = self.data_path.read_dir() else {
+            return vec![];
+        };
 
         dir.flatten()
             .map(|file| file.path())

server/src/utils.rs

@@ -43,7 +43,9 @@ pub fn capitalize_ascii(s: &str) -> String {
 }
 
 pub fn validate_path_is_writeable(path: &Path) -> anyhow::Result<()> {
-    let Ok(md) = std::fs::metadata(path) else { anyhow::bail!("Could not read metadata for staging dir") };
+    let Ok(md) = std::fs::metadata(path) else {
+        anyhow::bail!("Could not read metadata for staging dir")
+    };
     let permissions = md.permissions();
     if permissions.readonly() {
         anyhow::bail!("Staging directory {} is unwritable", path.display())

server/src/utils/json/flatten.rs

@@ -32,7 +32,9 @@ pub fn flatten(nested_value: Value, separator: &str) -> Result<Value, anyhow::Er
             for _value in &mut arr {
                 let value = std::mem::replace(_value, Value::Null);
                 let mut map = Map::new();
-                let Value::Object(obj) = value else { return Err(anyhow!("Expected object in array of objects")) };
+                let Value::Object(obj) = value else {
+                    return Err(anyhow!("Expected object in array of objects"));
+                };
                 flatten_object(&mut map, None, obj, separator)?;
                 *_value = Value::Object(map);
             }
@@ -255,7 +257,9 @@ mod tests {
             null,
             {"p": 2, "q": 2},
             {"q": 3},
-        ]) else { unreachable!() };
+        ]) else {
+            unreachable!()
+        };
 
         let mut map = Map::new();
         flatten_array_objects(&mut map, "key", arr, ".").unwrap();
@@ -267,7 +271,9 @@ mod tests {
 
     #[test]
     fn flatten_array_objects_nulls_at_end() {
-        let Value::Array(arr) = json!([{"a": 1, "b": 1}, {"a": 2}, null]) else { unreachable!() };
+        let Value::Array(arr) = json!([{"a": 1, "b": 1}, {"a": 2}, null]) else {
+            unreachable!()
+        };
 
         let mut map = Map::new();
         flatten_array_objects(&mut map, "key", arr, ".").unwrap();
@@ -279,7 +285,9 @@ mod tests {
 
     #[test]
     fn flatten_array_objects_nulls_in_middle() {
-        let Value::Array(arr) = json!([{"a": 1, "b": 1}, null, {"a": 3, "c": 3}]) else { unreachable!() };
+        let Value::Array(arr) = json!([{"a": 1, "b": 1}, null, {"a": 3, "c": 3}]) else {
+            unreachable!()
+        };
 
         let mut map = Map::new();
         flatten_array_objects(&mut map, "key", arr, ".").unwrap();
@@ -296,7 +304,9 @@ mod tests {
             {"p": 1, "q": 1},
             {"r": 2, "q": 2},
             {"p": 3, "r": 3}
-        ]) else { unreachable!() };
+        ]) else {
+            unreachable!()
+        };
 
         let mut map = Map::new();
         flatten_array_objects(&mut map, "key", arr, ".").unwrap();
@@ -313,7 +323,9 @@ mod tests {
             {"p": 1, "q": [{"x": 1}, {"x": 2}]},
             {"r": 2, "q": [{"x": 1}]},
             {"p": 3, "r": 3}
-        ]) else { unreachable!() };
+        ]) else {
+            unreachable!()
+        };
 
         let mut map = Map::new();
         flatten_array_objects(&mut map, "key", arr, ".").unwrap();