perf: batch runs-service NOTIFY payloads

runs/repository/impl/action.go

@@ -25,6 +25,8 @@ import (
 
 const rootActionName = "a0"
 
+const notifyBatchMaxPayloads = 1024
+
 // actionRepo implements actionRepo interface using PostgreSQL
 type actionRepo struct {
 	db       *sqlx.DB
@@ -1033,6 +1035,33 @@ func isConnError(err error) bool {
 	return false
 }
 
+func buildNotifyBatchQuery(channel string, payloads []string) (string, []any) {
+	notifyCalls := make([]string, 0, len(payloads))
+	args := make([]any, 0, len(payloads)+1)
+	args = append(args, channel)
+	for i, payload := range payloads {
+		notifyCalls = append(notifyCalls, fmt.Sprintf("pg_notify($1, $%d)", i+2))
+		args = append(args, payload)
+	}
+	return "SELECT " + strings.Join(notifyCalls, ", "), args
+}
+
+func drainNotifyBatch(firstPayload string, ch <-chan string) []string {
+	payloads := []string{firstPayload}
+	for len(payloads) < notifyBatchMaxPayloads {
+		select {
+		case payload, ok := <-ch:
+			if !ok {
+				return payloads
+			}
+			payloads = append(payloads, payload)
+		default:
+			return payloads
+		}
+	}
+	return payloads
+}
+
 // runNotifyLoop processes notify channels using the given connection.
 // On connection errors it attempts to reconnect; if reconnection fails
 // the error is logged and the notification is skipped.
@@ -1059,37 +1088,40 @@ func (r *actionRepo) runNotifyLoop(sqlDB *sql.DB, conn *sql.Conn) {
 		}
 	}
 
-	execNotify := func(channel, payload string) {
-		if conn == nil {
-			reconnect()
-		}
-		if conn == nil {
-			logger.Errorf(context.Background(), "No NOTIFY connection available, dropping %s notification", channel)
+	execNotifyBatch := func(channel string, payloads []string) {
+		if len(payloads) == 0 {
 			return
 		}
-		if _, err := conn.ExecContext(context.Background(), "SELECT pg_notify($1, $2)", channel, payload); err != nil {
-			logger.Errorf(context.Background(), "Failed to NOTIFY %s: %v", channel, err)
-			if isConnError(err) {
-				reconnect()
-			}
-		}
-	}
 
-	drainAndExec := func(channel, firstPayload string, ch <-chan string) {
-		execNotify(channel, firstPayload)
+		query, args := buildNotifyBatchQuery(channel, payloads)
 		for {
-			select {
-			case payload, ok := <-ch:
-				if !ok {
+			if conn == nil {
+				reconnect()
+			}
+			if conn == nil {
+				logger.Errorf(context.Background(), "No NOTIFY connection available for %s batch of %d notifications", channel, len(payloads))
+				if sqlDB == nil {
 					return
 				}
-				execNotify(channel, payload)
-			default:
-				return
+				time.Sleep(100 * time.Millisecond)
+				continue
 			}
+			if _, err := conn.ExecContext(context.Background(), query, args...); err != nil {
+				logger.Errorf(context.Background(), "Failed to NOTIFY %s batch of %d notifications: %v", channel, len(payloads), err)
+				if isConnError(err) {
+					reconnect()
+				}
+				time.Sleep(100 * time.Millisecond)
+				continue
+			}
+			return
 		}
 	}
 
+	drainAndExec := func(channel, firstPayload string, ch <-chan string) {
+		execNotifyBatch(channel, drainNotifyBatch(firstPayload, ch))
+	}
+
 	for {
 		select {
 		case payload, ok := <-r.actionNotifyCh:

runs/repository/impl/action_test.go

@@ -878,6 +878,48 @@ func TestNotifyRunUpdate_PayloadWithSpecialChars(t *testing.T) {
 	}
 }
 
+func TestBuildNotifyBatchQuery(t *testing.T) {
+	query, args := buildNotifyBatchQuery("action_updates", []string{
+		"proj/domain/run/action",
+		"proj/domain/run'; DROP TABLE actions; --/action",
+		"proj/domain/run/next-action",
+	})
+
+	assert.Equal(t, "SELECT pg_notify($1, $2), pg_notify($1, $3), pg_notify($1, $4)", query)
+	assert.Equal(t, []any{
+		"action_updates",
+		"proj/domain/run/action",
+		"proj/domain/run'; DROP TABLE actions; --/action",
+		"proj/domain/run/next-action",
+	}, args)
+}
+
+func TestDrainNotifyBatchDrainsBufferedPayloads(t *testing.T) {
+	ch := make(chan string, 3)
+	ch <- "second"
+	ch <- "third"
+	ch <- "fourth"
+
+	payloads := drainNotifyBatch("first", ch)
+
+	assert.Equal(t, []string{"first", "second", "third", "fourth"}, payloads)
+	assert.Empty(t, ch)
+}
+
+func TestDrainNotifyBatchRespectsBatchCap(t *testing.T) {
+	ch := make(chan string, notifyBatchMaxPayloads+5)
+	for i := 1; i < notifyBatchMaxPayloads+6; i++ {
+		ch <- fmt.Sprintf("payload-%d", i)
+	}
+
+	payloads := drainNotifyBatch("payload-0", ch)
+
+	assert.Len(t, payloads, notifyBatchMaxPayloads)
+	assert.Equal(t, "payload-0", payloads[0])
+	assert.Equal(t, fmt.Sprintf("payload-%d", notifyBatchMaxPayloads-1), payloads[len(payloads)-1])
+	assert.Len(t, ch, 6)
+}
+
 func TestIsConnError(t *testing.T) {
 	tests := []struct {
 		name   string