From 7c48ee256b054ad1b95bcebb232814604a92298e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Kutryj?= Date: Wed, 2 Sep 2026 23:53:21 +0200 Subject: [PATCH 1/7] fix(velocity): reuse one AMQP connection in the pending metrics emitter tackle.PublishMessage dials, handshakes and closes a connection per message. The emitter publishes every pending metric on each run with 20 workers, so a run with thousands of metrics opened thousands of short-lived connections; on a node behind a shared NAT the ports linger in TIME_WAIT and run out, the next dials time out, and the emitter keeps retrying into the same wall. Keep one publisher (connection + channel + exchange declared once) for the emitter's lifetime, guarded by a mutex since the workers publish concurrently. On a publish error reconnect once and retry; if that fails, drop the publisher so the next call reconnects. Co-Authored-By: Claude Code --- ee/velocity/pkg/emitter/pending_metrics.go | 62 +++++++++++++++++++++- 1 file changed, 60 insertions(+), 2 deletions(-) diff --git a/ee/velocity/pkg/emitter/pending_metrics.go b/ee/velocity/pkg/emitter/pending_metrics.go index 3dca1943b..5359cbccf 100644 --- a/ee/velocity/pkg/emitter/pending_metrics.go +++ b/ee/velocity/pkg/emitter/pending_metrics.go @@ -24,6 +24,9 @@ type PendingMetricsEmitter struct { crontab string options tackle.Options projectHubClient *service.ProjectHubGrpcClient + + publisherMu sync.Mutex + publisher *tackle.Publisher } func NewPendingMetricsEmitter(options tackle.Options, projectHubServiceClient *service.ProjectHubGrpcClient, crontab string) *PendingMetricsEmitter { @@ -198,6 +201,9 @@ func (emitter *PendingMetricsEmitter) emit(pendingMetric entity.PendingMetric, o return nil } +// publishMessage sends on one long-lived AMQP connection shared by all emitter +// workers. Opening a connection per message (tackle.PublishMessage) exhausts the +// egress NAT's ports on the node when a run has thousands of pending metrics. func (emitter *PendingMetricsEmitter) publishMessage(message []byte) (err error) { params := &tackle.PublishParams{ Body: message, @@ -206,14 +212,66 @@ func (emitter *PendingMetricsEmitter) publishMessage(message []byte) (err error) Exchange: emitter.options.RemoteExchange, } - if err = tackle.PublishMessage(params); HasError(err) { - log.Printf("failed to publish message, %v", err) + emitter.publisherMu.Lock() + defer emitter.publisherMu.Unlock() + + if err = emitter.ensurePublisher(); HasError(err) { + log.Printf("failed to connect publisher, %v", err) return } + if err = emitter.publisher.Publish(params); HasError(err) { + emitter.resetPublisher() + + if err = emitter.ensurePublisher(); HasError(err) { + log.Printf("failed to reconnect publisher, %v", err) + return + } + + err = emitter.publisher.Publish(params) + } + + if HasError(err) { + log.Printf("failed to publish message, %v", err) + emitter.resetPublisher() + } + return } +func (emitter *PendingMetricsEmitter) ensurePublisher() error { + if emitter.publisher != nil { + return nil + } + + publisher, err := tackle.NewPublisher(emitter.options.URL) + if err != nil { + return err + } + + publisher.SetConnectionName("velocity-pending-metrics-emitter") + + if err = publisher.Connect(); err != nil { + publisher.Close() + return err + } + + if err = publisher.ExchangeDeclare(emitter.options.RemoteExchange); err != nil { + publisher.Close() + return err + } + + emitter.publisher = publisher + return nil +} + +func (emitter *PendingMetricsEmitter) resetPublisher() { + if emitter.publisher != nil { + emitter.publisher.Close() + emitter.publisher = nil + } +} + func (emitter *PendingMetricsEmitter) buildMessage(pendingMetric entity.PendingMetric, orgID string, branchName string) (message []byte, err error) { if len(orgID) == 0 { log.Printf("organizationId is empty, projectId: %s", pendingMetric.ProjectId.String()) From 25656990a33a17c4c2ad8f09270d61c933b04486 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Kutryj?= Date: Thu, 3 Sep 2026 00:01:25 +0200 Subject: [PATCH 2/7] fix(velocity): drop narrative comment from publishMessage The rationale lives in the previous commit message and the PR body. Co-Authored-By: Claude Code --- ee/velocity/pkg/emitter/pending_metrics.go | 3 --- 1 file changed, 3 deletions(-) diff --git a/ee/velocity/pkg/emitter/pending_metrics.go b/ee/velocity/pkg/emitter/pending_metrics.go index 5359cbccf..9919ca268 100644 --- a/ee/velocity/pkg/emitter/pending_metrics.go +++ b/ee/velocity/pkg/emitter/pending_metrics.go @@ -201,9 +201,6 @@ func (emitter *PendingMetricsEmitter) emit(pendingMetric entity.PendingMetric, o return nil } -// publishMessage sends on one long-lived AMQP connection shared by all emitter -// workers. Opening a connection per message (tackle.PublishMessage) exhausts the -// egress NAT's ports on the node when a run has thousands of pending metrics. func (emitter *PendingMetricsEmitter) publishMessage(message []byte) (err error) { params := &tackle.PublishParams{ Body: message, From 2908972a6bcdc16433f811eb902f1a354a8c1e48 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Kutryj?= Date: Thu, 3 Sep 2026 13:21:01 +0200 Subject: [PATCH 3/7] fix(velocity): reuse a bounded, race-safe go-tackle publisher MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the hand-rolled persistent publisher with go-tackle's own Publisher, addressing the review on semaphoreio/semaphore#1213. - Bump go-tackle to v0.0.0-20231226193542-c913a4af4f94 (matches self_hosted_hub). Its Publisher dials with a bounded 5s timeout (net.DialTimeout + handshake deadline) and opens a fresh channel per publish over one shared connection, so we drop the hand-rolled ensurePublisher/resetPublisher and the amqp091 default 30s+30s dial. - Connect once at the start of each tick and Close() at the end, instead of holding a connection idle for ~24h between daily ticks. One connection per run: no Cloud NAT idle-reap, no orphan-on-rollout, and the unpinned heartbeat becomes irrelevant. The per-publish channel close-ok plus the end-of-tick connection close restore a flush barrier, so a mid-flight connection death surfaces as an error rather than a silent publish_message{success}. - Publish via PublishWithContext with a bounded 15s ctx instead of Publish()'s context.Background(), which would otherwise retry every 1s forever against a down broker. Keep a mutex, but scoped to only the PublishWithContext call. go-tackle's shared Publisher has a data race on its reconnect path: reconnect() reassigns p.connectOnce = sync.Once{} under reconnectionLock (publisher.go:232) while getConnection() reads that same Once via .Do() (publisher.go:179) without the lock. The emitter is the first caller to fan multiple goroutines at one shared Publisher, so it is the first to hit it; -race flags it deterministically once the broker drops mid-tick. The mutex serialises only the fast channel-open/publish/channel-close — the dial happens once per tick in openPublisher(), outside the lock, and describeProject()'s gRPC calls stay parallel across the worker pool. A go-tackle fix for the connectOnce race can let us drop the mutex later. Co-Authored-By: Claude Opus 4.8 (1M context) --- ee/velocity/go.mod | 2 +- ee/velocity/go.sum | 4 +- ee/velocity/pkg/emitter/pending_metrics.go | 86 ++++++++++------------ 3 files changed, 41 insertions(+), 51 deletions(-) diff --git a/ee/velocity/go.mod b/ee/velocity/go.mod index 4d99339ef..da5192078 100644 --- a/ee/velocity/go.mod +++ b/ee/velocity/go.mod @@ -12,7 +12,7 @@ require ( github.com/google/go-cmp v0.7.0 github.com/google/uuid v1.6.0 github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 - github.com/renderedtext/go-tackle v0.0.0-20231218124313-26ee274af69d + github.com/renderedtext/go-tackle v0.0.0-20231226193542-c913a4af4f94 github.com/renderedtext/go-watchman v0.0.0-20221222100224-451a6f3c8d92 github.com/samber/lo v1.38.1 github.com/semaphoreci/test-results v0.6.10-0.20231114151005-04b9d9ca32c9 diff --git a/ee/velocity/go.sum b/ee/velocity/go.sum index 35f3320a1..ee8be5461 100644 --- a/ee/velocity/go.sum +++ b/ee/velocity/go.sum @@ -241,8 +241,8 @@ github.com/prometheus/procfs v0.8.0 h1:ODq8ZFEaYeCaZOJlZZdJA2AbQR98dSHSM1KW/You5 github.com/prometheus/procfs v0.8.0/go.mod h1:z7EfXMXOkbkqb9IINtpCn86r/to3BnA0uaxHdg830/4= github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA= github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= -github.com/renderedtext/go-tackle v0.0.0-20231218124313-26ee274af69d h1:YtBUCNI/kgrJHf8lz/RE9Od8Agoqwujf0W2G68Ee1XM= -github.com/renderedtext/go-tackle v0.0.0-20231218124313-26ee274af69d/go.mod h1:IfWH6x6erQ2Y4C7+BdP/fzOCN+8Szs2atOc6vUDOnvY= +github.com/renderedtext/go-tackle v0.0.0-20231226193542-c913a4af4f94 h1:XynJJlfKWESMTlCM1fc7LDlPiQTvOPrRDQTiX6nyQiY= +github.com/renderedtext/go-tackle v0.0.0-20231226193542-c913a4af4f94/go.mod h1:IfWH6x6erQ2Y4C7+BdP/fzOCN+8Szs2atOc6vUDOnvY= github.com/renderedtext/go-watchman v0.0.0-20221222100224-451a6f3c8d92 h1:OmDghaSHy96nHV+ZnXBKQnXBLvuSQNdFZRYIQiDDXsg= github.com/renderedtext/go-watchman v0.0.0-20221222100224-451a6f3c8d92/go.mod h1:Z+qanDzSoUGCbcrTM7G6YCA9ST2KBdte7sCz+HQAp7I= github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= diff --git a/ee/velocity/pkg/emitter/pending_metrics.go b/ee/velocity/pkg/emitter/pending_metrics.go index 9919ca268..950ec9b95 100644 --- a/ee/velocity/pkg/emitter/pending_metrics.go +++ b/ee/velocity/pkg/emitter/pending_metrics.go @@ -2,6 +2,7 @@ package emitter import ( + "context" "fmt" "log" "sync" @@ -19,14 +20,21 @@ import ( "google.golang.org/protobuf/types/known/timestamppb" ) +const ( + defaultConnectionTimeout = 5 * time.Second + defaultPublishTimeout = 15 * time.Second +) + type PendingMetricsEmitter struct { Name string crontab string options tackle.Options projectHubClient *service.ProjectHubGrpcClient - publisherMu sync.Mutex - publisher *tackle.Publisher + publisherOptions tackle.PublisherOptions + publishTimeout time.Duration + publisher *tackle.Publisher + publisherMu sync.Mutex } func NewPendingMetricsEmitter(options tackle.Options, projectHubServiceClient *service.ProjectHubGrpcClient, crontab string) *PendingMetricsEmitter { @@ -35,6 +43,11 @@ func NewPendingMetricsEmitter(options tackle.Options, projectHubServiceClient *s crontab: crontab, options: options, projectHubClient: projectHubServiceClient, + publisherOptions: tackle.PublisherOptions{ + ConnectionName: options.ConnectionName, + ConnectionTimeout: defaultConnectionTimeout, + }, + publishTimeout: defaultPublishTimeout, } } @@ -68,6 +81,12 @@ func (emitter *PendingMetricsEmitter) PublishPendingMetrics() (err error) { CleanDatabase() log.Println(`Finished database cleanup`) + if err = emitter.openPublisher(); HasError(err) { + log.Printf("failed to connect publisher, %v", err) + return err + } + defer emitter.closePublisher() + wg := new(sync.WaitGroup) workerCount := 20 @@ -201,58 +220,27 @@ func (emitter *PendingMetricsEmitter) emit(pendingMetric entity.PendingMetric, o return nil } -func (emitter *PendingMetricsEmitter) publishMessage(message []byte) (err error) { - params := &tackle.PublishParams{ - Body: message, - AmqpURL: emitter.options.URL, - RoutingKey: emitter.options.RoutingKey, - Exchange: emitter.options.RemoteExchange, - } +func (emitter *PendingMetricsEmitter) publishMessage(message []byte) error { + ctx, cancel := context.WithTimeout(context.Background(), emitter.publishTimeout) + defer cancel() emitter.publisherMu.Lock() defer emitter.publisherMu.Unlock() - if err = emitter.ensurePublisher(); HasError(err) { - log.Printf("failed to connect publisher, %v", err) - return - } - - if err = emitter.publisher.Publish(params); HasError(err) { - emitter.resetPublisher() - - if err = emitter.ensurePublisher(); HasError(err) { - log.Printf("failed to reconnect publisher, %v", err) - return - } - - err = emitter.publisher.Publish(params) - } - - if HasError(err) { - log.Printf("failed to publish message, %v", err) - emitter.resetPublisher() - } - - return + return emitter.publisher.PublishWithContext(ctx, &tackle.PublishParams{ + Body: message, + AmqpURL: emitter.options.URL, + RoutingKey: emitter.options.RoutingKey, + Exchange: emitter.options.RemoteExchange, + }) } -func (emitter *PendingMetricsEmitter) ensurePublisher() error { - if emitter.publisher != nil { - return nil - } - - publisher, err := tackle.NewPublisher(emitter.options.URL) +func (emitter *PendingMetricsEmitter) openPublisher() error { + publisher, err := tackle.NewPublisher(emitter.options.URL, emitter.publisherOptions) if err != nil { return err } - publisher.SetConnectionName("velocity-pending-metrics-emitter") - - if err = publisher.Connect(); err != nil { - publisher.Close() - return err - } - if err = publisher.ExchangeDeclare(emitter.options.RemoteExchange); err != nil { publisher.Close() return err @@ -262,11 +250,13 @@ func (emitter *PendingMetricsEmitter) ensurePublisher() error { return nil } -func (emitter *PendingMetricsEmitter) resetPublisher() { - if emitter.publisher != nil { - emitter.publisher.Close() - emitter.publisher = nil +func (emitter *PendingMetricsEmitter) closePublisher() { + if emitter.publisher == nil { + return } + + emitter.publisher.Close() + emitter.publisher = nil } func (emitter *PendingMetricsEmitter) buildMessage(pendingMetric entity.PendingMetric, orgID string, branchName string) (message []byte, err error) { From 7e368bbfc75729e2b1f980c949e28e53203d8ad4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Kutryj?= Date: Thu, 3 Sep 2026 13:21:13 +0200 Subject: [PATCH 4/7] test(velocity): cover the pending-metrics emitter publisher Add pkg/emitter/pending_metrics_test.go, driving the publisher through tackle's PublisherOptions.ConnectFunc hook against the compose RabbitMQ: - TestPublisherDialsOncePerTick: 200 concurrent publishes over one connection dial exactly once. - TestPublisherReconnectsAfterConnectionDrop: a dropped connection is re-dialled and the retry succeeds. - TestPublisherClosesConnectionAtEndOfTick: closePublisher() actually closes the underlying connection. - TestOpenPublisherFailsFastWhenBrokerIsUnreachable: a blackholed broker aborts within the bounded connection timeout, not amqp091's 30s default. - TestPublishStaysBoundedWhenBrokerDiesMidTick: 20 concurrent publishes against a broker that dies mid-tick return within a bounded wall-clock instead of N x timeout. The suite passes under CI's make test (-p 1) and, with the scoped publish mutex, is clean under -race (run natively; the amd64 image's ThreadSanitizer aborts under emulation). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../pkg/emitter/pending_metrics_test.go | 168 ++++++++++++++++++ 1 file changed, 168 insertions(+) create mode 100644 ee/velocity/pkg/emitter/pending_metrics_test.go diff --git a/ee/velocity/pkg/emitter/pending_metrics_test.go b/ee/velocity/pkg/emitter/pending_metrics_test.go new file mode 100644 index 000000000..bb4c28e6e --- /dev/null +++ b/ee/velocity/pkg/emitter/pending_metrics_test.go @@ -0,0 +1,168 @@ +package emitter + +import ( + "errors" + "os" + "sync" + "sync/atomic" + "testing" + "time" + + rabbit "github.com/rabbitmq/amqp091-go" + "github.com/renderedtext/go-tackle" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var errBrokerDown = errors.New("broker is down") + +func testEmitterOptions() tackle.Options { + return tackle.Options{ + URL: os.Getenv("RABBITMQ_URL"), + ConnectionName: "velocity.pending_metrics_emitter.test", + RemoteExchange: "velocity_emitter_test_exchange", + RoutingKey: "done", + } +} + +func testEmitter(connect func() (*rabbit.Connection, error)) *PendingMetricsEmitter { + emitter := NewPendingMetricsEmitter(testEmitterOptions(), nil, "0 8 * * *") + emitter.publisherOptions.ConnectFunc = connect + return emitter +} + +func TestPublisherDialsOncePerTick(t *testing.T) { + var dials int32 + + emitter := testEmitter(func() (*rabbit.Connection, error) { + atomic.AddInt32(&dials, 1) + return rabbit.Dial(testEmitterOptions().URL) + }) + + require.NoError(t, emitter.openPublisher()) + defer emitter.closePublisher() + + wg := new(sync.WaitGroup) + errs := make(chan error, 200) + + for i := 0; i < 200; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if err := emitter.publishMessage([]byte("pending metric")); err != nil { + errs <- err + } + }() + } + + wg.Wait() + close(errs) + + for err := range errs { + t.Errorf("publish failed: %v", err) + } + + assert.Equal(t, int32(1), atomic.LoadInt32(&dials)) +} + +func TestPublisherReconnectsAfterConnectionDrop(t *testing.T) { + conns := make([]*rabbit.Connection, 0, 2) + + emitter := testEmitter(func() (*rabbit.Connection, error) { + conn, err := rabbit.Dial(testEmitterOptions().URL) + if err != nil { + return nil, err + } + + conns = append(conns, conn) + return conn, nil + }) + + require.NoError(t, emitter.openPublisher()) + defer emitter.closePublisher() + + require.NoError(t, emitter.publishMessage([]byte("before the drop"))) + require.Len(t, conns, 1) + + require.NoError(t, conns[0].Close()) + + require.NoError(t, emitter.publishMessage([]byte("after the drop"))) + assert.Len(t, conns, 2) +} + +func TestPublisherClosesConnectionAtEndOfTick(t *testing.T) { + var conn *rabbit.Connection + + emitter := testEmitter(func() (*rabbit.Connection, error) { + var err error + conn, err = rabbit.Dial(testEmitterOptions().URL) + return conn, err + }) + + require.NoError(t, emitter.openPublisher()) + require.NoError(t, emitter.publishMessage([]byte("pending metric"))) + + emitter.closePublisher() + + assert.Nil(t, emitter.publisher) + assert.True(t, conn.IsClosed()) +} + +func TestOpenPublisherFailsFastWhenBrokerIsUnreachable(t *testing.T) { + options := testEmitterOptions() + options.URL = "amqp://guest:guest@10.255.255.1:5672" + + emitter := NewPendingMetricsEmitter(options, nil, "0 8 * * *") + require.Equal(t, defaultConnectionTimeout, emitter.publisherOptions.ConnectionTimeout) + emitter.publisherOptions.ConnectionTimeout = 300 * time.Millisecond + + start := time.Now() + err := emitter.openPublisher() + elapsed := time.Since(start) + + require.Error(t, err) + assert.Nil(t, emitter.publisher) + assert.Less(t, elapsed, 3*time.Second) +} + +func TestPublishStaysBoundedWhenBrokerDiesMidTick(t *testing.T) { + var dials int32 + var conn *rabbit.Connection + + emitter := testEmitter(func() (*rabbit.Connection, error) { + if atomic.AddInt32(&dials, 1) > 1 { + return nil, errBrokerDown + } + + var err error + conn, err = rabbit.Dial(testEmitterOptions().URL) + return conn, err + }) + emitter.publishTimeout = time.Second + + require.NoError(t, emitter.openPublisher()) + defer emitter.closePublisher() + require.NoError(t, conn.Close()) + + wg := new(sync.WaitGroup) + failures := make(chan error, 20) + + start := time.Now() + for i := 0; i < 20; i++ { + wg.Add(1) + go func() { + defer wg.Done() + failures <- emitter.publishMessage([]byte("pending metric")) + }() + } + + wg.Wait() + close(failures) + elapsed := time.Since(start) + + for err := range failures { + assert.Error(t, err) + } + + assert.Less(t, elapsed, 5*time.Second) +} From f3092926b1e3cde6f550bd0db64c421f9c0c9d88 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Kutryj?= Date: Thu, 3 Sep 2026 13:44:07 +0200 Subject: [PATCH 5/7] test(velocity): harden emitter publisher tests after review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Applies the agreed points from a 4-reviewer pass on semaphoreio/semaphore#1213: - Skip the broker-backed tests when RABBITMQ_URL is unset instead of hard-failing, so `go test ./...` on a host without the compose stack skips rather than reporting spurious failures. - Add TestConcurrentReconnectAfterLiveDropIsRaceFree: establish a live connection, drop it, then fan 50 concurrent publishes that must all reconnect. Verified load-bearing — with the publish mutex removed it trips -race in go-tackle's reconnectAndPublish (the connectOnce reset outside reconnectionLock); with the mutex it is clean and re-dials exactly once. The prior suite could pass with the mutex removed, so it did not guard the headline fix; this test does. Co-Authored-By: Claude Opus 4.8 (1M context) --- ee/velocity/go.mod | 2 +- .../pkg/emitter/pending_metrics_test.go | 68 +++++++++++++++++++ 2 files changed, 69 insertions(+), 1 deletion(-) diff --git a/ee/velocity/go.mod b/ee/velocity/go.mod index da5192078..15e2461b6 100644 --- a/ee/velocity/go.mod +++ b/ee/velocity/go.mod @@ -12,6 +12,7 @@ require ( github.com/google/go-cmp v0.7.0 github.com/google/uuid v1.6.0 github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 + github.com/rabbitmq/amqp091-go v1.13.0 github.com/renderedtext/go-tackle v0.0.0-20231226193542-c913a4af4f94 github.com/renderedtext/go-watchman v0.0.0-20221222100224-451a6f3c8d92 github.com/samber/lo v1.38.1 @@ -47,7 +48,6 @@ require ( github.com/prometheus/client_model v0.3.0 // indirect github.com/prometheus/common v0.37.0 // indirect github.com/prometheus/procfs v0.8.0 // indirect - github.com/rabbitmq/amqp091-go v1.13.0 // indirect github.com/robfig/cron/v3 v3.0.1 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect golang.org/x/arch v0.18.0 // indirect diff --git a/ee/velocity/pkg/emitter/pending_metrics_test.go b/ee/velocity/pkg/emitter/pending_metrics_test.go index bb4c28e6e..ad2d42593 100644 --- a/ee/velocity/pkg/emitter/pending_metrics_test.go +++ b/ee/velocity/pkg/emitter/pending_metrics_test.go @@ -31,7 +31,15 @@ func testEmitter(connect func() (*rabbit.Connection, error)) *PendingMetricsEmit return emitter } +func requireBroker(t *testing.T) { + if os.Getenv("RABBITMQ_URL") == "" { + t.Skip("RABBITMQ_URL not set; skipping broker integration test") + } +} + func TestPublisherDialsOncePerTick(t *testing.T) { + requireBroker(t) + var dials int32 emitter := testEmitter(func() (*rabbit.Connection, error) { @@ -66,6 +74,8 @@ func TestPublisherDialsOncePerTick(t *testing.T) { } func TestPublisherReconnectsAfterConnectionDrop(t *testing.T) { + requireBroker(t) + conns := make([]*rabbit.Connection, 0, 2) emitter := testEmitter(func() (*rabbit.Connection, error) { @@ -91,6 +101,8 @@ func TestPublisherReconnectsAfterConnectionDrop(t *testing.T) { } func TestPublisherClosesConnectionAtEndOfTick(t *testing.T) { + requireBroker(t) + var conn *rabbit.Connection emitter := testEmitter(func() (*rabbit.Connection, error) { @@ -126,6 +138,8 @@ func TestOpenPublisherFailsFastWhenBrokerIsUnreachable(t *testing.T) { } func TestPublishStaysBoundedWhenBrokerDiesMidTick(t *testing.T) { + requireBroker(t) + var dials int32 var conn *rabbit.Connection @@ -166,3 +180,57 @@ func TestPublishStaysBoundedWhenBrokerDiesMidTick(t *testing.T) { assert.Less(t, elapsed, 5*time.Second) } + +func TestConcurrentReconnectAfterLiveDropIsRaceFree(t *testing.T) { + requireBroker(t) + + var mu sync.Mutex + var conns []*rabbit.Connection + + emitter := testEmitter(func() (*rabbit.Connection, error) { + conn, err := rabbit.Dial(testEmitterOptions().URL) + if err != nil { + return nil, err + } + + mu.Lock() + conns = append(conns, conn) + mu.Unlock() + return conn, nil + }) + + require.NoError(t, emitter.openPublisher()) + defer emitter.closePublisher() + + require.NoError(t, emitter.publishMessage([]byte("establish connection"))) + mu.Lock() + require.Len(t, conns, 1) + first := conns[0] + mu.Unlock() + + require.NoError(t, first.Close()) + + wg := new(sync.WaitGroup) + errs := make(chan error, 50) + + for i := 0; i < 50; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if err := emitter.publishMessage([]byte("after live drop")); err != nil { + errs <- err + } + }() + } + + wg.Wait() + close(errs) + + for err := range errs { + t.Errorf("publish failed after live drop: %v", err) + } + + mu.Lock() + assert.Equal(t, 2, len(conns), "one initial dial plus exactly one reconnect") + mu.Unlock() +} From 70e402aca75356efc0d47291c89168183cfe76a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Kutryj?= Date: Wed, 23 Sep 2026 12:42:14 +0200 Subject: [PATCH 6/7] fix(velocity): drop the publish mutex, bump go-tackle past the reconnect race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit renderedtext/go-tackle#14 replaces the connectOnce-reset dance in the publisher's reconnect path with a mutex-guarded connection, so a shared Publisher is safe to fan multiple goroutines at. The app-level mutex here existed only to work around that race, so it goes — which also restores real concurrency across the emitter's 20 publish workers. Measured with -race against a live broker, mutex removed in both arms: the pre-#14 pin reports 7 data races and fails TestConcurrentReconnectAfterLiveDropIsRaceFree and TestPublishStaysBoundedWhenBrokerDiesMidTick; the post-#14 pin is clean. Full compose suite passes (236 tests). Co-Authored-By: Claude Opus 5 (1M context) --- ee/velocity/go.mod | 2 +- ee/velocity/go.sum | 4 ++-- ee/velocity/pkg/emitter/pending_metrics.go | 4 ---- 3 files changed, 3 insertions(+), 7 deletions(-) diff --git a/ee/velocity/go.mod b/ee/velocity/go.mod index 15e2461b6..13d707d72 100644 --- a/ee/velocity/go.mod +++ b/ee/velocity/go.mod @@ -13,7 +13,7 @@ require ( github.com/google/uuid v1.6.0 github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 github.com/rabbitmq/amqp091-go v1.13.0 - github.com/renderedtext/go-tackle v0.0.0-20231226193542-c913a4af4f94 + github.com/renderedtext/go-tackle v0.0.0-20260921080357-d04bdecbe4a8 github.com/renderedtext/go-watchman v0.0.0-20221222100224-451a6f3c8d92 github.com/samber/lo v1.38.1 github.com/semaphoreci/test-results v0.6.10-0.20231114151005-04b9d9ca32c9 diff --git a/ee/velocity/go.sum b/ee/velocity/go.sum index ee8be5461..24cf333ad 100644 --- a/ee/velocity/go.sum +++ b/ee/velocity/go.sum @@ -241,8 +241,8 @@ github.com/prometheus/procfs v0.8.0 h1:ODq8ZFEaYeCaZOJlZZdJA2AbQR98dSHSM1KW/You5 github.com/prometheus/procfs v0.8.0/go.mod h1:z7EfXMXOkbkqb9IINtpCn86r/to3BnA0uaxHdg830/4= github.com/rabbitmq/amqp091-go v1.13.0 h1:L8NA1WtF76C6KA3LAoufjfLgbist/If1UQYcsOjtxXA= github.com/rabbitmq/amqp091-go v1.13.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= -github.com/renderedtext/go-tackle v0.0.0-20231226193542-c913a4af4f94 h1:XynJJlfKWESMTlCM1fc7LDlPiQTvOPrRDQTiX6nyQiY= -github.com/renderedtext/go-tackle v0.0.0-20231226193542-c913a4af4f94/go.mod h1:IfWH6x6erQ2Y4C7+BdP/fzOCN+8Szs2atOc6vUDOnvY= +github.com/renderedtext/go-tackle v0.0.0-20260921080357-d04bdecbe4a8 h1:Tf7XF17DPdAJMZ0o08LRQd0aOhooNuIq1yNMUHaQ0mA= +github.com/renderedtext/go-tackle v0.0.0-20260921080357-d04bdecbe4a8/go.mod h1:S2Q09FoUIsc0TmOZL3yyTTTcytnpoqD+YXFJAodLqhU= github.com/renderedtext/go-watchman v0.0.0-20221222100224-451a6f3c8d92 h1:OmDghaSHy96nHV+ZnXBKQnXBLvuSQNdFZRYIQiDDXsg= github.com/renderedtext/go-watchman v0.0.0-20221222100224-451a6f3c8d92/go.mod h1:Z+qanDzSoUGCbcrTM7G6YCA9ST2KBdte7sCz+HQAp7I= github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= diff --git a/ee/velocity/pkg/emitter/pending_metrics.go b/ee/velocity/pkg/emitter/pending_metrics.go index 950ec9b95..e2f7a2650 100644 --- a/ee/velocity/pkg/emitter/pending_metrics.go +++ b/ee/velocity/pkg/emitter/pending_metrics.go @@ -34,7 +34,6 @@ type PendingMetricsEmitter struct { publisherOptions tackle.PublisherOptions publishTimeout time.Duration publisher *tackle.Publisher - publisherMu sync.Mutex } func NewPendingMetricsEmitter(options tackle.Options, projectHubServiceClient *service.ProjectHubGrpcClient, crontab string) *PendingMetricsEmitter { @@ -224,9 +223,6 @@ func (emitter *PendingMetricsEmitter) publishMessage(message []byte) error { ctx, cancel := context.WithTimeout(context.Background(), emitter.publishTimeout) defer cancel() - emitter.publisherMu.Lock() - defer emitter.publisherMu.Unlock() - return emitter.publisher.PublishWithContext(ctx, &tackle.PublishParams{ Body: message, AmqpURL: emitter.options.URL, From 0313e98bdf372f003cd49b8bfcd5bf6e07a26cf5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Kutryj?= Date: Wed, 23 Sep 2026 14:18:54 +0200 Subject: [PATCH 7/7] fix(velocity): reuse one AMQP connection in the summary processors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit job_summary_processor and pipeline_summary_processor published through tackle.PublishMessage, which builds a Publisher, publishes once and closes it — one AMQP connection per published message. Prod runs ~630k deliveries a day through these two consumers at a measured 0.75 publish ratio, so they open on the order of 475k connections a day against the egress NAT. Both now hold a shared publisher for the consumer's lifetime and publish through it under a bounded context, matching the emitter. No mutex is needed: the go-tackle bump in this branch carries renderedtext/go-tackle#14, which makes a shared Publisher's reconnect path race-free. Opening the publisher retries on the same schedule as the consumer, so a broker that is not up yet still waits rather than crash-looping. Covered by TestPublisherIsSharedAcrossDeliveries (100 concurrent publishes must share exactly one dial) and TestOpenPublisherDeclaresExchangeAndPublishes. Full compose suite passes (238 tests), clean under -race. Co-Authored-By: Claude Opus 5 (1M context) --- .../pkg/proc/summary/job_summary_processor.go | 18 +++- .../summary/pipeline_summary_processor.go | 18 +++- ee/velocity/pkg/proc/summary/publisher.go | 37 ++++++++ .../pkg/proc/summary/publisher_test.go | 91 +++++++++++++++++++ 4 files changed, 160 insertions(+), 4 deletions(-) create mode 100644 ee/velocity/pkg/proc/summary/publisher.go create mode 100644 ee/velocity/pkg/proc/summary/publisher_test.go diff --git a/ee/velocity/pkg/proc/summary/job_summary_processor.go b/ee/velocity/pkg/proc/summary/job_summary_processor.go index f0cb2ea61..832c40eec 100644 --- a/ee/velocity/pkg/proc/summary/job_summary_processor.go +++ b/ee/velocity/pkg/proc/summary/job_summary_processor.go @@ -30,6 +30,7 @@ type JobSummarySetupOptions struct { type JobSummaryProcessor struct { amqp tackle.Options + publisher *tackle.Publisher serverFarmClient service.ServerFarmClient projectHubClient service.ProjectHubClient reportFetcherClient service.ReportFetcherClient @@ -109,14 +110,27 @@ func (p *JobSummaryProcessor) Process(delivery tackle.Delivery) (err error) { Exchange: p.amqp.RemoteExchange, } - return tackle.PublishMessage(¶ms) + return publish(p.publisher, ¶ms) } func StartJobSummaryProcessor(o *JobSummarySetupOptions) { log.Println("starting job summary processor") + var publisher *tackle.Publisher + + err := retry.WithConstantWait("RabbitMQ connection", 20, 2*time.Second, func() error { + var err error + publisher, err = openPublisher(o.OutOptions) + return err + }) + + if err != nil { + log.Fatalf("failed to open publisher for job summary processor: %v", err) + } + processor := &JobSummaryProcessor{ amqp: o.OutOptions, + publisher: publisher, serverFarmClient: o.FarmClient, projectHubClient: o.ProjectClient, reportFetcherClient: o.ReportFetcherClient, @@ -124,7 +138,7 @@ func StartJobSummaryProcessor(o *JobSummarySetupOptions) { consumer := tackle.NewConsumer() - err := retry.WithConstantWait("RabbitMQ connection", 20, 2*time.Second, func() error { + err = retry.WithConstantWait("RabbitMQ connection", 20, 2*time.Second, func() error { return consumer.Start(&o.InOptions, processor.Process) }) diff --git a/ee/velocity/pkg/proc/summary/pipeline_summary_processor.go b/ee/velocity/pkg/proc/summary/pipeline_summary_processor.go index 2cd63b6f8..ed234bb6f 100644 --- a/ee/velocity/pkg/proc/summary/pipeline_summary_processor.go +++ b/ee/velocity/pkg/proc/summary/pipeline_summary_processor.go @@ -19,6 +19,7 @@ import ( type PipelineSummaryProcessor struct { amqp tackle.Options + publisher *tackle.Publisher plumberClient service.PlumberClient projectHubClient service.ProjectHubClient reportFetcherClient service.ReportFetcherClient @@ -112,7 +113,7 @@ func (p *PipelineSummaryProcessor) Process(delivery tackle.Delivery) (err error) Exchange: p.amqp.RemoteExchange, } - return tackle.PublishMessage(¶ms) + return publish(p.publisher, ¶ms) } // StartPipelineSummaryProcessor @@ -120,8 +121,21 @@ func StartPipelineSummaryProcessor(inOptions, outOptions tackle.Options, plumberClient service.PlumberClient, projectClient service.ProjectHubClient, reportFetcherClient service.ReportFetcherClient) { log.Println("starting pipeline summary processor") + var publisher *tackle.Publisher + + err := retry.WithConstantWait("RabbitMQ connection", 20, 2*time.Second, func() error { + var err error + publisher, err = openPublisher(outOptions) + return err + }) + + if err != nil { + log.Fatalf("failed to open publisher for pipeline summary processor: %v", err) + } + processor := &PipelineSummaryProcessor{ amqp: outOptions, + publisher: publisher, plumberClient: plumberClient, projectHubClient: projectClient, reportFetcherClient: reportFetcherClient, @@ -129,7 +143,7 @@ func StartPipelineSummaryProcessor(inOptions, outOptions tackle.Options, consumer := tackle.NewConsumer() - err := retry.WithConstantWait("RabbitMQ connection", 20, 2*time.Second, func() error { + err = retry.WithConstantWait("RabbitMQ connection", 20, 2*time.Second, func() error { return consumer.Start(&inOptions, processor.Process) }) diff --git a/ee/velocity/pkg/proc/summary/publisher.go b/ee/velocity/pkg/proc/summary/publisher.go new file mode 100644 index 000000000..4b0663603 --- /dev/null +++ b/ee/velocity/pkg/proc/summary/publisher.go @@ -0,0 +1,37 @@ +package summary + +import ( + "context" + "time" + + "github.com/renderedtext/go-tackle" +) + +const ( + publisherConnectionTimeout = 5 * time.Second + publishTimeout = 15 * time.Second +) + +func openPublisher(options tackle.Options) (*tackle.Publisher, error) { + publisher, err := tackle.NewPublisher(options.URL, tackle.PublisherOptions{ + ConnectionName: options.ConnectionName, + ConnectionTimeout: publisherConnectionTimeout, + }) + if err != nil { + return nil, err + } + + if err := publisher.ExchangeDeclare(options.RemoteExchange); err != nil { + publisher.Close() + return nil, err + } + + return publisher, nil +} + +func publish(publisher *tackle.Publisher, params *tackle.PublishParams) error { + ctx, cancel := context.WithTimeout(context.Background(), publishTimeout) + defer cancel() + + return publisher.PublishWithContext(ctx, params) +} diff --git a/ee/velocity/pkg/proc/summary/publisher_test.go b/ee/velocity/pkg/proc/summary/publisher_test.go new file mode 100644 index 000000000..6d58c9089 --- /dev/null +++ b/ee/velocity/pkg/proc/summary/publisher_test.go @@ -0,0 +1,91 @@ +package summary + +import ( + "os" + "sync" + "sync/atomic" + "testing" + + rabbit "github.com/rabbitmq/amqp091-go" + "github.com/renderedtext/go-tackle" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func testSummaryOptions() tackle.Options { + return tackle.Options{ + URL: os.Getenv("RABBITMQ_URL"), + ConnectionName: "velocity.summary_processor.test", + RemoteExchange: "velocity_summary_test_exchange", + RoutingKey: "done", + } +} + +func testPublishParams(options tackle.Options) *tackle.PublishParams { + return &tackle.PublishParams{ + Body: []byte("summary event"), + AmqpURL: options.URL, + RoutingKey: options.RoutingKey, + Exchange: options.RemoteExchange, + } +} + +func requireBroker(t *testing.T) { + if os.Getenv("RABBITMQ_URL") == "" { + t.Skip("RABBITMQ_URL not set; skipping broker integration test") + } +} + +func TestOpenPublisherDeclaresExchangeAndPublishes(t *testing.T) { + requireBroker(t) + + options := testSummaryOptions() + + publisher, err := openPublisher(options) + require.NoError(t, err) + defer publisher.Close() + + require.NoError(t, publish(publisher, testPublishParams(options))) +} + +func TestPublisherIsSharedAcrossDeliveries(t *testing.T) { + requireBroker(t) + + options := testSummaryOptions() + + var dials int32 + + publisher, err := tackle.NewPublisher(options.URL, tackle.PublisherOptions{ + ConnectionName: options.ConnectionName, + ConnectionTimeout: publisherConnectionTimeout, + ConnectFunc: func() (*rabbit.Connection, error) { + atomic.AddInt32(&dials, 1) + return rabbit.Dial(options.URL) + }, + }) + require.NoError(t, err) + defer publisher.Close() + require.NoError(t, publisher.ExchangeDeclare(options.RemoteExchange)) + + wg := new(sync.WaitGroup) + errs := make(chan error, 100) + + for i := 0; i < 100; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if err := publish(publisher, testPublishParams(options)); err != nil { + errs <- err + } + }() + } + + wg.Wait() + close(errs) + + for err := range errs { + t.Errorf("publish failed: %v", err) + } + + assert.Equal(t, int32(1), atomic.LoadInt32(&dials), "100 deliveries must share one AMQP connection") +}