[Bridge] Added the contract and protos of Core's stream Nexus calls.

crates/protos/build.rs

@@ -59,6 +59,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     let protos = &[
         "./protos/local/temporal/sdk/core/core_interface.proto",
         "./protos/local/temporal/sdk/core/streams/streams.proto",
+        "./protos/local/temporal/sdk/core/streams/nexus.proto",
         "./protos/api_upstream/temporal/api/sdk/v1/workflow_metadata.proto",
         "./protos/api_upstream/temporal/api/workflowservice/v1/service.proto",
         "./protos/api_upstream/temporal/api/nexusservices/workerservice/v1/request_response.proto",

crates/protos/protos/local/temporal/sdk/core/streams/nexus.proto

@@ -0,0 +1,93 @@
+syntax = "proto3";
+
+package coresdk.streams;
+option ruby_package = "Temporalio::Internal::Bridge::Api::Streams";
+
+import "google/protobuf/timestamp.proto";
+import "temporal/api/common/v1/message.proto";
+import "temporal/api/failure/v1/message.proto";
+
+// The stream service's Nexus operations, served by Core. Lang forwards each call's input in the
+// N0a contract's JSON transfer form, runs its payload codec when Core asks for it, and hands the
+// failure Core answers with to its Nexus handler. The contract is
+// `crates/streams/nexus/temporal_streams.nexusrpc.yaml`.
+
+// The calls Bridge adds beside `StreamService`, dispatched by name through the same entry.
+service StreamNexusService {
+  // The stream service's Nexus `append` operation, over its JSON transfer form.
+  rpc NexusAppend(NexusCall) returns (NexusStep);
+  // The stream service's Nexus `read` operation, over its JSON transfer form.
+  rpc NexusRead(NexusCall) returns (NexusStep);
+  // Goes on with a call that asked for codec work.
+  rpc NexusResume(NexusResumeRequest) returns (NexusStep);
+}
+
+// One Nexus operation call.
+message NexusCall {
+  // The operation's input, the N0a model in its JSON transfer form.
+  bytes input = 1;
+  // When the Nexus request times out. Core answers a read in time for its answer to travel back.
+  google.protobuf.Timestamp deadline = 2;
+  NexusLimits limits = 3;
+  // The namespace of the lang Worker's client, which streams the call names live in.
+  string namespace = 4;
+}
+
+message NexusLimits {
+  // The most bytes one read answer takes as JSON past its first record, with its tokens and state.
+  // Zero takes the default of 1 MiB, which keeps an answer below the server's blob size limit.
+  uint64 max_answer_bytes = 1;
+  // The most bytes one record may take in an answer. A record above it can never be read, so an
+  // append with one is refused. Zero takes the default of 1.5 MiB.
+  uint64 max_record_bytes = 2;
+}
+
+// What Core answers a call or a resume with. A codec step comes before the output on a call whose
+// bodies need lang's payload codec, and lang resumes it with the results.
+message NexusStep {
+  oneof step {
+    // The operation's output, the N0a model in its JSON transfer form.
+    bytes output = 1;
+    CodecWork codec = 2;
+    NexusFailure failure = 3;
+  }
+}
+
+// Bodies for lang's payload codec. Lang runs the namespace's codec over every payload, in order,
+// and resumes with one result per payload.
+message CodecWork {
+  // True encodes the bodies of an append, and false decodes the bodies of a read.
+  bool encode = 1;
+  // The stream's namespace and owner, for the codec's serialization context.
+  string namespace = 2;
+  string workflow_id = 3;
+  repeated temporal.api.common.v1.Payload payloads = 4;
+  // Hand back in `NexusResumeRequest`. Core forgets a continuation 60 s after it asked.
+  uint64 continuation = 5;
+}
+
+message NexusResumeRequest {
+  uint64 continuation = 1;
+  // One per payload of the `CodecWork`, in order.
+  repeated CodecResult results = 2;
+}
+
+message CodecResult {
+  oneof result {
+    temporal.api.common.v1.Payload payload = 1;
+    // Why the codec failed on this payload. A read hands the record over as an error at its
+    // cursor, and an append fails.
+    string error = 2;
+  }
+}
+
+// How a Nexus operation failed, ready for lang's Nexus handler error.
+message NexusFailure {
+  // The Nexus `HandlerErrorType` name, such as `BAD_REQUEST`.
+  string handler_error_type = 1;
+  bool retryable = 2;
+  string message = 3;
+  // The failure that crosses to the caller, with the stream condition's class as its cause's
+  // application error type.
+  temporal.api.failure.v1.Failure failure = 4;
+}

crates/protos/protos/local/temporal/sdk/core/streams/streams.proto

@@ -769,8 +769,8 @@ enum StreamFailureKind {
   // This release or this store does not offer the capability.
   STREAM_FAILURE_KIND_UNSUPPORTED = 11;
   // The call's input breaks the contract, such as a missing member or a value out of range, so
-  // the same call fails the same way. A lang built before this kind maps it, as any kind it
-  // doesn't know, to a storage failure.
+  // the same call fails the same way. The Nexus calls answer it, as BAD_REQUEST. A lang built
+  // before this kind maps it, as any kind it doesn't know, to a storage failure.
   STREAM_FAILURE_KIND_INVALID_ARGUMENT = 12;
   // A replay read found other records than the read it repeats: other content, another count
   // or another end. Lang reports it as nondeterminism.