From 8996506d43383bee05afa954d76d81c9d8eaeb98 Mon Sep 17 00:00:00 2001 From: Jon Ayers Date: Tue, 7 Jul 2026 00:25:42 -0500 Subject: [PATCH] fix: use a unique channel name per latency measurement (#27040) --- coderd/database/pubsub/latency.go | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/coderd/database/pubsub/latency.go b/coderd/database/pubsub/latency.go index b8c14eec4f..4ab9dc5335 100644 --- a/coderd/database/pubsub/latency.go +++ b/coderd/database/pubsub/latency.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "sync/atomic" "time" "github.com/google/uuid" @@ -18,6 +19,10 @@ type LatencyMeasurer struct { // Create unique pubsub channel names so that multiple coderd replicas do not clash when performing latency measurements. channel uuid.UUID logger slog.Logger + // seq distinguishes consecutive measurements from each other so that a + // subscription whose teardown is still in flight cannot receive (and + // count) the next measurement's message. + seq atomic.Int64 } // LatencyMessageLength is the length of a UUIDv4 encoded to hex. @@ -40,7 +45,8 @@ func (lm *LatencyMeasurer) Measure(ctx context.Context, p Pubsub) (send, recv ti msg := []byte(uuid.New().String()) lm.logger.Debug(ctx, "performing measurement", slog.F("msg", msg)) - cancel, err := p.Subscribe(lm.latencyChannelName(), func(ctx context.Context, in []byte) { + channel := lm.nextChannelName() + cancel, err := p.Subscribe(channel, func(ctx context.Context, in []byte) { if !bytes.Equal(in, msg) { lm.logger.Warn(ctx, "received unexpected message", slog.F("got", in), slog.F("expected", msg)) return @@ -54,7 +60,7 @@ func (lm *LatencyMeasurer) Measure(ctx context.Context, p Pubsub) (send, recv ti defer cancel() start = time.Now() - err = p.Publish(lm.latencyChannelName(), msg) + err = p.Publish(channel, msg) if err != nil { return -1, -1, xerrors.Errorf("failed to publish: %w", err) } @@ -69,6 +75,12 @@ func (lm *LatencyMeasurer) Measure(ctx context.Context, p Pubsub) (send, recv ti } } -func (lm *LatencyMeasurer) latencyChannelName() string { - return fmt.Sprintf("latency-measure:%s", lm.channel) +// nextChannelName returns a channel name unique to this measurement. +// Uniqueness across replicas comes from the channel UUID; uniqueness +// across consecutive measurements of the same replica comes from the +// sequence number. The name must stay within Postgres's 63-byte +// identifier limit: 16 (prefix) + 36 (UUID) + 1 (dot) leaves 10 digits +// for the sequence. +func (lm *LatencyMeasurer) nextChannelName() string { + return fmt.Sprintf("latency-measure:%s.%d", lm.channel, lm.seq.Add(1)) }