Always advance generation when creating a new cluster in scheduling (ArroyoSystems#1065)

Author: mwylde

crates/arroyo-controller/src/states/mod.rs

@@ -194,13 +194,7 @@ impl TransitionTo<Compiling> for Stopped {}
 
 impl TransitionTo<Compiling> for Scheduling {}
 
-impl TransitionTo<Scheduling> for Compiling {
-    fn update_status(&self) -> TransitionFn {
-        Box::new(|ctx| {
-            ctx.status.generation += 1;
-        })
-    }
-}
+impl TransitionTo<Scheduling> for Compiling {}
 
 impl TransitionTo<Running> for Scheduling {
     fn update_status(&self) -> TransitionFn {
@@ -257,13 +251,7 @@ impl TransitionTo<Recovering> for Scheduling {
 
 impl TransitionTo<Rescaling> for Running {}
 
-impl TransitionTo<Scheduling> for Rescaling {
-    fn update_status(&self) -> TransitionFn {
-        Box::new(|ctx| {
-            ctx.status.generation += 1;
-        })
-    }
-}
+impl TransitionTo<Scheduling> for Rescaling {}
 
 impl TransitionTo<Compiling> for Recovering {}
 impl TransitionTo<Compiling> for Failed {
@@ -352,13 +340,7 @@ impl TransitionTo<Stopping> for LeaderFinishing {}
 
 impl TransitionTo<LeaderCheckpointStopping> for LeaderRestarting {}
 impl TransitionTo<LeaderRestarting> for LeaderRestarting {}
-impl TransitionTo<Scheduling> for LeaderRestarting {
-    fn update_status(&self) -> TransitionFn {
-        Box::new(|ctx| {
-            ctx.status.generation += 1;
-        })
-    }
-}
+impl TransitionTo<Scheduling> for LeaderRestarting {}
 
 impl TransitionTo<Recovering> for LeaderRestarting {
     fn update_status(&self) -> TransitionFn {
@@ -369,13 +351,8 @@ impl TransitionTo<Recovering> for LeaderRestarting {
 }
 impl TransitionTo<LeaderStopping> for LeaderRestarting {}
 
-impl TransitionTo<Scheduling> for LeaderRescaling {
-    fn update_status(&self) -> TransitionFn {
-        Box::new(|ctx| {
-            ctx.status.generation += 1;
-        })
-    }
-}
+impl TransitionTo<Scheduling> for LeaderRescaling {}
+
 impl TransitionTo<Recovering> for LeaderRescaling {
     fn update_status(&self) -> TransitionFn {
         Box::new(|ctx| {
@@ -406,13 +383,7 @@ impl TransitionTo<Restarting> for Running {
     }
 }
 impl TransitionTo<Restarting> for Restarting {}
-impl TransitionTo<Scheduling> for Restarting {
-    fn update_status(&self) -> TransitionFn {
-        Box::new(|ctx| {
-            ctx.status.generation += 1;
-        })
-    }
-}
+impl TransitionTo<Scheduling> for Restarting {}
 impl TransitionTo<Stopping> for Restarting {}
 impl TransitionTo<CheckpointStopping> for Restarting {}
 

crates/arroyo-controller/src/states/scheduling.rs

@@ -608,6 +608,17 @@ impl State for Scheduling {
         // to schedule
         stop_if_desired_non_running!(self, &ctx.config);
 
+        // update the generation for this scheduling attempt
+        ctx.status.generation += 1;
+        if let Err(e) = ctx.status.update_db(&ctx.db).await {
+            return Err(ctx.retryable(
+                self,
+                "failed to advance generation for scheduling retry",
+                anyhow!("{}", e),
+                10,
+            ));
+        }
+
         // clear out any existing workers for this job
         if let Err(e) = ctx.scheduler.stop_workers(&ctx.config.id, None, true).await {
             warn!(
@@ -683,7 +694,12 @@ impl State for Scheduling {
 
         for h in handles {
             if let Err(e) = h.await {
-                return Err(fatal("Failed to start cluster for pipeline", e.into()));
+                return Err(ctx.retryable(
+                    self,
+                    "Failed to start cluster for pipeline",
+                    e.into(),
+                    10,
+                ));
             }
         }