fix(pubsub/pstest): make Message.Modacks thread safe (#2755)

* fix(pubsub/pstest): make Message.Modacks thread safe

* restore original modack test

* address review comments

* replace todo contexts

* use t.Logf instead of fmt

pubsub/pstest/fake.go

@@ -160,15 +160,15 @@ type Message struct {
 	Data        []byte
 	Attributes  map[string]string
 	PublishTime time.Time
-	Deliveries  int // number of times delivery of the message was attempted
-	Acks        int // number of acks received from clients
+	Deliveries  int      // number of times delivery of the message was attempted
+	Acks        int      // number of acks received from clients
+	Modacks     []Modack // modacks received by server for this message
 	OrderingKey string
 
 	// protected by server mutex
 	deliveries int
 	acks       int
-	Modacks    []Modack // modacks received by server for this message
-
+	modacks    []Modack
 }
 
 // Modack represents a modack sent to the server.
@@ -187,6 +187,7 @@ func (s *Server) Messages() []*Message {
 	for _, m := range s.GServer.msgs {
 		m.Deliveries = m.deliveries
 		m.Acks = m.acks
+		m.Modacks = append([]Modack(nil), m.modacks...)
 		msgs = append(msgs, m)
 	}
 	return msgs
@@ -202,6 +203,7 @@ func (s *Server) Message(id string) *Message {
 	if m != nil {
 		m.Deliveries = m.deliveries
 		m.Acks = m.acks
+		m.Modacks = append([]Modack(nil), m.modacks...)
 	}
 	return m
 }
@@ -657,7 +659,7 @@ func (s *GServer) ModifyAckDeadline(_ context.Context, req *pb.ModifyAckDeadline
 	}
 	now := time.Now()
 	for _, id := range req.AckIds {
-		s.msgsByID[id].Modacks = append(s.msgsByID[id].Modacks, Modack{AckID: id, AckDeadline: req.AckDeadlineSeconds, ReceivedAt: now})
+		s.msgsByID[id].modacks = append(s.msgsByID[id].modacks, Modack{AckID: id, AckDeadline: req.AckDeadlineSeconds, ReceivedAt: now})
 	}
 	dur := secsToDur(req.AckDeadlineSeconds)
 	for _, id := range req.AckIds {

pubsub/pstest/fake_test.go

@@ -705,6 +705,48 @@ func TestTimeNowFunc(t *testing.T) {
 	}
 }
 
+func TestModAck_Race(t *testing.T) {
+	ctx := context.Background()
+	pclient, sclient, server, cleanup := newFake(ctx, t)
+	defer cleanup()
+
+	top := mustCreateTopic(ctx, t, pclient, &pb.Topic{Name: "projects/P/topics/T"})
+	sub := mustCreateSubscription(ctx, t, sclient, &pb.Subscription{
+		Name:               "projects/P/subscriptions/S",
+		Topic:              top.Name,
+		AckDeadlineSeconds: 10,
+	})
+
+	publish(t, pclient, top, []*pb.PubsubMessage{
+		{Data: []byte("d1")},
+		{Data: []byte("d2")},
+		{Data: []byte("d3")},
+	})
+	msgs := streamingPullN(ctx, t, 3, sclient, sub)
+	var ackIDs []string
+	for _, m := range msgs {
+		ackIDs = append(ackIDs, m.AckId)
+	}
+
+	// Try to access m.Modacks while simultaneously calling ModifyAckDeadline
+	// so as to try and create a race condition.
+	// Invoke ModifyAckDeadline from the server rather than the client
+	// to increase replicability of simultaneous data access.
+	go func() {
+		req := &pb.ModifyAckDeadlineRequest{
+			Subscription:       sub.Name,
+			AckIds:             ackIDs,
+			AckDeadlineSeconds: 0,
+		}
+		server.GServer.ModifyAckDeadline(ctx, req)
+	}()
+
+	sm := server.Messages()
+	for _, m := range sm {
+		t.Logf("got modacks: %v\n", m.Modacks)
+	}
+}
+
 func TestUpdateDeadLetterPolicy(t *testing.T) {
 	pclient, sclient, _, cleanup := newFake(context.TODO(), t)
 	defer cleanup()