fix(ci): bound queue DB connections to stop E2E 1040 stalls

.github/actions/run-bazel-test/action.yml

@@ -17,3 +17,7 @@ runs:
       env:
         TARGET: ${{ inputs.target }}
       run: ./tool/bazel test "$TARGET" --test_output=streamed
+
+    - name: Upload Bazel failure logs
+      if: ${{ failure() }}
+      uses: ./.github/actions/upload-testlogs

.github/actions/upload-testlogs/action.yml

@@ -0,0 +1,13 @@
+name: Upload Bazel test logs
+description: Upload bazel-testlogs on job failure so truncated CI output is not the only diagnostic.
+
+runs:
+  using: composite
+  steps:
+    - name: Upload Bazel failure logs
+      uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+      with:
+        name: testlogs-${{ github.job }}-${{ github.run_id }}-${{ github.run_attempt }}
+        path: bazel-testlogs/**/test.log
+        if-no-files-found: warn
+        retention-days: 7

.github/workflows/ci.yml

@@ -88,6 +88,10 @@ jobs:
       - name: Run unit tests
         run: make test
 
+      - name: Upload Bazel failure logs
+        if: ${{ failure() }}
+        uses: ./.github/actions/upload-testlogs
+
   # ---------------------------------------------------------------------------
   # INTEGRATION TESTS (e2e, gateway, orchestrator)
   # ---------------------------------------------------------------------------
@@ -106,6 +110,10 @@ jobs:
       - name: Run E2E tests
         run: make e2e-test
 
+      - name: Upload Bazel failure logs
+        if: ${{ failure() }}
+        uses: ./.github/actions/upload-testlogs
+
   gateway-integration-test:
     name: Gateway Integration Test
     if: ${{ github.event_name != 'pull_request' || github.event.pull_request.draft == false }}
@@ -121,6 +129,10 @@ jobs:
       - name: Run Gateway integration tests
         run: make integration-test-submitqueue-gateway
 
+      - name: Upload Bazel failure logs
+        if: ${{ failure() }}
+        uses: ./.github/actions/upload-testlogs
+
   orchestrator-integration-test:
     name: Orchestrator Integration Test
     if: ${{ github.event_name != 'pull_request' || github.event.pull_request.draft == false }}
@@ -136,6 +148,10 @@ jobs:
       - name: Run Orchestrator integration tests
         run: make integration-test-submitqueue-orchestrator
 
+      - name: Upload Bazel failure logs
+        if: ${{ failure() }}
+        uses: ./.github/actions/upload-testlogs
+
   # ---------------------------------------------------------------------------
   # EXTENSION TESTS
   # ---------------------------------------------------------------------------

service/messagequeue/BUILD.bazel

@@ -2,16 +2,23 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["tenant.go"],
+    srcs = [
+        "database.go",
+        "tenant.go",
+    ],
     importpath = "github.com/uber/submitqueue/service/messagequeue",
     visibility = ["//visibility:public"],
 )
 
 go_test(
     name = "go_default_test",
-    srcs = ["tenant_test.go"],
+    srcs = [
+        "database_test.go",
+        "tenant_test.go",
+    ],
     embed = [":go_default_library"],
     deps = [
+        "@com_github_go_sql_driver_mysql//:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_stretchr_testify//require:go_default_library",
     ],

service/messagequeue/database.go

@@ -0,0 +1,44 @@
+// Copyright (c) 2026 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package messagequeue
+
+import (
+	"database/sql"
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+// DefaultQueueMySQLMaxOpenConnections bounds queue database concurrency when
+// the service does not provide an explicit limit.
+const DefaultQueueMySQLMaxOpenConnections = 16
+
+// ConfigureQueueMySQLConnectionPool applies the configured maximum open
+// connection count and keeps that many connections idle. Empty input
+// selects DefaultQueueMySQLMaxOpenConnections.
+func ConfigureQueueMySQLConnectionPool(db *sql.DB, value string) error {
+	maxOpenConnections := DefaultQueueMySQLMaxOpenConnections
+	trimmed := strings.TrimSpace(value)
+	if trimmed != "" {
+		parsed, err := strconv.Atoi(trimmed)
+		if err != nil || parsed <= 0 {
+			return fmt.Errorf("maximum open connections must be a positive integer")
+		}
+		maxOpenConnections = parsed
+	}
+	db.SetMaxOpenConns(maxOpenConnections)
+	db.SetMaxIdleConns(maxOpenConnections)
+	return nil
+}

service/messagequeue/database_test.go

@@ -0,0 +1,56 @@
+// Copyright (c) 2026 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package messagequeue
+
+import (
+	"database/sql"
+	"testing"
+
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConfigureQueueMySQLConnectionPool(t *testing.T) {
+	tests := []struct {
+		name    string
+		value   string
+		want    int
+		wantErr bool
+	}{
+		{name: "empty uses default", want: DefaultQueueMySQLMaxOpenConnections},
+		{name: "explicit value", value: "24", want: 24},
+		{name: "surrounding whitespace", value: " 8 ", want: 8},
+		{name: "zero", value: "0", wantErr: true},
+		{name: "negative", value: "-1", wantErr: true},
+		{name: "non-integer", value: "many", wantErr: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			db, err := sql.Open("mysql", "")
+			require.NoError(t, err)
+			t.Cleanup(func() { require.NoError(t, db.Close()) })
+
+			err = ConfigureQueueMySQLConnectionPool(db, tt.value)
+			if tt.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, db.Stats().MaxOpenConnections)
+		})
+	}
+}

service/runway/server/docker-compose.yml

@@ -50,6 +50,7 @@ services:
       - MERGER=${SQ_RUNWAY_MERGER:-}
       # Queue infrastructure connection
       - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true
+      - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16}
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}

