[Brook] Removed the worker extension and fixed a zero-cache eviction.

crates/sdk-core/CHANGELOG.md

@@ -34,10 +34,6 @@ relevant information.
 ## Unreleased
 
 ### Added
-* A worker can take a `WorkerExtension`, a plugin component that the replay worker leaves out.
-* A `WorkerExtension` has `call` and `call_async`, which take a method name and encoded input. A
-  bridge can route a generic lang call to them, so the extension needs no bridge function of its
-  own. Core never calls them.
 * A Local Activity's `ActivityTask.Start` has `workflow_task` (field 20), with the started event
   id, attempt and start time of the Workflow Task in which Core dispatched it. Retries that Core
   runs on its own keep the value.
@@ -54,6 +50,8 @@ relevant information.
   nondeterminism while retaining compatibility with histories recorded before these checks.
 
 ### Fixed
+* With the workflow cache off, the eviction after a terminal completion that the server accepted
+  has reason `WORKFLOW_EXECUTION_ENDING`, as with a cache, not `CACHE_FULL`.
 * Dev and test server startup now stops waiting at the configured deadline even if a readiness
   request stalls. Timeout messages include the last connection failure when available.
 * Workers with large workflow caches avoid unnecessary CPU usage when no workflow tasks are waiting

crates/sdk-core/src/core_tests/workflow_tasks.rs

@@ -4278,3 +4278,64 @@ async fn cached_run_fetch_failure_doesnt_spam_task_fails() {
         2
     );
 }
+
+/// The eviction that follows a terminal completion the server accepted, with the cache disabled.
+async fn eviction_after_completing_workflow_with_zero_cache(
+    completion_result: Result<(), tonic::Status>,
+) -> EvictionReason {
+    let wfid = "fake_wf_id";
+    let mut t = TestHistoryBuilder::default();
+    t.add_by_type(EventType::WorkflowExecutionStarted);
+    t.add_full_wf_task();
+    let mut mock_cfg = MockPollCfg::from_resp_batches(wfid, t, [1], mock_worker_client());
+    mock_cfg.make_poll_stream_interminable = true;
+    let mut completion_result = Some(completion_result);
+    mock_cfg.completion_mock_fn = Some(Box::new(move |_| {
+        completion_result
+            .take()
+            .unwrap_or(Ok(()))
+            .map(|()| Default::default())
+    }));
+    let mut mock = build_mock_pollers(mock_cfg);
+    mock.worker_cfg(|wc| wc.max_cached_workflows = 0);
+    let core = mock_worker(mock);
+
+    let activation = core.poll_workflow_activation().await.unwrap();
+    core.complete_workflow_activation(WorkflowActivationCompletion::from_cmd(
+        activation.run_id,
+        CompleteWorkflowExecution { result: None }.into(),
+    ))
+    .await
+    .unwrap();
+    let eviction = core.poll_workflow_activation().await.unwrap();
+    let reason = assert_matches!(
+        eviction.jobs.as_slice(),
+        [WorkflowActivationJob {
+            variant: Some(workflow_activation_job::Variant::RemoveFromCache(r)),
+        }] => r.reason()
+    );
+    core.complete_workflow_activation(WorkflowActivationCompletion::empty(eviction.run_id))
+        .await
+        .unwrap();
+    core.drain_pollers_and_shutdown().await;
+    reason
+}
+
+#[tokio::test]
+async fn accepted_terminal_completion_with_zero_cache_evicts_as_execution_ending() {
+    assert_eq!(
+        eviction_after_completing_workflow_with_zero_cache(Ok(())).await,
+        EvictionReason::WorkflowExecutionEnding
+    );
+}
+
+#[tokio::test]
+async fn rejected_terminal_completion_with_zero_cache_evicts_with_rejection_reason() {
+    assert_eq!(
+        eviction_after_completing_workflow_with_zero_cache(Err(tonic::Status::not_found(
+            "Workflow task not found."
+        )))
+        .await,
+        EvictionReason::TaskNotFound
+    );
+}

crates/sdk-core/src/lib.rs

@@ -50,7 +50,6 @@ pub use worker::{
     WorkerConfigBuilder, WorkerTuner, WorkerValidationError, WorkerVersioningStrategy,
     WorkflowErrorType, WorkflowSlotKind,
 };
