opensource/coder
3
0
Fork 0
mirror of https://github.com/coder/coder.git synced 2026-10-29 17:44:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: use a unique channel name per latency measurement (#27040)

Browse Source
This commit is contained in:
Jon Ayers
2026-07-07 00:25:42 -05:00
committed by GitHub
parent 8b60d1d877
commit 8996506d43
1 changed files with 16 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coderd/database/pubsub/latency.go
+16 -4
View File
@@ -4,6 +4,7 @@ import (
"bytes" "bytes"
"context" "context"
"fmt" "fmt"
"sync/atomic"
"time" "time"
"github.com/google/uuid" "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. // Create unique pubsub channel names so that multiple coderd replicas do not clash when performing latency measurements.
channel uuid.UUID channel uuid.UUID
logger slog.Logger 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. // 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()) msg := []byte(uuid.New().String())
lm.logger.Debug(ctx, "performing measurement", slog.F("msg", msg)) 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) { if !bytes.Equal(in, msg) {
lm.logger.Warn(ctx, "received unexpected message", slog.F("got", in), slog.F("expected", msg)) lm.logger.Warn(ctx, "received unexpected message", slog.F("got", in), slog.F("expected", msg))
return return
@@ -54,7 +60,7 @@ func (lm *LatencyMeasurer) Measure(ctx context.Context, p Pubsub) (send, recv ti
defer cancel() defer cancel()
start = time.Now() start = time.Now()
err = p.Publish(lm.latencyChannelName(), msg) err = p.Publish(channel, msg)
if err != nil { if err != nil {
return -1, -1, xerrors.Errorf("failed to publish: %w", err) 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 { // nextChannelName returns a channel name unique to this measurement.
return fmt.Sprintf("latency-measure:%s", lm.channel) // 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))
} }