service/runway/server/main.go

@@ -140,6 +140,9 @@ func run() error {
 		return fmt.Errorf("failed to open queue database: %w", err)
 	}
 	defer queueDB.Close()
+	if err := servicemq.ConfigureQueueMySQLConnectionPool(queueDB, os.Getenv("QUEUE_MYSQL_MAX_OPEN_CONNECTIONS")); err != nil {
+		return fmt.Errorf("failed to configure queue database pool: %w", err)
+	}
 
 	tenants, err := servicemq.ParseRequiredTenants(os.Getenv("MQ_TENANTS"))
 	if err != nil {

service/stovepipe/docker-compose.yml

@@ -67,6 +67,7 @@ services:
       - PORT=:8080
       - STORAGE_MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true
       - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true
+      - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16}
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}

service/stovepipe/server/main.go

@@ -273,6 +273,9 @@ func run() error {
 		return fmt.Errorf("failed to open queue database: %w", err)
 	}
 	defer queueDB.Close()
+	if err := servicemq.ConfigureQueueMySQLConnectionPool(queueDB, os.Getenv("QUEUE_MYSQL_MAX_OPEN_CONNECTIONS")); err != nil {
+		return fmt.Errorf("failed to configure queue database pool: %w", err)
+	}
 
 	tenants, err := servicemq.ParseRequiredTenants(os.Getenv("MQ_TENANTS"))
 	if err != nil {

service/submitqueue/docker-compose.yml

@@ -73,6 +73,7 @@ services:
       - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true
       # Queue infrastructure connection (separate database)
       - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true
+      - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16}
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}
@@ -105,6 +106,7 @@ services:
       - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true
       # Queue infrastructure connection (separate database)
       - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true
+      - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16}
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}
@@ -137,6 +139,7 @@ services:
       - PORT=:8080
       # Queue infrastructure connection (shared with the orchestrator)
       - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true
+      - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16}
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}

service/submitqueue/gateway/server/docker-compose.yml

@@ -62,6 +62,7 @@ services:
       - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true
       # Queue infrastructure connection (separate database)
       - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true
+      - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16}
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}

service/submitqueue/gateway/server/main.go

@@ -241,6 +241,9 @@ func run() error {
 		return fmt.Errorf("failed to open queue database: %w", err)
 	}
 	defer queueDB.Close()
+	if err := servicemq.ConfigureQueueMySQLConnectionPool(queueDB, os.Getenv("QUEUE_MYSQL_MAX_OPEN_CONNECTIONS")); err != nil {
+		return fmt.Errorf("failed to configure queue database pool: %w", err)
+	}
 
 	// Load queue configurations from YAML. Path is required so the gateway
 	// can reject requests for unknown queues at the edge.

service/submitqueue/orchestrator/server/docker-compose.yml

@@ -62,6 +62,7 @@ services:
       - MYSQL_DSN=root:root@tcp(mysql-app:3306)/submitqueue?parseTime=true
       # Queue infrastructure connection (separate database)
       - QUEUE_MYSQL_DSN=root:root@tcp(mysql-queue:3306)/submitqueue?parseTime=true
