Temporarily avoid InvocationStatus::Invoked to ::Inboxed state transition when using vqueues

While it is correct that we should go from InvocationStatus::Invoked to ::Inboxed when resuming
an invocation using vqueues, it is currently supported by the state machine. For it to be fully
supported, we need to support handling journal completions when being ::Inboxed. Morever, we
need to handle cancellations of ::Inboxed items depending on whether the invocation was running
before or not. Additionally, we need to support the hotfix_apply_cancellation_after_deployment_is_pinned
when going from ::Suspended to ::Inbox, for example. Until this is done, we stay in ::Invoked when
resuming an invocation even when using vqueues.

Author: tillrohrmann

crates/worker/src/partition/state_machine/lifecycle/resume.rs

@@ -9,7 +9,7 @@
 // by the Apache License, Version 2.0.
 
 use restate_invoker_api::InvokeInputJournal;
-use restate_storage_api::invocation_status_table::{InboxedInvocation, InvocationStatus};
+use restate_storage_api::invocation_status_table::InvocationStatus;
 use restate_storage_api::vqueue_table::{ReadVQueueTable, WriteVQueueTable};
 use restate_types::config::Configuration;
 use restate_types::identifiers::InvocationId;
@@ -45,12 +45,15 @@ where
         if Configuration::pinned().common.experimental_enable_vqueues {
             ctx.vqueue_move_invocation_to_inbox_stage(&self.invocation_id)
                 .await?;
-            // When moving an invocation back into the inbox stage, we have to update the
-            // InvocationStatus accordingly. Otherwise, we miss running changes on the VQueues
-            // inbox.
-            *self.invocation_status = InvocationStatus::Inboxed(
-                InboxedInvocation::from_in_flight_invocation_metadata(metadata.clone()),
-            );
+
+            // todo enable once we properly support handling completions when being in status Inboxed
+            // // When moving an invocation back into the inbox stage, we have to update the
+            // // InvocationStatus accordingly. Otherwise, we miss running changes on the VQueues
+            // // inbox.
+            // *self.invocation_status = InvocationStatus::Inboxed(
+            //     InboxedInvocation::from_in_flight_invocation_metadata(metadata.clone()),
+            // );
+            *self.invocation_status = InvocationStatus::Invoked(metadata.clone());
         } else {
             ctx.action_collector.push(Action::Invoke {
                 invocation_id: self.invocation_id,

crates/worker/src/partition/state_machine/mod.rs

@@ -2905,18 +2905,32 @@ impl<S> StateMachineApplyContext<'_, S> {
                 )
                 .await?;
             }
-            InvocationStatus::Invoked(metadata) if self.is_leader => {
-                // just send to invoker
-                debug_if_leader!(self.is_leader, "Invoke");
-                self.action_collector.push(Action::VQInvoke {
+            InvocationStatus::Invoked(metadata) => {
+                // Temporary fix to transition the vqueue item back to the running stage. This is
+                // needed because we don't support yet going back from InvocationStatus::Invoked
+                // to InvocationStatus::Inboxed and then accepting journal completions.
+                // See https://github.com/restatedev/restate/blob/86d4d055ad8f3aa8b426c06486d52382a76bf9dd/crates/worker/src/partition/state_machine/lifecycle/resume.rs#L53
+                VQueues::new(
                     qid,
-                    item_hash: card.unique_hash(),
-                    invocation_id,
-                    invocation_target: metadata.invocation_target,
-                    invoke_input_journal: InvokeInputJournal::NoCachedJournal,
-                });
+                    self.storage,
+                    self.vqueues_cache,
+                    self.is_leader.then_some(self.action_collector),
+                )
+                .attempt_to_run(at, card)
+                .await?;
+
+                if self.is_leader {
+                    // just send to invoker
+                    debug_if_leader!(self.is_leader, "Invoke");
+                    self.action_collector.push(Action::VQInvoke {
+                        qid,
+                        item_hash: card.unique_hash(),
+                        invocation_id,
+                        invocation_target: metadata.invocation_target,
+                        invoke_input_journal: InvokeInputJournal::NoCachedJournal,
+                    });
+                }
             }
-            InvocationStatus::Invoked(_) => { /* do nothing when not leader */ }
             // Suspended invocations must first be put back on inbox. On wake-up, they
             // transition back into Invoked state. So seeing a suspended invocation
             // here means that some state transition is missing.