Remove ResumeAlreadyRunning as it's not going to be feasible

AhmedSoliman · AhmedSoliman · commit 2883d050548c · 2026-03-26T18:57:09.000Z
diff --git a/crates/vqueues/src/metric_definitions.rs b/crates/vqueues/src/metric_definitions.rs
@@ -24,7 +24,6 @@ pub const VQUEUE_GLOBAL_THROTTLE_WAIT_MS: &str =
 pub const VQUEUE_LOCAL_THROTTLE_WAIT_MS: &str = "restate.vqueue.scheduler.vqueue_throttle_ms.total";
 
 pub const ACTION_YIELD: &str = "yield";
-pub const ACTION_RESUME: &str = "resume";
 pub const ACTION_RUN: &str = "run";
 pub const ACTION_START: &str = "start";
 
@@ -72,14 +71,8 @@ pub fn describe_metrics() {
     );
 }
 
-pub fn publish_scheduler_decision_metrics(
-    num_start: u16,
-    num_run: u16,
-    num_yield: u16,
-    num_resume: u16,
-) {
+pub fn publish_scheduler_decision_metrics(num_start: u16, num_run: u16, num_yield: u16) {
     counter!(VQUEUE_SCHEDULER_DECISION, "action" => ACTION_START).increment(num_start as u64);
     counter!(VQUEUE_SCHEDULER_DECISION, "action" => ACTION_RUN).increment(num_run as u64);
-    counter!(VQUEUE_SCHEDULER_DECISION, "action" => ACTION_RESUME).increment(num_resume as u64);
     counter!(VQUEUE_SCHEDULER_DECISION, "action" => ACTION_YIELD).increment(num_yield as u64);
 }
diff --git a/crates/vqueues/src/scheduler.rs b/crates/vqueues/src/scheduler.rs
@@ -237,9 +237,7 @@ impl<Item> AssignmentSegment<Item> {
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
 pub enum Action {
     /// Items are in inbox, let's move them to running queue.
-    MoveToRunning,
-    /// Items are already in running queue, execute them now.
-    ResumeAlreadyRunning,
+    MoveToRun,
     /// Items was already in running queue and we want them to yield back to the inbox
     Yield,
 }
@@ -252,8 +250,6 @@ pub struct Decision<Item> {
     num_start: u16,
     /// running items previously started
     num_run: u16,
-    /// running items that should continue to run
-    num_resume: u16,
     /// Items in run queue that need to go back to waiting inbox
     num_yield: u16,
 }
@@ -264,7 +260,6 @@ impl<Item> Default for Decision<Item> {
             q: HashMap::default(),
             num_start: 0,
             num_run: 0,
-            num_resume: 0,
             num_yield: 0,
         }
     }
@@ -281,10 +276,9 @@ impl<Item: VQueueEntry> Decision<Item> {
         let assignments = self.q.entry_ref(qid).or_default();
         assignments.set_latest_run_tb_zero_time(updated_zt);
         match action {
-            Action::ResumeAlreadyRunning => self.num_resume += 1,
             Action::Yield => self.num_yield += 1,
-            Action::MoveToRunning if entry.item.priority().is_new() => self.num_start += 1,
-            Action::MoveToRunning => self.num_run += 1,
+            Action::MoveToRun if entry.item.priority().is_new() => self.num_start += 1,
+            Action::MoveToRun => self.num_run += 1,
         }
         assignments.push(action, entry);
     }
@@ -313,31 +307,18 @@ impl<Item: VQueueEntry> Decision<Item> {
         self.num_run as usize
     }
 
-    #[cfg(test)]
-    pub fn num_resume(&self) -> usize {
-        self.num_resume as usize
-    }
-
     #[cfg(test)]
     pub fn num_yield(&self) -> usize {
         self.num_yield as usize
     }
 
     /// Total number of items in all queues
     pub fn total_items(&self) -> usize {
-        self.num_start as usize
-            + self.num_run as usize
-            + self.num_resume as usize
-            + self.num_yield as usize
+        self.num_start as usize + self.num_run as usize + self.num_yield as usize
     }
 
     pub fn report_metrics(&self) {
-        publish_scheduler_decision_metrics(
-            self.num_start,
-            self.num_run,
-            self.num_yield,
-            self.num_resume,
-        );
+        publish_scheduler_decision_metrics(self.num_start, self.num_run, self.num_yield);
     }
 }
 
