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 7599af14fe..df2fce5fa6 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 @@ -2046,8 +2046,9 @@ service WorkflowService { // starts the notifier when the stream has none. // // (-- 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 @@ -2055,8 +2056,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"; } // WaitStreamCallback asks the notifier of a stream to wake one idle callback once, when the @@ -2081,15 +2083,16 @@ service WorkflowService { // when the stream has none, so a callback that attaches later still gets the close outcome. // // (-- 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) { } @@ -2098,10 +2101,11 @@ service WorkflowService { // (`streamnotifier.pollMaxWait`). A client that reads a stream calls it after a read that // found nothing, in place of another blocking read on the store. It starts the notifier when the // stream has none. The client reads the stream once more after the poll starts, since a - // notification that came before the notifier existed was dropped. + // notification that came before the notifier existed was dropped. 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) { }