Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions crates/protos/build.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
20 changes: 0 additions & 20 deletions crates/protos/protos/api_upstream/buf.gen.yaml

This file was deleted.

Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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 {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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 {
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
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";

import "google/protobuf/empty.proto";

// EXPERIMENTAL. Subject to redesign or removal.
// 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 {
// 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.
// 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 {
// 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 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;

// Future: references to other types of sessions.
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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 {
Expand Down
6 changes: 6 additions & 0 deletions crates/protos/src/protos/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1915,6 +1915,7 @@ pub mod temporal {
request_eager_execution: !s.do_not_eagerly_execute,
use_workflow_build_id,
priority: s.priority,
session_options: s.session_options,
},
)
}
Expand Down Expand Up @@ -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");
Expand Down
8 changes: 8 additions & 0 deletions crates/sdk-core/CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,14 @@ 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 without a placement use normal
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
Expand Down
5 changes: 2 additions & 3 deletions crates/sdk-core/src/abstractions.rs
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,7 @@ pub(crate) struct MeteredPermitDealer<SK: SlotKind> {
max_permits: Option<usize>,
metrics_ctx: MetricsContext,
meter: Option<TemporalMeter>,
/// 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<PermitDealerContextData>,
}
Expand Down Expand Up @@ -198,7 +197,7 @@ where
}
}

impl MeteredPermitDealer<WorkflowSlotKind> {
impl<SK: SlotKind> MeteredPermitDealer<SK> {
pub(crate) fn into_sticky(mut self) -> Self {
self.is_sticky_poller = true;
self
Expand Down
10 changes: 5 additions & 5 deletions crates/sdk-core/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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::{
Expand Down
4 changes: 3 additions & 1 deletion crates/sdk-core/src/pollers/mod.rs
Original file line number Diff line number Diff line change
@@ -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};

Expand Down
Loading
Loading