From af5a195d1d1bab2e6026c25eb673f1538d977778 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 22:50:10 -0700 Subject: [PATCH] Vendored the System Nexus WIT for the stream notifier RPCs. AttachStreamCallback, DetachStreamCallback and NotifyStream carry the exposed tag again, with the WIT and the model refs that the SDK generators read. It matches api moe/bridge/v2/n3-notifier-wit. --- crates/protos/protos/api_upstream/Makefile | 11 ++- .../nexus/temporal-proto-models-nexusrpc.yaml | 45 +++++++++ .../api_upstream/nexus/workflow-service.wit | 97 +++++++++++++++++++ .../api/workflowservice/v1/service.proto | 18 ++-- 4 files changed, 161 insertions(+), 10 deletions(-) diff --git a/crates/protos/protos/api_upstream/Makefile b/crates/protos/protos/api_upstream/Makefile index f6fcf899d8..39f8b6b25a 100644 --- a/crates/protos/protos/api_upstream/Makefile +++ b/crates/protos/protos/api_upstream/Makefile @@ -35,7 +35,12 @@ PROTO_PATHS = paths=source_relative:$(PROTO_OUT) OAPI_OUT := openapi OAPI3_PATH := .components.schemas.Payload -NEX_GEN ?= nex-gen +# nexgen escapes WIT keywords in field names and scaffolds proto maps and bytes in the +# forms its generators accept. Pinned by revision so every machine writes the same WIT. +NEXGEN_GIT ?= https://github.com/moetemp/nexgen +NEXGEN_REV ?= 6f46948678f764c68bb76f103ab43713047187b2 +NEXGEN_ROOT := $(STAMPDIR)/nexgen-$(NEXGEN_REV) +NEX_GEN := $(NEXGEN_ROOT)/bin/nexgen $(PROTO_OUT): mkdir $(PROTO_OUT) @@ -156,8 +161,8 @@ system-nexus-wit-install: @cd cmd/protoc-gen-system-nexus-wit && go install . nex-gen-install: - printf $(COLOR) "Install nex-gen if missing..." - command -v $(NEX_GEN) >/dev/null || cargo install nex-gen --version ^0.1 + printf $(COLOR) "Install nexgen $(NEXGEN_REV) if missing..." + command -v $(NEX_GEN) >/dev/null || CARGO_NET_GIT_FETCH_WITH_CLI=true cargo install --locked --git $(NEXGEN_GIT) --rev $(NEXGEN_REV) --features advanced --root $(NEXGEN_ROOT) nexgen ##### Clean ##### clean: diff --git a/crates/protos/protos/api_upstream/nexus/temporal-proto-models-nexusrpc.yaml b/crates/protos/protos/api_upstream/nexus/temporal-proto-models-nexusrpc.yaml index ebc74b68b8..e611f78575 100644 --- a/crates/protos/protos/api_upstream/nexus/temporal-proto-models-nexusrpc.yaml +++ b/crates/protos/protos/api_upstream/nexus/temporal-proto-models-nexusrpc.yaml @@ -17,3 +17,48 @@ services: $pythonRef: temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse $rubyRef: Temporalio::Api::WorkflowService::V1::SignalWithStartWorkflowExecutionResponse $typescriptRef: '@temporalio/api/workflowservice/v1.SignalWithStartWorkflowExecutionResponse' + AttachStreamCallback: + input: + $dotnetRef: Temporalio.Api.WorkflowService.V1.AttachStreamCallbackRequest + $goRef: go.temporal.io/api/workflowservice/v1.AttachStreamCallbackRequest + $javaRef: io.temporal.api.workflowservice.v1.AttachStreamCallbackRequest + $pythonRef: temporalio.api.workflowservice.v1.AttachStreamCallbackRequest + $rubyRef: Temporalio::Api::WorkflowService::V1::AttachStreamCallbackRequest + $typescriptRef: '@temporalio/api/workflowservice/v1.AttachStreamCallbackRequest' + output: + $dotnetRef: Temporalio.Api.WorkflowService.V1.AttachStreamCallbackResponse + $goRef: go.temporal.io/api/workflowservice/v1.AttachStreamCallbackResponse + $javaRef: io.temporal.api.workflowservice.v1.AttachStreamCallbackResponse + $pythonRef: temporalio.api.workflowservice.v1.AttachStreamCallbackResponse + $rubyRef: Temporalio::Api::WorkflowService::V1::AttachStreamCallbackResponse + $typescriptRef: '@temporalio/api/workflowservice/v1.AttachStreamCallbackResponse' + DetachStreamCallback: + input: + $dotnetRef: Temporalio.Api.WorkflowService.V1.DetachStreamCallbackRequest + $goRef: go.temporal.io/api/workflowservice/v1.DetachStreamCallbackRequest + $javaRef: io.temporal.api.workflowservice.v1.DetachStreamCallbackRequest + $pythonRef: temporalio.api.workflowservice.v1.DetachStreamCallbackRequest + $rubyRef: Temporalio::Api::WorkflowService::V1::DetachStreamCallbackRequest + $typescriptRef: '@temporalio/api/workflowservice/v1.DetachStreamCallbackRequest' + output: + $dotnetRef: Temporalio.Api.WorkflowService.V1.DetachStreamCallbackResponse + $goRef: go.temporal.io/api/workflowservice/v1.DetachStreamCallbackResponse + $javaRef: io.temporal.api.workflowservice.v1.DetachStreamCallbackResponse + $pythonRef: temporalio.api.workflowservice.v1.DetachStreamCallbackResponse + $rubyRef: Temporalio::Api::WorkflowService::V1::DetachStreamCallbackResponse + $typescriptRef: '@temporalio/api/workflowservice/v1.DetachStreamCallbackResponse' + NotifyStream: + input: + $dotnetRef: Temporalio.Api.WorkflowService.V1.NotifyStreamRequest + $goRef: go.temporal.io/api/workflowservice/v1.NotifyStreamRequest + $javaRef: io.temporal.api.workflowservice.v1.NotifyStreamRequest + $pythonRef: temporalio.api.workflowservice.v1.NotifyStreamRequest + $rubyRef: Temporalio::Api::WorkflowService::V1::NotifyStreamRequest + $typescriptRef: '@temporalio/api/workflowservice/v1.NotifyStreamRequest' + output: + $dotnetRef: Temporalio.Api.WorkflowService.V1.NotifyStreamResponse + $goRef: go.temporal.io/api/workflowservice/v1.NotifyStreamResponse + $javaRef: io.temporal.api.workflowservice.v1.NotifyStreamResponse + $pythonRef: temporalio.api.workflowservice.v1.NotifyStreamResponse + $rubyRef: Temporalio::Api::WorkflowService::V1::NotifyStreamResponse + $typescriptRef: '@temporalio/api/workflowservice/v1.NotifyStreamResponse' diff --git a/crates/protos/protos/api_upstream/nexus/workflow-service.wit b/crates/protos/protos/api_upstream/nexus/workflow-service.wit index caf3842a3b..fa0aa9aae2 100644 --- a/crates/protos/protos/api_upstream/nexus/workflow-service.wit +++ b/crates/protos/protos/api_upstream/nexus/workflow-service.wit @@ -138,4 +138,101 @@ interface workflow-service { signal-with-start-workflow: func( request: signal-with-start-workflow-request, ) -> signal-with-start-workflow-response; + /// @nexus.proto "temporal.api.enums.v1.StreamOwnerKind" + enum stream-owner-kind { + stream-owner-kind-unspecified, + stream-owner-kind-workflow, + } + + /// @nexus.proto "temporal.api.stream.v1.StreamReference" + record stream-reference { + owner-kind: option, + workflow-id: string, + /// @nexus.doc "The id of the first run in the owning Workflow's run chain. It is required, since the server refuses an empty run id." + run-id: option, + topic: string, + } + + /// @nexus.proto "temporal.api.common.v1.Callback.Nexus" + record callback-nexus { + url: string, + header: map, + } + + /// @nexus.proto "temporal.api.workflowservice.v1.AttachStreamCallbackRequest" + record attach-stream-callback-request { + /// @nexus.source python="workflow_namespace()" typescript="workflowNamespace()" go="workflow.GetInfo(ctx).Namespace" dotnet="TemporalWorkflowContext.WorkflowNamespace()" + /// @nexus.doc "Namespace of the stream." + namespace: string, + stream-ref: option, + request-id: string, + callback: option, + /// @nexus.omit + identity: placeholder, + operation-token: string, + start-time: option, + /// @nexus.omit + workflow-target: placeholder, + } + + /// @nexus.proto "temporal.api.workflowservice.v1.AttachStreamCallbackResponse" + record attach-stream-callback-response { + } + + attach-stream-callback: func( + request: attach-stream-callback-request, + ) -> attach-stream-callback-response; + /// @nexus.proto "temporal.api.workflowservice.v1.DetachStreamCallbackRequest" + record detach-stream-callback-request { + /// @nexus.source python="workflow_namespace()" typescript="workflowNamespace()" go="workflow.GetInfo(ctx).Namespace" dotnet="TemporalWorkflowContext.WorkflowNamespace()" + /// @nexus.doc "Namespace of the stream." + namespace: string, + stream-ref: option, + request-id: string, + /// @nexus.omit + identity: placeholder, + } + + /// @nexus.proto "temporal.api.workflowservice.v1.DetachStreamCallbackResponse" + record detach-stream-callback-response { + } + + detach-stream-callback: func( + request: detach-stream-callback-request, + ) -> detach-stream-callback-response; + use nexus:temporal-types/model@1.0.0.{ + payload, + }; + + /// @nexus.proto "temporal.api.workflowservice.v1.NotifyStreamRequest" + record notify-stream-request { + /// @nexus.source python="workflow_namespace()" typescript="workflowNamespace()" go="workflow.GetInfo(ctx).Namespace" dotnet="TemporalWorkflowContext.WorkflowNamespace()" + /// @nexus.doc "Namespace of the stream." + namespace: string, + stream-ref: option, + position: string, + counter: option, + metadata: map, + close: option, + close-result: option, + /// @nexus.omit + identity: placeholder, + close-failure: option, + } + + /// @nexus.proto "temporal.api.workflowservice.v1.NotifyStreamResponse" + record notify-stream-response { + } + + notify-stream: func( + request: notify-stream-request, + ) -> notify-stream-response; + /// @nexus.proto "google.protobuf.Timestamp" + record timestamp { + seconds: option, + nanos: option, + } + use nexus:temporal-types/model@1.0.0.{ + failure, + }; } diff --git a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto index 7aec05b140..25cb8db62b 100644 --- a/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto +++ b/crates/protos/protos/api_upstream/temporal/api/workflowservice/v1/service.proto @@ -2045,8 +2045,9 @@ service WorkflowService { // operation returns a stream calls it with the caller's callback when the operation starts. // // (-- api-linter: core::0127::http-annotation=disabled - // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) rpc AttachStreamCallback (AttachStreamCallbackRequest) returns (AttachStreamCallbackResponse) { + option (nexusannotations.v1.operation).tags = "exposed"; } // DetachStreamCallback removes a callback from the notifier of a stream. The notifier completes @@ -2054,8 +2055,9 @@ service WorkflowService { // when the caller cancels the operation. // // (-- api-linter: core::0127::http-annotation=disabled - // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) rpc DetachStreamCallback (DetachStreamCallbackRequest) returns (DetachStreamCallbackResponse) { + option (nexusannotations.v1.operation).tags = "exposed"; } // NotifyStream tells the notifier of a stream that the stream moved, or that it closed. The @@ -2063,25 +2065,27 @@ service WorkflowService { // completes them. Producers fold their notifications and keep at most one call in flight. // // (-- api-linter: core::0127::http-annotation=disabled - // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) rpc NotifyStream (NotifyStreamRequest) returns (NotifyStreamResponse) { + option (nexusannotations.v1.operation).tags = "exposed"; } // DescribeStreamNotifier returns the state of the notifier of a stream and its callbacks. It is - // for operators and tools. + // for operators and tools, so it is not exposed through System Nexus. // // (-- api-linter: core::0127::http-annotation=disabled - // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) rpc DescribeStreamNotifier (DescribeStreamNotifierRequest) returns (DescribeStreamNotifierResponse) { } // PollStreamNotifier waits until the notifier of a stream has a counter above `after_counter`, // or the stream is closed. It returns with no change after a wait limit set by the server // (`streamnotifier.pollMaxWait`). A client that reads a stream calls it after a read that - // found nothing and armed the topic, in place of another blocking read on the store. + // found nothing and armed the topic, in place of another blocking read on the store. It is a + // long poll, so it is not exposed through System Nexus. // // (-- api-linter: core::0127::http-annotation=disabled - // aip.dev/not-precedent: SDKs reach the stream notifier through gRPC. --) + // aip.dev/not-precedent: SDKs reach the stream notifier through System Nexus or gRPC. --) rpc PollStreamNotifier (PollStreamNotifierRequest) returns (PollStreamNotifierResponse) { } }