use .flatten instead of .map_err(...)?

src/rust/fs/store/src/local.rs

@@ -215,8 +215,8 @@ impl ShardedFSDB {
                     Ok((src_file, dst_dir))
                 })
                 .await
-                .map_err(|e| format!("Failed to join Tokio task: {e}"))?
-                .map_err(|e: String| format!("hardlink canary temp files task failed: {e}"))?;
+                .map_err(|e| format!("hardlink canary temp files task failed: {e}"))
+                .flatten()?;
             let dst_file = dst_dir.path().join("hard_link");
             let is_hardlinkable = hard_link(src_file, dst_file).await.is_ok();
             log::debug!("{src_display} -> {dst_display} hardlinkable: {is_hardlinkable}");
@@ -295,8 +295,8 @@ impl ShardedFSDB {
                         .map_err(|e| format!("Failed to create temp file: {e}"))
                 })
                 .await
-                .map_err(|e| format!("Failed to join Tokio task: {e}"))?
-                .map_err(|e| format!("temp file creation task failed: {e}"))?;
+                .map_err(|e| format!("temp file creation task failed: {e}"))
+                .flatten()?;
             let (std_file, tmp_path) = named_temp_file
                 .keep()
                 .map_err(|e| format!("Failed to keep temp file: {e}"))?;
@@ -369,8 +369,8 @@ impl UnderlyingByteStore for ShardedFSDB {
                     .map_err(|e| format!("Failed to extend mtime of {path:?}: {e}"))
             })
             .await
-            .map_err(|e| format!("Failed to join Tokio task: {e}"))?
             .map_err(|e| format!("`lease` task failed: {e}"))
+            .flatten()
     }
 
     async fn remove(&self, fingerprint: Fingerprint) -> Result<bool, String> {
@@ -516,8 +516,8 @@ impl UnderlyingByteStore for ShardedFSDB {
                 Ok(fingerprints)
             })
             .await
-            .map_err(|e| format!("Failed to join Tokio task: {e}"))?
-            .map_err(|e: String| format!("`aged_fingerprints` task failed: {e}"))
+            .map_err(|e| format!("`aged_fingerprints` task failed: {e}"))
+            .flatten()
     }
 }
 

src/rust/process_execution/docker/src/docker.rs

@@ -316,8 +316,8 @@ async fn credentials_for_image(
             Ok(Some(bollard_credentials))
         })
         .await
-        .map_err(|e| format!("Failed to join Tokio task: {e}"))?
-        .map_err(|e: String| format!("Credentials task failed: {e}"))
+        .map_err(|e| format!("Credentials task failed: {e}"))
+        .flatten()
 }
 
 /// Pull an image given its name and the image pull policy. This method is debounced by

src/rust/process_execution/pe_nailgun/src/nailgun_pool.rs

@@ -394,8 +394,8 @@ impl NailgunProcess {
                 move || spawn_and_read_port(startup_options, workdir)
             })
             .await
-            .map_err(|e| format!("Failed to join Tokio task: {e}"))?
-            .map_err(|e| format!("Nailgun spawn task failed: {e}"))?;
+            .map_err(|e| format!("Nailgun spawn task failed: {e}"))
+            .flatten()?;
         debug!(
             "Created nailgun server process with pid {} and port {}",
             child.id(),

src/rust/sharded_lmdb/src/lib.rs

@@ -296,8 +296,8 @@ impl ShardedLmdb {
                 }
             })
             .await
-            .map_err(|e| format!("Failed to join Tokio task: {e}"))?
             .map_err(|e| format!("`remove` task failed: {e}"))
+            .flatten()
     }
 
     ///
@@ -366,8 +366,8 @@ impl ShardedLmdb {
                 Ok(exists)
             })
             .await
-            .map_err(|e| format!("Failed to join Tokio task: {e}"))?
-            .map_err(|e: String| format!("`exists_batch` task failed: {e}"))
+            .map_err(|e| format!("`exists_batch` task failed: {e}"))
+            .flatten()
     }
 
     ///
@@ -426,8 +426,8 @@ impl ShardedLmdb {
                 Ok(fingerprints)
             })
             .await
-            .map_err(|e| format!("Failed to join Tokio task: {e}"))?
-            .map_err(|e: String| format!("`all_fingerprints` task failed: {e}"))
+            .map_err(|e| format!("`all_fingerprints` task failed: {e}"))
+            .flatten()
     }
 
     ///
@@ -513,8 +513,8 @@ impl ShardedLmdb {
                 Ok(())
             })
             .await
-            .map_err(|e| format!("Failed to join Tokio task: {e}"))?
-            .map_err(|e: String| format!("`store_bytes_batch` task failed: {e}"))
+            .map_err(|e| format!("`store_bytes_batch` task failed: {e}"))
+            .flatten()
     }
 
     ///
@@ -609,8 +609,8 @@ impl ShardedLmdb {
                 }
             })
             .await
-            .map_err(|e| format!("Failed to join Tokio task: {e}"))?
             .map_err(|e| format!("`store` task failed: {e}"))
+            .flatten()
     }
 
     pub async fn lease(&self, fingerprint: Fingerprint) -> Result<(), String> {
@@ -632,8 +632,8 @@ impl ShardedLmdb {
                     .map_err(|e| format!("Error leasing {fingerprint:?}: {e}"))
             })
             .await
-            .map_err(|e| format!("Failed to join Tokio task: {e}"))?
             .map_err(|e| format!("`lease` task failed: {e}"))
+            .flatten()
     }
 
     fn lease_inner(
@@ -685,8 +685,8 @@ impl ShardedLmdb {
                 }
             })
             .await
-            .map_err(|e| format!("Failed to join Tokio task: {e}"))?
             .map_err(|e| format!("`load_bytes_with` task failed: {e}"))
+            .flatten()
     }
 
     #[allow(clippy::useless_conversion)] // False positive: https://github.com/rust-lang/rust-clippy/issues/3913

src/rust/watch/src/lib.rs

@@ -298,8 +298,8 @@ impl InvalidationWatcher {
                     .map_err(|e| maybe_enrich_notify_error(&path, e))
             })
             .await
-            .map_err(|e| format!("Failed to join Tokio task: {e}"))?
             .map_err(|e| format!("Watch attempt failed: {e}"))
+            .flatten()
     }
 }