Better report errors when a writing task fails

Author: sandhose

crates/syn2mas/src/migration.rs

@@ -15,7 +15,7 @@ use std::{pin::pin, time::Instant};
 
 use chrono::{DateTime, Utc};
 use compact_str::CompactString;
-use futures_util::{SinkExt, StreamExt as _, TryStreamExt as _};
+use futures_util::{SinkExt, StreamExt as _, TryFutureExt, TryStreamExt as _};
 use mas_storage::Clock;
 use rand::{RngCore, SeedableRng};
 use thiserror::Error;
@@ -310,15 +310,20 @@ async fn migrate_users(
         .instrument(tracing::info_span!("ingest_task")),
     );
 
-    synapse
+    // In case this has an error, we still want to join the task, so we look at the
+    // error later
+    let res = synapse
         .read_users()
         .with_progress_bar(count_hint, 10_000)
         .map_err(|e| e.into_synapse("reading users"))
         .forward(PollSender::new(tx).sink_map_err(|_| Error::ChannelClosed))
-        .await?;
+        .inspect_err(|e| tracing::error!(error = e as &dyn std::error::Error))
+        .await;
 
     let (mas, state) = task.await.into_join("user write task")??;
 
+    res?;
+
     info!(
         "users migrated in {:.1}s",
         Instant::now().duration_since(start).as_secs_f64()
@@ -626,15 +631,20 @@ async fn migrate_devices(
         .instrument(tracing::info_span!("ingest_task")),
     );
 
-    synapse
+    // In case this has an error, we still want to join the task, so we look at the
+    // error later
+    let res = synapse
         .read_devices()
         .with_progress_bar(count_hint, 10_000)
         .map_err(|e| e.into_synapse("reading devices"))
         .forward(PollSender::new(tx).sink_map_err(|_| Error::ChannelClosed))
-        .await?;
+        .inspect_err(|e| tracing::error!(error = e as &dyn std::error::Error))
+        .await;
 
     let (mas, state) = task.await.into_join("device write task")??;
 
+    res?;
+
     info!(
         "devices migrated in {:.1}s",
         Instant::now().duration_since(start).as_secs_f64()
@@ -763,15 +773,20 @@ async fn migrate_unrefreshable_access_tokens(
         .instrument(tracing::info_span!("ingest_task")),
     );
 
-    synapse
+    // In case this has an error, we still want to join the task, so we look at the
+    // error later
+    let res = synapse
         .read_unrefreshable_access_tokens()
         .with_progress_bar(count_hint, 10_000)
         .map_err(|e| e.into_synapse("reading tokens"))
         .forward(PollSender::new(tx).sink_map_err(|_| Error::ChannelClosed))
-        .await?;
+        .inspect_err(|e| tracing::error!(error = e as &dyn std::error::Error))
+        .await;
 
     let (mas, state) = task.await.into_join("token write task")??;
 
+    res?;
+
     info!(
         "non-refreshable access tokens migrated in {:.1}s",
         Instant::now().duration_since(start).as_secs_f64()