fix(velocity): reuse one AMQP connection in the pending metrics emitter

ee/velocity/go.mod

@@ -12,7 +12,8 @@ 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/rabbitmq/amqp091-go v1.13.0
+	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
@@ -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

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-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=

ee/velocity/pkg/emitter/pending_metrics.go

@@ -2,6 +2,7 @@
 package emitter
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"sync"
@@ -19,11 +20,20 @@ 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
+
+	publisherOptions tackle.PublisherOptions
+	publishTimeout   time.Duration
+	publisher        *tackle.Publisher
 }
 
 func NewPendingMetricsEmitter(options tackle.Options, projectHubServiceClient *service.ProjectHubGrpcClient, crontab string) *PendingMetricsEmitter {
@@ -32,6 +42,11 @@ func NewPendingMetricsEmitter(options tackle.Options, projectHubServiceClient *s
 		crontab:          crontab,
 		options:          options,
 		projectHubClient: projectHubServiceClient,
+		publisherOptions: tackle.PublisherOptions{
+			ConnectionName:    options.ConnectionName,
+			ConnectionTimeout: defaultConnectionTimeout,
+		},
+		publishTimeout: defaultPublishTimeout,
 	}
 }
 
@@ -65,6 +80,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
 
@@ -198,20 +219,40 @@ func (emitter *PendingMetricsEmitter) emit(pendingMetric entity.PendingMetric, o
 	return nil
 }
 
-func (emitter *PendingMetricsEmitter) publishMessage(message []byte) (err error) {
-	params := &tackle.PublishParams{
+func (emitter *PendingMetricsEmitter) publishMessage(message []byte) error {
+	ctx, cancel := context.WithTimeout(context.Background(), emitter.publishTimeout)
+	defer cancel()
+
+	return emitter.publisher.PublishWithContext(ctx, &tackle.PublishParams{
 		Body:       message,
 		AmqpURL:    emitter.options.URL,
 		RoutingKey: emitter.options.RoutingKey,
 		Exchange:   emitter.options.RemoteExchange,
+	})
+}
+
+func (emitter *PendingMetricsEmitter) openPublisher() error {
+	publisher, err := tackle.NewPublisher(emitter.options.URL, emitter.publisherOptions)
+	if err != nil {
+		return err
+	}
+
+	if err = publisher.ExchangeDeclare(emitter.options.RemoteExchange); err != nil {
+		publisher.Close()
+		return err
 	}
 
-	if err = tackle.PublishMessage(params); HasError(err) {
-		log.Printf("failed to publish message, %v", err)
+	emitter.publisher = publisher
+	return nil
+}
+
+func (emitter *PendingMetricsEmitter) closePublisher() {
+	if emitter.publisher == nil {
 		return
 	}
 
-	return
+	emitter.publisher.Close()
+	emitter.publisher = nil
 }
 
 func (emitter *PendingMetricsEmitter) buildMessage(pendingMetric entity.PendingMetric, orgID string, branchName string) (message []byte, err error) {

ee/velocity/pkg/emitter/pending_metrics_test.go

@@ -0,0 +1,236 @@
+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 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) {
+		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) {
+	requireBroker(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) {
+	requireBroker(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) {
+	requireBroker(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)
+}
+
+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()
+}