From b59281514300e86093ffe3f5e1a7495488f09335 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 15 Oct 2026 04:52:44 -0700 Subject: [PATCH] Vendored the Workflow Activity owner kind. An Activity that a Workflow run scheduled owns streams in Jetty. The API and the stream protos name it by kind, with the Workflow id, the scheduling run and the Activity id. --- .../temporal/api/enums/v1/stream.proto | 7 +++++++ .../temporal/api/stream/v1/message.proto | 11 ++++++---- .../temporal/sdk/core/streams/streams.proto | 20 +++++++++++++------ crates/sdk-core/CHANGELOG.md | 5 +++++ 4 files changed, 33 insertions(+), 10 deletions(-) diff --git a/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto b/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto index 6c2dfa80c5..4f53e16d8b 100644 --- a/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto +++ b/crates/protos/protos/api_upstream/temporal/api/enums/v1/stream.proto @@ -22,4 +22,11 @@ enum StreamOwnerKind { // `streamnotifier.ownerCheckInterval`. A completed run gives no result. A failed, canceled, // terminated or timed-out run gives the run's own failure. STREAM_OWNER_KIND_ACTIVITY = 2; + // An Activity that a Workflow scheduled owns the stream. `workflow_id` names the Workflow, + // `run_id` the run that scheduled the Activity, and `activity_id` the Activity. The stream + // closes when the Workflow sees the Activity end. If the Activity is no longer pending in the + // run and no close came, the stream notifier closes the stream with a failure, after + // `streamnotifier.ownerStartGrace` from its start and `streamnotifier.ownerCloseGrace` from + // the first check that found the Activity gone. + STREAM_OWNER_KIND_WORKFLOW_ACTIVITY = 3; } diff --git a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto index 755fa7c7bb..41e4158ca2 100644 --- a/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto +++ b/crates/protos/protos/api_upstream/temporal/api/stream/v1/message.proto @@ -18,17 +18,20 @@ import "temporal/api/failure/v1/message.proto"; // Names one stream within a namespace: its owner and its topic. message StreamReference { temporal.api.enums.v1.StreamOwnerKind owner_kind = 1; - // The owning Workflow, for `STREAM_OWNER_KIND_WORKFLOW`. Empty for other kinds. + // The owning Workflow, for `STREAM_OWNER_KIND_WORKFLOW`. The Workflow that scheduled the + // owning Activity, for `STREAM_OWNER_KIND_WORKFLOW_ACTIVITY`. Empty for other kinds. string workflow_id = 2; // For `STREAM_OWNER_KIND_WORKFLOW`, the id of the first run in the owning Workflow's run // chain. The stream notifier is keyed by it, so one notifier serves the stream across the // chain's runs, and a stream that reuses the Workflow id in a new chain gets a notifier of its - // own. For `STREAM_OWNER_KIND_ACTIVITY`, the id of the owning Activity run. An empty run id is - // refused. + // own. For `STREAM_OWNER_KIND_ACTIVITY`, the id of the owning Activity run. For + // `STREAM_OWNER_KIND_WORKFLOW_ACTIVITY`, the id of the run that scheduled the Activity, not + // the chain's first run. An empty run id is refused. string run_id = 3; // The stream's topic within its owner. string topic = 4; - // The owning standalone Activity, for `STREAM_OWNER_KIND_ACTIVITY`. Empty for other kinds. + // The owning standalone Activity, for `STREAM_OWNER_KIND_ACTIVITY`. The owning Activity's id + // in the scheduling run, for `STREAM_OWNER_KIND_WORKFLOW_ACTIVITY`. Empty for other kinds. string activity_id = 5; } diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto index 78f1d94b26..99ab860cc7 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -123,29 +123,36 @@ enum StreamOwnerKind { STREAM_OWNER_KIND_WORKFLOW = 1; // A standalone Activity run. It is a chain of one run, so its streams end with the run. STREAM_OWNER_KIND_ACTIVITY = 2; + // An Activity that a Workflow run scheduled. It is a chain of one, keyed by the scheduling run + // and the Activity id, so its streams end with the Activity. + STREAM_OWNER_KIND_WORKFLOW_ACTIVITY = 3; } // One stream, a topic of an owner in a namespace. message StreamAddress { string namespace = 1; StreamOwnerKind owner_kind = 2; - // The owning Workflow. Empty for an Activity owner. + // The owning Workflow, or the Workflow that scheduled a Workflow Activity. Empty for a + // standalone Activity owner. string workflow_id = 3; // Pins the stream to the run chain that holds this run. Empty follows the chain that holds the // Workflow id when the call starts. For an Activity owner, the Activity run, and empty names - // the newest run of the Activity id when the call starts. + // the newest run of the Activity id when the call starts. For a Workflow Activity owner, the + // run that scheduled the Activity. It is required, since an Activity id names nothing across + // runs. string run_id = 4; string topic = 5; - // The owning standalone Activity. Empty for a Workflow owner. + // The owning Activity. Empty for a Workflow owner. string activity_id = 6; } // A Workflow's run chain. Its streams are keyed by its first run, so they follow // Continue-as-New, and a new chain on the same Workflow id gets streams of its own. A standalone -// Activity run is a chain of one run, with the run as `first_run_id`. +// Activity run is a chain of one run, with the run as `first_run_id`. A Workflow Activity is a +// chain of one, with the scheduling run as `first_run_id`. message ChainId { string namespace = 1; - // Empty for an Activity owner. + // Empty for a standalone Activity owner. string workflow_id = 2; string first_run_id = 3; // `UNSPECIFIED` is a Workflow owner, so a chain from before the Activity kind stays valid. @@ -435,7 +442,8 @@ message RunStreamDriverResponse {} message DeleteOwnerRequest { string namespace = 1; StreamOwnerKind owner_kind = 2; - // The Workflow whose streams go, for a Workflow owner. + // The Workflow whose streams go, for a Workflow owner. The streams of every Activity that the + // Workflow scheduled go too. string workflow_id = 3; // The standalone Activity whose streams go, across all its runs, for an Activity owner. string activity_id = 4; diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 6de48a00ec..d5fec1df53 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -172,6 +172,11 @@ relevant information. fits 53 bits and a JSON number holds it exactly. The Nexus contract's models and JavaScript refused the earlier 16-bit form. A start time from May 2109 on is refused. The vector file holds the new keys. +* Experimental: the stream protos name a Workflow Activity owner. `StreamOwnerKind` gains + `WORKFLOW_ACTIVITY`, an Activity that a Workflow run scheduled. Its `StreamAddress` and + `ChainId` set `workflow_id`, the scheduling run and `activity_id`, with no new field. The + vendored API's owner kind gains `WORKFLOW_ACTIVITY`. Core still refuses the kind as + `UNSUPPORTED`. ### Breaking Changes * Replay detects changes to the service or operation of scheduled Nexus operations, reporting