Add support for starting/restarting without state in leader mode

crates/arroyo-api/queries/api_queries.sql

@@ -183,7 +183,6 @@ UPDATE job_configs
 SET
    updated_at = :updated_at,
    updated_by = :updated_by,
-   stop = 'none',
    restart_nonce = restart_nonce + 1,
    restart_mode = :mode,
    ignore_state_before_epoch = :ignore_state_before_epoch

crates/arroyo-api/src/pipelines.rs

@@ -869,43 +869,39 @@ pub async fn restart_pipeline(
 ) -> Result<Json<Pipeline>, ErrorResp> {
     let auth_data = authenticate(&state.database, bearer_auth).await?;
 
+    if req.ignore_state.unwrap_or(false)
+        && matches!(config().job_controller, JobControllerMode::Worker)
+    {
+        jobs::replace_job_without_state(&state.database, &id, &auth_data).await?;
+
+        let db = state.database.client().await?;
+        let pipeline = query_pipeline_by_pub_id(&id, &db, &auth_data).await?;
+        return Ok(Json(pipeline));
+    }
+
     let db = state.database.client().await?;
 
-    let job = api_queries::fetch_get_pipeline_jobs(&db, &auth_data.organization_id, &id)
+    let job_id = api_queries::fetch_get_pipeline_jobs(&db, &auth_data.organization_id, &id)
         .await?
         .into_iter()
         .next()
-        .ok_or_else(|| bad_request("No jobs for pipeline"))?;
+        .ok_or_else(|| bad_request("No jobs for pipeline"))?
+        .id;
 
     let mode = if req.force == Some(true) {
         RestartMode::force
     } else {
         RestartMode::safe
     };
 
+    // If user wants to ignore state, query max checkpoint epoch and compute threshold
     let ignore_before_epoch = if req.ignore_state.unwrap_or(false) {
-        match config().job_controller {
-            JobControllerMode::Controller => {
-                // Controller mode uses this as an epoch threshold.
-                api_queries::fetch_max_checkpoint_epoch(&db, &job.id, &auth_data.organization_id)
-                    .await?
-                    .into_iter()
-                    .next()
-                    .and_then(|r| r.max_epoch)
-                    .map(|max_epoch| max_epoch + 1)
-            }
-            JobControllerMode::Worker => {
-                // Leader mode uses this as the generation that should start without state.
-                Some(
-                    job.run_id
-                        .unwrap_or(0)
-                        .max(0)
-                        .checked_add(1)
-                        .and_then(|generation| generation.try_into().ok())
-                        .ok_or_else(|| bad_request("Job generation is too large to restart"))?,
-                )
-            }
-        }
+        api_queries::fetch_max_checkpoint_epoch(&db, &job_id, &auth_data.organization_id)
+            .await?
+            .into_iter()
+            .next()
+            .and_then(|r| r.max_epoch)
+            .map(|max_epoch| max_epoch + 1)
     } else {
         None
     };
@@ -916,7 +912,7 @@ pub async fn restart_pipeline(
         &auth_data.user_id,
         &mode,
         &ignore_before_epoch,
-        &job.id,
+        &job_id,
         &auth_data.organization_id,
     )
     .await?;

crates/arroyo-api/src/rest_utils.rs

@@ -168,6 +168,13 @@ pub(crate) fn bad_request(message: impl Into<String>) -> ErrorResp {
     }
 }
 
+pub(crate) fn conflict(message: impl Into<String>) -> ErrorResp {
+    ErrorResp {
+        status_code: StatusCode::CONFLICT,
+        message: message.into(),
+    }
+}
+
 pub(crate) fn service_unavailable(object: &str) -> ErrorResp {
     ErrorResp {
         status_code: StatusCode::SERVICE_UNAVAILABLE,

crates/arroyo-controller/src/lib.rs

@@ -111,8 +111,6 @@ pub struct JobConfig {
     parallelism_overrides: HashMap<u32, usize>,
     restart_nonce: i32,
     restart_mode: RestartMode,
-    /// Minimum checkpoint epoch in controller mode; generation to start without state in leader
-    /// mode.
     ignore_state_before_epoch: Option<i32>,
     /// Per-job environment variables forwarded to workers at scheduling time.
     env_vars: serde_json::Value,

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

@@ -472,7 +472,6 @@ async fn get_and_register_checkpoint_info_leader<'a>(
             job_id: JobId(ctx.config.id.clone()),
             generation: Generation(ctx.status.generation),
             updated_at: SystemTime::now(),
-            ignore_state,
         },
         true,
     )

crates/arroyo-state-protocol/src/lib.rs

@@ -759,7 +759,6 @@ mod tests {
                 job_id: JobId::new("J"),
                 generation: Generation(1),
                 updated_at: from_micros(123),
-                ignore_state: false,
             },
             false,
         )
@@ -815,7 +814,6 @@ mod tests {
                 job_id: JobId::new("J"),
                 generation: Generation(2),
                 updated_at: from_micros(456),
-                ignore_state: false,
             },
             false,
         )
@@ -847,62 +845,6 @@ mod tests {
         assert_eq!(written_manifest, expected_manifest);
     }
 