-pub use worker::{ExtensionError, WorkerExtension};
 
 use crate::{
     replay::{HistoryForReplay, ReplayWorkerInput},

crates/sdk-core/src/replay/mod.rs

@@ -79,8 +79,6 @@ where
         self.config.workflow_task_poller_behavior = Some(PollerBehavior::SimpleMaximum(1));
         self.config.task_types = WorkerTaskTypes::workflow_only();
         self.config.skip_client_worker_set_check = true;
-        // Replay must not repeat the side effects of the hooks.
-        self.config.extension = None;
         let historator = Historator::new(self.history_stream);
         let post_activate = historator.get_post_activate_hook();
         let shutdown_tok = historator.get_shutdown_setter();

crates/sdk-core/src/worker/mod.rs

@@ -1,6 +1,5 @@
 mod activities;
 pub(crate) mod client;
-mod extension;
 pub(crate) mod heartbeat;
 mod nexus;
 mod slot_provider;
@@ -11,7 +10,6 @@ mod workflow;
 /// the error limit, shared so every conversion site reports the same identifier.
 pub(crate) const PAYLOADS_TOO_LARGE_FAILURE_TYPE: &str = "PayloadsTooLarge";
 
-pub use extension::{ExtensionError, WorkerExtension};
 use temporalio_client::{Connection, PayloadErrorLimits, worker::NamespaceDescriptionSource};
 use temporalio_common::{
     protos::{
@@ -301,9 +299,6 @@ pub struct WorkerConfig {
     /// NOTE: Experimental
     #[builder(default = false)]
     pub disable_payload_error_limit: bool,
-
-    /// A plugin component that lang calls by method name. See [WorkerExtension].
-    pub extension: Option<Arc<dyn WorkerExtension>>,
 }
 
 impl WorkerConfig {

crates/sdk-core/src/worker/workflow/managed_run.rs

@@ -973,6 +973,24 @@ impl ManagedRun {
         self.task_buffer.has_tasks()
     }
 
+    /// Reports an eviction already requested because the cache is full (or disabled) as
+    /// [EvictionReason::WorkflowExecutionEnding], once the server has accepted the run's terminal
+    /// completion. With the cache disabled the cache-full eviction is requested before the
+    /// completion is reported, so without this lang would never see that the execution ended.
+    ///
+    /// The caller cannot tell acceptance from rejection: [WFTReportStatus::Reported] is also the
+    /// status of a completion the server rejected. This relies on a rejection having already
+    /// replaced the cache-full reason: the rejection's eviction request is sent before the
+    /// post-activation message, and [Self::request_eviction] keeps the most recent reason.
+    pub(super) fn report_cache_full_eviction_as_execution_ending(&mut self) {
+        if let Some(info) = self.trying_to_evict.as_mut()
+            && info.reason == EvictionReason::CacheFull
+        {
+            info.reason = EvictionReason::WorkflowExecutionEnding;
+            info.message = "Workflow completed".to_string();
+        }
+    }
+
     pub(super) fn request_eviction(&mut self, info: RequestEvictMsg) -> EvictionRequestResult {
         // If we were waiting on a page fetch and we're getting evicted because fetching failed,
         // then make sure we allow the completion to proceed, otherwise we're stuck waiting forever.

crates/sdk-core/src/worker/workflow/workflow_stream.rs

@@ -441,13 +441,15 @@ impl WFStream {
         if res.is_none()
             && let Some(rh) = self.runs.get_mut(run_id)
         {
+            let completed_workflow = rh.workflow_is_finished()
+                && matches!(report.wft_report_status, WFTReportStatus::Reported { .. });
+            if completed_workflow {
+                rh.report_cache_full_eviction_as_execution_ending();
+            }
             // Attempt to produce the next activation if needed
             res = rh.check_more_activations();
             // If there's no more work and we reported workflow completion to server, evict.
-            if res.is_none()
-                && rh.workflow_is_finished()
-                && matches!(report.wft_report_status, WFTReportStatus::Reported { .. })
-            {
+            if res.is_none() && completed_workflow {
                 res = rh
                     .request_eviction(RequestEvictMsg {
                         run_id: run_id.to_string(),