fix(pubsub/pstest): start DeliveryAttempt at 1

Per [the documentation](https://github.com/googleapis/google-cloud-go/blob/74da335fea6cd70b27808507f2e58ae53f5f4910/internal/pubsub/message.go#L63) `DeliveryAttempt` should start at 1 when a dead-letter topic is enabled, not 0 as pstest currently returns

Add `DeliveryAttempt` to streaming pull responses.

pubsub/pstest/fake.go

@@ -1037,10 +1037,10 @@ func (s *subscription) pull(max int) []*pb.ReceivedMessage {
 			s.publishToDeadLetter(m)
 			continue
 		}
+		(*m.deliveries)++
 		if s.proto.DeadLetterPolicy != nil {
 			m.proto.DeliveryAttempt = int32(*m.deliveries)
 		}
-		(*m.deliveries)++
 		m.ackDeadline = now.Add(s.ackTimeout)
 		msgs = append(msgs, m.proto)
 		if len(msgs) >= max {
@@ -1122,6 +1122,13 @@ func (s *subscription) deliver() {
 //
 // Must be called with the lock held.
 func (s *subscription) tryDeliverMessage(m *message, start int, now time.Time) (int, bool) {
+	// Optimistically increment DeliveryAttempt assuming we'll be able to deliver the message.  This is
+	// safe since the lock is held for the duration of this function, and the channel receiver does not
+	// modify the message.
+	if s.proto.DeadLetterPolicy != nil {
+		m.proto.DeliveryAttempt = int32(*m.deliveries) + 1
+	}
+
 	for i := 0; i < len(s.streams); i++ {
 		idx := (i + start) % len(s.streams)
 
@@ -1139,6 +1146,10 @@ func (s *subscription) tryDeliverMessage(m *message, start int, now time.Time) (
 		default:
 		}
 	}
+	// Restore the correct value of DeliveryAttempt if we were not able to deliver the message.
+	if s.proto.DeadLetterPolicy != nil {
+		m.proto.DeliveryAttempt = int32(*m.deliveries)
+	}
 	return 0, false
 }
 

pubsub/pstest/fake_test.go

@@ -346,8 +346,8 @@ func TestSubscriptionDeadLetter(t *testing.T) {
 
 			}
 			for _, m := range pull.ReceivedMessages {
-				if int32(i) != m.DeliveryAttempt {
-					t.Fatalf("message delivery attempt not the expected one. expected: %d, actual: %d", i, m.DeliveryAttempt)
+				if int32(i+1) != m.DeliveryAttempt {
+					t.Fatalf("message delivery attempt not the expected one. expected: %d, actual: %d", i+1, m.DeliveryAttempt)
 				}
 				_, err := server.GServer.ModifyAckDeadline(ctx, &pb.ModifyAckDeadlineRequest{
 					Subscription:       sub.Name,
@@ -551,19 +551,33 @@ func TestStreamingPull(t *testing.T) {
 	pclient, sclient, srv, cleanup := newFake(context.TODO(), t)
 	defer cleanup()
 
+	deadLetterTopic := mustCreateTopic(context.TODO(), t, pclient, &pb.Topic{
+		Name: "projects/P/topics/deadLetter",
+	})
+
 	top := mustCreateTopic(context.TODO(), t, pclient, &pb.Topic{Name: "projects/P/topics/T"})
 	sub := mustCreateSubscription(context.TODO(), t, sclient, &pb.Subscription{
 		Name:               "projects/P/subscriptions/S",
 		Topic:              top.Name,
 		AckDeadlineSeconds: 10,
+		DeadLetterPolicy: &pb.DeadLetterPolicy{
+			DeadLetterTopic:     deadLetterTopic.Name,
+			MaxDeliveryAttempts: 3,
+		},
 	})
 
 	want := publish(t, srv, pclient, top, []*pb.PubsubMessage{
 		{Data: []byte("d1")},
 		{Data: []byte("d2")},
 		{Data: []byte("d3")},
 	})
-	got := pubsubMessages(streamingPullN(context.TODO(), t, len(want), sclient, sub))
+	received := streamingPullN(context.TODO(), t, len(want), sclient, sub)
+	for _, m := range received {
+		if m.DeliveryAttempt != 1 {
+			t.Errorf("got DeliveryAttempt==%d, want 1", m.DeliveryAttempt)
+		}
+	}
+	got := pubsubMessages(received)
 	if diff := testutil.Diff(got, want); diff != "" {
 		t.Error(diff)
 	}