From 989b0c331e42ff0886db2f61a82036e10f67337f Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 23 Oct 2026 03:26:40 -0700 Subject: [PATCH 1/2] Removed the worker extension, since the bridge binds the crate itself. Core only passed the extension through to lang. A bridge can build its own object per client connection, so Core needs no trait or config field for it. --- crates/sdk-core/CHANGELOG.md | 4 - crates/sdk-core/src/lib.rs | 1 - crates/sdk-core/src/replay/mod.rs | 2 - crates/sdk-core/src/worker/extension.rs | 132 ------------------------ crates/sdk-core/src/worker/mod.rs | 5 - 5 files changed, 144 deletions(-) delete mode 100644 crates/sdk-core/src/worker/extension.rs diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 8f4a809b02..b9db5edcea 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/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. diff --git a/crates/sdk-core/src/lib.rs b/crates/sdk-core/src/lib.rs index c3984975b4..cf0f68c112 100644 --- a/crates/sdk-core/src/lib.rs +++ b/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}, diff --git a/crates/sdk-core/src/replay/mod.rs b/crates/sdk-core/src/replay/mod.rs index 11990d094a..6e4cadc506 100644 --- a/crates/sdk-core/src/replay/mod.rs +++ b/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(); diff --git a/crates/sdk-core/src/worker/extension.rs b/crates/sdk-core/src/worker/extension.rs deleted file mode 100644 index 4ff8a91807..0000000000 --- a/crates/sdk-core/src/worker/extension.rs +++ /dev/null @@ -1,132 +0,0 @@ -//! A plugin component that a worker carries, which lang calls by method name. - -/// A component that lang passes to the worker through [crate::WorkerConfig::extension]. -/// -/// Core never calls [WorkerExtension::call] or [WorkerExtension::call_async]. A bridge exposes -/// them to lang, so lang code reaches the extension in the same native library without a bridge -/// function of its own. -#[async_trait::async_trait] -pub trait WorkerExtension: Send + Sync + 'static { - /// A name that identifies the extension in logs. - fn name(&self) -> &str; - - /// Runs `method` with encoded input and returns the encoded output. It must not block, since - /// lang may call it from workflow code. - fn call(&self, method: &str, _input: &[u8]) -> Result, ExtensionError> { - Err(ExtensionError::unknown_method(method)) - } - - /// Like [WorkerExtension::call], for work that waits on I/O. - async fn call_async(&self, method: String, _input: Vec) -> Result, ExtensionError> { - Err(ExtensionError::unknown_method(&method)) - } -} - -/// The error of a [WorkerExtension] call. It has the fields of a gRPC status, so a bridge can turn -/// it into the same lang error as a failed service call. -#[derive(Clone, Debug, thiserror::Error)] -#[error("{code:?}: {message}")] -#[non_exhaustive] -pub struct ExtensionError { - /// The status code. - pub code: tonic::Code, - /// The message for the caller. - pub message: String, - /// Encoded error details, empty if there are none. - pub details: Vec, -} - -impl ExtensionError { - /// Creates an error with no details. - pub fn new(code: tonic::Code, message: impl Into) -> Self { - Self { - code, - message: message.into(), - details: vec![], - } - } - - /// Adds encoded error details. - pub fn with_details(mut self, details: Vec) -> Self { - self.details = details; - self - } - - /// The error for a method the extension does not have. - pub fn unknown_method(method: &str) -> Self { - Self::new( - tonic::Code::Unimplemented, - format!("Unknown extension method {method}"), - ) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::sync::Arc; - - struct Named; - - impl WorkerExtension for Named { - fn name(&self) -> &str { - "named" - } - } - - #[tokio::test] - async fn calls_to_an_extension_without_methods_are_unimplemented() { - let extension: Arc = Arc::new(Named); - let err = extension.call("Append", b"x").unwrap_err(); - assert_eq!(err.code, tonic::Code::Unimplemented); - assert!(err.message.contains("Append")); - let err = extension - .call_async("Append".to_string(), vec![]) - .await - .unwrap_err(); - assert_eq!(err.code, tonic::Code::Unimplemented); - } - - struct Echo; - - #[async_trait::async_trait] - impl WorkerExtension for Echo { - fn name(&self) -> &str { - "echo" - } - - fn call(&self, method: &str, input: &[u8]) -> Result, ExtensionError> { - match method { - "Echo" => Ok(input.to_vec()), - _ => { - Err(ExtensionError::new(tonic::Code::NotFound, "no stream") - .with_details(vec![7])) - } - } - } - - async fn call_async( - &self, - method: String, - input: Vec, - ) -> Result, ExtensionError> { - self.call(&method, &input) - } - } - - #[tokio::test] - async fn calls_reach_the_extension_methods() { - let extension: Arc = Arc::new(Echo); - assert_eq!(extension.call("Echo", b"abc").unwrap(), b"abc"); - assert_eq!( - extension - .call_async("Echo".to_string(), b"abc".to_vec()) - .await - .unwrap(), - b"abc" - ); - let err = extension.call("Read", b"").unwrap_err(); - assert_eq!(err.code, tonic::Code::NotFound); - assert_eq!(err.details, [7]); - } -} diff --git a/crates/sdk-core/src/worker/mod.rs b/crates/sdk-core/src/worker/mod.rs index a37defa00d..d31addaf47 100644 --- a/crates/sdk-core/src/worker/mod.rs +++ b/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>, } impl WorkerConfig { From b565c433547496f2c14a911549c288a1076e72b6 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 23 Oct 2026 03:26:41 -0700 Subject: [PATCH 2/2] Reported a completed run's eviction as ending with the cache off. With no workflow cache, Core asked for a cache-full eviction before it reported the completion, so lang never saw that the execution ended. A worker that acts after an accepted completion needs that reason. Co-authored-by: Dan Davison --- crates/sdk-core/CHANGELOG.md | 2 + .../sdk-core/src/core_tests/workflow_tasks.rs | 61 +++++++++++++++++++ .../src/worker/workflow/managed_run.rs | 18 ++++++ .../src/worker/workflow/workflow_stream.rs | 10 +-- 4 files changed, 87 insertions(+), 4 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index b9db5edcea..5423ac4e72 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -50,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 diff --git a/crates/sdk-core/src/core_tests/workflow_tasks.rs b/crates/sdk-core/src/core_tests/workflow_tasks.rs index 2cc51f4380..f7c68629b5 100644 --- a/crates/sdk-core/src/core_tests/workflow_tasks.rs +++ b/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 + ); +} diff --git a/crates/sdk-core/src/worker/workflow/managed_run.rs b/crates/sdk-core/src/worker/workflow/managed_run.rs index 0c8d2aa247..77b5a70d87 100644 --- a/crates/sdk-core/src/worker/workflow/managed_run.rs +++ b/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. diff --git a/crates/sdk-core/src/worker/workflow/workflow_stream.rs b/crates/sdk-core/src/worker/workflow/workflow_stream.rs index d00173f64e..2cd0397456 100644 --- a/crates/sdk-core/src/worker/workflow/workflow_stream.rs +++ b/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(),