From 8911d163bb3265adf01718fdda497326b0d039e3 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Mon, 5 Oct 2026 16:57:09 -0700 Subject: [PATCH 1/3] Pinned proto/api to the stream record envelope. --- .gitmodules | 2 +- proto/api | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.gitmodules b/.gitmodules index c4a68cc48..934a40e24 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,6 +1,6 @@ [submodule "proto/api"] path = proto/api - url = https://github.com/temporalio/api + url = https://github.com/moedash/api [submodule "proto/api-cloud"] path = proto/api-cloud url = https://github.com/temporalio/api-cloud diff --git a/proto/api b/proto/api index 1c27468c7..ec2c8f635 160000 --- a/proto/api +++ b/proto/api @@ -1 +1 @@ -Subproject commit 1c27468c756fc8abc800235c33b74c6eba638c88 +Subproject commit ec2c8f635d4ccd49bd2262026a432220084d9555 From f3058496b94e14fbafee62e52cd1a60cfc822578 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Mon, 5 Oct 2026 16:58:41 -0700 Subject: [PATCH 2/3] Regenerated for the stream record envelope. --- proxy/interceptor.go | 33 ++++ stream/v1/message.go-helpers.pb.go | 64 +++++++ stream/v1/message.pb.go | 276 +++++++++++++++++++++++++++++ 3 files changed, 373 insertions(+) create mode 100644 stream/v1/message.go-helpers.pb.go create mode 100644 stream/v1/message.pb.go diff --git a/proxy/interceptor.go b/proxy/interceptor.go index 8ef2313da..1758201f8 100644 --- a/proxy/interceptor.go +++ b/proxy/interceptor.go @@ -28,6 +28,7 @@ import ( "go.temporal.io/api/query/v1" "go.temporal.io/api/schedule/v1" "go.temporal.io/api/sdk/v1" + "go.temporal.io/api/stream/v1" "go.temporal.io/api/update/v1" "go.temporal.io/api/workflow/v1" workflowservice "go.temporal.io/api/workflowservice/v1" @@ -3579,6 +3580,38 @@ func visitPayloads( ctx.Context = prevCtx + case *stream.StreamRecord: + + if o == nil { + continue + } + + prevCtx := ctx.Context + if options.ContextHook != nil { + var hookErr error + if ctx.Context, hookErr = options.ContextHook(prevCtx, o); hookErr != nil { + return hookErr + } + } + + if o.Body != nil { + if err := visitPayload(ctx, options, o, concState, &o.Body); err != nil { + return err + } + } + + if err := visitPayloads( + ctx, + options, + o, + concState, + o.GetMetadata(), + ); err != nil { + return err + } + + ctx.Context = prevCtx + case *update.Acceptance: if o == nil { diff --git a/stream/v1/message.go-helpers.pb.go b/stream/v1/message.go-helpers.pb.go new file mode 100644 index 000000000..ab1894ae1 --- /dev/null +++ b/stream/v1/message.go-helpers.pb.go @@ -0,0 +1,64 @@ +// Code generated by protoc-gen-go-helpers. DO NOT EDIT. +package stream + +import ( + "fmt" + + "google.golang.org/protobuf/proto" +) + +// Marshal an object of type StreamRecord to the protobuf v3 wire format +func (val *StreamRecord) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type StreamRecord from the protobuf v3 wire format +func (val *StreamRecord) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *StreamRecord) Size() int { + return proto.Size(val) +} + +// Equal returns whether two StreamRecord values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *StreamRecord) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *StreamRecord + switch t := that.(type) { + case *StreamRecord: + that1 = t + case StreamRecord: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} + +var ( + StreamRecordKind_shorthandValue = map[string]int32{ + "Unspecified": 0, + "Data": 1, + "Finish": 2, + } +) + +// StreamRecordKindFromString parses a StreamRecordKind value from either the protojson +// canonical SCREAMING_CASE enum or the traditional temporal PascalCase enum to StreamRecordKind +func StreamRecordKindFromString(s string) (StreamRecordKind, error) { + if v, ok := StreamRecordKind_value[s]; ok { + return StreamRecordKind(v), nil + } else if v, ok := StreamRecordKind_shorthandValue[s]; ok { + return StreamRecordKind(v), nil + } + return StreamRecordKind(0), fmt.Errorf("%s is not a valid StreamRecordKind", s) +} diff --git a/stream/v1/message.pb.go b/stream/v1/message.pb.go new file mode 100644 index 000000000..c84aa3373 --- /dev/null +++ b/stream/v1/message.pb.go @@ -0,0 +1,276 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// plugins: +// protoc-gen-go +// protoc +// source: temporal/api/stream/v1/message.proto + +package stream + +import ( + reflect "reflect" + "strconv" + sync "sync" + unsafe "unsafe" + + v1 "go.temporal.io/api/common/v1" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// What a record means to a reader. Kept on the record itself so every store +// and every language reads it the same way without a private envelope. +type StreamRecordKind int32 + +const ( + // Read as DATA. + STREAM_RECORD_KIND_UNSPECIFIED StreamRecordKind = 0 + // A value the producer published; `body` carries it. + STREAM_RECORD_KIND_DATA StreamRecordKind = 1 + // The producer named by `producer_id` writes nothing more on `topic`. + // Says nothing about that producer's outcome and does not end the stream. + STREAM_RECORD_KIND_FINISH StreamRecordKind = 2 +) + +// Enum value maps for StreamRecordKind. +var ( + StreamRecordKind_name = map[int32]string{ + 0: "STREAM_RECORD_KIND_UNSPECIFIED", + 1: "STREAM_RECORD_KIND_DATA", + 2: "STREAM_RECORD_KIND_FINISH", + } + StreamRecordKind_value = map[string]int32{ + "STREAM_RECORD_KIND_UNSPECIFIED": 0, + "STREAM_RECORD_KIND_DATA": 1, + "STREAM_RECORD_KIND_FINISH": 2, + } +) + +func (x StreamRecordKind) Enum() *StreamRecordKind { + p := new(StreamRecordKind) + *p = x + return p +} + +func (x StreamRecordKind) String() string { + switch x { + case STREAM_RECORD_KIND_UNSPECIFIED: + return "Unspecified" + case STREAM_RECORD_KIND_DATA: + return "Data" + case STREAM_RECORD_KIND_FINISH: + return "Finish" + default: + return strconv.Itoa(int(x)) + } + +} + +func (StreamRecordKind) Descriptor() protoreflect.EnumDescriptor { + return file_temporal_api_stream_v1_message_proto_enumTypes[0].Descriptor() +} + +func (StreamRecordKind) Type() protoreflect.EnumType { + return &file_temporal_api_stream_v1_message_proto_enumTypes[0] +} + +func (x StreamRecordKind) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use StreamRecordKind.Descriptor instead. +func (StreamRecordKind) EnumDescriptor() ([]byte, []int) { + return file_temporal_api_stream_v1_message_proto_rawDescGZIP(), []int{0} +} + +// One entry in a stream. The record is the wire format: stores keep it +// serialized as is and readers in every language decode the same bytes. +type StreamRecord struct { + state protoimpl.MessageState `protogen:"open.v1"` + // The value the producer published, stored as sent. A stream provider + // writes and reads it as part of the record, so applying a payload codec + // to it is the provider's job. + Body *v1.Payload `protobuf:"bytes,1,opt,name=body,proto3" json:"body,omitempty"` + // Producer-supplied provenance, stored as sent. + Metadata map[string]*v1.Payload `protobuf:"bytes,2,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // Producer-supplied grouping label, stored as sent. + Topic string `protobuf:"bytes,3,opt,name=topic,proto3" json:"topic,omitempty"` + // How to read this record. Unspecified is read as DATA. + Kind StreamRecordKind `protobuf:"varint,4,opt,name=kind,proto3,enum=temporal.api.stream.v1.StreamRecordKind" json:"kind,omitempty"` + // Who wrote the record. Empty when the owning Workflow did. + ProducerId string `protobuf:"bytes,5,opt,name=producer_id,json=producerId,proto3" json:"producer_id,omitempty"` + // The producer's attempt. Readers treat a later attempt by the same + // producer as superseding what the earlier one wrote. + Attempt int64 `protobuf:"varint,6,opt,name=attempt,proto3" json:"attempt,omitempty"` + // The producer's position within its attempt, zero when it does not number + // its records. Stored as sent; the server does not assign, validate or + // order by it, and the stream's own offsets are what order a read. + Sequence int64 `protobuf:"varint,7,opt,name=sequence,proto3" json:"sequence,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *StreamRecord) Reset() { + *x = StreamRecord{} + mi := &file_temporal_api_stream_v1_message_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *StreamRecord) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*StreamRecord) ProtoMessage() {} + +func (x *StreamRecord) ProtoReflect() protoreflect.Message { + mi := &file_temporal_api_stream_v1_message_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use StreamRecord.ProtoReflect.Descriptor instead. +func (*StreamRecord) Descriptor() ([]byte, []int) { + return file_temporal_api_stream_v1_message_proto_rawDescGZIP(), []int{0} +} + +func (x *StreamRecord) GetBody() *v1.Payload { + if x != nil { + return x.Body + } + return nil +} + +func (x *StreamRecord) GetMetadata() map[string]*v1.Payload { + if x != nil { + return x.Metadata + } + return nil +} + +func (x *StreamRecord) GetTopic() string { + if x != nil { + return x.Topic + } + return "" +} + +func (x *StreamRecord) GetKind() StreamRecordKind { + if x != nil { + return x.Kind + } + return STREAM_RECORD_KIND_UNSPECIFIED +} + +func (x *StreamRecord) GetProducerId() string { + if x != nil { + return x.ProducerId + } + return "" +} + +func (x *StreamRecord) GetAttempt() int64 { + if x != nil { + return x.Attempt + } + return 0 +} + +func (x *StreamRecord) GetSequence() int64 { + if x != nil { + return x.Sequence + } + return 0 +} + +var File_temporal_api_stream_v1_message_proto protoreflect.FileDescriptor + +const file_temporal_api_stream_v1_message_proto_rawDesc = "" + + "\n" + + "$temporal/api/stream/v1/message.proto\x12\x16temporal.api.stream.v1\x1a$temporal/api/common/v1/message.proto\"\x9c\x03\n" + + "\fStreamRecord\x123\n" + + "\x04body\x18\x01 \x01(\v2\x1f.temporal.api.common.v1.PayloadR\x04body\x12N\n" + + "\bmetadata\x18\x02 \x03(\v22.temporal.api.stream.v1.StreamRecord.MetadataEntryR\bmetadata\x12\x14\n" + + "\x05topic\x18\x03 \x01(\tR\x05topic\x12<\n" + + "\x04kind\x18\x04 \x01(\x0e2(.temporal.api.stream.v1.StreamRecordKindR\x04kind\x12\x1f\n" + + "\vproducer_id\x18\x05 \x01(\tR\n" + + "producerId\x12\x18\n" + + "\aattempt\x18\x06 \x01(\x03R\aattempt\x12\x1a\n" + + "\bsequence\x18\a \x01(\x03R\bsequence\x1a\\\n" + + "\rMetadataEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x125\n" + + "\x05value\x18\x02 \x01(\v2\x1f.temporal.api.common.v1.PayloadR\x05value:\x028\x01*r\n" + + "\x10StreamRecordKind\x12\"\n" + + "\x1eSTREAM_RECORD_KIND_UNSPECIFIED\x10\x00\x12\x1b\n" + + "\x17STREAM_RECORD_KIND_DATA\x10\x01\x12\x1d\n" + + "\x19STREAM_RECORD_KIND_FINISH\x10\x02B\x89\x01\n" + + "\x19io.temporal.api.stream.v1B\fMessageProtoP\x01Z#go.temporal.io/api/stream/v1;stream\xaa\x02\x18Temporalio.Api.Stream.V1\xea\x02\x1bTemporalio::Api::Stream::V1b\x06proto3" + +var ( + file_temporal_api_stream_v1_message_proto_rawDescOnce sync.Once + file_temporal_api_stream_v1_message_proto_rawDescData []byte +) + +func file_temporal_api_stream_v1_message_proto_rawDescGZIP() []byte { + file_temporal_api_stream_v1_message_proto_rawDescOnce.Do(func() { + file_temporal_api_stream_v1_message_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_temporal_api_stream_v1_message_proto_rawDesc), len(file_temporal_api_stream_v1_message_proto_rawDesc))) + }) + return file_temporal_api_stream_v1_message_proto_rawDescData +} + +var file_temporal_api_stream_v1_message_proto_enumTypes = make([]protoimpl.EnumInfo, 1) +var file_temporal_api_stream_v1_message_proto_msgTypes = make([]protoimpl.MessageInfo, 2) +var file_temporal_api_stream_v1_message_proto_goTypes = []any{ + (StreamRecordKind)(0), // 0: temporal.api.stream.v1.StreamRecordKind + (*StreamRecord)(nil), // 1: temporal.api.stream.v1.StreamRecord + nil, // 2: temporal.api.stream.v1.StreamRecord.MetadataEntry + (*v1.Payload)(nil), // 3: temporal.api.common.v1.Payload +} +var file_temporal_api_stream_v1_message_proto_depIdxs = []int32{ + 3, // 0: temporal.api.stream.v1.StreamRecord.body:type_name -> temporal.api.common.v1.Payload + 2, // 1: temporal.api.stream.v1.StreamRecord.metadata:type_name -> temporal.api.stream.v1.StreamRecord.MetadataEntry + 0, // 2: temporal.api.stream.v1.StreamRecord.kind:type_name -> temporal.api.stream.v1.StreamRecordKind + 3, // 3: temporal.api.stream.v1.StreamRecord.MetadataEntry.value:type_name -> temporal.api.common.v1.Payload + 4, // [4:4] is the sub-list for method output_type + 4, // [4:4] is the sub-list for method input_type + 4, // [4:4] is the sub-list for extension type_name + 4, // [4:4] is the sub-list for extension extendee + 0, // [0:4] is the sub-list for field type_name +} + +func init() { file_temporal_api_stream_v1_message_proto_init() } +func file_temporal_api_stream_v1_message_proto_init() { + if File_temporal_api_stream_v1_message_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_api_stream_v1_message_proto_rawDesc), len(file_temporal_api_stream_v1_message_proto_rawDesc)), + NumEnums: 1, + NumMessages: 2, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_temporal_api_stream_v1_message_proto_goTypes, + DependencyIndexes: file_temporal_api_stream_v1_message_proto_depIdxs, + EnumInfos: file_temporal_api_stream_v1_message_proto_enumTypes, + MessageInfos: file_temporal_api_stream_v1_message_proto_msgTypes, + }.Build() + File_temporal_api_stream_v1_message_proto = out.File + file_temporal_api_stream_v1_message_proto_goTypes = nil + file_temporal_api_stream_v1_message_proto_depIdxs = nil +} From 4b4e92ffc37fd3400d3316140990a8c4d4afe4d9 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 6 Oct 2026 19:38:16 -0700 Subject: [PATCH 3/3] Pointed the proto submodule at the renamed fork account. --- .gitmodules | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.gitmodules b/.gitmodules index 934a40e24..fa15217e2 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,6 +1,6 @@ [submodule "proto/api"] path = proto/api - url = https://github.com/moedash/api + url = https://github.com/moetemp/api [submodule "proto/api-cloud"] path = proto/api-cloud url = https://github.com/temporalio/api-cloud