feat: add opt-in consumer auto-ack protocol

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
+```
+

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);
-}
+}

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")
+	}
+}