JoinHandle is infallible

Tasks can fail in theory for two ways:
- Panic! However, we disable panics in production.
  (And for tests, we percolate the panic upward.)
- Task is explicitly aborted. This is statically ruled out by the
  wrapper.
- Runtime is shutting down. This is not interesting - it means the
  process is shutting down - and there's no value to percolating errors
  in that case. Here, we just remain pending indefinitely until the
  runtime dies.

Author: bkirwi

src/adapter/src/catalog.rs

@@ -3372,7 +3372,7 @@ mod tests {
 
         let handles = Catalog::with_debug(|catalog| async { inner(catalog) }).await;
         for handle in handles {
-            handle.await.expect("must succeed");
+            handle.await;
         }
     }
 

src/adapter/src/catalog/apply.rs

@@ -1420,7 +1420,7 @@ impl CatalogState {
             // Wait for a view to be ready.
             let (handle, _idx, remaining) = future::select_all(handles).await;
             handles = remaining;
-            let (id, global_id, res) = handle.expect("must join");
+            let (id, global_id, res) = handle;
             let mut insert_cached_expr = |cached_expr| {
                 if let Some(cached_expr) = cached_expr {
                     local_expression_cache.insert_cached_expression(global_id, cached_expr);

src/adapter/src/coord.rs

@@ -2553,15 +2553,12 @@ impl Coordinator {
 
         let retractions_res = futures::future::join_all(retraction_tasks).await;
         for retractions in retractions_res {
-            let retractions = retractions.expect("cannot fail to fetch snapshot");
             builtin_table_updates.push(retractions);
         }
 
         let audit_join_start = Instant::now();
         info!("startup: coordinator init: bootstrap: join audit log deserialization beginning");
-        let audit_log_updates = audit_log_task
-            .await
-            .expect("cannot fail to fetch audit log updates");
+        let audit_log_updates = audit_log_task.await;
         let audit_log_builtin_table_updates = self
             .catalog()
             .state()

src/adapter/src/coord/sequencer/inner.rs

@@ -252,16 +252,7 @@ impl Coordinator {
         spawn(|| "sequence_staged", async move {
             tokio::select! {
                 res = handle => {
-                    let next = match res {
-                        Ok(next) => return_if_err!(next, ctx),
-                        Err(err) => {
-                            tracing::error!("sequence_staged join error {err}");
-                            ctx.retire(Err(AdapterError::Internal(
-                                "sequence_staged join error".into(),
-                            )));
-                            return;
-                        }
-                    };
+                    let next = return_if_err!(res, ctx);
                     f(ctx, next);
                 }
                 _ = rx, if cancel_enabled => {

src/adapter/src/explain/insights.rs

@@ -132,7 +132,7 @@ impl PlanInsights {
             });
         for task in tasks {
             let res = task.await;
-            let Ok(Ok((name, plan))) = res else {
+            let Ok((name, plan)) = res else {
                 continue;
             };
             let (plan, _, _) = plan.unapply();

src/adapter/src/webhook.rs

@@ -223,7 +223,7 @@ impl AppendWebhookValidator {
         .map_err(|e| {
             tracing::error!("Failed to run validation for webhook, {e}");
             AppendWebhookError::ValidationError
-        })??;
+        })?;
 
         valid
     }

src/aws-util/src/s3_uploader.rs

@@ -20,7 +20,7 @@ use bytes::{Bytes, BytesMut};
 use bytesize::ByteSize;
 use mz_ore::cast::CastFrom;
 use mz_ore::error::ErrorExt;
-use mz_ore::task::{JoinHandle, JoinHandleExt, spawn};
+use mz_ore::task::{JoinHandle, spawn};
 
 /// A multi part uploader which can upload a single object across multiple parts
 /// and keeps track of state to eventually finish the upload process.
@@ -226,7 +226,7 @@ impl S3MultiPartUploader {
 
         let mut parts: Vec<CompletedPart> = Vec::with_capacity(self.upload_handles.len());
         for handle in self.upload_handles {
-            let (etag, part_num) = handle.wait_and_assert_finished().await?;
+            let (etag, part_num) = handle.await?;
             match etag {
                 Some(etag) => {
                     parts.push(

src/balancerd/tests/server.rs

@@ -334,7 +334,7 @@ async fn test_balancer() {
                     handles.push(handle);
                 }
                 for handle in handles {
-                    handle.await.unwrap();
+                    handle.await;
                 }
                 let end_auth_count = *frontegg_server.auth_requests.lock().unwrap();
                 // We expect that the auth count increased by fewer than the number of connections.

src/catalog/tests/debug.rs

@@ -589,7 +589,7 @@ async fn test_concurrent_debugs(state_builder: TestCatalogStateBuilder) {
         .await
         .unwrap();
 
-    state_handle.await.unwrap();
+    state_handle.await;
 
     let mut state = state_builder
         .clone()
@@ -625,7 +625,7 @@ async fn test_concurrent_debugs(state_builder: TestCatalogStateBuilder) {
         .await
         .unwrap();
 
-    state_handle.await.unwrap();
+    state_handle.await;
 
     let configs = state_builder
         .clone()

src/catalog/tests/open.rs

@@ -970,7 +970,7 @@ async fn test_concurrent_open(state_builder: TestCatalogStateBuilder) {
         .unwrap()
         .0;
 
-    state_handle.await.unwrap();
+    state_handle.await;
 
     // Open again to ensure that we didn't commit an invalid retraction.
     let _state = state_builder