From d8c6174a041b5bd40310748165705172c6b7c017 Mon Sep 17 00:00:00 2001 From: AmHonored Date: Sat, 7 Nov 2026 11:26:54 +0330 Subject: [PATCH] feat: add opt-in consumer auto-ack protocol --- README.md | 21 +++++++++++++++++++++ proto/consumer.proto | 13 +++++++++++-- proto/go/consumer.pb.go | 28 ++++++++++++++++++++++------ proto/go/consumer_test.go | 36 ++++++++++++++++++++++++++++++++++++ 4 files changed, 90 insertions(+), 8 deletions(-) create mode 100644 proto/go/consumer_test.go diff --git a/README.md b/README.md index 76f1cf4..7affedc 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,23 @@ # Protocol +`proto/consumer.proto` defines the consumer stream. `SubscribeInit.auto_ack` +defaults to `false`, preserving manual ACK/NACK behavior. A worker may opt in +with `true`; the broker can reject the stream with `PermissionDenied` according +to its configuration. The setting belongs to the stream, so reconnects must +request it again; durable subscription creation is unchanged. + +An accepting broker sends initial metadata `subscription-auto-ack: true` along +with `subscription-generation`. Clients must require this confirmation before +omitting ACK/NACK frames, so older brokers cannot silently ignore the option. +On an accepted auto-ack stream the broker completes each recipient after a +successful gRPC send, without waiting for application processing, and ignores +all manual ACKs and NACKs. Handler errors cannot request redelivery. Other +recipients retain their independent delivery obligations. A failed send is not +acknowledged; a crash before completion is persisted can still cause duplicates. + +Regenerate the consumer Go bindings with protoc and protoc-gen-go v1.36.11: + +```sh +protoc -I proto --go_out=proto/go --go_opt=paths=source_relative proto/consumer.proto +``` + diff --git a/proto/consumer.proto b/proto/consumer.proto index e540428..468b088 100644 --- a/proto/consumer.proto +++ b/proto/consumer.proto @@ -20,9 +20,18 @@ message SubscribeInit { // one of group_id and subscriber_id. CreateSubscription must precede this stream. string subscriber_id = 3; + // Opt in to broker-side acknowledgment after each successful stream send. + // Defaults to false (manual ACK/NACK). This is per worker connection, not + // durable subscription state. The broker may reject it with PermissionDenied. + // When accepted, initial metadata includes subscription-auto-ack: true; + // clients must require that confirmation before suppressing ACK/NACK frames. + // Handler failures cannot request redelivery; a crash before completion is + // persisted may still cause duplicates. Other recipients are unaffected. + bool auto_ack = 4; } // AckRequest is sent by the consumer to acknowledge or reject a message. +// Brokers ignore both ACKs and NACKs on streams bound with auto_ack=true. message AckRequest { // success=true completes this subscription's delivery. The shared message // is deleted only after all destination subscriptions finish. @@ -54,7 +63,7 @@ message QueueMessage { bytes payload = 2; // delivery_tag is an opaque server-assigned identifier (the raw Pebble - // key). The consumer must echo this in AckRequest.delivery_tag. + // key). Manual-ack consumers must echo this in AckRequest.delivery_tag. bytes delivery_tag = 3; // enqueued_at_unix_ms is the broker wall-clock time (Unix ms) when the @@ -92,4 +101,4 @@ service FutureQConsumer { // AckRequest). The server pushes QueueMessage frames whenever messages // become ready for the subscribed topic and group. rpc Subscribe(stream ConsumerFrame) returns (stream QueueMessage); -} \ No newline at end of file +} diff --git a/proto/go/consumer.pb.go b/proto/go/consumer.pb.go index 7d0cb0f..10f294d 100644 --- a/proto/go/consumer.pb.go +++ b/proto/go/consumer.pb.go @@ -34,7 +34,15 @@ type SubscribeInit struct { GroupId string `protobuf:"bytes,2,opt,name=group_id,json=groupId,proto3" json:"group_id,omitempty"` // Stable identity for an independent universal subscription. Set exactly // one of group_id and subscriber_id. CreateSubscription must precede this stream. - SubscriberId string `protobuf:"bytes,3,opt,name=subscriber_id,json=subscriberId,proto3" json:"subscriber_id,omitempty"` + SubscriberId string `protobuf:"bytes,3,opt,name=subscriber_id,json=subscriberId,proto3" json:"subscriber_id,omitempty"` + // Opt in to broker-side acknowledgment after each successful stream send. + // Defaults to false (manual ACK/NACK). This is per worker connection, not + // durable subscription state. The broker may reject it with PermissionDenied. + // When accepted, initial metadata includes subscription-auto-ack: true; + // clients must require that confirmation before suppressing ACK/NACK frames. + // Handler failures cannot request redelivery; a crash before completion is + // persisted may still cause duplicates. Other recipients are unaffected. + AutoAck bool `protobuf:"varint,4,opt,name=auto_ack,json=autoAck,proto3" json:"auto_ack,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -90,14 +98,21 @@ func (x *SubscribeInit) GetSubscriberId() string { return "" } +func (x *SubscribeInit) GetAutoAck() bool { + if x != nil { + return x.AutoAck + } + return false +} + // AckRequest is sent by the consumer to acknowledge or reject a message. +// Brokers ignore both ACKs and NACKs on streams bound with auto_ack=true. type AckRequest struct { state protoimpl.MessageState `protogen:"open.v1"` // success=true completes this subscription's delivery. The shared message // is deleted only after all destination subscriptions finish. // success=false: message was rejected (NACK); the broker will re-dispatch - // - // it to another consumer on the next dispatcher tick. + // it to another consumer on the next dispatcher tick. Success bool `protobuf:"varint,1,opt,name=success,proto3" json:"success,omitempty"` // delivery_tag must be the exact bytes received in QueueMessage.delivery_tag. DeliveryTag []byte `protobuf:"bytes,2,opt,name=delivery_tag,json=deliveryTag,proto3" json:"delivery_tag,omitempty"` @@ -243,7 +258,7 @@ type QueueMessage struct { // payload is the application data originally sent by the producer. Payload []byte `protobuf:"bytes,2,opt,name=payload,proto3" json:"payload,omitempty"` // delivery_tag is an opaque server-assigned identifier (the raw Pebble - // key). The consumer must echo this in AckRequest.delivery_tag. + // key). Manual-ack consumers must echo this in AckRequest.delivery_tag. DeliveryTag []byte `protobuf:"bytes,3,opt,name=delivery_tag,json=deliveryTag,proto3" json:"delivery_tag,omitempty"` // enqueued_at_unix_ms is the broker wall-clock time (Unix ms) when the // message was first received. Useful for computing actual delivery latency. @@ -490,11 +505,12 @@ var File_consumer_proto protoreflect.FileDescriptor const file_consumer_proto_rawDesc = "" + "\n" + - "\x0econsumer.proto\x12\afutureq\"e\n" + + "\x0econsumer.proto\x12\afutureq\"\x80\x01\n" + "\rSubscribeInit\x12\x14\n" + "\x05topic\x18\x01 \x01(\tR\x05topic\x12\x19\n" + "\bgroup_id\x18\x02 \x01(\tR\agroupId\x12#\n" + - "\rsubscriber_id\x18\x03 \x01(\tR\fsubscriberId\"I\n" + + "\rsubscriber_id\x18\x03 \x01(\tR\fsubscriberId\x12\x19\n" + + "\bauto_ack\x18\x04 \x01(\bR\aautoAck\"I\n" + "\n" + "AckRequest\x12\x18\n" + "\asuccess\x18\x01 \x01(\bR\asuccess\x12!\n" + diff --git a/proto/go/consumer_test.go b/proto/go/consumer_test.go new file mode 100644 index 0000000..4d006e1 --- /dev/null +++ b/proto/go/consumer_test.go @@ -0,0 +1,36 @@ +package _go + +import ( + "testing" + + "google.golang.org/protobuf/proto" +) + +func TestSubscribeInitAutoAckWireCompatibility(t *testing.T) { + for _, autoAck := range []bool{false, true} { + original := &SubscribeInit{Topic: "orders", GroupId: "workers", AutoAck: autoAck} + data, err := proto.Marshal(original) + if err != nil { + t.Fatal(err) + } + var decoded SubscribeInit + if err := proto.Unmarshal(data, &decoded); err != nil { + t.Fatal(err) + } + if !proto.Equal(original, &decoded) { + t.Fatalf("round trip: got %v, want %v", &decoded, original) + } + } + // Field 4 is a varint bool; older handshakes omit it and stay manual. + var decoded SubscribeInit + if err := proto.Unmarshal([]byte{0x20, 0x01}, &decoded); err != nil { + t.Fatal(err) + } + if !decoded.GetAutoAck() { + t.Fatal("field 4 did not enable auto-ack") + } + decoded.Reset() + if decoded.GetAutoAck() || (*SubscribeInit)(nil).GetAutoAck() { + t.Fatal("auto-ack must default to false") + } +}