From e2d07be73763b5ec5cf1e956f1e8521196041e4c Mon Sep 17 00:00:00 2001 From: Thomas Hardy Date: Tue, 20 Oct 2026 11:26:52 -0400 Subject: [PATCH 1/5] Import revised worker session API Import temporalio/api sticky-activity-session-options at 6811161a3c59c7773ac236dbf9adf9eb359f9288 as an exact subtree snapshot. Register session messages and keep scheduling and poll session options absent until feature plumbing is implemented. git-subtree-dir: crates/protos/protos/api_upstream git-subtree-split: 6811161a3c59c7773ac236dbf9adf9eb359f9288 --- crates/protos/build.rs | 1 + .../protos/protos/api_upstream/buf.gen.yaml | 20 --------- .../temporal/api/command/v1/message.proto | 3 ++ .../temporal/api/history/v1/message.proto | 4 ++ .../temporal/api/session/v1/message.proto | 44 +++++++++++++++++++ .../workflowservice/v1/request_response.proto | 7 +++ crates/protos/src/protos/mod.rs | 6 +++ crates/sdk-core/src/worker/client.rs | 2 + 8 files changed, 67 insertions(+), 20 deletions(-) delete mode 100644 crates/protos/protos/api_upstream/buf.gen.yaml create mode 100644 crates/protos/protos/api_upstream/temporal/api/session/v1/message.proto diff --git a/crates/protos/build.rs b/crates/protos/build.rs index ddcdebf38..68f59ffcb 100644 --- a/crates/protos/build.rs +++ b/crates/protos/build.rs @@ -39,6 +39,7 @@ const SERDE_DERIVE_PREFIXES: &[&str] = &[ ".temporal.api.replication", ".temporal.api.rules", ".temporal.api.schedule", + ".temporal.api.session", ".temporal.api.taskqueue", ".temporal.api.testservice", ".temporal.api.update", diff --git a/crates/protos/protos/api_upstream/buf.gen.yaml b/crates/protos/protos/api_upstream/buf.gen.yaml deleted file mode 100644 index 555cd008d..000000000 --- a/crates/protos/protos/api_upstream/buf.gen.yaml +++ /dev/null @@ -1,20 +0,0 @@ -version: v1 -plugins: - - plugin: buf.build/protocolbuffers/go:v1.31.0 - out: ./ - opt: - - paths=source_relative - - plugin: buf.build/grpc/go:v1.3.0 - out: ./ - opt: - - paths=source_relative - - plugin: buf.build/grpc-ecosystem/gateway:v2.18.0 - out: ./ - opt: - - paths=source_relative - - allow_patch_feature=false - - name: go-helpers - out: ./ - path: ["go", "run", "./protoc-gen-go-helpers"] - opt: - - paths=source_relative diff --git a/crates/protos/protos/api_upstream/temporal/api/command/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/command/v1/message.proto index ee839115b..b8be64315 100644 --- a/crates/protos/protos/api_upstream/temporal/api/command/v1/message.proto +++ b/crates/protos/protos/api_upstream/temporal/api/command/v1/message.proto @@ -16,6 +16,7 @@ import "temporal/api/enums/v1/command_type.proto"; import "temporal/api/common/v1/message.proto"; import "temporal/api/failure/v1/message.proto"; import "temporal/api/taskqueue/v1/message.proto"; +import "temporal/api/session/v1/message.proto"; import "temporal/api/workflow/v1/message.proto"; import "temporal/api/sdk/v1/user_metadata.proto"; import "temporal/api/sdk/v1/event_group_marker.proto"; @@ -68,6 +69,8 @@ message ScheduleActivityTaskCommandAttributes { // Priority metadata. If this message is not present, or any fields are not // present, they inherit the values from the workflow. temporal.api.common.v1.Priority priority = 14; + + temporal.api.session.v1.ActivitySessionOptions session_options = 15; } message RequestCancelActivityTaskCommandAttributes { diff --git a/crates/protos/protos/api_upstream/temporal/api/history/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/history/v1/message.proto index 1c6efd305..8f5a233bf 100644 --- a/crates/protos/protos/api_upstream/temporal/api/history/v1/message.proto +++ b/crates/protos/protos/api_upstream/temporal/api/history/v1/message.proto @@ -20,6 +20,7 @@ import "temporal/api/common/v1/message.proto"; import "temporal/api/deployment/v1/message.proto"; import "temporal/api/failure/v1/message.proto"; import "temporal/api/nexus/v1/message.proto"; +import "temporal/api/session/v1/message.proto"; import "temporal/api/taskqueue/v1/message.proto"; import "temporal/api/update/v1/message.proto"; import "temporal/api/workflow/v1/message.proto"; @@ -465,6 +466,9 @@ message ActivityTaskScheduledEventAttributes { // Priority metadata. If this message is not present, or any fields are not // present, they inherit the values from the workflow. temporal.api.common.v1.Priority priority = 14; + // Preserves the scheduling command's session routing options so activity state + // can be reconstructed from history. + temporal.api.session.v1.ActivitySessionOptions session_options = 15; } message ActivityTaskStartedEventAttributes { diff --git a/crates/protos/protos/api_upstream/temporal/api/session/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/session/v1/message.proto new file mode 100644 index 000000000..4a7cdb2e5 --- /dev/null +++ b/crates/protos/protos/api_upstream/temporal/api/session/v1/message.proto @@ -0,0 +1,44 @@ +syntax = "proto3"; + +package temporal.api.session.v1; + +option go_package = "go.temporal.io/api/session/v1;session"; +option java_package = "io.temporal.api.session.v1"; +option java_multiple_files = true; +option java_outer_classname = "MessageProto"; +option ruby_package = "Temporalio::Api::Session::V1"; +option csharp_namespace = "Temporalio.Api.Session.V1"; + +// EXPERIMENTAL. Subject to redesign or removal. +// Configures a worker's polling for sessions that keep workflow and activity +// tasks on a stable worker destination. Activities opt into session routing +// through ActivitySessionOptions on their scheduling commands. +message WorkerSessionOptions { + oneof isolation { + SingleSessionOptions single_session = 1; + } +} + +// EXPERIMENTAL. Subject to redesign or removal. +// Dedicates a sticky destination to the first workflow matched to it. +// Workflow sticky polls may be forwarded to the normal queue until the first +// workflow task is dispatched. The server persists that workflow ID and does +// not forward subsequent polls, including after workflow completion. +// Activity sticky polls are never forwarded to the normal queue. +message SingleSessionOptions { + // Stable sticky task queue name, shared by workflow and activity polls. + // Reuse this value across worker restarts within the same session, but not + // across concurrently running workers. This is distinct from worker identity. + // The SDK must also set the polled sticky task queue's name to this value. + string stable_identity = 1; +} + +message ActivitySessionOptions { + oneof session { + // Execute this activity as part of its workflow's session. + // Currently routes to the workflow's sticky task queue when true. + bool use_workflow_session = 1; + + // later session_id, etc + } +} diff --git a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto index 2ad815a5e..e570a9466 100644 --- a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto +++ b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/request_response.proto @@ -38,6 +38,7 @@ import "temporal/api/replication/v1/message.proto"; import "temporal/api/rules/v1/message.proto"; import "temporal/api/sdk/v1/worker_config.proto"; import "temporal/api/schedule/v1/message.proto"; +import "temporal/api/session/v1/message.proto"; import "temporal/api/taskqueue/v1/message.proto"; import "temporal/api/update/v1/message.proto"; import "temporal/api/version/v1/message.proto"; @@ -323,6 +324,9 @@ message PollWorkflowTaskQueueRequest { // Removed in 1.55.0; was temporal.api.worker.v1.WorkerHeartbeat worker_heartbeat reserved 7; reserved "worker_heartbeat"; + + // EXPERIMENTAL. Subject to redesign or removal. Only applicable to sticky polls. + temporal.api.session.v1.WorkerSessionOptions session_options = 11; } message PollWorkflowTaskQueueResponse { @@ -553,6 +557,9 @@ message PollActivityTaskQueueRequest { // Removed in 1.55.0; was temporal.api.worker.v1.WorkerHeartbeat worker_heartbeat reserved 7; reserved "worker_heartbeat"; + + // EXPERIMENTAL. Subject to redesign or removal. Only applicable to sticky polls. + temporal.api.session.v1.WorkerSessionOptions session_options = 11; } message PollActivityTaskQueueResponse { diff --git a/crates/protos/src/protos/mod.rs b/crates/protos/src/protos/mod.rs index 8a7c8badd..0c6eb6e4b 100644 --- a/crates/protos/src/protos/mod.rs +++ b/crates/protos/src/protos/mod.rs @@ -1915,6 +1915,7 @@ pub mod temporal { request_eager_execution: !s.do_not_eagerly_execute, use_workflow_build_id, priority: s.priority, + session_options: None, }, ) } @@ -2619,6 +2620,11 @@ pub mod temporal { include_proto_with_serde!("temporal.api.sdk.v1"); } } + pub mod session { + pub mod v1 { + tonic::include_proto!("temporal.api.session.v1"); + } + } pub mod taskqueue { pub mod v1 { tonic::include_proto!("temporal.api.taskqueue.v1"); diff --git a/crates/sdk-core/src/worker/client.rs b/crates/sdk-core/src/worker/client.rs index 00ba75f62..c332c2ff7 100644 --- a/crates/sdk-core/src/worker/client.rs +++ b/crates/sdk-core/src/worker/client.rs @@ -493,6 +493,7 @@ impl WorkerClient for WorkerClientBag { worker_instance_key: self.worker_instance_key.to_string(), worker_control_task_queue: self.worker_control_task_queue(), poller_group_id: Default::default(), + session_options: None, } .into_request(); request.extensions_mut().insert(IsWorkerTaskLongPoll); @@ -533,6 +534,7 @@ impl WorkerClient for WorkerClientBag { worker_instance_key: self.worker_instance_key.to_string(), worker_control_task_queue: self.worker_control_task_queue(), poller_group_id: Default::default(), + session_options: None, } .into_request(); request.extensions_mut().insert(IsWorkerTaskLongPoll); From 75d42a7df9eb27be1131c711ea3187791932bc08 Mon Sep 17 00:00:00 2001 From: Thomas Hardy Date: Wed, 21 Oct 2026 09:36:41 -0400 Subject: [PATCH 2/5] Add sticky activity session polling and routing --- .../workflow_commands/workflow_commands.proto | 2 + crates/protos/src/protos/mod.rs | 2 +- crates/sdk-core/CHANGELOG.md | 5 + crates/sdk-core/src/abstractions.rs | 5 +- crates/sdk-core/src/lib.rs | 10 +- crates/sdk-core/src/pollers/mod.rs | 4 +- crates/sdk-core/src/pollers/poll_buffer.rs | 56 +++-- crates/sdk-core/src/pollers/sticky.rs | 16 +- crates/sdk-core/src/replay/mod.rs | 62 +++++- crates/sdk-core/src/telemetry/metrics.rs | 8 + crates/sdk-core/src/worker/activities.rs | 10 +- .../src/worker/activities/act_poller.rs | 206 ++++++++++++++++++ crates/sdk-core/src/worker/client.rs | 42 ++-- crates/sdk-core/src/worker/mod.rs | 106 ++++++--- .../workflow/machines/workflow_machines.rs | 116 +++++++++- .../src/worker/workflow/wft_poller.rs | 8 +- 16 files changed, 572 insertions(+), 86 deletions(-) create mode 100644 crates/sdk-core/src/worker/activities/act_poller.rs diff --git a/crates/protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto b/crates/protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto index 2db4ccfb6..19d1291b8 100644 --- a/crates/protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto +++ b/crates/protos/protos/local/temporal/sdk/core/workflow_commands/workflow_commands.proto @@ -16,6 +16,7 @@ import "temporal/api/enums/v1/workflow.proto"; import "temporal/api/failure/v1/message.proto"; import "temporal/api/sdk/v1/user_metadata.proto"; import "temporal/api/sdk/v1/event_group_marker.proto"; +import "temporal/api/session/v1/message.proto"; import "temporal/api/workflow/v1/message.proto"; import "temporal/sdk/core/child_workflow/child_workflow.proto"; import "temporal/sdk/core/nexus/nexus.proto"; @@ -106,6 +107,7 @@ message ScheduleActivity { coresdk.common.VersioningIntent versioning_intent = 15; // The Priority to use for this activity temporal.api.common.v1.Priority priority = 16; + temporal.api.session.v1.ActivitySessionOptions session_options = 17; } message ScheduleLocalActivity { diff --git a/crates/protos/src/protos/mod.rs b/crates/protos/src/protos/mod.rs index 0c6eb6e4b..9efaaf77e 100644 --- a/crates/protos/src/protos/mod.rs +++ b/crates/protos/src/protos/mod.rs @@ -1915,7 +1915,7 @@ pub mod temporal { request_eager_execution: !s.do_not_eagerly_execute, use_workflow_build_id, priority: s.priority, - session_options: None, + session_options: s.session_options, }, ) } diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 9cd893121..db35d64c1 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -40,6 +40,11 @@ relevant information. name the run's last completed workflow task, and `failed_tasks`, the started event ids of the run's last 100 failed or timed out workflow tasks. Retries that Core runs locally keep the value. Core leaves it unset for other activities. +* Experimental session workers can poll workflow-affine activity queues; individual activities opt + into their workflow's session, while normal activity routing remains the default. This increment + supports fixed-size activity slots with capacity of at least two; resource-based and custom + activity suppliers are not supported yet. Activity session options must select workflow-session + routing; empty options or an explicit false selection are rejected. ### Security * Ephemeral server downloads now use `zip` 9.0 for ZIP extraction, replacing the 8.x dependency diff --git a/crates/sdk-core/src/abstractions.rs b/crates/sdk-core/src/abstractions.rs index eee40b04b..93f53bc48 100644 --- a/crates/sdk-core/src/abstractions.rs +++ b/crates/sdk-core/src/abstractions.rs @@ -35,8 +35,7 @@ pub(crate) struct MeteredPermitDealer { max_permits: Option, metrics_ctx: MetricsContext, meter: Option, - /// Only applies to permit dealers for workflow tasks. True if this permit dealer is associated - /// with a sticky queue poller. + /// True if this permit dealer is associated with a sticky queue poller. is_sticky_poller: bool, context_data: Arc, } @@ -198,7 +197,7 @@ where } } -impl MeteredPermitDealer { +impl MeteredPermitDealer { pub(crate) fn into_sticky(mut self) -> Self { self.is_sticky_poller = true; self diff --git a/crates/sdk-core/src/lib.rs b/crates/sdk-core/src/lib.rs index cf0f68c11..f13ed11d9 100644 --- a/crates/sdk-core/src/lib.rs +++ b/crates/sdk-core/src/lib.rs @@ -44,11 +44,11 @@ pub use worker::{ FixedSizeSlotSupplier, LocalActivitySlotKind, NamespaceCapabilities, NexusSlotKind, PollError, PollerBehavior, ResourceBasedSlotsOptions, ResourceBasedSlotsOptionsBuilder, ResourceBasedTuner, ResourceBasedTunerConfig, ResourceController, ResourceSlotOptions, - SlotInfo, SlotInfoTrait, SlotKind, SlotKindType, SlotMarkUsedContext, SlotReleaseContext, - SlotReservationContext, SlotSupplier, SlotSupplierOptions, SlotSupplierPermit, TunerBuilder, - TunerHolder, TunerHolderOptions, TunerHolderOptionsBuilder, Worker, WorkerConfig, - WorkerConfigBuilder, WorkerTuner, WorkerValidationError, WorkerVersioningStrategy, - WorkflowErrorType, WorkflowSlotKind, + SessionWorkerOptions, SessionWorkerOptionsBuilder, SlotInfo, SlotInfoTrait, SlotKind, + SlotKindType, SlotMarkUsedContext, SlotReleaseContext, SlotReservationContext, SlotSupplier, + SlotSupplierOptions, SlotSupplierPermit, TunerBuilder, TunerHolder, TunerHolderOptions, + TunerHolderOptionsBuilder, Worker, WorkerConfig, WorkerConfigBuilder, WorkerTuner, + WorkerValidationError, WorkerVersioningStrategy, WorkflowErrorType, WorkflowSlotKind, }; use crate::{ diff --git a/crates/sdk-core/src/pollers/mod.rs b/crates/sdk-core/src/pollers/mod.rs index e6f555762..64df2a19d 100644 --- a/crates/sdk-core/src/pollers/mod.rs +++ b/crates/sdk-core/src/pollers/mod.rs @@ -1,7 +1,9 @@ mod poll_buffer; mod sticky; -pub(crate) use poll_buffer::{ActivityTaskOptions, LongPollBuffer, WorkflowTaskOptions}; +pub(crate) use poll_buffer::{ + ActivityTaskOptions, LongPollBuffer, PollRateLimiter, WorkflowTaskOptions, +}; pub(crate) use sticky::{StickyPollerShared, StickyTaskPoller, sticky_poller_behavior}; pub use temporalio_client::{Client, ClientOptions, ClientTlsOptions, RetryOptions, TlsOptions}; diff --git a/crates/sdk-core/src/pollers/poll_buffer.rs b/crates/sdk-core/src/pollers/poll_buffer.rs index 12772bad2..b4ab589eb 100644 --- a/crates/sdk-core/src/pollers/poll_buffer.rs +++ b/crates/sdk-core/src/pollers/poll_buffer.rs @@ -68,13 +68,13 @@ const PERSISTENT_POLL_ERROR_WARN_BACKOFF: ExponentialBuilder = ExponentialBuilde type PollReceiver = Mutex)>>>; -struct PollRateLimiter { +pub(crate) struct PollRateLimiter { interval: Duration, next_allowed_at: Mutex, } impl PollRateLimiter { - fn new(polls_per_second: f64) -> Self { + pub(crate) fn new(polls_per_second: f64) -> Self { Self { interval: Duration::from_secs_f64(polls_per_second.recip()), next_allowed_at: Mutex::new(Instant::now()), @@ -106,11 +106,13 @@ pub(crate) struct WorkflowTaskOptions { pub(crate) wft_poller_shared: Option>, } +#[derive(Default)] pub(crate) struct ActivityTaskOptions { /// Local ratelimit - pub(crate) max_worker_acts_per_second: Option, + pub(crate) rate_limiter: Option>, /// Server ratelimit pub(crate) max_tps: Option, + pub(crate) poller_shared: Option>, } impl LongPollBuffer { @@ -202,6 +204,7 @@ impl LongPollBuffer { pub(crate) fn new_activity_task( client: Arc, task_queue: String, + sticky_queue: Option, poller_behavior: PollerBehavior, permit_dealer: MeteredPermitDealer, shutdown: CancellationToken, @@ -210,15 +213,37 @@ impl LongPollBuffer { last_successful_poll_time: Arc>>, capabilities: Arc, ) -> Self { - let pre_permit_delay = options - .max_worker_acts_per_second - .map(|ps| Arc::new(PollRateLimiter::new(ps))) - .map(|rl| { - move || { - let rl = rl.clone(); - async move { rl.wait().await }.boxed() + let is_sticky = sticky_queue.is_some(); + let poll_scaler = PollScaler::new( + poller_behavior, + num_pollers_handler, + shutdown.clone(), + last_successful_poll_time, + capabilities.clone(), + ); + if let Some(shared) = &options.poller_shared { + if is_sticky { + shared.set_sticky_active( + poll_scaler.active_rx.clone(), + poll_scaler.report_handle.target.subscribe(), + ); + } else { + shared.set_non_sticky_active(poll_scaler.active_rx.clone()); + } + } + let pre_permit_delay = Some(move || { + let shared = options.poller_shared.clone(); + let rate_limiter = options.rate_limiter.clone(); + async move { + if let Some(shared) = shared { + // Activity responses have no sticky backlog hint; backlog priority stays inactive. + shared.wait_if_needed(is_sticky).await; } - }); + if let Some(rate_limiter) = rate_limiter { + rate_limiter.wait().await; + } + } + }); let no_retry = if matches!(poller_behavior, PollerBehavior::Autoscaling { .. }) { Some(NoRetryOnMatching { predicate: poll_scaling_error_matcher, @@ -229,6 +254,7 @@ impl LongPollBuffer { let poll_fn = move |timeout_override: Option| { let client = client.clone(); let task_queue = task_queue.clone(); + let sticky_queue_name = sticky_queue.clone(); async move { client .poll_activity_task( @@ -239,19 +265,13 @@ impl LongPollBuffer { }, PollActivityOptions { max_tasks_per_sec: options.max_tps, + sticky_queue_name, }, ) .await } }; - let poll_scaler = PollScaler::new( - poller_behavior, - num_pollers_handler, - shutdown.clone(), - last_successful_poll_time, - capabilities.clone(), - ); Self::new( poll_fn, permit_dealer, diff --git a/crates/sdk-core/src/pollers/sticky.rs b/crates/sdk-core/src/pollers/sticky.rs index 332dfbe64..c4ad5c6ca 100644 --- a/crates/sdk-core/src/pollers/sticky.rs +++ b/crates/sdk-core/src/pollers/sticky.rs @@ -209,11 +209,7 @@ mod tests { #[test] fn max_polls_calculated_properly() { - let cfg = { - let mut cfg = test_worker_cfg().build().unwrap(); - cfg.workflow_task_poller_behavior = Some(PollerBehavior::SimpleMaximum(5_usize)); - cfg - }; + let cfg = test_worker_cfg().build().unwrap(); assert_eq!( sticky_poller_behavior(PollerBehavior::SimpleMaximum(5), &cfg, false), PollerBehavior::SimpleMaximum(1) @@ -222,5 +218,15 @@ mod tests { sticky_poller_behavior(PollerBehavior::SimpleMaximum(5), &cfg, true), PollerBehavior::SimpleMaximum(4) ); + let autoscaling = PollerBehavior::Autoscaling { + minimum: 1, + maximum: 10, + initial: 3, + }; + assert_eq!( + sticky_poller_behavior(autoscaling, &cfg, false), + autoscaling + ); + assert_eq!(sticky_poller_behavior(autoscaling, &cfg, true), autoscaling); } } diff --git a/crates/sdk-core/src/replay/mod.rs b/crates/sdk-core/src/replay/mod.rs index 6e4cadc50..251d767d8 100644 --- a/crates/sdk-core/src/replay/mod.rs +++ b/crates/sdk-core/src/replay/mod.rs @@ -75,6 +75,8 @@ where } pub(crate) fn into_core_worker(mut self) -> Result { + // Offline replay consumes supplied history and does not execute activities. + self.config.session_worker_options = None; self.config.max_cached_workflows = 1; self.config.workflow_task_poller_behavior = Some(PollerBehavior::SimpleMaximum(1)); self.config.task_types = WorkerTaskTypes::workflow_only(); @@ -150,8 +152,66 @@ where #[cfg(test)] mod tests { use super::*; - use crate::test_help::test_worker_cfg; + use crate::test_help::{WorkerExt, test_worker_cfg}; use futures_util::{FutureExt, stream}; + use temporalio_common::protos::{ + coresdk::{ + workflow_activation::workflow_activation_job, + workflow_commands::{CompleteWorkflowExecution, ScheduleActivity}, + workflow_completion::WorkflowActivationCompletion, + }, + temporal::api::session::v1::{ActivitySessionOptions, activity_session_options}, + }; + + #[tokio::test] + async fn replay_session_activity_without_session_worker() { + let mut history = canned_histories::single_activity("activity"); + history.add_workflow_task_completed(); + history.add_workflow_execution_completed(); + let mut client = mock_manual_worker_client(); + client.expect_fail_workflow_task().times(0); + let mut input = ReplayWorkerInput::new( + test_worker_cfg().build().unwrap(), + stream::iter([history.into()]), + ); + input.client_override = Some(client); + let worker = input.into_core_worker().unwrap(); + let start = worker.poll_workflow_activation().await.unwrap(); + assert!(start.is_replaying); + worker + .complete_workflow_activation(WorkflowActivationCompletion::from_cmds( + start.run_id, + vec![ + ScheduleActivity { + seq: 1, + activity_id: "activity".to_owned(), + activity_type: DEFAULT_ACTIVITY_TYPE.to_owned(), + session_options: Some(ActivitySessionOptions { + session: Some(activity_session_options::Session::UseWorkflowSession( + true, + )), + }), + ..Default::default() + } + .into(), + ], + )) + .await + .unwrap(); + let resolved = worker.poll_workflow_activation().await.unwrap(); + assert!(matches!( + resolved.jobs[0].variant, + Some(workflow_activation_job::Variant::ResolveActivity(_)) + )); + worker + .complete_workflow_activation(WorkflowActivationCompletion::from_cmds( + resolved.run_id, + vec![CompleteWorkflowExecution::default().into()], + )) + .await + .unwrap(); + worker.drain_pollers_and_shutdown().await; + } #[tokio::test] async fn client_override_describe_namespace_precedes_fallback() { diff --git a/crates/sdk-core/src/telemetry/metrics.rs b/crates/sdk-core/src/telemetry/metrics.rs index d95301d6f..2caf3b887 100644 --- a/crates/sdk-core/src/telemetry/metrics.rs +++ b/crates/sdk-core/src/telemetry/metrics.rs @@ -647,6 +647,7 @@ pub(crate) struct NumPollersMetric { pub wft_current_pollers: HeartbeatMetric, pub sticky_wft_current_pollers: HeartbeatMetric, pub activity_current_pollers: HeartbeatMetric, + pub sticky_activity_current_pollers: HeartbeatMetric, pub nexus_current_pollers: HeartbeatMetric, } @@ -659,6 +660,10 @@ impl NumPollersMetric { self.sticky_wft_current_pollers.clone(), ), ("activity_task", self.activity_current_pollers.clone()), + ( + "sticky_activity_task", + self.sticky_activity_current_pollers.clone(), + ), ("nexus_task", self.nexus_current_pollers.clone()), ]) } @@ -760,6 +765,9 @@ pub(crate) fn workflow_sticky_poller() -> MetricKeyValue { pub(crate) fn activity_poller() -> MetricKeyValue { MetricKeyValue::new(KEY_POLLER_TYPE, "activity_task") } +pub(crate) fn activity_sticky_poller() -> MetricKeyValue { + MetricKeyValue::new(KEY_POLLER_TYPE, "sticky_activity_task") +} pub(crate) fn nexus_poller() -> MetricKeyValue { MetricKeyValue::new(KEY_POLLER_TYPE, "nexus_task") } diff --git a/crates/sdk-core/src/worker/activities.rs b/crates/sdk-core/src/worker/activities.rs index a167d6a43..14b0d0678 100644 --- a/crates/sdk-core/src/worker/activities.rs +++ b/crates/sdk-core/src/worker/activities.rs @@ -1,3 +1,4 @@ +pub(crate) mod act_poller; mod activity_heartbeat_manager; mod local_activities; @@ -1030,7 +1031,7 @@ mod tests { use super::*; use crate::{ abstractions::tests::fixed_size_permit_dealer, - pollers::{ActivityTaskOptions, LongPollBuffer}, + pollers::{ActivityTaskOptions, LongPollBuffer, PollRateLimiter}, prost_dur, worker::{ NamespaceCapabilities, PollerBehavior, @@ -1049,13 +1050,14 @@ mod tests { let ap = LongPollBuffer::new_activity_task( mock_client.clone(), "tq".to_string(), + None, PollerBehavior::SimpleMaximum(1), sem.clone(), shutdown_token, None::, ActivityTaskOptions { - max_worker_acts_per_second: None, max_tps: None, + ..Default::default() }, Arc::new(AtomicCell::new(None)), Arc::new(NamespaceCapabilities::default()), @@ -1135,14 +1137,16 @@ mod tests { let ap = LongPollBuffer::new_activity_task( mock_client.clone(), "tq".to_string(), + None, // Lots of concurrent pollers, to ensure we don't poll to much when that's the case PollerBehavior::SimpleMaximum(5), sem.clone(), shutdown_token.clone(), None::, ActivityTaskOptions { - max_worker_acts_per_second: Some(2.0), + rate_limiter: Some(Arc::new(PollRateLimiter::new(2.0))), max_tps: None, + ..Default::default() }, Arc::new(AtomicCell::new(None)), Arc::new(NamespaceCapabilities::default()), diff --git a/crates/sdk-core/src/worker/activities/act_poller.rs b/crates/sdk-core/src/worker/activities/act_poller.rs new file mode 100644 index 000000000..9525e6170 --- /dev/null +++ b/crates/sdk-core/src/worker/activities/act_poller.rs @@ -0,0 +1,206 @@ +use crate::{ + MetricsContext, WorkerConfig, + abstractions::MeteredPermitDealer, + pollers::{ + ActivityTaskOptions, LongPollBuffer, PollRateLimiter, StickyPollerShared, StickyTaskPoller, + sticky_poller_behavior, + }, + telemetry::metrics::{activity_poller, activity_sticky_poller}, + worker::{ActivitySlotKind, NamespaceCapabilities, client::WorkerClient}, +}; +use crossbeam_utils::atomic::AtomicCell; +use std::{sync::Arc, time::SystemTime}; +use temporalio_common::protos::temporal::api::workflowservice::v1::PollActivityTaskQueueResponse; +use tokio_util::sync::CancellationToken; + +#[allow(clippy::too_many_arguments)] +pub(crate) fn make_activity_poller( + config: &WorkerConfig, + sticky_queue: Option<&str>, + client: &Arc, + metrics: &MetricsContext, + shutdown_token: &CancellationToken, + act_slots: &MeteredPermitDealer, + last_successful_poll_time: Arc>>, + capabilities: Arc, +) -> StickyTaskPoller { + let effective_behavior = crate::worker::resolve_effective_behavior( + config.activity_task_poller_behavior, + &capabilities, + ); + let poller_shared = + sticky_queue.map(|_| Arc::new(StickyPollerShared::new(act_slots.available_permits()))); + let rate_limiter = config + .max_worker_activities_per_second + .map(|rate| Arc::new(PollRateLimiter::new(rate))); + let act_metrics = metrics.with_new_attrs([activity_poller()]); + let normal_poller = LongPollBuffer::new_activity_task( + client.clone(), + config.task_queue.clone(), + None, + if sticky_queue.is_some() { + sticky_poller_behavior(effective_behavior, config, false) + } else { + effective_behavior + }, + act_slots.clone(), + shutdown_token.child_token(), + Some(move |np| act_metrics.record_num_pollers(np)), + ActivityTaskOptions { + rate_limiter: rate_limiter.clone(), + max_tps: config.max_task_queue_activities_per_second, + poller_shared: poller_shared.clone(), + }, + last_successful_poll_time.clone(), + capabilities.clone(), + ); + let sticky_poller = sticky_queue.map(|sqn| { + let sticky_metrics = metrics.with_new_attrs([activity_sticky_poller()]); + LongPollBuffer::new_activity_task( + client.clone(), + config.task_queue.clone(), + Some(sqn.to_owned()), + sticky_poller_behavior(effective_behavior, config, true), + act_slots.clone().into_sticky(), + shutdown_token.child_token(), + Some(move |np| sticky_metrics.record_num_pollers(np)), + ActivityTaskOptions { + rate_limiter, + max_tps: config.max_task_queue_activities_per_second, + poller_shared, + }, + last_successful_poll_time, + capabilities, + ) + }); + StickyTaskPoller::new(normal_poller, sticky_poller) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + abstractions::tests::fixed_size_permit_dealer, + pollers::Poller, + test_help::test_worker_cfg, + worker::{PollerBehavior, client::mocks::mock_manual_worker_client}, + }; + use futures_util::FutureExt; + use std::{future::pending, time::Duration}; + use tokio::sync::mpsc::unbounded_channel; + + #[tokio::test(start_paused = true)] + async fn session_activity_pollers_share_capacity() { + let (tx, mut rx) = unbounded_channel(); + let mut client = mock_manual_worker_client(); + client + .expect_poll_activity_task() + .times(2) + .returning(move |poll, options| { + tx.send(( + poll.task_queue, + options.sticky_queue_name, + tokio::time::Instant::now(), + )) + .unwrap(); + pending().boxed() + }); + let client: Arc = Arc::new(client); + let config = WorkerConfig::builder() + .namespace("default") + .task_queue("normal") + .versioning_strategy(crate::WorkerVersioningStrategy::None { + build_id: String::new(), + }) + .task_types(temporalio_common::worker::WorkerTaskTypes::all()) + .max_cached_workflows(2_usize) + .workflow_task_poller_behavior(PollerBehavior::SimpleMaximum(2)) + .activity_task_poller_behavior(PollerBehavior::SimpleMaximum(10)) + .max_worker_activities_per_second(2.0) + .session_worker_options(crate::SessionWorkerOptions::default()) + .build() + .unwrap(); + let slots = fixed_size_permit_dealer(2); + let poller = Arc::new(make_activity_poller( + &config, + Some("sticky"), + &client, + &MetricsContext::no_op(), + &CancellationToken::new(), + &slots, + Arc::new(AtomicCell::new(None)), + Arc::new(NamespaceCapabilities::default()), + )); + let polling = tokio::spawn({ + let poller = poller.clone(); + async move { poller.poll().await } + }); + let first = tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await + .unwrap() + .unwrap(); + let second = tokio::time::timeout(Duration::from_secs(5), rx.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(first.0, config.task_queue); + assert_eq!(second.0, config.task_queue); + assert_ne!(first.1, second.1); + assert!(second.2.duration_since(first.2) >= Duration::from_millis(500)); + assert!(first.1.as_deref() == Some("sticky") || second.1.as_deref() == Some("sticky")); + assert_eq!(slots.available_permits(), Some(0)); + poller.notify_shutdown(); + assert!(polling.await.unwrap().is_none()); + match Arc::try_unwrap(poller) { + Ok(poller) => poller.shutdown().await, + Err(_) => panic!("poller still referenced"), + } + assert_eq!(slots.available_permits(), Some(2)); + } + + #[tokio::test] + async fn ordinary_activity_polling_keeps_full_poller_budget() { + tokio::time::timeout(Duration::from_secs(5), async { + let (tx, mut rx) = unbounded_channel(); + let mut client = mock_manual_worker_client(); + client + .expect_poll_activity_task() + .times(5) + .returning(move |_, options| { + assert!(options.sticky_queue_name.is_none()); + tx.send(()).unwrap(); + pending().boxed() + }); + let client: Arc = Arc::new(client); + let config = test_worker_cfg() + .activity_task_poller_behavior(PollerBehavior::SimpleMaximum(5)) + .build() + .unwrap(); + let poller = Arc::new(make_activity_poller( + &config, + None, + &client, + &MetricsContext::no_op(), + &CancellationToken::new(), + &fixed_size_permit_dealer(10), + Arc::new(AtomicCell::new(None)), + Arc::new(NamespaceCapabilities::default()), + )); + let polling = tokio::spawn({ + let poller = poller.clone(); + async move { poller.poll().await } + }); + for _ in 0..5 { + rx.recv().await.unwrap(); + } + poller.notify_shutdown(); + assert!(polling.await.unwrap().is_none()); + match Arc::try_unwrap(poller) { + Ok(poller) => poller.shutdown().await, + Err(_) => panic!("poller still referenced"), + } + }) + .await + .expect("ordinary workers should start the full configured poller budget"); + } +} diff --git a/crates/sdk-core/src/worker/client.rs b/crates/sdk-core/src/worker/client.rs index c332c2ff7..b9d50d875 100644 --- a/crates/sdk-core/src/worker/client.rs +++ b/crates/sdk-core/src/worker/client.rs @@ -453,6 +453,8 @@ pub struct PollWorkflowOptions { pub struct PollActivityOptions { /// Optional rate limit (tasks per second) for activity polling pub max_tasks_per_sec: Option, + /// Poll this sticky destination instead of the normal activity queue. + pub sticky_queue_name: Option, } /// Additional options specific to Nexus task polling #[derive(Debug, Clone, Default)] @@ -462,6 +464,21 @@ pub struct PollNexusOptions { pub worker_commands_queue: bool, } +fn poll_task_queue(normal_name: String, sticky_name: Option) -> TaskQueue { + match sticky_name { + Some(name) => TaskQueue { + name, + kind: TaskQueueKind::Sticky.into(), + normal_name, + }, + None => TaskQueue { + name: normal_name, + kind: TaskQueueKind::Normal.into(), + normal_name: String::new(), + }, + } +} + #[async_trait::async_trait] impl WorkerClient for WorkerClientBag { async fn poll_workflow_task( @@ -469,19 +486,7 @@ impl WorkerClient for WorkerClientBag { poll_options: PollOptions, wf_options: PollWorkflowOptions, ) -> Result { - let task_queue = if let Some(sticky) = wf_options.sticky_queue_name { - TaskQueue { - name: sticky, - kind: TaskQueueKind::Sticky.into(), - normal_name: poll_options.task_queue, - } - } else { - TaskQueue { - name: poll_options.task_queue, - kind: TaskQueueKind::Normal.into(), - normal_name: "".to_string(), - } - }; + let task_queue = poll_task_queue(poll_options.task_queue, wf_options.sticky_queue_name); #[allow(deprecated)] // want to list all fields explicitly let mut request = PollWorkflowTaskQueueRequest { namespace: self.namespace.clone(), @@ -493,6 +498,7 @@ impl WorkerClient for WorkerClientBag { worker_instance_key: self.worker_instance_key.to_string(), worker_control_task_queue: self.worker_control_task_queue(), poller_group_id: Default::default(), + // Activity routing does not opt this worker into the deferred isolation policy. session_options: None, } .into_request(); @@ -520,11 +526,10 @@ impl WorkerClient for WorkerClientBag { #[allow(deprecated)] // want to list all fields explicitly let mut request = PollActivityTaskQueueRequest { namespace: self.namespace.clone(), - task_queue: Some(TaskQueue { - name: poll_options.task_queue, - kind: TaskQueueKind::Normal as i32, - normal_name: "".to_string(), - }), + task_queue: Some(poll_task_queue( + poll_options.task_queue, + act_options.sticky_queue_name, + )), identity: self.identity(), task_queue_metadata: act_options.max_tasks_per_sec.map(|tps| TaskQueueMetadata { max_tasks_per_second: Some(tps), @@ -534,6 +539,7 @@ impl WorkerClient for WorkerClientBag { worker_instance_key: self.worker_instance_key.to_string(), worker_control_task_queue: self.worker_control_task_queue(), poller_group_id: Default::default(), + // Activity routing does not opt this worker into the deferred isolation policy. session_options: None, } .into_request(); diff --git a/crates/sdk-core/src/worker/mod.rs b/crates/sdk-core/src/worker/mod.rs index 46c2c85a1..43bd2dac5 100644 --- a/crates/sdk-core/src/worker/mod.rs +++ b/crates/sdk-core/src/worker/mod.rs @@ -45,14 +45,17 @@ pub use workflow::LEGACY_QUERY_ID; use crate::{ ActivityHeartbeat, abstractions::{MeteredPermitDealer, PermitDealerContextData, dbg_panic}, - pollers::{ActivityTaskOptions, BoxedActPoller, BoxedNexusPoller, LongPollBuffer}, + pollers::{BoxedActPoller, BoxedNexusPoller, LongPollBuffer}, protosext::validate_activity_completion, telemetry::metrics::{ - MetricsContext, WorkerHeartbeatMetrics, activity_poller, activity_worker_type, - local_activity_worker_type, nexus_poller, nexus_worker_type, workflow_worker_type, + MetricsContext, WorkerHeartbeatMetrics, activity_worker_type, local_activity_worker_type, + nexus_poller, nexus_worker_type, workflow_worker_type, }, worker::{ - activities::{LACompleteAction, LocalActivityManager, NextPendingLAAction}, + activities::{ + LACompleteAction, LocalActivityManager, NextPendingLAAction, + act_poller::make_activity_poller, + }, client::WorkerClient, heartbeat::SharedNamespaceWorker, nexus::NexusManager, @@ -126,6 +129,13 @@ use { }, }; +/// Experimental options enabling activity polling on the workflow worker's sticky destination. +/// Currently supported only with fixed-size activity slots and capacity of at least two. +/// Resource-based and custom activity slot suppliers are not supported in this increment. +#[derive(Clone, Debug, Default, bon::Builder)] +#[non_exhaustive] +pub struct SessionWorkerOptions {} + /// Defines per-worker configuration options #[derive(Clone, bon::Builder)] #[builder(on(String, into), state_mod(vis = "pub"), finish_fn(vis = "", name = build_internal))] @@ -147,6 +157,9 @@ pub struct WorkerConfig { /// or failures. #[builder(default = 0)] pub max_cached_workflows: usize, + /// Experimental: enables sticky activity polling. Requires workflows, remote activities, + /// and a nonzero workflow cache. Activities still opt into session routing individually. + pub session_worker_options: Option, /// Set a [crate::WorkerTuner] for this worker. Either this or at least one of the /// `max_outstanding_*` fields must be set. pub tuner: Option>, @@ -166,6 +179,7 @@ pub struct WorkerConfig { /// queue will allow 4 max pollers while the nonsticky queue will allow one. The minimum for /// either poller is 1, so if the maximum allowed is 1 and sticky queues are enabled, there will /// be 2 concurrent polls. + /// Session workers apply the same split to normal and sticky activity polling. #[builder(default = 0.2)] pub nonsticky_to_sticky_poll_ratio: f32, /// Maximum number of concurrent poll activity task requests we will perform at a time on this @@ -346,6 +360,13 @@ impl WorkerConfigBuilder { "`task_types` cannot enable local activities without workflows".to_string(), ); } + if config.session_worker_options.is_some() + && (!task_types.enable_workflows + || !task_types.enable_remote_activities + || config.max_cached_workflows == 0) + { + return Err("`session_worker_options` requires workflows, remote activities, and a nonzero workflow cache".to_string()); + } if let Some(b) = config.workflow_task_poller_behavior { b.validate()?; @@ -371,6 +392,13 @@ impl WorkerConfigBuilder { if matches!(config.max_outstanding_activities, Some(v) if v == 0) { return Err("`max_outstanding_activities` must be > 0".to_string()); } + if config.session_worker_options.is_some() + && matches!(config.max_outstanding_activities, Some(v) if v < 2) + { + return Err( + "Paired session activity polling requires at least 2 activity slots".to_string(), + ); + } if matches!(config.max_outstanding_local_activities, Some(v) if v == 0) { return Err("`max_outstanding_local_activities` must be > 0".to_string()); } @@ -930,7 +958,7 @@ impl Worker { let wft_stream = if config.task_types.enable_workflows { let stream = make_wft_poller( &config, - &sticky_queue_name, + sticky_queue_name.as_deref(), &client, &metrics, &shutdown_token, @@ -957,22 +985,17 @@ impl Worker { }; let act_poll_buffer = if config.task_types.enable_remote_activities { - let act_metrics = metrics.with_new_attrs([activity_poller()]); - let ap = LongPollBuffer::new_activity_task( - client.clone(), - config.task_queue.clone(), - resolve_effective_behavior( - config.activity_task_poller_behavior, - &capabilities, - ), - act_slots.clone(), - shutdown_token.child_token(), - Some(move |np| act_metrics.record_num_pollers(np)), - ActivityTaskOptions { - max_worker_acts_per_second: config - .max_worker_activities_per_second, - max_tps: config.max_task_queue_activities_per_second, - }, + let activity_sticky_queue = config + .session_worker_options + .as_ref() + .and(sticky_queue_name.as_deref()); + let ap = make_activity_poller( + &config, + activity_sticky_queue, + &client, + &metrics, + &shutdown_token, + &act_slots, act_last_suc_poll_time.clone(), capabilities.clone(), ); @@ -1857,7 +1880,9 @@ pub trait WorkerTuner { &self, ) -> Arc + Send + Sync>; - /// Return a [SlotSupplier] for activity tasks + /// Return a [SlotSupplier] for activity tasks. + /// Experimental session workers currently require a fixed-size activity supplier with + /// at least two slots. Resource-based and custom suppliers are not yet supported for sessions. fn activity_task_slot_supplier( &self, ) -> Arc + Send + Sync>; @@ -1930,7 +1955,7 @@ pub trait SlotReservationContext: Send + Sync { /// Returns the number of currently outstanding slot permits, whether used or un-used. fn num_issued_slots(&self) -> usize; - /// Returns true iff this is a sticky poll for a workflow task + /// Returns true iff this is a sticky poll for a workflow or activity task. fn is_sticky(&self) -> bool; /// Returns the metrics meter if metrics are enabled @@ -2406,10 +2431,14 @@ impl WorkerHeartbeatManager { .is_autoscaling(), }); worker_heartbeat.activity_poller_info = Some(WorkerPollerInfo { - current_pollers: in_mem + current_pollers: (in_mem .num_pollers .activity_current_pollers - .load(Ordering::Relaxed) as i32, + .load(Ordering::Relaxed) + + in_mem + .num_pollers + .sticky_activity_current_pollers + .load(Ordering::Relaxed)) as i32, last_successful_poll_time: heartbeat_manager_metrics .act_last_suc_poll_time .load() @@ -2621,6 +2650,33 @@ mod tests { ); } + #[test] + fn session_worker_requires_both_task_types_cache_and_paired_capacity() { + for (workflows, activities, cache, slots, valid) in [ + (true, true, 2, 2_usize, true), + (false, true, 2, 2, false), + (true, false, 2, 2, false), + (true, true, 0, 2, false), + (true, true, 2, 1, false), + ] { + let config = WorkerConfig::builder() + .namespace("default") + .task_queue("queue") + .versioning_strategy(default_versioning_strategy()) + .task_types(WorkerTaskTypes { + enable_workflows: workflows, + enable_remote_activities: activities, + enable_local_activities: false, + enable_nexus: false, + }) + .max_cached_workflows(cache) + .max_outstanding_activities(slots) + .session_worker_options(SessionWorkerOptions::default()) + .build(); + assert_eq!(config.is_ok(), valid); + } + } + fn default_versioning_strategy() -> WorkerVersioningStrategy { WorkerVersioningStrategy::None { build_id: String::new(), diff --git a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs index cb3da1bac..ac534f25e 100644 --- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs +++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs @@ -61,7 +61,9 @@ use temporalio_common::{ workflow_activation::{ self, NotifyHasPatch, UpdateRandomSeed, WorkflowActivation, workflow_activation_job, }, - workflow_commands::{ContinueAsNewWorkflowExecution, update_response}, + workflow_commands::{ + ContinueAsNewWorkflowExecution, ScheduleActivity, update_response, + }, }, temporal::api::{ command::v1::{Command as ProtoCommand, command::Attributes as ProtoCmdAttrs}, @@ -70,6 +72,7 @@ use temporalio_common::{ history::v1::{HistoryEvent, history_event}, protocol::v1::{Message as ProtocolMessage, message::SequencingId}, sdk::v1::WorkflowTaskCompletedMetadata, + session::v1::activity_session_options, }, }, worker::WorkerDeploymentVersion, @@ -1503,7 +1506,8 @@ impl WorkflowMachines { annotations ); } - WFCommandVariant::AddActivity(attrs) => { + WFCommandVariant::AddActivity(mut attrs) => { + prepare_session_activity(&mut attrs, &self.worker_config, self.replaying)?; let seq = attrs.seq; let use_compat = self.determine_use_compatible_flag( attrs.versioning_intent(), @@ -1986,6 +1990,114 @@ fn patch_marker_handling( } } +fn prepare_session_activity( + attrs: &mut ScheduleActivity, + worker: &WorkerConfig, + replaying: bool, +) -> Result<()> { + let Some(options) = &attrs.session_options else { + return Ok(()); + }; + if !matches!( + options.session, + Some(activity_session_options::Session::UseWorkflowSession(true)) + ) { + return Err(fatal!( + "Activity '{}' (seq {}) has invalid session_options: use_workflow_session must be true", + attrs.activity_id, + attrs.seq + )); + } + if attrs.task_queue.is_empty() { + attrs.task_queue.clone_from(&worker.task_queue); + } + // Replay must not depend on this worker's activity execution capabilities. + if replaying { + return Ok(()); + } + if worker.session_worker_options.is_none() { + return Err(fatal!( + "Activity '{}' (seq {}) requests a workflow session, but session_worker_options is not enabled", + attrs.activity_id, + attrs.seq + )); + } + if attrs.task_queue != worker.task_queue { + return Err(fatal!( + "Activity '{}' (seq {}) requests a workflow session on queue '{}', but must use '{}'", + attrs.activity_id, + attrs.seq, + attrs.task_queue, + worker.task_queue + )); + } + Ok(()) +} + +#[cfg(test)] +mod session_tests { + use super::*; + use temporalio_common::protos::temporal::api::session::v1::ActivitySessionOptions; + + #[rstest::rstest] + #[case(true, Some(true), "", false, true)] + #[case(true, Some(true), "other", false, false)] + #[case(false, Some(true), "", false, false)] + #[case(false, Some(false), "", false, false)] + #[case(false, None, "", false, false)] + #[case(false, Some(false), "", true, false)] + #[case(false, None, "", true, false)] + #[case(false, Some(true), "other", true, true)] + fn session_activity_validation( + #[case] enabled: bool, + #[case] use_session: Option, + #[case] queue: &str, + #[case] replaying: bool, + #[case] valid: bool, + ) { + let config = WorkerConfig::builder() + .namespace("default") + .task_queue("queue") + .versioning_strategy(crate::WorkerVersioningStrategy::None { + build_id: String::new(), + }) + .task_types(temporalio_common::worker::WorkerTaskTypes::all()) + .max_cached_workflows(2_usize) + .maybe_session_worker_options(enabled.then(crate::SessionWorkerOptions::default)) + .build() + .unwrap(); + let expected_queue = config.task_queue.clone(); + let mut attrs = ScheduleActivity { + seq: 1, + activity_id: "activity".to_owned(), + task_queue: queue.to_owned(), + session_options: Some(ActivitySessionOptions { + session: use_session.map(activity_session_options::Session::UseWorkflowSession), + }), + ..Default::default() + }; + let result = prepare_session_activity(&mut attrs, &config, replaying); + assert_eq!(result.is_ok(), valid, "{result:?}"); + if valid { + assert_eq!( + attrs.task_queue, + if use_session == Some(true) && queue.is_empty() { + expected_queue.as_str() + } else { + queue + } + ); + } else { + assert!( + result + .unwrap_err() + .to_string() + .contains("Activity 'activity' (seq 1)") + ); + } + } +} + #[derive(derive_more::From)] enum CommandIdKind { /// A normal command, requested by lang diff --git a/crates/sdk-core/src/worker/workflow/wft_poller.rs b/crates/sdk-core/src/worker/workflow/wft_poller.rs index 8737e8159..02097fe6b 100644 --- a/crates/sdk-core/src/worker/workflow/wft_poller.rs +++ b/crates/sdk-core/src/worker/workflow/wft_poller.rs @@ -18,7 +18,7 @@ use tokio_util::sync::CancellationToken; #[allow(clippy::too_many_arguments)] pub(crate) fn make_wft_poller( config: &WorkerConfig, - sticky_queue_name: &Option, + sticky_queue_name: Option<&str>, client: &Arc, metrics: &MetricsContext, shutdown_token: &CancellationToken, @@ -68,12 +68,12 @@ pub(crate) fn make_wft_poller( last_successful_poll_time, capabilities.clone(), ); - let sticky_queue_poller = sticky_queue_name.as_ref().map(|sqn| { + let sticky_queue_poller = sticky_queue_name.map(|sqn| { let sticky_metrics = metrics.with_new_attrs([workflow_sticky_poller()]); LongPollBuffer::new_workflow_task( client.clone(), config.task_queue.clone(), - Some(sqn.clone()), + Some(sqn.to_owned()), sticky_poller_behavior(effective_behavior, config, true), wft_slots.clone().into_sticky(), shutdown_token.child_token(), @@ -314,7 +314,7 @@ mod tests { let client: Arc = Arc::new(client); let stream = make_wft_poller( &cfg, - &Some("sticky-q".to_string()), + Some("sticky-q"), &client, &MetricsContext::no_op(), &CancellationToken::new(), From 44362d9636fb319a96580d0d59cee531d2f22c50 Mon Sep 17 00:00:00 2001 From: Thomas Hardy Date: Tue, 27 Oct 2026 13:13:55 -0400 Subject: [PATCH 3/5] Update session API to temporalio/api d183f54c Replace crates/protos/protos/api_upstream with an exact snapshot of temporalio/api sticky-activity-session-options at d183f54c642fe320218bc3bf8592b2533540f050, which merges the revised session API shape (temporalio/api#893) and current upstream main. --- .../temporal/api/enums/v1/failed_cause.proto | 2 + .../temporal/api/session/v1/message.proto | 48 ++++++++++++------- 2 files changed, 32 insertions(+), 18 deletions(-) diff --git a/crates/protos/protos/api_upstream/temporal/api/enums/v1/failed_cause.proto b/crates/protos/protos/api_upstream/temporal/api/enums/v1/failed_cause.proto index f902aa9a1..cb14de64d 100644 --- a/crates/protos/protos/api_upstream/temporal/api/enums/v1/failed_cause.proto +++ b/crates/protos/protos/api_upstream/temporal/api/enums/v1/failed_cause.proto @@ -155,6 +155,8 @@ enum ResourceExhaustedCause { RESOURCE_EXHAUSTED_CAUSE_WORKER_DEPLOYMENT_LIMITS = 10; // Namespace exceeds bandwidth limit. RESOURCE_EXHAUSTED_CAUSE_BANDWIDTH_LIMIT = 11; + // Caller exceeds Nexus requests per second limit. + RESOURCE_EXHAUSTED_CAUSE_CALLER_RPS_LIMIT = 12; } enum ResourceExhaustedScope { diff --git a/crates/protos/protos/api_upstream/temporal/api/session/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/session/v1/message.proto index 4a7cdb2e5..7291ab7fa 100644 --- a/crates/protos/protos/api_upstream/temporal/api/session/v1/message.proto +++ b/crates/protos/protos/api_upstream/temporal/api/session/v1/message.proto @@ -9,36 +9,48 @@ option java_outer_classname = "MessageProto"; option ruby_package = "Temporalio::Api::Session::V1"; option csharp_namespace = "Temporalio.Api.Session.V1"; +import "google/protobuf/empty.proto"; + // EXPERIMENTAL. Subject to redesign or removal. -// Configures a worker's polling for sessions that keep workflow and activity -// tasks on a stable worker destination. Activities opt into session routing -// through ActivitySessionOptions on their scheduling commands. +// Sent on workflow and activity polls. Presence means this worker can serve +// sessions. Currently, the only supported Session type is Workflow Session +// (aka Workflow-as-Session). A Workflow Session can receive both workflow +// tasks and activity tasks of an execution. message WorkerSessionOptions { - oneof isolation { + // Admission control, if unset the worker accepts an arbitrary number of + // sessions (bounded by session slots in the future) + oneof admission { + // If specified, the worker only polls for work belonging to its session. SingleSessionOptions single_session = 1; } } // EXPERIMENTAL. Subject to redesign or removal. -// Dedicates a sticky destination to the first workflow matched to it. -// Workflow sticky polls may be forwarded to the normal queue until the first -// workflow task is dispatched. The server persists that workflow ID and does -// not forward subsequent polls, including after workflow completion. -// Activity sticky polls are never forwarded to the normal queue. +// Serve exactly one session for the worker's lifetime, including across +// process restarts with the same stable_identity. The server binds the worker +// to the first session that needs a worker, and dispatches no other work here, +// even after that session completes. message SingleSessionOptions { - // Stable sticky task queue name, shared by workflow and activity polls. - // Reuse this value across worker restarts within the same session, but not - // across concurrently running workers. This is distinct from worker identity. - // The SDK must also set the polled sticky task queue's name to this value. + // Identity the binding is stored under. Reuse it across restarts of the + // same session, not across concurrently running workers. If multiple + // workers poll with the same stable_identity concurrently, the server + // treats them as pollers of a single worker and process affinity is lost. + // + // Note: this effectively acts as a configurable sticky queue. + // Note: we are currently unsure how this will evolve. May be deprecated as + // the session model develops. string stable_identity = 1; } +// EXPERIMENTAL. Subject to redesign or removal. +// Resolved activity placement. Unset = normal task-queue routing. message ActivitySessionOptions { - oneof session { - // Execute this activity as part of its workflow's session. - // Currently routes to the workflow's sticky task queue when true. - bool use_workflow_session = 1; + oneof placement { + // Run in the workflow's current session (on its sticky queue). + // Currently requires the activity's task queue to be the workflow's, + // and the workflow's worker to serve sessions. + google.protobuf.Empty workflow_session = 1; - // later session_id, etc + // Future: references to other types of sessions. } } From 57bd0635cda81bb83b9f635103930df5f4bd16a6 Mon Sep 17 00:00:00 2001 From: Thomas Hardy Date: Tue, 27 Oct 2026 13:13:55 -0400 Subject: [PATCH 4/5] Adopt revised session API semantics - Activity session placement uses the WorkflowSession variant; options with no placement keep normal task-queue routing instead of failing. - Session workers send WorkerSessionOptions on sticky polls, advertising that they can serve sessions. --- crates/sdk-core/CHANGELOG.md | 4 +- crates/sdk-core/src/pollers/poll_buffer.rs | 18 ++++++++- crates/sdk-core/src/replay/mod.rs | 4 +- .../src/worker/activities/act_poller.rs | 8 ++++ crates/sdk-core/src/worker/client.rs | 12 ++++-- crates/sdk-core/src/worker/mod.rs | 9 +++++ .../workflow/machines/workflow_machines.rs | 39 ++++++++----------- .../src/worker/workflow/wft_poller.rs | 6 ++- 8 files changed, 68 insertions(+), 32 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index db35d64c1..914caf2c3 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -43,8 +43,8 @@ relevant information. * Experimental session workers can poll workflow-affine activity queues; individual activities opt into their workflow's session, while normal activity routing remains the default. This increment supports fixed-size activity slots with capacity of at least two; resource-based and custom - activity suppliers are not supported yet. Activity session options must select workflow-session - routing; empty options or an explicit false selection are rejected. + activity suppliers are not supported yet. Activity session options without a placement use normal + task-queue routing. Session workers advertise session support on their sticky polls. ### Security * Ephemeral server downloads now use `zip` 9.0 for ZIP extraction, replacing the 8.x dependency diff --git a/crates/sdk-core/src/pollers/poll_buffer.rs b/crates/sdk-core/src/pollers/poll_buffer.rs index b4ab589eb..f196f9006 100644 --- a/crates/sdk-core/src/pollers/poll_buffer.rs +++ b/crates/sdk-core/src/pollers/poll_buffer.rs @@ -26,6 +26,7 @@ use temporalio_client::{ ERROR_RETURNED_DUE_TO_SHORT_CIRCUIT, jittered, request_extensions::NoRetryOnMatching, }; use temporalio_common::protos::temporal::api::{ + session::v1::WorkerSessionOptions, taskqueue::v1::PollerScalingDecision, workflowservice::v1::{ PollActivityTaskQueueResponse, PollNexusTaskQueueResponse, PollWorkflowTaskQueueResponse, @@ -102,8 +103,10 @@ pub(crate) struct LongPollBuffer { did_start: AtomicBool, } +#[derive(Default)] pub(crate) struct WorkflowTaskOptions { pub(crate) wft_poller_shared: Option>, + pub(crate) session_options: Option, } #[derive(Default)] @@ -113,6 +116,7 @@ pub(crate) struct ActivityTaskOptions { /// Server ratelimit pub(crate) max_tps: Option, pub(crate) poller_shared: Option>, + pub(crate) session_options: Option, } impl LongPollBuffer { @@ -174,6 +178,7 @@ impl LongPollBuffer { let client = client.clone(); let task_queue = task_queue.clone(); let sticky_queue_name = sticky_queue.clone(); + let session_options = options.session_options.clone(); async move { client .poll_workflow_task( @@ -182,7 +187,10 @@ impl LongPollBuffer { no_retry, timeout_override, }, - PollWorkflowOptions { sticky_queue_name }, + PollWorkflowOptions { + sticky_queue_name, + session_options, + }, ) .await } @@ -255,6 +263,7 @@ impl LongPollBuffer { let client = client.clone(); let task_queue = task_queue.clone(); let sticky_queue_name = sticky_queue.clone(); + let session_options = options.session_options.clone(); async move { client .poll_activity_task( @@ -266,6 +275,7 @@ impl LongPollBuffer { PollActivityOptions { max_tasks_per_sec: options.max_tps, sticky_queue_name, + session_options, }, ) .await @@ -915,6 +925,7 @@ mod tests { None::, WorkflowTaskOptions { wft_poller_shared: Some(Arc::new(StickyPollerShared::new(Some(10)))), + ..Default::default() }, Arc::new(AtomicCell::new(None)), Arc::new(NamespaceCapabilities::default()), @@ -972,6 +983,7 @@ mod tests { None::, WorkflowTaskOptions { wft_poller_shared: Some(Arc::new(StickyPollerShared::new(Some(1)))), + ..Default::default() }, Arc::new(AtomicCell::new(None)), Arc::new(NamespaceCapabilities::default()), @@ -1049,6 +1061,7 @@ mod tests { None::, WorkflowTaskOptions { wft_poller_shared: Some(Arc::new(StickyPollerShared::new(Some(10)))), + ..Default::default() }, Arc::new(AtomicCell::new(None)), Arc::new(NamespaceCapabilities::default()), @@ -1158,6 +1171,7 @@ mod tests { None::, WorkflowTaskOptions { wft_poller_shared: Some(Arc::new(StickyPollerShared::new(Some(INITIAL_POLLERS)))), + ..Default::default() }, Arc::new(AtomicCell::new(None)), Arc::new(NamespaceCapabilities::default()), @@ -1239,6 +1253,7 @@ mod tests { }), WorkflowTaskOptions { wft_poller_shared: None, + ..Default::default() }, Arc::new(AtomicCell::new(None)), Arc::new(NamespaceCapabilities::resolved(Capabilities { @@ -1317,6 +1332,7 @@ mod tests { None::, WorkflowTaskOptions { wft_poller_shared: None, + ..Default::default() }, Arc::new(AtomicCell::new(None)), Arc::new(NamespaceCapabilities::resolved(Capabilities { diff --git a/crates/sdk-core/src/replay/mod.rs b/crates/sdk-core/src/replay/mod.rs index 251d767d8..2724c77d2 100644 --- a/crates/sdk-core/src/replay/mod.rs +++ b/crates/sdk-core/src/replay/mod.rs @@ -187,8 +187,8 @@ mod tests { activity_id: "activity".to_owned(), activity_type: DEFAULT_ACTIVITY_TYPE.to_owned(), session_options: Some(ActivitySessionOptions { - session: Some(activity_session_options::Session::UseWorkflowSession( - true, + placement: Some(activity_session_options::Placement::WorkflowSession( + (), )), }), ..Default::default() diff --git a/crates/sdk-core/src/worker/activities/act_poller.rs b/crates/sdk-core/src/worker/activities/act_poller.rs index 9525e6170..f37878847 100644 --- a/crates/sdk-core/src/worker/activities/act_poller.rs +++ b/crates/sdk-core/src/worker/activities/act_poller.rs @@ -50,6 +50,7 @@ pub(crate) fn make_activity_poller( rate_limiter: rate_limiter.clone(), max_tps: config.max_task_queue_activities_per_second, poller_shared: poller_shared.clone(), + ..Default::default() }, last_successful_poll_time.clone(), capabilities.clone(), @@ -68,6 +69,7 @@ pub(crate) fn make_activity_poller( rate_limiter, max_tps: config.max_task_queue_activities_per_second, poller_shared, + session_options: config.poll_session_options(), }, last_successful_poll_time, capabilities, @@ -101,6 +103,7 @@ mod tests { poll.task_queue, options.sticky_queue_name, tokio::time::Instant::now(), + options.session_options, )) .unwrap(); pending().boxed() @@ -148,6 +151,10 @@ mod tests { assert_ne!(first.1, second.1); assert!(second.2.duration_since(first.2) >= Duration::from_millis(500)); assert!(first.1.as_deref() == Some("sticky") || second.1.as_deref() == Some("sticky")); + // Only sticky polls advertise session support. + for (_, sticky, _, session_options) in [&first, &second] { + assert_eq!(sticky.is_some(), session_options.is_some()); + } assert_eq!(slots.available_permits(), Some(0)); poller.notify_shutdown(); assert!(polling.await.unwrap().is_none()); @@ -168,6 +175,7 @@ mod tests { .times(5) .returning(move |_, options| { assert!(options.sticky_queue_name.is_none()); + assert!(options.session_options.is_none()); tx.send(()).unwrap(); pending().boxed() }); diff --git a/crates/sdk-core/src/worker/client.rs b/crates/sdk-core/src/worker/client.rs index b9d50d875..8469cabd5 100644 --- a/crates/sdk-core/src/worker/client.rs +++ b/crates/sdk-core/src/worker/client.rs @@ -447,6 +447,9 @@ pub struct PollOptions { pub struct PollWorkflowOptions { /// Optional sticky queue name for session‐based workflow polling pub sticky_queue_name: Option, + /// Experimental server session options, expected to change with the session API. + pub session_options: + Option, } /// Additional options specific to activity task polling #[derive(Debug, Clone)] @@ -455,6 +458,9 @@ pub struct PollActivityOptions { pub max_tasks_per_sec: Option, /// Poll this sticky destination instead of the normal activity queue. pub sticky_queue_name: Option, + /// Experimental server session options, expected to change with the session API. + pub session_options: + Option, } /// Additional options specific to Nexus task polling #[derive(Debug, Clone, Default)] @@ -498,8 +504,7 @@ impl WorkerClient for WorkerClientBag { worker_instance_key: self.worker_instance_key.to_string(), worker_control_task_queue: self.worker_control_task_queue(), poller_group_id: Default::default(), - // Activity routing does not opt this worker into the deferred isolation policy. - session_options: None, + session_options: wf_options.session_options, } .into_request(); request.extensions_mut().insert(IsWorkerTaskLongPoll); @@ -539,8 +544,7 @@ impl WorkerClient for WorkerClientBag { worker_instance_key: self.worker_instance_key.to_string(), worker_control_task_queue: self.worker_control_task_queue(), poller_group_id: Default::default(), - // Activity routing does not opt this worker into the deferred isolation policy. - session_options: None, + session_options: act_options.session_options, } .into_request(); request.extensions_mut().insert(IsWorkerTaskLongPoll); diff --git a/crates/sdk-core/src/worker/mod.rs b/crates/sdk-core/src/worker/mod.rs index 43bd2dac5..eeac790e2 100644 --- a/crates/sdk-core/src/worker/mod.rs +++ b/crates/sdk-core/src/worker/mod.rs @@ -102,6 +102,7 @@ use temporalio_common::{ deployment, enums::v1::{TaskQueueKind, TaskQueueType, WorkerStatus}, namespace::v1::namespace_info::Capabilities as ApiNamespaceCapabilities, + session::v1::WorkerSessionOptions as ApiWorkerSessionOptions, taskqueue::v1::{StickyExecutionAttributes, TaskQueue}, worker::v1::{ EnvironmentInfo, WorkerHeartbeat, WorkerHostInfo, WorkerPollerInfo, WorkerSlotsInfo, @@ -314,6 +315,14 @@ pub struct WorkerConfig { } impl WorkerConfig { + /// Session options for sticky polls. Their presence tells the server this worker can serve + /// sessions. + pub(crate) fn poll_session_options(&self) -> Option { + self.session_worker_options + .as_ref() + .map(|_| ApiWorkerSessionOptions::default()) + } + /// Returns true if the configuration specifies we should fail a workflow on a certain error /// type rather than failing the workflow task. pub fn should_fail_workflow( diff --git a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs index ac534f25e..03a474399 100644 --- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs +++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs @@ -1995,19 +1995,14 @@ fn prepare_session_activity( worker: &WorkerConfig, replaying: bool, ) -> Result<()> { - let Some(options) = &attrs.session_options else { + // Unset placement means normal task-queue routing. + let Some(activity_session_options::Placement::WorkflowSession(())) = attrs + .session_options + .as_ref() + .and_then(|options| options.placement.as_ref()) + else { return Ok(()); }; - if !matches!( - options.session, - Some(activity_session_options::Session::UseWorkflowSession(true)) - ) { - return Err(fatal!( - "Activity '{}' (seq {}) has invalid session_options: use_workflow_session must be true", - attrs.activity_id, - attrs.seq - )); - } if attrs.task_queue.is_empty() { attrs.task_queue.clone_from(&worker.task_queue); } @@ -2040,17 +2035,16 @@ mod session_tests { use temporalio_common::protos::temporal::api::session::v1::ActivitySessionOptions; #[rstest::rstest] - #[case(true, Some(true), "", false, true)] - #[case(true, Some(true), "other", false, false)] - #[case(false, Some(true), "", false, false)] - #[case(false, Some(false), "", false, false)] - #[case(false, None, "", false, false)] - #[case(false, Some(false), "", true, false)] - #[case(false, None, "", true, false)] - #[case(false, Some(true), "other", true, true)] + #[case(true, true, "", false, true)] + #[case(true, true, "other", false, false)] + #[case(true, false, "other", false, true)] + #[case(false, true, "", false, false)] + #[case(false, false, "", false, true)] + #[case(false, false, "", true, true)] + #[case(false, true, "other", true, true)] fn session_activity_validation( #[case] enabled: bool, - #[case] use_session: Option, + #[case] workflow_session: bool, #[case] queue: &str, #[case] replaying: bool, #[case] valid: bool, @@ -2072,7 +2066,8 @@ mod session_tests { activity_id: "activity".to_owned(), task_queue: queue.to_owned(), session_options: Some(ActivitySessionOptions { - session: use_session.map(activity_session_options::Session::UseWorkflowSession), + placement: workflow_session + .then_some(activity_session_options::Placement::WorkflowSession(())), }), ..Default::default() }; @@ -2081,7 +2076,7 @@ mod session_tests { if valid { assert_eq!( attrs.task_queue, - if use_session == Some(true) && queue.is_empty() { + if workflow_session && queue.is_empty() { expected_queue.as_str() } else { queue diff --git a/crates/sdk-core/src/worker/workflow/wft_poller.rs b/crates/sdk-core/src/worker/workflow/wft_poller.rs index 02097fe6b..50c2a3ec5 100644 --- a/crates/sdk-core/src/worker/workflow/wft_poller.rs +++ b/crates/sdk-core/src/worker/workflow/wft_poller.rs @@ -64,6 +64,7 @@ pub(crate) fn make_wft_poller( }), WorkflowTaskOptions { wft_poller_shared: wft_poller_shared.clone(), + ..Default::default() }, last_successful_poll_time, capabilities.clone(), @@ -80,7 +81,10 @@ pub(crate) fn make_wft_poller( Some(move |np| { sticky_metrics.record_num_pollers(np); }), - WorkflowTaskOptions { wft_poller_shared }, + WorkflowTaskOptions { + wft_poller_shared, + session_options: config.poll_session_options(), + }, sticky_last_successful_poll_time, capabilities, ) From a7fe573770f905eddaa8352000245e7626d5b553 Mon Sep 17 00:00:00 2001 From: Thomas Hardy Date: Wed, 28 Oct 2026 16:05:57 -0400 Subject: [PATCH 5/5] Require session workers to replay session activities Session workflows should only run, including replay, on workers that serve sessions; replaying elsewhere silently drops the activity affinity the workflow expects. Validate workflow-session activities during replay too, and keep the replay worker's session options instead of clearing them, so a replay worker without session options fails the workflow task. --- crates/sdk-core/CHANGELOG.md | 5 ++- crates/sdk-core/src/replay/mod.rs | 44 +++++++++++++++---- .../workflow/machines/workflow_machines.rs | 30 +++++-------- 3 files changed, 51 insertions(+), 28 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 914caf2c3..3ecb6ce3e 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -44,7 +44,10 @@ relevant information. into their workflow's session, while normal activity routing remains the default. This increment supports fixed-size activity slots with capacity of at least two; resource-based and custom activity suppliers are not supported yet. Activity session options without a placement use normal - task-queue routing. Session workers advertise session support on their sticky polls. + task-queue routing. Session workers advertise session support on their sticky polls. A workflow + task that schedules a session activity fails on a worker without session support, including + during replay, so replay workers must be configured with session options to replay such + workflows. ### Security * Ephemeral server downloads now use `zip` 9.0 for ZIP extraction, replacing the 8.x dependency diff --git a/crates/sdk-core/src/replay/mod.rs b/crates/sdk-core/src/replay/mod.rs index 2724c77d2..5d893fc7f 100644 --- a/crates/sdk-core/src/replay/mod.rs +++ b/crates/sdk-core/src/replay/mod.rs @@ -75,8 +75,8 @@ where } pub(crate) fn into_core_worker(mut self) -> Result { - // Offline replay consumes supplied history and does not execute activities. - self.config.session_worker_options = None; + // Offline replay consumes supplied history and does not execute activities. Session + // options are kept: workflows with session activities only replay on session workers. self.config.max_cached_workflows = 1; self.config.workflow_task_poller_behavior = Some(PollerBehavior::SimpleMaximum(1)); self.config.task_types = WorkerTaskTypes::workflow_only(); @@ -163,17 +163,30 @@ mod tests { temporal::api::session::v1::{ActivitySessionOptions, activity_session_options}, }; + #[rstest::rstest] + #[case::session_worker(true)] + #[case::non_session_worker(false)] #[tokio::test] - async fn replay_session_activity_without_session_worker() { + async fn replay_session_activity_requires_session_worker(#[case] session_worker: bool) { let mut history = canned_histories::single_activity("activity"); history.add_workflow_task_completed(); history.add_workflow_execution_completed(); let mut client = mock_manual_worker_client(); - client.expect_fail_workflow_task().times(0); - let mut input = ReplayWorkerInput::new( - test_worker_cfg().build().unwrap(), - stream::iter([history.into()]), - ); + if session_worker { + client.expect_fail_workflow_task().times(0); + } + let config = WorkerConfig::builder() + .namespace("default") + .task_queue("queue") + .versioning_strategy(crate::WorkerVersioningStrategy::None { + build_id: String::new(), + }) + .task_types(WorkerTaskTypes::all()) + .max_cached_workflows(2_usize) + .maybe_session_worker_options(session_worker.then(crate::SessionWorkerOptions::default)) + .build() + .unwrap(); + let mut input = ReplayWorkerInput::new(config, stream::iter([history.into()])); input.client_override = Some(client); let worker = input.into_core_worker().unwrap(); let start = worker.poll_workflow_activation().await.unwrap(); @@ -198,6 +211,21 @@ mod tests { )) .await .unwrap(); + if !session_worker { + let evicted = worker.poll_workflow_activation().await.unwrap(); + assert_eq!(evicted.eviction_reason(), Some(EvictionReason::Fatal)); + assert_matches!( + &evicted.jobs[0].variant, + Some(workflow_activation_job::Variant::RemoveFromCache(r)) + if r.message.contains("session_worker_options is not enabled") + ); + worker + .complete_workflow_activation(WorkflowActivationCompletion::empty(evicted.run_id)) + .await + .unwrap(); + worker.drain_pollers_and_shutdown().await; + return; + } let resolved = worker.poll_workflow_activation().await.unwrap(); assert!(matches!( resolved.jobs[0].variant, diff --git a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs index 03a474399..693b95ad8 100644 --- a/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs +++ b/crates/sdk-core/src/worker/workflow/machines/workflow_machines.rs @@ -1507,7 +1507,7 @@ impl WorkflowMachines { ); } WFCommandVariant::AddActivity(mut attrs) => { - prepare_session_activity(&mut attrs, &self.worker_config, self.replaying)?; + prepare_session_activity(&mut attrs, &self.worker_config)?; let seq = attrs.seq; let use_compat = self.determine_use_compatible_flag( attrs.versioning_intent(), @@ -1990,11 +1990,7 @@ fn patch_marker_handling( } } -fn prepare_session_activity( - attrs: &mut ScheduleActivity, - worker: &WorkerConfig, - replaying: bool, -) -> Result<()> { +fn prepare_session_activity(attrs: &mut ScheduleActivity, worker: &WorkerConfig) -> Result<()> { // Unset placement means normal task-queue routing. let Some(activity_session_options::Placement::WorkflowSession(())) = attrs .session_options @@ -2006,10 +2002,8 @@ fn prepare_session_activity( if attrs.task_queue.is_empty() { attrs.task_queue.clone_from(&worker.task_queue); } - // Replay must not depend on this worker's activity execution capabilities. - if replaying { - return Ok(()); - } + // Checked during replay too: session workflows only run on session workers, since + // running elsewhere would drop the affinity their activities expect. if worker.session_worker_options.is_none() { return Err(fatal!( "Activity '{}' (seq {}) requests a workflow session, but session_worker_options is not enabled", @@ -2035,18 +2029,16 @@ mod session_tests { use temporalio_common::protos::temporal::api::session::v1::ActivitySessionOptions; #[rstest::rstest] - #[case(true, true, "", false, true)] - #[case(true, true, "other", false, false)] - #[case(true, false, "other", false, true)] - #[case(false, true, "", false, false)] - #[case(false, false, "", false, true)] - #[case(false, false, "", true, true)] - #[case(false, true, "other", true, true)] + #[case(true, true, "", true)] + #[case(true, true, "other", false)] + #[case(true, false, "other", true)] + #[case(false, true, "", false)] + #[case(false, true, "other", false)] + #[case(false, false, "", true)] fn session_activity_validation( #[case] enabled: bool, #[case] workflow_session: bool, #[case] queue: &str, - #[case] replaying: bool, #[case] valid: bool, ) { let config = WorkerConfig::builder() @@ -2071,7 +2063,7 @@ mod session_tests { }), ..Default::default() }; - let result = prepare_session_activity(&mut attrs, &config, replaying); + let result = prepare_session_activity(&mut attrs, &config); assert_eq!(result.is_ok(), valid, "{result:?}"); if valid { assert_eq!(