Evictions in their own activations (#712)

Author: Sushisource

.github/workflows/per-pr.yml

@@ -22,7 +22,7 @@ jobs:
       - uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: 1.74.0
+          toolchain: 1.77.0
           override: true
       - name: Install protoc
         uses: arduino/setup-protoc@v1
@@ -55,7 +55,7 @@ jobs:
       - uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: 1.74.0
+          toolchain: 1.77.0
           override: true
       - name: Install protoc
         uses: arduino/setup-protoc@v1
@@ -85,7 +85,7 @@ jobs:
       - uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: 1.74.0
+          toolchain: 1.77.0
           override: true
       - name: Install protoc
         uses: arduino/setup-protoc@v1

core/src/telemetry/mod.rs

@@ -121,7 +121,8 @@ impl TelemetryInstance {
 }
 
 thread_local! {
-    static SUB_GUARD: RefCell<Option<tracing::subscriber::DefaultGuard>> = RefCell::new(None);
+    static SUB_GUARD: RefCell<Option<tracing::subscriber::DefaultGuard>> =
+        const { RefCell::new(None) };
 }
 /// Set the trace subscriber for the current thread. This must be done in every thread which uses
 /// core stuff, otherwise traces/logs will not be collected on that thread. For example, if using

core/src/test_help/mod.rs

@@ -37,7 +37,7 @@ use temporal_sdk_core_api::{
 };
 use temporal_sdk_core_protos::{
     coresdk::{
-        workflow_activation::WorkflowActivation,
+        workflow_activation::{workflow_activation_job, WorkflowActivation},
         workflow_commands::workflow_command,
         workflow_completion::{self, workflow_activation_completion, WorkflowActivationCompletion},
     },
@@ -831,18 +831,23 @@ pub(crate) async fn poll_and_reply_clears_outstanding_evicts<'a>(
             }
 
             let mut res = worker.poll_workflow_activation().await.unwrap();
-            let contains_eviction = res.eviction_index();
+            if res.jobs.iter().any(|j| {
+                matches!(
+                    j.variant,
+                    Some(workflow_activation_job::Variant::RemoveFromCache(_))
+                )
+            }) && res.jobs.len() > 1
+            {
+                panic!("Saw an activation with an eviction & other work! {res:?}");
+            }
+            let is_eviction = res.is_only_eviction();
 
             let mut do_release = false;
-            if let Some(eviction_job_ix) = contains_eviction {
-                // If the job list has an eviction, make sure it was the last item in the list
-                // then remove it, since in the tests we don't explicitly specify evict assertions
-                assert_eq!(
-                    eviction_job_ix,
-                    res.jobs.len() - 1,
-                    "Eviction job was not last job in job list"
-                );
-                res.jobs.remove(eviction_job_ix);
+
+            if is_eviction {
+                // If the job is an eviction, clear it, since in the tests we don't explicitly
+                // specify evict assertions
+                res.jobs.clear();
                 do_release = true;
             }
 
@@ -873,7 +878,7 @@ pub(crate) async fn poll_and_reply_clears_outstanding_evicts<'a>(
             }
             // Restart assertions from the beginning if it was an eviction (and workflow execution
             // isn't over)
-            if contains_eviction.is_some() && !ends_execution {
+            if is_eviction && !ends_execution {
                 continue 'outer;
             }
 

core/src/worker/workflow/managed_run.rs

@@ -31,7 +31,7 @@ use temporal_sdk_core_protos::{
     coresdk::{
         workflow_activation::{
             create_evict_activation, query_to_job, remove_from_cache::EvictionReason,
-            workflow_activation_job, RemoveFromCache, WorkflowActivation,
+            workflow_activation_job, WorkflowActivation,
         },
         workflow_commands::{FailWorkflowExecution, QueryResult},
         workflow_completion,
@@ -347,15 +347,8 @@ impl ManagedRun {
                 }
             }
             if let Some(wte) = self.trying_to_evict.clone() {
-                let mut act = self.wfm.machines.get_wf_activation();
-                // No other jobs make any sense to send if we encountered an error.
-                if self.am_broken {
-                    act.jobs = vec![];
-                }
-                act.append_evict_job(RemoveFromCache {
-                    message: wte.message,
-                    reason: wte.reason as i32,
-                });
+                let act =
+                    create_evict_activation(self.run_id().to_string(), wte.message, wte.reason);
                 Ok(Some(ActivationOrAuto::LangActivation(act)))
             } else {
                 Ok(None)
@@ -375,7 +368,7 @@ impl ManagedRun {
         used_flags: Vec<u32>,
         resp_chan: Option<oneshot::Sender<ActivationCompleteResult>>,
     ) -> Result<RunUpdateAct, NextPageReq> {
-        let activation_was_only_eviction = self.activation_has_only_eviction();
+        let activation_was_only_eviction = self.activation_is_eviction();
         let (task_token, has_pending_query, start_time) = if let Some(entry) = self.wft.as_ref() {
             (
                 entry.info.task_token.clone(),
@@ -444,15 +437,14 @@ impl ManagedRun {
                 .collect();
 
             if activation_was_only_eviction && !commands.is_empty() {
-                dbg_panic!("Reply to an eviction only containing an eviction included commands");
+                dbg_panic!("Reply to an eviction included commands");
             }
 
             let rac = RunActivationCompletion {
                 task_token,
                 start_time,
                 commands,
-                activation_was_eviction: self.activation_has_eviction(),
-                activation_was_only_eviction,
+                activation_was_eviction: self.activation_is_eviction(),
                 has_pending_query,
                 query_responses,
                 used_flags,
@@ -623,7 +615,8 @@ impl ManagedRun {
     ) -> (bool, BufferedTasks) {
         let evict = if self.activation().map(pred).unwrap_or_default() {
             let act = self.activation.take();
-            act.map(|a| a.has_eviction()).unwrap_or_default()
+            act.map(|a| matches!(a, OutstandingActivation::Eviction))
+                .unwrap_or_default()
         } else {
             false
         };
@@ -655,14 +648,14 @@ impl ManagedRun {
             task_token: completion.task_token,
             query_responses: completion.query_responses,
             has_pending_query: completion.has_pending_query,
-            activation_was_only_eviction: completion.activation_was_only_eviction,
+            activation_was_eviction: completion.activation_was_eviction,
         };
 
         self.wfm.machines.add_lang_used_flags(completion.used_flags);
 
-        // If this is just bookkeeping after a reply to an only-eviction activation, we can bypass
+        // If this is just bookkeeping after a reply to an eviction activation, we can bypass
         // everything, since there is no reason to continue trying to update machines.
-        if completion.activation_was_only_eviction {
+        if completion.activation_was_eviction {
             return Ok(Some(self.prepare_complete_resp(
                 completion.resp_chan,
                 data,
@@ -793,11 +786,9 @@ impl ManagedRun {
             self.trying_to_evict.is_some()
         };
         let act_work = if ignore_evicts {
-            if let Some(ref act) = self.activation {
-                !act.has_only_eviction()
-            } else {
-                false
-            }
+            self.activation
+                .map(|a| !matches!(a, OutstandingActivation::Eviction))
+                .unwrap_or_default()
         } else {
             self.activation.is_some()
         };
@@ -854,7 +845,7 @@ impl ManagedRun {
             return EvictionRequestResult::EvictionRequested(attempts, run_upd);
         }
 
-        if !self.activation_has_eviction() && self.trying_to_evict.is_none() {
+        if !self.activation_is_eviction() && self.trying_to_evict.is_none() {
             debug!(run_id=%info.run_id, reason=%info.message, "Eviction requested");
             self.trying_to_evict = Some(info);
             EvictionRequestResult::EvictionRequested(attempts, self.check_more_activations())
@@ -990,13 +981,12 @@ impl ManagedRun {
     fn insert_outstanding_activation(&mut self, act: &ActivationOrAuto) {
         let act_type = match &act {
             ActivationOrAuto::LangActivation(act) | ActivationOrAuto::ReadyForQueries(act) => {
-                if act.is_legacy_query() {
+                if act.is_only_eviction() {
+                    OutstandingActivation::Eviction
+                } else if act.is_legacy_query() {
                     OutstandingActivation::LegacyQuery
                 } else {
-                    OutstandingActivation::Normal {
-                        contains_eviction: act.eviction_index().is_some(),
-                        num_jobs: act.jobs.len(),
-                    }
+                    OutstandingActivation::Normal
                 }
             }
             ActivationOrAuto::Autocomplete { .. } | ActivationOrAuto::AutoFail { .. } => {
@@ -1021,7 +1011,7 @@ impl ManagedRun {
         due_to_heartbeat_timeout: bool,
     ) -> FulfillableActivationComplete {
         let mut machines_wft_response = self.wfm.prepare_for_wft_response();
-        if data.activation_was_only_eviction
+        if data.activation_was_eviction
             && (machines_wft_response.commands().peek().is_some()
                 || machines_wft_response.has_messages())
             && !self.am_broken
@@ -1045,7 +1035,7 @@ impl ManagedRun {
         let should_respond = !(machines_wft_response.has_pending_jobs
             || machines_wft_response.replaying
             || is_query_playback
-            || data.activation_was_only_eviction
+            || data.activation_was_eviction
             || machines_wft_response.have_seen_terminal_event);
         // If there are pending LA resolutions, and we're responding to a query here,
         // we want to make sure to force a new task, as otherwise once we tell lang about
@@ -1058,7 +1048,7 @@ impl ManagedRun {
         let outcome = if should_respond || has_query_responses {
             // If we broke there could be commands or messages in the pipe that we didn't
             // get a chance to handle properly during replay. Don't send them.
-            let (commands, messages) = if self.am_broken && data.activation_was_only_eviction {
+            let (commands, messages) = if self.am_broken && data.activation_was_eviction {
                 (vec![], vec![])
             } else {
                 (
@@ -1168,15 +1158,9 @@ impl ManagedRun {
         self.wfm.machines.last_processed_event
     }
 
-    fn activation_has_eviction(&mut self) -> bool {
-        self.activation
-            .map(OutstandingActivation::has_eviction)
-            .unwrap_or_default()
-    }
-
-    fn activation_has_only_eviction(&mut self) -> bool {
+    fn activation_is_eviction(&mut self) -> bool {
         self.activation
-            .map(OutstandingActivation::has_only_eviction)
+            .map(|a| matches!(a, OutstandingActivation::Eviction))
             .unwrap_or_default()
     }
 
@@ -1245,7 +1229,7 @@ struct CompletionDataForWFT {
     task_token: TaskToken,
     query_responses: Vec<QueryResult>,
     has_pending_query: bool,
-    activation_was_only_eviction: bool,
+    activation_was_eviction: bool,
 }
 
 /// Manages an instance of a [WorkflowMachines], which is not thread-safe, as well as other data
@@ -1385,7 +1369,6 @@ struct RunActivationCompletion {
     start_time: Instant,
     commands: Vec<WFCommand>,
     activation_was_eviction: bool,
-    activation_was_only_eviction: bool,
     has_pending_query: bool,
     query_responses: Vec<QueryResult>,
     used_flags: Vec<u32>,

core/src/worker/workflow/mod.rs

@@ -817,39 +817,15 @@ impl OutstandingTask {
 #[derive(Copy, Clone, Debug)]
 pub(crate) enum OutstandingActivation {
     /// A normal activation with a joblist
-    Normal {
-        /// True if there is an eviction in the joblist
-        contains_eviction: bool,
-        /// Number of jobs in the activation
-        num_jobs: usize,
-    },
+    Normal,
+    /// An eviction job
+    Eviction,
     /// An activation for a legacy query
     LegacyQuery,
     /// A fake activation which is never sent to lang, but used internally
     Autocomplete,
 }
 
-impl OutstandingActivation {
-    pub(crate) const fn has_only_eviction(self) -> bool {
-        matches!(
-            self,
-            OutstandingActivation::Normal {
-                contains_eviction: true,
-                num_jobs: nj
-            }
-        if nj == 1)
-    }
-    pub(crate) const fn has_eviction(self) -> bool {
-        matches!(
-            self,
-            OutstandingActivation::Normal {
-                contains_eviction: true,
-                ..
-            }
-        )
-    }
-}
-
 /// Contains important information about a given workflow task that we need to memorize while
 /// lang handles it.
 #[derive(Clone, Debug)]

fsm/rustfsm_procmacro/tests/trybuild/no_handle_conversions_require_into_fail.stderr

@@ -2,7 +2,7 @@ error[E0277]: the trait bound `One: From<Two>` is not satisfied
   --> tests/trybuild/no_handle_conversions_require_into_fail.rs:11:5
    |
 11 |     Two --(B)--> One;
-   |     ^^^ the trait `From<Two>` is not implemented for `One`
+   |     ^^^ the trait `From<Two>` is not implemented for `One`, which is required by `Two: Into<One>`
    |
    = note: required for `Two` to implement `Into<One>`
 note: required by a bound in `TransitionResult::<Sm, Ds>::from`

sdk-core-protos/protos/local/temporal/sdk/core/workflow_activation/workflow_activation.proto

@@ -30,6 +30,7 @@ import "temporal/sdk/core/common/common.proto";
 // * Signal and update handlers should be invoked before workflow routines are iterated. That is to
 //   say before the users' main workflow function and anything spawned by it is allowed to continue.
 // * Queries always go last (and, in fact, always come in their own activation)
+// * Evictions also always come in their own activation
 //
 // The downside of this reordering is that a signal or update handler may not observe that some
 // other event had already happened (ex: an activity completed) when it is first invoked, though it
@@ -41,11 +42,9 @@ import "temporal/sdk/core/common/common.proto";
 //
 // ## Evictions
 //
-// Activations that contain only a `remove_from_cache` job should not cause the workflow code
-// to be invoked and may be responded to with an empty command list. Eviction jobs may also
-// appear with other jobs, but will always appear last in the job list. In this case it is
-// expected that the workflow code will be invoked, and the response produced as normal, but
-// the caller should evict the run after doing so.
+// Evictions appear as an activations that contains only a `remove_from_cache` job. Such activations
+// should not cause the workflow code to be invoked and may be responded to with an empty command
+// list.
 message WorkflowActivation {
     // The id of the currently active run of the workflow. Also used as a cache key. There may
     // only ever be one active workflow task (and hence activation) of a run at one time.
@@ -81,7 +80,8 @@ message WorkflowActivationJob {
         FireTimer fire_timer = 2;
         // Workflow was reset. The randomness seed must be updated.
         UpdateRandomSeed update_random_seed = 4;
-        // A request to query the workflow was received.
+        // A request to query the workflow was received. It is guaranteed that queries (one or more)
+        // always come in their own activation after other mutating jobs.
         QueryWorkflow query_workflow = 5;
         // A request to cancel the workflow was received.
         CancelWorkflow cancel_workflow = 6;
@@ -103,11 +103,9 @@ message WorkflowActivationJob {
         ResolveRequestCancelExternalWorkflow resolve_request_cancel_external_workflow = 13;
         // A request to handle a workflow update.
         DoUpdate do_update = 14;
-        // Remove the workflow identified by the [WorkflowActivation] containing this job from the cache
-        // after performing the activation.
-        //
-        // If other job variant are present in the list, this variant will be the last job in the
-        // job list. The string value is a reason for eviction.
+        // Remove the workflow identified by the [WorkflowActivation] containing this job from the
+        // cache after performing the activation. It is guaranteed that this will be the only job
+        // in the activation if present.
         RemoveFromCache remove_from_cache = 50;
     }
 }
@@ -208,7 +206,7 @@ message ResolveChildWorkflowExecutionStartFailure {
 // `failure` should be ChildWorkflowFailure with cause set to CancelledFailure.
 // The failure is constructed in core for lang's convenience.
 message ResolveChildWorkflowExecutionStartCancelled {
-  temporal.api.failure.v1.Failure failure = 1;
+    temporal.api.failure.v1.Failure failure = 1;
 }
 
 // Notify a workflow that a child workflow execution has been resolved