pubsub: Publish latency spikes in certain messages (low throughput publisher)

[agis]

Client

cloud.google.com/go/pubsub v1.44.0

Environment

Cloud Run
go 1.23.4

Code and Dependencies

var (
  wg             sync.WaitGroup
  publishResults sync.Map
)

for _, job := range jobs {
  result := pubsub.Publish(ctx, topic, job)

  wg.Add(1)
  go func(res *pubsub.PublishResult) {
    defer wg.Done()

    publishStart := d.clock.Now().UTC()

    err := res.Get(ctx)
    if err != nil {
      // report to external monitoring system
      return
    }

    publishResults.Store(job.ID, true)

    _ = statsd.Distribution("publish_latency", time.Now().UTC().Sub(publishStart).Seconds(), 1)
  }(result)
}

wg.Wait()

Expected behavior

We expect all messages to be published within 100ms.

Actual behavior

We have a few outliers: while most messages are published within 20ms, a few of them take up to 5s.

Screenshots

The following are traces from Datadog, gathered via https://github.com/DataDog/dd-trace-go/tree/main/contrib/cloud.google.com/go/pubsub.v1:

This is consistent with what GCP's topic.send_request_latencies.avg reports, pretty much the same time:

Additional context

• We use the default publish settings
• Our typical batch size is 2-10 (low-throughput use case)
• Seems similar to pubsub: so slow when publish message #1961

[hongalex]

Could you show more of the code, specifically on how the topic object is passed around? Historically an issue of higher latency and Cloud Run (or Cloud Functions) is that the object isn't properly cached.

[agis]

We have a thin client wrapper, which keeps a reference to the created topic. We use that same reference throughout the code, and we initialize the client once in our main, so I believe we do use it as intented.

import (
	"context"
	"encoding/json"
	"errors"

	"cloud.google.com/go/pubsub"
	pubsubtrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/cloud.google.com/go/pubsub.v1"
)


type Client struct {
	pubsubClient *pubsub.Client
	topic        *pubsub.Topic
}

func NewClient(ctx context.Context, topicName string) *Client {
	client := &Client{}
        client.pubsubClient = createPubsubClient(ctx)

	client.topic = client.pubsubClient.Topic(topicName)
	client.topic.PublishSettings = pubsub.PublishSettings{
		ByteThreshold:  DefaultByteThreshold,
		CountThreshold: DefaultCountThreshold,
	}
	
	return client
}

func (c *Client) PublishAsync(ctx context.Context, message Message) *AsyncPublishResult {
	messageJSON, err := json.Marshal(message)
	if err != nil {
		return nil, err
	}

	msg := &pubsub.Message{
		Data:       messageJSON,
		Attributes: message.Attributes(),
	}

	// Publish the message
	result := pubsubtrace.Publish(ctx, c.topic, msg)
	return &AsyncPublishResult{
		result: result,
	}
}

type AsyncPublishResult struct {
	err    error
	result *pubsubtrace.PublishResult
}

func (r *AsyncPublishResult) Get(ctx context.Context) error {
	if r.err != nil {
		return r.err
	}
	if r.result != nil {
		_, err := r.result.Get(ctx)
		if err != nil {
			return err
		}
	}
	return nil
}

Note that it's the call to result.Get that is slow.