-    #[tokio::test]
-    async fn initialize_generation_can_ignore_previous_checkpoint() {
-        let store = MemoryProtocolStore::default();
-        let paths = ProtocolPaths::new(PipelineId::new("P"), JobId::new("J"));
-        write_current_generation(&store, &paths, Generation(2)).await;
-
-        let checkpoint_ref = paths.checkpoint_manifest(Generation(1), Epoch(1));
-        let checkpoint = checkpoint_for_generation(Generation(1), 1, None, false);
-        write_canonical_checkpoint(&store, &paths, &checkpoint_ref, &checkpoint).await;
-        let previous_manifest =
-            generation_manifest_for_generation(Generation(1), None, Some(checkpoint_ref));
-        put_json(
-            &store,
-            &paths.generation_manifest(Generation(1)),
-            &previous_manifest,
-        )
-        .await
-        .unwrap();
-
-        let initialization = initialize_generation(
-            &store,
-            InitializeGenerationRequest {
-                pipeline_id: PipelineId::new("P"),
-                job_id: JobId::new("J"),
-                generation: Generation(2),
-                updated_at: from_micros(456),
-                ignore_state: true,
-            },
-            false,
-        )
-        .await
-        .unwrap();
-
-        let expected_manifest = GenerationManifest::new(
-            PipelineId::new("P"),
-            JobId::new("J"),
-            Generation(2),
-            None,
-            456,
-        );
-        assert_eq!(
-            initialization,
-            GenerationInitialization::Initialized {
-                generation_manifest: expected_manifest.clone(),
-                recovery: GenerationRecovery::NoCheckpoint,
-            }
-        );
-
-        let written_manifest: GenerationManifest =
-            read_json(&store, &paths.generation_manifest(Generation(2)))
-                .await
-                .unwrap()
-                .expect("new generation manifest should be written");
-        assert_eq!(written_manifest, expected_manifest);
-    }
-
     #[tokio::test]
     async fn initialize_generation_restores_previous_checkpoint_requiring_commit_replay() {
         let store = MemoryProtocolStore::default();
@@ -929,7 +871,6 @@ mod tests {
                 job_id: JobId::new("J"),
                 generation: Generation(2),
                 updated_at: from_micros(456),
-                ignore_state: false,
             },
             false,
         )
@@ -978,7 +919,6 @@ mod tests {
                 job_id: JobId::new("J"),
                 generation: Generation(3),
                 updated_at: from_micros(789),
-                ignore_state: false,
             },
             false,
         )
@@ -1026,7 +966,6 @@ mod tests {
                 job_id: JobId::new("J"),
                 generation: Generation(2),
                 updated_at: from_micros(456),
-                ignore_state: false,
             },
             false,
         )
@@ -1083,7 +1022,6 @@ mod tests {
                 job_id: JobId::new("J"),
                 generation: Generation(3),
                 updated_at: from_micros(456),
-                ignore_state: false,
             },
             false,
         )
@@ -1149,7 +1087,6 @@ mod tests {
                 job_id: JobId::new("J"),
                 generation: Generation(3),
                 updated_at: from_micros(456),
-                ignore_state: false,
             },
             false,
         )
@@ -1187,7 +1124,6 @@ mod tests {
                 job_id: JobId::new("J"),
                 generation: Generation(2),
                 updated_at: from_micros(456),
-                ignore_state: false,
             },
             false,
         )

crates/arroyo-state-protocol/src/workflow.rs

@@ -71,8 +71,6 @@ pub struct InitializeGenerationRequest {
     pub job_id: JobId,
     pub generation: Generation,
     pub updated_at: SystemTime,
-    /// Start this generation without restoring a checkpoint from an earlier generation.
-    pub ignore_state: bool,
 }
 
 /// Checkpoint, if any, that a newly initialized generation should restore from.
@@ -261,11 +259,7 @@ where
         });
     }
 
-    let recovery = if request.ignore_state {
-        RecoverySearch::Found(GenerationRecovery::NoCheckpoint)
-    } else {
-        find_recovery_checkpoint(store, &paths, request.generation).await?
-    };
+    let recovery = find_recovery_checkpoint(store, &paths, request.generation).await?;
     let base_checkpoint_ref = match &recovery {
         RecoverySearch::Found(recovery) => match recovery {
             GenerationRecovery::NoCheckpoint => None,

crates/arroyo-worker/src/job_controller/controller.rs

@@ -160,9 +160,6 @@ impl WorkerJobController {
                 job_id: worker_context.job_id.clone(),
                 generation: Generation(worker_context.generation),
                 updated_at: SystemTime::now(),
-                // The controller passes no parent checkpoint when this generation should start
-                // without state (or when there is no state available).
-                ignore_state: parent_ref.is_none(),
             },
             false,
         )

webui/src/lib/data_fetching.ts

@@ -524,7 +524,7 @@ export const usePipeline = (pipelineId?: string, refresh: boolean = false) => {
       params: { path: { id: pipelineId } },
       body: { ignore_state: ignoreState ?? false },
     });
-    await mutate();
+    await Promise.all([mutate(), globalMutate(pipelineJobsKey(pipelineId))]);
   };
 
   const deletePipeline = async () => {