replicators: When restarting replicator, log the position we were at.

When we restart the replicator, we start from the last position we
applied to a table. In case of an instance where we do not snapshot
all tables, the position we error out vs the position we restart at
might be different. For a better debugging experience, log the
last parsed position before erroring out.

Release-Note-Core: Improve logging when restarting replicator to
  print the last replication position.
Change-Id: I1fac045c7253e8ca141b295899618d61226369c1
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/9739
Tested-by: Buildkite CI
Reviewed-by: Michael Zink <michael.z@readyset.io>

Author: altmannmarcelo

replicators/src/mysql_connector/snapshot.rs

@@ -716,7 +716,9 @@ impl MySqlReplicator<'_> {
                             "Compacting table",
                             table = %table.display(readyset_sql::Dialect::MySQL)
                         );
-                        span.in_scope(|| info!("Setting replication offset"));
+                        span.in_scope(
+                            || info!(repl_offset = %repl_offset, "Setting replication offset"),
+                        );
                         noria_table
                             .set_replication_offset(repl_offset)
                             .map_err(log_err)

replicators/src/noria_adapter.rs

@@ -469,9 +469,13 @@ impl<'a> NoriaAdapter<'a> {
             Some(max) if max > &current_pos => {
                 info!(start = %current_pos, end = %max, "Catching up");
                 let max = max.clone();
-                adapter
+                if let Err(err) = adapter
                     .main_loop(&mut current_pos, Some(max), controller_tx, replicator_rx)
-                    .await?;
+                    .await
+                {
+                    warn!(position = %current_pos, "Replicator stopped during catch-up phase");
+                    return Err(err);
+                }
             }
             _ => {}
         }
@@ -481,9 +485,13 @@ impl<'a> NoriaAdapter<'a> {
         let _ = controller_tx.send(ControllerMessage::SnapshotDone);
 
         info!(position = %current_pos, "Streaming replication started");
-        adapter
+        if let Err(err) = adapter
             .main_loop(&mut current_pos, None, controller_tx, replicator_rx)
-            .await?;
+            .await
+        {
+            warn!(position = %current_pos, "Replicator stopped during streaming replication");
+            return Err(err);
+        }
 
         unreachable!("`main_loop` will never stop with an Ok status if `until = None`");
     }
@@ -719,19 +727,27 @@ impl<'a> NoriaAdapter<'a> {
 
         if min_pos != max_pos {
             info!(start = %min_pos, end = %max_pos, "Catching up");
-            adapter
+            if let Err(err) = adapter
                 .main_loop(&mut min_pos, Some(max_pos), controller_tx, replicator_rx)
-                .await?;
+                .await
+            {
+                warn!(position = %min_pos, "Replicator stopped during catch-up phase");
+                return Err(err);
+            }
         }
 
         // Let Controller know that the initial snapshotting is complete. Ignores the error, which
         // will not occur unless the Controller dropped the rx half of this channel.
         let _ = controller_tx.send(ControllerMessage::SnapshotDone);
 
         info!(position = %min_pos, "Streaming replication started");
-        adapter
+        if let Err(err) = adapter
             .main_loop(&mut min_pos, None, controller_tx, replicator_rx)
-            .await?;
+            .await
+        {
+            warn!(position = %min_pos, "Replicator stopped during streaming replication");
+            return Err(err);
+        }
 
         unreachable!("`main_loop` will never stop with an Ok status if `until = None`");
     }