replicators: Fix race between domain recovery and dropping table

When a domain fails, we will replace the domain in a background
thread. However, if the failure is due to a replication issue, like
not being able to find a row in the source table, we will remove
the table from readyset. Those two operations will race with each
other.

We should wait for the new domain to be ready before removing the
table. This can be done by checking the replication offsets via the
RPC call. We just need to make the caller wait for the RPC to
succeed instead of returning immediately after the first error.

We use the same approach when we are booting up for the first time
in the noria_adapter.

Closes: REA-5563
Fixes: #1484

Release-Note-Core: Fix a race between domain recovery and dropping
  a table after replication failure.
Change-Id: I9274eca5fcf256ce37bdd4b2b2bbfda946f2952e
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/9146
Tested-by: Buildkite CI
Reviewed-by: Jason Brown <jason.b@readyset.io>

Author: altmannmarcelo

replicators/src/noria_adapter.rs

@@ -1159,7 +1159,19 @@ impl<'a> NoriaAdapter<'a> {
             table = %table.display(readyset_sql::Dialect::PostgreSQL),
             "Removing table state from readyset"
         );
-        self.noria.replication_offsets().await?;
+
+        // In case of a domain failure, we might be replacing the failed domain.
+        // and at the same time attempting to remove a table from readyset.
+        // We need to wait for the new domain to be ready before removing the table.
+        retry_with_exponential_backoff(
+            || async {
+                let mut noria = self.noria.clone();
+                noria.replication_offsets().await
+            },
+            5,
+            Duration::from_millis(250),
+        )
+        .await?;
         self.replication_offsets.tables.remove(&table);
         self.mutator_map.remove(&table);
         // Dropping the table cleans up any dataflow state that may have been made as well as