diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 8f4a809b0..5423ac4e7 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. @@ -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 diff --git a/crates/sdk-core/src/core_tests/workflow_tasks.rs b/crates/sdk-core/src/core_tests/workflow_tasks.rs index 2cc51f438..f7c68629b 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/lib.rs b/crates/sdk-core/src/lib.rs index c3984975b..cf0f68c11 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 11990d094..6e4cadc50 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 4ff8a9180..000000000 --- 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 a37defa00..d31addaf4 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 { diff --git a/crates/sdk-core/src/worker/workflow/managed_run.rs b/crates/sdk-core/src/worker/workflow/managed_run.rs index 0c8d2aa24..77b5a70d8 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 d00173f64..2cd039745 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(),