+      - QUEUE_MYSQL_MAX_OPEN_CONNECTIONS=${QUEUE_MYSQL_MAX_OPEN_CONNECTIONS:-16}
       # Level for the queue's own logs; info by default so its per-message
       # chatter does not bury the rest of the service at debug.
       - QUEUE_LOG_LEVEL=${QUEUE_LOG_LEVEL:-}

service/submitqueue/orchestrator/server/main.go

@@ -156,6 +156,9 @@ func run() error {
 		return fmt.Errorf("failed to open queue database: %w", err)
 	}
 	defer queueDB.Close()
+	if err := servicemq.ConfigureQueueMySQLConnectionPool(queueDB, os.Getenv("QUEUE_MYSQL_MAX_OPEN_CONNECTIONS")); err != nil {
+		return fmt.Errorf("failed to configure queue database pool: %w", err)
+	}
 
 	// Build per-queue extension profiles (host-private). Each queue resolves
 	// to its own set of extension implementations (conflict analyzer, …),

test/integration/submitqueue/gateway/suite_test.go

@@ -67,16 +67,9 @@ func TestGatewayIntegration(t *testing.T) {
 	suite.Run(t, new(GatewayIntegrationSuite))
 }
 
-// The log consumer runs inside the gateway-service container, so this suite can
-// only observe persistence black-box through the request-summary RPC — there is no
-// in-process channel/HookSignal to wait on across the container boundary. A
-// bounded poll is therefore the deterministic-enough analog: persistTimeout is a
-// safety net (a failure here means something is genuinely stuck, not a timing
-// race), and persistPollInterval bounds how often we re-query.
-const (
-	persistTimeout      = 30 * time.Second
-	persistPollInterval = 500 * time.Millisecond
-)
+// The container boundary leaves the request-summary RPC as the only signal that
+// the gateway consumer persisted a log entry.
+const persistPollInterval = 500 * time.Millisecond
 
 func (s *GatewayIntegrationSuite) SetupSuite() {
 	t := s.T()
@@ -163,7 +156,7 @@ func (s *GatewayIntegrationSuite) TestLandAPI() {
 
 	// Verify message published to queue
 	var msgCount int
-	err = s.queueDB.QueryRow("SELECT COUNT(*) FROM queue_messages WHERE id = ?", resp.Sqid).Scan(&msgCount)
+	err = s.queueDB.QueryRow("SELECT COUNT(*) FROM queue_messages WHERE tenant = ? AND id = ?", req.Queue, resp.Sqid).Scan(&msgCount)
 	require.NoError(t, err, "failed to query queue messages")
 	assert.Equal(t, 1, msgCount, "should have 1 message in queue")
 }
@@ -265,8 +258,8 @@ func (s *GatewayIntegrationSuite) TestReadAPIErrorCodes() {
 // entry to storage, observable through the request-summary RPC.
 func (s *GatewayIntegrationSuite) TestRequestLogConsumer() {
 	t := s.T()
-	const sqid = "log-consumer-test/1"
-	const logQueue = "log-consumer-test"
+	const sqid = "test-queue/log-consumer-test"
+	const logQueue = "test-queue"
 
 	// Build a publisher against the shared queue database. NewQueue only wires up
 	// stores; nothing consumes until a subscriber is started, so this publish-only
@@ -300,14 +293,15 @@ func (s *GatewayIntegrationSuite) TestRequestLogConsumer() {
 
 	s.log.Logf("Published 'started' log for sqid=%s; waiting for gateway consumer to persist it", sqid)
 
-	require.Eventually(t, func() bool {
-		resp, statusErr := s.client.GetRequestSummaryByID(s.ctx, &pb.GetRequestSummaryByIDRequest{Sqid: sqid, Queue: "log-consumer-test"})
-		if statusErr != nil {
-			return false
+	ticker := time.NewTicker(persistPollInterval)
+	defer ticker.Stop()
+	for {
+		resp, statusErr := s.client.GetRequestSummaryByID(s.ctx, &pb.GetRequestSummaryByIDRequest{Sqid: sqid, Queue: logQueue})
+		if statusErr == nil && resp.Request != nil && resp.Request.Status == string(entity.RequestStatusStarted) {
+			break
 		}
-		return resp.Request != nil && resp.Request.Status == string(entity.RequestStatusStarted)
-	}, persistTimeout, persistPollInterval,
-		"gateway log consumer should persist the published request log for sqid=%s", sqid)
+		<-ticker.C
+	}
 
 	s.log.Logf("Request log consumer test passed: entry persisted and readable via GetRequestSummaryByID")
 }