Add debug logging when delete_objects return Err (#985)

Author: dcherian

icechunk/src/ops/gc.rs

@@ -303,7 +303,7 @@ async fn gc_chunks(
     Ok(storage.delete_chunks(storage_settings, to_delete).await?)
 }
 
-#[instrument(skip(storage, storage_settings, config, keep_ids), fields(keep_ids.len = keep_ids.len()))]
+#[instrument(skip(asset_manager, storage, storage_settings, config, keep_ids), fields(keep_ids.len = keep_ids.len()))]
 async fn gc_manifests(
     asset_manager: &AssetManager,
     storage: &(dyn Storage + Send + Sync),
@@ -331,7 +331,7 @@ async fn gc_manifests(
     Ok(storage.delete_manifests(storage_settings, to_delete).await?)
 }
 
-#[instrument(skip(storage, storage_settings, config, keep_ids), fields(keep_ids.len = keep_ids.len()))]
+#[instrument(skip(asset_manager, storage, storage_settings, config, keep_ids), fields(keep_ids.len = keep_ids.len()))]
 async fn gc_snapshots(
     asset_manager: &AssetManager,
     storage: &(dyn Storage + Send + Sync),
@@ -359,7 +359,7 @@ async fn gc_snapshots(
     Ok(storage.delete_snapshots(storage_settings, to_delete).await?)
 }
 
-#[instrument(skip(storage, storage_settings, config, keep_ids), fields(keep_ids.len = keep_ids.len()))]
+#[instrument(skip(asset_manager, storage, storage_settings, config, keep_ids), fields(keep_ids.len = keep_ids.len()))]
 async fn gc_transaction_logs(
     asset_manager: &AssetManager,
     storage: &(dyn Storage + Send + Sync),

icechunk/src/storage/mod.rs

@@ -33,7 +33,7 @@ use std::{
 };
 use tokio::io::AsyncRead;
 use tokio_util::io::SyncIoBridge;
-use tracing::instrument;
+use tracing::{instrument, warn};
 
 use async_trait::async_trait;
 use bytes::{Buf, Bytes};
@@ -497,9 +497,12 @@ pub trait Storage: fmt::Debug + fmt::Display + private::Sealed + Sync + Send {
             .for_each_concurrent(10, |batch| {
                 let res = Arc::clone(&res);
                 async move {
-                    // FIXME: handle error instead of skipping
                     let new_deletes =
-                        self.delete_batch(prefix, batch).await.unwrap_or_default();
+                        self.delete_batch(prefix, batch).await.unwrap_or_else(|_| {
+                            // FIXME: handle error instead of skipping
+                            warn!("ignoring error in Storage::delete_batch");
+                            Default::default()
+                        });
                     #[allow(clippy::expect_used)]
                     res.lock().expect("Bug in delete objects").merge(&new_deletes);
                 }