diff --git a/subscription_test.go b/subscription_test.go index 0ebf0ad..037ccd4 100644 --- a/subscription_test.go +++ b/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