From 1c3ee2ff136f068fe04361d4983027327c2698e2 Mon Sep 17 00:00:00 2001 From: Kaien Date: Sun, 9 Aug 2026 22:31:32 +0300 Subject: [PATCH] fix(api): add subscribe ready marker to generated proto Co-authored-by: Cursor --- chat.pb.go | 48 +++++++++++++++++++++++++++++++++++++++--------- chat.proto | 14 ++++++++++++++ chat_grpc.pb.go | 26 +++++++++++++------------- 3 files changed, 66 insertions(+), 22 deletions(-) diff --git a/chat.pb.go b/chat.pb.go index 2555a37..e4f0d6d 100644 --- a/chat.pb.go +++ b/chat.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 -// protoc v7.35.1 +// protoc-gen-go v1.36.8 +// protoc (unknown) // source: chat.proto package proto @@ -1039,8 +1039,15 @@ type DeliveryMessage struct { // Subscribe stream. MUST be echoed in Acknowledge; a stale id (from a // replaced/closed stream) is rejected. SubscriptionId string `protobuf:"bytes,3,opt,name=subscription_id,json=subscriptionId,proto3" json:"subscription_id,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // true on the FIRST message of a stream (message is empty). The server + // guarantees that at this point the subscription is registered and its + // durable cursor resolved, so every message committed AFTER the ready + // marker is delivered to this subscription (never skipped by the + // head-initialization race). Old messages are NOT replayed before the marker: + // a brand-new consumer still starts at the topic head. + Ready bool `protobuf:"varint,4,opt,name=ready,proto3" json:"ready,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *DeliveryMessage) Reset() { @@ -1094,6 +1101,13 @@ func (x *DeliveryMessage) GetSubscriptionId() string { return "" } +func (x *DeliveryMessage) GetReady() bool { + if x != nil { + return x.Ready + } + return false +} + type AcknowledgeRequest struct { state protoimpl.MessageState `protogen:"open.v1"` TopicId string `protobuf:"bytes,1,opt,name=topic_id,json=topicId,proto3" json:"topic_id,omitempty"` @@ -1102,8 +1116,15 @@ type AcknowledgeRequest struct { // The subscription_id of the stream that delivered the message. Must be // the CURRENT active subscription of this consumer in this topic. SubscriptionId string `protobuf:"bytes,3,opt,name=subscription_id,json=subscriptionId,proto3" json:"subscription_id,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // true on the FIRST message of a stream (message is empty). The server + // guarantees that at this point the subscription is registered and its + // durable cursor resolved, so every message committed AFTER the ready + // marker is delivered to this subscription (never skipped by the + // head-initialization race). Old messages are NOT replayed before the marker: + // a brand-new consumer still starts at the topic head. + Ready bool `protobuf:"varint,4,opt,name=ready,proto3" json:"ready,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *AcknowledgeRequest) Reset() { @@ -1157,6 +1178,13 @@ func (x *AcknowledgeRequest) GetSubscriptionId() string { return "" } +func (x *AcknowledgeRequest) GetReady() bool { + if x != nil { + return x.Ready + } + return false +} + type AcknowledgeResponse struct { state protoimpl.MessageState `protogen:"open.v1"` LastAckSequence int64 `protobuf:"varint,1,opt,name=last_ack_sequence,json=lastAckSequence,proto3" json:"last_ack_sequence,omitempty"` // the cursor as stored (never lower than before) @@ -1277,15 +1305,17 @@ const file_chat_proto_rawDesc = "" + "\bmessages\x18\x01 \x03(\v2\r.chat.MessageR\bmessages\x12\x19\n" + "\bhas_more\x18\x02 \x01(\bR\ahasMore\"-\n" + "\x10SubscribeRequest\x12\x19\n" + - "\btopic_id\x18\x01 \x01(\tR\atopicId\"{\n" + + "\btopic_id\x18\x01 \x01(\tR\atopicId\"\x91\x01\n" + "\x0fDeliveryMessage\x12'\n" + "\amessage\x18\x01 \x01(\v2\r.chat.MessageR\amessage\x12\x16\n" + "\x06replay\x18\x02 \x01(\bR\x06replay\x12'\n" + - "\x0fsubscription_id\x18\x03 \x01(\tR\x0esubscriptionId\"\x84\x01\n" + + "\x0fsubscription_id\x18\x03 \x01(\tR\x0esubscriptionId\x12\x14\n" + + "\x05ready\x18\x04 \x01(\bR\x05ready\"\x9a\x01\n" + "\x12AcknowledgeRequest\x12\x19\n" + "\btopic_id\x18\x01 \x01(\tR\atopicId\x12*\n" + "\x11last_ack_sequence\x18\x02 \x01(\x03R\x0flastAckSequence\x12'\n" + - "\x0fsubscription_id\x18\x03 \x01(\tR\x0esubscriptionId\"A\n" + + "\x0fsubscription_id\x18\x03 \x01(\tR\x0esubscriptionId\x12\x14\n" + + "\x05ready\x18\x04 \x01(\bR\x05ready\"A\n" + "\x13AcknowledgeResponse\x12*\n" + "\x11last_ack_sequence\x18\x01 \x01(\x03R\x0flastAckSequence*\x8c\x01\n" + "\n" + diff --git a/chat.proto b/chat.proto index d2eb8ed..7de4d5d 100644 --- a/chat.proto +++ b/chat.proto @@ -204,6 +204,13 @@ message DeliveryMessage { // Subscribe stream. MUST be echoed in Acknowledge; a stale id (from a // replaced/closed stream) is rejected. string subscription_id = 3; + // true on the FIRST message of a stream (message is empty). The server + // guarantees that at this point the subscription is registered and its + // durable cursor resolved, so every message committed AFTER the ready + // marker is delivered to this subscription (never skipped by the + // head-initialization race). Old messages are NOT replayed before the marker: + // a brand-new consumer still starts at the topic head. + bool ready = 4; } message AcknowledgeRequest { @@ -213,6 +220,13 @@ message AcknowledgeRequest { // The subscription_id of the stream that delivered the message. Must be // the CURRENT active subscription of this consumer in this topic. string subscription_id = 3; + // true on the FIRST message of a stream (message is empty). The server + // guarantees that at this point the subscription is registered and its + // durable cursor resolved, so every message committed AFTER the ready + // marker is delivered to this subscription (never skipped by the + // head-initialization race). Old messages are NOT replayed before the marker: + // a brand-new consumer still starts at the topic head. + bool ready = 4; } message AcknowledgeResponse { diff --git a/chat_grpc.pb.go b/chat_grpc.pb.go index 05f4d94..bbe67ca 100644 --- a/chat_grpc.pb.go +++ b/chat_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.2 -// - protoc v7.35.1 +// - protoc-gen-go-grpc v1.5.1 +// - protoc (unknown) // source: chat.proto package proto @@ -277,34 +277,34 @@ type ChatServiceServer interface { type UnimplementedChatServiceServer struct{} func (UnimplementedChatServiceServer) ListTopics(context.Context, *ListTopicsRequest) (*ListTopicsResponse, error) { - return nil, status.Error(codes.Unimplemented, "method ListTopics not implemented") + return nil, status.Errorf(codes.Unimplemented, "method ListTopics not implemented") } func (UnimplementedChatServiceServer) GetTopic(context.Context, *GetTopicRequest) (*Topic, error) { - return nil, status.Error(codes.Unimplemented, "method GetTopic not implemented") + return nil, status.Errorf(codes.Unimplemented, "method GetTopic not implemented") } func (UnimplementedChatServiceServer) CreateTopic(context.Context, *CreateTopicRequest) (*Topic, error) { - return nil, status.Error(codes.Unimplemented, "method CreateTopic not implemented") + return nil, status.Errorf(codes.Unimplemented, "method CreateTopic not implemented") } func (UnimplementedChatServiceServer) UpdateTopic(context.Context, *UpdateTopicRequest) (*Topic, error) { - return nil, status.Error(codes.Unimplemented, "method UpdateTopic not implemented") + return nil, status.Errorf(codes.Unimplemented, "method UpdateTopic not implemented") } func (UnimplementedChatServiceServer) PostMessage(context.Context, *PostMessageRequest) (*PostMessageResponse, error) { - return nil, status.Error(codes.Unimplemented, "method PostMessage not implemented") + return nil, status.Errorf(codes.Unimplemented, "method PostMessage not implemented") } func (UnimplementedChatServiceServer) AgentPostMessage(context.Context, *AgentPostMessageRequest) (*PostMessageResponse, error) { - return nil, status.Error(codes.Unimplemented, "method AgentPostMessage not implemented") + return nil, status.Errorf(codes.Unimplemented, "method AgentPostMessage not implemented") } func (UnimplementedChatServiceServer) GetMessage(context.Context, *GetMessageRequest) (*Message, error) { - return nil, status.Error(codes.Unimplemented, "method GetMessage not implemented") + return nil, status.Errorf(codes.Unimplemented, "method GetMessage not implemented") } func (UnimplementedChatServiceServer) History(context.Context, *HistoryRequest) (*HistoryResponse, error) { - return nil, status.Error(codes.Unimplemented, "method History not implemented") + return nil, status.Errorf(codes.Unimplemented, "method History not implemented") } func (UnimplementedChatServiceServer) Subscribe(*SubscribeRequest, grpc.ServerStreamingServer[DeliveryMessage]) error { - return status.Error(codes.Unimplemented, "method Subscribe not implemented") + return status.Errorf(codes.Unimplemented, "method Subscribe not implemented") } func (UnimplementedChatServiceServer) Acknowledge(context.Context, *AcknowledgeRequest) (*AcknowledgeResponse, error) { - return nil, status.Error(codes.Unimplemented, "method Acknowledge not implemented") + return nil, status.Errorf(codes.Unimplemented, "method Acknowledge not implemented") } func (UnimplementedChatServiceServer) mustEmbedUnimplementedChatServiceServer() {} func (UnimplementedChatServiceServer) testEmbeddedByValue() {} @@ -317,7 +317,7 @@ type UnsafeChatServiceServer interface { } func RegisterChatServiceServer(s grpc.ServiceRegistrar, srv ChatServiceServer) { - // If the following call panics, it indicates UnimplementedChatServiceServer was + // If the following call pancis, it indicates UnimplementedChatServiceServer was // embedded by pointer and is nil. This will cause panics if an // unimplemented method is ever invoked, so we test this at initialization // time to prevent it from happening at runtime later due to I/O.