From 4f53abf5c8923fad12ef1364c736b4a85de68500 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 20 Oct 2026 19:45:51 -0700 Subject: [PATCH 1/2] Added generic call methods to the worker extension. A bridge can route one generic lang call to the extension by method name, so a plugin's native code needs no bridge function of its own. --- crates/sdk-core/CHANGELOG.md | 3 + crates/sdk-core/src/lib.rs | 2 +- crates/sdk-core/src/worker/extension.rs | 124 +++++++++++++++++++++++- crates/sdk-core/src/worker/mod.rs | 2 +- 4 files changed, 128 insertions(+), 3 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 6d24face6..94342e43d 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -41,6 +41,9 @@ relevant information. the hooks. * The info a `WorkflowTaskInterceptor` gets names the run's last completed workflow task and the tasks that history shows as failed or timed out since then. +* 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. ### Security * Ephemeral server downloads now use `zip` 9.0 for ZIP extraction, replacing the 8.x dependency diff --git a/crates/sdk-core/src/lib.rs b/crates/sdk-core/src/lib.rs index db9075f9b..fd22a00e9 100644 --- a/crates/sdk-core/src/lib.rs +++ b/crates/sdk-core/src/lib.rs @@ -51,7 +51,7 @@ pub use worker::{ WorkflowErrorType, WorkflowSlotKind, }; pub use worker::{ - RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch, + ExtensionError, RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch, WorkflowTaskFailedInfo, WorkflowTaskInterceptError, WorkflowTaskInterceptor, }; diff --git a/crates/sdk-core/src/worker/extension.rs b/crates/sdk-core/src/worker/extension.rs index 45d7521d8..e641468ad 100644 --- a/crates/sdk-core/src/worker/extension.rs +++ b/crates/sdk-core/src/worker/extension.rs @@ -8,7 +8,10 @@ use std::{ /// A component that lang passes to the worker through [crate::WorkerConfig::extension]. /// -/// Core never inspects the extension beyond the hooks it asks for here. +/// Core never inspects the extension beyond the hooks it asks for here. 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; @@ -18,6 +21,56 @@ pub trait WorkerExtension: Send + Sync + 'static { fn workflow_task_interceptor(&self) -> Option> { None } + + /// 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}"), + ) + } } /// Hooks around each `RespondWorkflowTaskCompleted` call that Core makes for a run. @@ -144,3 +197,72 @@ impl WorkflowTaskInterceptError { } } } + +#[cfg(test)] +mod tests { + use super::*; + + 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 84701de3c..fcf58bca4 100644 --- a/crates/sdk-core/src/worker/mod.rs +++ b/crates/sdk-core/src/worker/mod.rs @@ -12,7 +12,7 @@ mod workflow; pub(crate) const PAYLOADS_TOO_LARGE_FAILURE_TYPE: &str = "PayloadsTooLarge"; pub use extension::{ - RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch, + ExtensionError, RunCloseKind, WorkerExtension, WorkflowTaskCompletionInfo, WorkflowTaskEpoch, WorkflowTaskFailedInfo, WorkflowTaskInterceptError, WorkflowTaskInterceptor, }; use temporalio_client::{Connection, PayloadErrorLimits, worker::NamespaceDescriptionSource}; From 9a34fce508fb94b6ccf3feac7fcc50f31c672f8d Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 20 Oct 2026 20:39:43 -0700 Subject: [PATCH 2/2] Removed stray merge lines from the changelog. --- crates/sdk-core/CHANGELOG.md | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index a08f181bb..aa4616945 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -37,19 +37,13 @@ relevant information. * A worker can take a `WorkerExtension` whose `WorkflowTaskInterceptor` runs before and after each Workflow Task completion Core sends, including the Local Activity heartbeat completion and the completion of output buffered before an eviction. An error from `before_complete`, or - a hook that misses the task's deadline, fails the workflow task. The replay worker never calls - the hooks. -* The info a `WorkflowTaskInterceptor` gets names the run's last completed workflow task and the - tasks that history shows as failed or timed out since then. -* 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. - tasks of the run that history shows as failed or timed out. - a hook that misses the task's deadline, fails the workflow task. Core waits at most five a hook that misses the task's deadline, fails the Workflow Task. Core waits at most five seconds on `after_complete`. The replay worker never calls the hooks. * The info a `WorkflowTaskInterceptor` gets has the run's last completed Workflow Task and the tasks of the run that history shows as failed or timed 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. ### Security * Ephemeral server downloads now use `zip` 9.0 for ZIP extraction, replacing the 8.x dependency