Replace panic on unfinished buffers with an error

This doesn't get clobbered by the progress bar

Author: reivilibre

crates/syn2mas/src/mas_writer/mod.rs

@@ -7,7 +7,14 @@
 //!
 //! This module is responsible for writing new records to MAS' database.
 
-use std::{fmt::Display, net::IpAddr};
+use std::{
+    fmt::Display,
+    net::IpAddr,
+    sync::{
+        Arc,
+        atomic::{AtomicU32, Ordering},
+    },
+};
 
 use chrono::{DateTime, Utc};
 use futures_util::{FutureExt, TryStreamExt, future::BoxFuture};
@@ -44,6 +51,9 @@ pub enum Error {
     #[error("inconsistent database: {0}")]
     Inconsistent(String),
 
+    #[error("bug in syn2mas: write buffers not finished")]
+    WriteBuffersNotFinished,
+
     #[error("{0}")]
     Multiple(MultipleErrors),
 }
@@ -188,12 +198,52 @@ impl WriterConnectionPool {
     }
 }
 
+/// Small utility to make sure `finish()` is called on all write buffers
+/// before committing to the database.
+#[derive(Default)]
+struct FinishChecker {
+    counter: Arc<AtomicU32>,
+}
+
+struct FinishCheckerHandle {
+    counter: Arc<AtomicU32>,
+}
+
+impl FinishChecker {
+    /// Acquire a new handle, for a task that should declare when it has
+    /// finished.
+    pub fn handle(&self) -> FinishCheckerHandle {
+        self.counter.fetch_add(1, Ordering::SeqCst);
+        FinishCheckerHandle {
+            counter: Arc::clone(&self.counter),
+        }
+    }
+
+    /// Check that all handles have been declared as finished.
+    pub fn check_all_finished(self) -> Result<(), Error> {
+        if self.counter.load(Ordering::SeqCst) == 0 {
+            Ok(())
+        } else {
+            Err(Error::WriteBuffersNotFinished)
+        }
+    }
+}
+
+impl FinishCheckerHandle {
+    /// Declare that the task this handle represents has been finished.
+    pub fn declare_finished(self) {
+        self.counter.fetch_sub(1, Ordering::SeqCst);
+    }
+}
+
 pub struct MasWriter {
     conn: LockedMasDatabase,
     writer_pool: WriterConnectionPool,
 
     indices_to_restore: Vec<IndexDescription>,
     constraints_to_restore: Vec<ConstraintDescription>,
+
+    write_buffer_finish_checker: FinishChecker,
 }
 
 pub struct MasNewUser {
@@ -453,6 +503,7 @@ impl MasWriter {
             writer_pool: WriterConnectionPool::new(writer_connections),
             indices_to_restore,
             constraints_to_restore,
+            write_buffer_finish_checker: FinishChecker::default(),
         })
     }
 
