Test ACK response loss reconnect behavior

Cover the ambiguous case where the server applies an ACK but the client sees a transient failure, ensuring reconnect follows the server durable cursor instead of local receive state.

Co-authored-by: Cursor <cursoragent@cursor.com>

subscription_test.go

@@ -28,6 +28,7 @@ type fakeChatServer struct {
 	genericPosts   int
 	ackCalls       []ackCall
 	ackErrors      map[int64][]error
+	ackFunc        func(context.Context, *pb.AcknowledgeRequest) (*pb.AcknowledgeResponse, error)
 	subscribeCount int
 	subscribeFunc  func(*pb.SubscribeRequest, pb.ChatService_SubscribeServer) error
 	historyPages   []*pb.HistoryResponse
@@ -114,6 +115,12 @@ func (f *fakeChatServer) Subscribe(req *pb.SubscribeRequest, stream pb.ChatServi
 func (f *fakeChatServer) Acknowledge(ctx context.Context, req *pb.AcknowledgeRequest) (*pb.AcknowledgeResponse, error) {
 	f.record(ctx, "Acknowledge")
 	f.mu.Lock()
+	fn := f.ackFunc
+	f.mu.Unlock()
+	if fn != nil {
+		return fn(ctx, req)
+	}
+	f.mu.Lock()
 	f.ackCalls = append(f.ackCalls, ackCall{topic: req.GetTopicId(), subID: req.GetSubscriptionId(), seq: req.GetLastAckSequence()})
 	errs := f.ackErrors[req.GetLastAckSequence()]
 	if len(errs) > 0 {
@@ -385,6 +392,61 @@ func TestACKTransientFailureIsRetriedWithoutLosingBoundary(t *testing.T) {
 	waitACKs(t, fake, 104, 104)
 }
 
+func TestACKResponseLostAmbiguityReconnectUsesServerCursor(t *testing.T) {
+	fake := newFake()
+	dropFirst := make(chan struct{})
+	var dropOnce sync.Once
+	var calls int
+	fake.subscribeFunc = func(_ *pb.SubscribeRequest, stream pb.ChatService_SubscribeServer) error {
+		calls++
+		_ = stream.Send(&pb.DeliveryMessage{Ready: true})
+		if calls == 1 {
+			_ = stream.Send(msg(100))
+			_ = stream.Send(msg(104))
+			<-dropFirst
+			return status.Error(codes.Unavailable, "ack response was lost with the stream")
+		}
+		dm := msg(111)
+		dm.SubscriptionId = "sub-2"
+		_ = stream.Send(dm)
+		<-stream.Context().Done()
+		return stream.Context().Err()
+	}
+	fake.ackFunc = func(_ context.Context, req *pb.AcknowledgeRequest) (*pb.AcknowledgeResponse, error) {
+		fake.mu.Lock()
+		fake.ackCalls = append(fake.ackCalls, ackCall{topic: req.GetTopicId(), subID: req.GetSubscriptionId(), seq: req.GetLastAckSequence()})
+		fake.mu.Unlock()
+		if req.GetLastAckSequence() == 104 {
+			// Simulate "server applied ACK, client lost response"; the next
+			// stream script starts after that durable cursor and does not
+			// replay 100/104.
+			dropOnce.Do(func() { close(dropFirst) })
+			return nil, status.Error(codes.Unavailable, "response lost")
+		}
+		return &pb.AcknowledgeResponse{LastAckSequence: req.GetLastAckSequence()}, nil
+	}
+	c := newClientForTest(t, fake,
+		WithReconnectBackoff(10*time.Millisecond, 10*time.Millisecond),
+		withSleeperForTest(sleepContext),
+	)
+	sub, _ := c.Subscribe(context.Background(), SubscribeRequest{TopicID: "topic"})
+	d100, d104 := recvDelivery(t, sub), recvDelivery(t, sub)
+	if err := sub.MarkProcessed(context.Background(), d104); err != nil {
+		t.Fatal(err)
+	}
+	if err := sub.MarkProcessed(context.Background(), d100); err != nil {
+		t.Fatal(err)
+	}
+	d111 := recvDelivery(t, sub)
+	if d111.Message.GetSequence() != 111 {
+		t.Fatalf("reconnect must follow server cursor, got seq %d", d111.Message.GetSequence())
+	}
+	if err := sub.MarkProcessed(context.Background(), d111); err != nil {
+		t.Fatal(err)
+	}
+	waitACKs(t, fake, 104, 104, 111)
+}
+
 func TestReconnectReplayAndStaleGenerationFencing(t *testing.T) {
 	fake := newFake()
 	var calls int