controller: Don't commit failed migrations

Prior to this commit, we were ignoring errors in the `migrate` function,
which meant that we would still call `Migration::commit` even if `f`
returned an Err. We would still see warnings and such later in the logs
because the error would get returned from `migrate` and then caught
elsewhere in the code, but calling `commit` would still cause some side
effects to happen that shouldn't.

This could cause strange effects, such as nodes being created in domains
but never registered in the controller as part of the graph. I'm not
sure whether this caused any problems in the past (other than leaking
resources) but it started causing snapshotting failures for certain
invalid queries after the merging of...

72f191c17 (dataflow: Fix replication offsets in memory state, 2022-12-01)

...because domains would attempt to return replication offsets for these
orphaned nodes, which would confuse the controller since it would have
no record of these nodes in its state.

To fix this, we immediately return an error if `f` fails in `migrate`,
thus skipping the call to `commit`.

This commit also contains a replicators test that reproduces the
problem prior to this fix, but which now passes with the fix in place.

Release-Note-Core: Fixed a bug that could cause snapshotting and
  replication to stall forever after encountering certain types of
  errors while loading a view from the upstream database.
Co-authored-by: Griffin Smith <[email protected]>
Fixes: ENG-2190
Change-Id: I3ac529de0d0be702172913225440d5f58f23bacf
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/4013
Reviewed-by: Griffin Smith <[email protected]>
Tested-by: Buildkite CI

Author: nickelization

readyset-server/src/controller/mod.rs

@@ -395,12 +395,15 @@ impl Controller {
                 if let Some(ref mut inner) = *guard {
                     let mut writer = inner.dataflow_state_handle.write().await;
                     let ds = writer.as_mut();
-                    let ret = ds.migrate(false, dialect, move |m| func(m)).await?;
-                    inner
-                        .dataflow_state_handle
-                        .commit(writer, &self.authority)
-                        .await?;
-                    if done_tx.send(ret).is_err() {
+                    let res = ds.migrate(false, dialect, move |m| func(m)).await;
+                    if res.is_ok() {
+                        inner
+                            .dataflow_state_handle
+                            .commit(writer, &self.authority)
+                            .await?;
+                    }
+
+                    if done_tx.send(res).is_err() {
                         warn!("handle-based migration sender hung up!");
                     }
                 } else {

readyset-server/src/controller/state.rs

@@ -839,14 +839,14 @@ impl DfState {
         dry_run: bool,
         dialect: Dialect,
         f: F,
-    ) -> Result<T, ReadySetError>
+    ) -> ReadySetResult<T>
     where
-        F: FnOnce(&mut Migration<'_>) -> T,
+        F: FnOnce(&mut Migration<'_>) -> ReadySetResult<T>,
     {
         debug!("starting migration");
         gauge!(recorded::CONTROLLER_MIGRATION_IN_PROGRESS, 1.0);
         let mut m = Migration::new(self, dialect);
-        let r = f(&mut m);
+        let r = f(&mut m)?;
         m.commit(dry_run).await?;
         debug!("finished migration");
         gauge!(recorded::CONTROLLER_MIGRATION_IN_PROGRESS, 0.0);
@@ -1149,11 +1149,11 @@ impl DfState {
             .migrate(dry_run, changelist.dialect, |mig| {
                 new.activate(mig, changelist)
             })
-            .await?;
+            .await;
 
-        match r {
+        match &r {
             Ok(_) => self.recipe = new,
-            Err(ref e) => {
+            Err(e) => {
                 tracing::
                     warn!(error = %e, "failed to apply recipe. Will retry periodically up to max_processing_minutes.");
             }

replicators/tests/tests.rs

@@ -1634,6 +1634,44 @@ async fn postgresql_non_base_offsets() {
         .unwrap();
 }
 
+#[tokio::test(flavor = "multi_thread")]
+#[serial_test::serial]
+async fn postgresql_orphaned_nodes() {
+    // This test checks for the error described in ENG-2190.
+    // The bug would be triggered when a view would fail to be added due to certain types of
+    // errors, because we would still call `commit` on the migration after the error occurred,
+    // resulting in nodes being created in the domain without being added to the controller.
+    readyset_tracing::init_test_logging();
+    let url = pgsql_url();
+    let mut client = DbConnection::connect(&url).await.unwrap();
+
+    // The important thing about creating RS_FAKE_TEST_FUN is that it's not a builtin function, so
+    // we will get a ReadySetError::NoSuchFunction during lowering and view creation will fail.
+    client
+        .query(
+            "CREATE OR REPLACE FUNCTION RS_FAKE_TEST_FUN(bigint) RETURNS bigint
+                 AS 'select $1' LANGUAGE SQL;
+             DROP TABLE IF EXISTS t1 CASCADE;
+             CREATE TABLE t1(i INTEGER);
+             CREATE OR REPLACE VIEW v1 AS
+                SELECT RS_FAKE_TEST_FUN(subq.my_count) FROM
+                    (SELECT count(*) AS my_count FROM t1 GROUP BY i) subq;
+             CREATE OR REPLACE VIEW check_t1 AS SELECT * FROM t1;
+             INSERT INTO t1 VALUES(99);",
+        )
+        .await
+        .unwrap();
+
+    let mut ctx = TestHandle::start_noria(url.to_string(), None)
+        .await
+        .unwrap();
+    ctx.ready_notify.as_ref().unwrap().notified().await;
+
+    ctx.check_results("check_t1", "Snapshot", &[&[99.into()]])
+        .await
+        .unwrap();
+}
+
 #[tokio::test(flavor = "multi_thread")]
 #[serial_test::serial]
 async fn postgresql_replicate_citext() {