Revert "Add support for starting/restarting without state in leader mode (#1111)"

This reverts commit fab1668.

Author: mwylde

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

@@ -52,7 +52,7 @@ use crate::rest_utils::{
 use crate::types::public::{PipelineType, RestartMode, StopMode};
 use crate::udfs::build_udf;
 use crate::{connection_tables, to_micros};
-use arroyo_rpc::config::{JobControllerMode, config};
+use arroyo_rpc::config::config;
 use arroyo_rpc::errors::ErrorDomain;
 use arroyo_types::to_millis;
 use cornucopia_async::{Database, DatabaseSource};
@@ -868,44 +868,29 @@ pub async fn restart_pipeline(
     WithRejection(Json(req), _): WithRejection<Json<PipelineRestart>, ApiError>,
 ) -> Result<Json<Pipeline>, ErrorResp> {
     let auth_data = authenticate(&state.database, bearer_auth).await?;
-
     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 +901,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-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-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/routes/pipelines/PipelineDetails.tsx

@@ -138,7 +138,7 @@ export function PipelineDetails() {
 
   async function updateJobState(stop: StopType) {
     console.log(`Setting pipeline stop_mode=${stop}`);
-    await updatePipeline({ stop });
+    updatePipeline({ stop });
   }
 
   async function updateJobParallelism(parallelism: number) {
@@ -352,29 +352,23 @@ export function PipelineDetails() {
   let actionButton = <></>;
   if (pipeline) {
     editPipelineButton = <Button onClick={onConfigModalOpen}>Edit</Button>;
-    const isStopped = job.state === 'Stopped' && pipeline.action != null;
-    if (job.state === 'Failed' || isStopped) {
-      const actionText = isStopped ? 'Start' : 'Restart';
+    if (job.state == 'Failed') {
       actionButton = (
         <Popover trigger="hover" placement="bottom-start">
           <PopoverTrigger>
             <Button
               onClick={async () => {
-                if (isStopped) {
-                  await updateJobState(pipeline.action!);
-                } else {
-                  await restartPipeline(false);
-                }
+                await restartPipeline(false);
               }}
             >
-              {actionText}
+              Restart
             </Button>
           </PopoverTrigger>
           <PopoverContent width="auto">
             <PopoverArrow />
             <PopoverBody p={4}>
               <Button size="sm" colorScheme="red" onClick={onRestartWithoutStateModalOpen}>
-                {actionText} Without State
+                Restart Without State
               </Button>
             </PopoverBody>
           </PopoverContent>
@@ -395,10 +389,6 @@ export function PipelineDetails() {
     }
   }
 
-  const startingWithoutState = job.state === 'Stopped';
-  const withoutStateAction = startingWithoutState ? 'Start' : 'Restart';
-  const withoutStateActionPresentParticiple = startingWithoutState ? 'Starting' : 'Restarting';
-
   const headerArea = (
     <Flex>
       <Box p={5}>
@@ -432,13 +422,13 @@ export function PipelineDetails() {
         <AlertDialogOverlay>
           <AlertDialogContent>
             <AlertDialogHeader fontSize="lg" fontWeight="bold">
-              {withoutStateAction} Without State
+              Restart Without State
             </AlertDialogHeader>
 
             <AlertDialogBody>
               <Text>
-                {withoutStateActionPresentParticiple} without state could lead to data loss,
-                duplication, and incorrect results. Are you sure you want to continue?
+                Restarting without state could lead to data loss, duplication, and incorrect
+                results. Are you sure you want to continue?
               </Text>
             </AlertDialogBody>
 
@@ -454,7 +444,7 @@ export function PipelineDetails() {
                 }}
                 ml={3}
               >
-                {withoutStateAction} Without State
+                Restart Without State
               </Button>
             </AlertDialogFooter>
           </AlertDialogContent>