@@ -520,6 +571,8 @@ impl MasWriter {
     /// - If the database connection experiences an error.
     #[tracing::instrument(skip_all)]
     pub async fn finish(mut self) -> Result<PgConnection, Error> {
+        self.write_buffer_finish_checker.check_all_finished()?;
+
         // Commit all writer transactions to the database.
         self.writer_pool
             .finish()
@@ -1033,28 +1086,24 @@ type WriteBufferFlusher<T> =
 
 /// A buffer for writing rows to the MAS database.
 /// Generic over the type of rows.
-///
-/// # Panics
-///
-/// Panics if dropped before `finish()` has been called.
 pub struct MasWriteBuffer<T> {
     rows: Vec<T>,
     flusher: WriteBufferFlusher<T>,
-    finished: bool,
+    finish_checker_handle: FinishCheckerHandle,
 }
 
 impl<T> MasWriteBuffer<T> {
-    pub fn new(flusher: WriteBufferFlusher<T>) -> Self {
+    pub fn new(writer: &MasWriter, flusher: WriteBufferFlusher<T>) -> Self {
         MasWriteBuffer {
             rows: Vec::with_capacity(WRITE_BUFFER_BATCH_SIZE),
             flusher,
-            finished: false,
+            finish_checker_handle: writer.write_buffer_finish_checker.handle(),
         }
     }
 
     pub async fn finish(mut self, writer: &mut MasWriter) -> Result<(), Error> {
-        self.finished = true;
         self.flush(writer).await?;
+        self.finish_checker_handle.declare_finished();
         Ok(())
     }
 
@@ -1077,12 +1126,6 @@ impl<T> MasWriteBuffer<T> {
     }
 }
 
-impl<T> Drop for MasWriteBuffer<T> {
-    fn drop(&mut self) {
-        assert!(self.finished, "MasWriteBuffer dropped but not finished!");
-    }
-}
-
 #[cfg(test)]
 mod test {
     use std::collections::{BTreeMap, BTreeSet};

crates/syn2mas/src/migration.rs

@@ -232,8 +232,8 @@ async fn migrate_users(
     span.pb_set_style(&ProgressStyle::default_bar());
     span.pb_set_length(count_hint as u64);
 
-    let mut user_buffer = MasWriteBuffer::new(MasWriter::write_users);
-    let mut password_buffer = MasWriteBuffer::new(MasWriter::write_passwords);
+    let mut user_buffer = MasWriteBuffer::new(&mas, MasWriter::write_users);
+    let mut password_buffer = MasWriteBuffer::new(&mas, MasWriter::write_passwords);
     let mut users_stream = pin!(synapse.read_users());
 
     while let Some(user_res) = users_stream.next().await {
@@ -312,8 +312,8 @@ async fn migrate_threepids(
     span.pb_set_style(&ProgressStyle::default_bar());
     span.pb_set_length(count_hint as u64);
 
-    let mut email_buffer = MasWriteBuffer::new(MasWriter::write_email_threepids);
-    let mut unsupported_buffer = MasWriteBuffer::new(MasWriter::write_unsupported_threepids);
+    let mut email_buffer = MasWriteBuffer::new(&mas, MasWriter::write_email_threepids);
+    let mut unsupported_buffer = MasWriteBuffer::new(&mas, MasWriter::write_unsupported_threepids);
     let mut users_stream = pin!(synapse.read_threepids());
 
     while let Some(threepid_res) = users_stream.next().await {
@@ -402,7 +402,7 @@ async fn migrate_external_ids(
     span.pb_set_style(&ProgressStyle::default_bar());
     span.pb_set_length(count_hint as u64);
 
-    let mut write_buffer = MasWriteBuffer::new(MasWriter::write_upstream_oauth_links);
+    let mut write_buffer = MasWriteBuffer::new(&mas, MasWriter::write_upstream_oauth_links);
     let mut extids_stream = pin!(synapse.read_user_external_ids());
 
     while let Some(extid_res) = extids_stream.next().await {
@@ -486,7 +486,7 @@ async fn migrate_devices(
     span.pb_set_length(count_hint as u64);
 
     let mut devices_stream = pin!(synapse.read_devices());
-    let mut write_buffer = MasWriteBuffer::new(MasWriter::write_compat_sessions);
+    let mut write_buffer = MasWriteBuffer::new(&mas, MasWriter::write_compat_sessions);
 
     while let Some(device_res) = devices_stream.next().await {
         span.pb_inc(1);
@@ -590,8 +590,9 @@ async fn migrate_unrefreshable_access_tokens(
     span.pb_set_length(count_hint as u64);
 
     let mut token_stream = pin!(synapse.read_unrefreshable_access_tokens());
-    let mut write_buffer = MasWriteBuffer::new(MasWriter::write_compat_access_tokens);
-    let mut deviceless_session_write_buffer = MasWriteBuffer::new(MasWriter::write_compat_sessions);
+    let mut write_buffer = MasWriteBuffer::new(&mas, MasWriter::write_compat_access_tokens);
+    let mut deviceless_session_write_buffer =
+        MasWriteBuffer::new(&mas, MasWriter::write_compat_sessions);
 
     while let Some(token_res) = token_stream.next().await {
         span.pb_inc(1);
@@ -708,9 +709,10 @@ async fn migrate_refreshable_token_pairs(
     span.pb_set_length(count_hint as u64);
 
     let mut token_stream = pin!(synapse.read_refreshable_token_pairs());
-    let mut access_token_write_buffer = MasWriteBuffer::new(MasWriter::write_compat_access_tokens);
+    let mut access_token_write_buffer =
+        MasWriteBuffer::new(&mas, MasWriter::write_compat_access_tokens);
     let mut refresh_token_write_buffer =
-        MasWriteBuffer::new(MasWriter::write_compat_refresh_tokens);
+        MasWriteBuffer::new(&mas, MasWriter::write_compat_refresh_tokens);
 
     while let Some(token_res) = token_stream.next().await {
         span.pb_inc(1);