Data race on the shared duration in Google Pub/Sub Subscribe

Follow-up from the review of #4055. That PR fixes the subStarted/receiveChan map race from #4054 and is correct; this is a separate, pre-existing race in the same function that it deliberately left in scope for later.

Describe the bug

googleClient.Subscribe shares one time.Duration with the long-lived receiver goroutine, and subscription.Receive invokes the callback on several goroutines. Two concurrent processMessage invocations then write the same variable with nothing ordering them, which is a write/write data race.

On development today:

processMessage := func(ctx context.Context, msg *gcPubSub.Message) {
    m := pubsub.NewMessage(ctx)
    end = time.Since(start)   // pkg/gofr/datasource/pubsub/google/google.go:186
    ...
}

After #4055 the same statement is *end = time.Since(start) at google.go:249, reached through the end *time.Duration parameter on startTopicSubscriber. The refactor preserved the original closure's behaviour exactly, so this is neither introduced nor worsened by that PR.

It is worth being precise about one thing, because the PR thread records the opposite: the channel does give a happens-before edge between the write and the Subscribe call that reads end, but it orders nothing between two processMessage invocations. That is where the race is.

To Reproduce

A single draining goroutine is enough — this is exactly gofr's own model, one goroutine per topic in startSubscriptions. Publish a batch first so several messages are in flight, then drain:

topicObj, _ := client.CreateTopic(ctx, topic)

for i := 0; i < 60; i++ {
    topicObj.Publish(ctx, &gcPubSub.Message{Data: []byte("m")})
}

go func() {                       // one drainer, as gofr has
    for {
        m, err := g.Subscribe(ctx, topic)
        if err != nil || m == nil {
            return
        }
    }
}()

Run with -race. Measured over five runs of that probe:

WARNING: DATA RACE
Write at 0x... by goroutine 116:
  ...google.go:249
Previous write at 0x... by goroutine 114:
  ...google.go:249

The existing package tests do not surface it, and CI does not pass -race anywhere in go.yml, which is why it has gone unnoticed.

Expected behavior

No data race, and the Time value in the SUB debug log should describe the message actually being logged.

Both are wrong today for a second reason worth fixing in the same change: only the first Subscribe caller's variable ever reaches the receiver goroutine, so every later call on that topic logs Time: 0 regardless.

Suggested fix

Stop sharing the duration. Measure it per Subscribe call, which removes the shared variable entirely and also makes the number meaningful — how long that call waited for a message — instead of "time since the subscriber was created":

 func (g *googleClient) Subscribe(ctx context.Context, topic string) (*pubsub.Message, error) {
-	var end time.Duration
-
 	if g.client == nil {
 		return nil, nil
 	}
@@
-	receiveChan, err := g.startTopicSubscriber(ctx, topic, &end)
+	start := time.Now()
+
+	receiveChan, err := g.startTopicSubscriber(ctx, topic)
 	if err != nil {
 		return nil, err
 	}
 
 	select {
 	case m := <-receiveChan:
+		end := time.Since(start)
+
 		// Create span with links to producer span from message attributes
@@
-func (g *googleClient) startTopicSubscriber(ctx context.Context, topic string, end *time.Duration) (chan *pubsub.Message, error) {
+func (g *googleClient) startTopicSubscriber(ctx context.Context, topic string) (chan *pubsub.Message, error) {
@@
-	start := time.Now()
-
 	receiveChan = make(chan *pubsub.Message)
@@
 	processMessage := func(ctx context.Context, msg *gcPubSub.Message) {
 		m := pubsub.NewMessage(ctx)
-		*end = time.Since(start)
 
 		m.Topic = topic

Verified on top of #4055: the probe above is clean over five runs, the full package passes under -race, and gofmt/go vet are clean. Net six insertions, seven deletions.

A regression test needs -race to mean anything, so it is only useful alongside a decision about running the suite with -race in CI, which is a separate question.

On the Close panic raised in the #4055 thread — it is not reachable, and should not be filed

The review thread flagged that Close closes receiveChan while processMessage may still be in its select-send, and agreed to open a follow-up. Having gone looking for it, that panic cannot happen, so this issue deliberately does not cover it:

func (g *googleClient) Close() error {
	if g.client != nil {
		return g.client.Close()   // always taken in a real deployment
	}

	g.mu.Lock()
	defer g.mu.Unlock()

	for _, c := range g.receiveChan {
		close(c)
	}

	return nil
}

The closing loop only runs when g.client == nil, and nothing sets client back to nil after construction — it is only ever assigned in New and in retryConnect. In that state receiveChan is also always empty, because Subscribe returns at its if g.client == nil guard before any channel is created. The two tests that cover the loop are named TestClose_ClientNil and TestClose_MultipleReceiveChans_ClientNil and construct a state that cannot arise at runtime.

Confirmed by running it on the real connected path: after Close() the receive channel is still open and blocking, so nothing is ever closed out from under a pending send. Shutdown is clean anyway — client.Close() terminates the receivers, with pubsub.(*Subscription).Receive goroutines going from 13 to 0.

The only real observation left there is that the close(c) loop is dead code on the live path. That is a tidy-up, not a bug, and not worth a separate issue.

Environments

• OS: macOS (darwin/arm64)
• gofr version: development, and #4055 head
• go version: 1.26.3