diff --git a/crates/vqueues/src/scheduler/drr.rs b/crates/vqueues/src/scheduler/drr.rs
@@ -646,7 +646,7 @@ mod tests {
 
         decision.into_iter().for_each(|(qid, assignments)| {
             for (action, mut items) in assignments.into_iter_per_action() {
-                assert_eq!(action, Action::MoveToRunning);
+                assert_eq!(action, Action::MoveToRun);
                 // for each vqueue, we should see the head item. The head item is the lowest
                 // ID, so we should expect the item ID to the same as the queue ID.
                 assert_eq!(items.len(), 1);
@@ -868,7 +868,7 @@ mod tests {
             // creation timestamp is identical (for the purpose of the test)
             for (action, entries) in assignments.iter() {
                 assert_eq!(entries.len(), 2);
-                assert_eq!(action, Action::MoveToRunning);
+                assert_eq!(action, Action::MoveToRun);
                 assert_eq!(entries[0].item.priority, EffectivePriority::UserDefault);
                 assert_eq!(entries[0].item.id, EntryId::new([1; 16]));
                 in_flight.push(entries[0].item.clone());
@@ -920,7 +920,7 @@ mod tests {
 
             for (action, entries) in assignments.iter() {
                 assert_eq!(entries.len(), 2);
-                assert_eq!(action, Action::MoveToRunning);
+                assert_eq!(action, Action::MoveToRun);
                 // Verify first item is high priority
                 assert_eq!(entries[0].item.priority, EffectivePriority::UserHigh);
                 assert_eq!(entries[0].item.id, EntryId::new([125; 16]));
@@ -973,7 +973,7 @@ mod tests {
         for (_, assignments) in decision {
             for (action, entries) in assignments.iter() {
                 assert_eq!(entries.len(), 2);
-                assert_eq!(action, Action::MoveToRunning);
+                assert_eq!(action, Action::MoveToRun);
                 // Verify first item is high priority
                 assert_eq!(entries[0].item.priority, EffectivePriority::UserDefault);
                 assert_eq!(entries[0].item.id, EntryId::new([4; 16]));
@@ -994,7 +994,7 @@ mod tests {
         for (_, assignments) in decision {
             for (action, entries) in assignments.iter() {
                 assert_eq!(entries.len(), 2);
-                assert_eq!(action, Action::MoveToRunning);
+                assert_eq!(action, Action::MoveToRun);
                 // Verify first item is high priority
                 assert_eq!(entries[0].item.priority, EffectivePriority::UserDefault);
                 assert_eq!(entries[0].item.id, EntryId::new([6; 16]));
diff --git a/crates/vqueues/src/scheduler/vqueue_state.rs b/crates/vqueues/src/scheduler/vqueue_state.rs
@@ -255,11 +255,6 @@ impl<S: VQueueStore> VQueueState<S> {
         }
 
         if is_running {
-            // switch between these to change the behavior as needed
-            // Action::ResumeAlreadyRunning;
-            // Note that resumption requires acquiring concurrency permits similar to
-            // MoveToRunning. This is currently not implemented since (at the moment) we
-            // only support yielding.
             let result = Pop::Item {
                 action: Action::Yield,
                 permit: None,
@@ -337,7 +332,7 @@ impl<S: VQueueStore> VQueueState<S> {
         self.unconfirmed_assignments
             .insert(inbox_head.unique_hash());
         let result = Pop::Item {
-            action: Action::MoveToRunning,
+            action: Action::MoveToRun,
             permit,
             entry: Entry {
                 item: inbox_head.clone(),
diff --git a/crates/worker/src/partition/leadership/leader_state.rs b/crates/worker/src/partition/leadership/leader_state.rs
@@ -267,7 +267,6 @@ impl LeaderState {
     pub async fn handle_action_effects(
         &mut self,
         action_effects: impl IntoIterator<Item = ActionEffect>,
-        // invoker_tx: &mut impl restate_invoker_api::InvokerHandle<InvokerStorageReader<PartitionStore>>,
     ) -> Result<(), Error> {
         for effect in action_effects {
             match effect {
@@ -285,7 +284,7 @@ impl LeaderState {
                                 assignment.push(item, stats);
                             }
                             match action {
-                                scheduler::Action::MoveToRunning => {
+                                scheduler::Action::MoveToRun => {
                                     let command = vqueues::VQWaitingToRunning {
                                         assignment,
                                         meta_updates: vqueues::MetaUpdates {
@@ -306,21 +305,6 @@ impl LeaderState {
                                         Command::VQYieldRunning(command),
                                     ));
                                 }
-                                scheduler::Action::ResumeAlreadyRunning => {
-                                    todo!(
-                                        "Unsupported: We don't support directly resuming at the moment"
-                                    )
-                                    // invoker_tx
-                                    //     .run(
-                                    //         self.partition_id,
-                                    //         self.leader_epoch,
-                                    //         inv_id,
-                                    //         // todo: fix/remove
-                                    //         SharedString::from_borrowed(""),
-                                    //         SharedString::from_borrowed(""),
-                                    //     )
-                                    //     .map_err(Error::Invoker)?;
-                                }
                             }
                         }
                     }
diff --git a/crates/worker/src/partition/leadership/mod.rs b/crates/worker/src/partition/leadership/mod.rs
@@ -640,9 +640,7 @@ where
                 // nothing to do :-)
             }
             State::Leader(leader_state) => {
-                leader_state
-                    .handle_action_effects(action_effects /*, &mut self.invoker_tx */)
-                    .await?
+                leader_state.handle_action_effects(action_effects).await?
             }
         }
 

Original file line number	Diff line number	Diff line change
`@@ -640,9 +640,7 @@ where`
`640`	`640`	`// nothing to do :-)`
`641`	`641`	`}`
`642`	`642`	`State::Leader(leader_state) => {`
`643`		`- leader_state`
`644`		`- .handle_action_effects(action_effects /, &mut self.invoker_tx /)`
`645`		`- .await?`
	`643`	`+ leader_state.handle_action_effects(action_effects).await?`
`646`	`644`	`}`
`647`	`645`	`}`
`648`	`646`