tests: add goleak leak detection and advisory race-detector CI (CON-179)

.claude/agents/tester.md

@@ -592,6 +592,29 @@ t.Cleanup(func() {
 - Do not use `tc := tc` in loop bodies. Go 1.22+ fixed loop variable scoping.
 - Use `t.Context()` for test contexts. Exception: in `t.Cleanup()` functions, use `context.Background()` because `t.Context()` is already canceled during cleanup.
 
+# Goroutine Leak Detection (goleak)
+
+Connector packages opt in to goroutine-leak detection by wiring `go.uber.org/goleak` into a package-level `TestMain` (one `TestMain` per package — check for an existing one first). See `internal/impl/protobuf` and `internal/impl/sql` for reference.
+
+```go
+// main_test.go (with the package's usual license header)
+func TestMain(m *testing.M) {
+    goleak.VerifyTestMain(m,
+        goleak.IgnoreCurrent(),
+        // Narrow, commented ignores for known-benign goroutines only, e.g.:
+        // database/sql keeps a connection-pool opener alive per *sql.DB.
+        goleak.IgnoreTopFunction("database/sql.(*DB).connectionOpener"),
+    )
+}
+```
+
+Adding it to a new package:
+
+1. Create `main_test.go` with the pattern above and the package's license header.
+2. Run `go test -count=1 ./internal/impl/<pkg>/...` at least twice. For every goleak failure, decide: real leak in the component (fix it) or a benign third-party goroutine (add a narrow `goleak.IgnoreTopFunction` with a comment naming the library).
+3. `VerifyTestMain` also runs after the package's integration tests in nightly CI, so verify those too (or tune ignores) before merging — the package must be green on a clean baseline, never aspirationally.
+4. Prefer `IgnoreTopFunction` over broad ignores; an ignore list that swallows everything defeats the check.
+
 # Running Tests
 
 ```bash

.github/race-blocking-packages.txt

@@ -0,0 +1,11 @@
+# Packages promoted to BLOCKING in the race-detector CI job (CON-179 R3).
+#
+# The race_test.yml workflow runs `go test -race` on every internal/impl
+# package a PR touches. For packages listed here (one directory name per
+# line, e.g. `sql`), a race failure fails the job; for everything else the
+# failure is an advisory warning only.
+#
+# Promotion rule: add a package here only after it holds a green -race
+# baseline (repeated clean runs of `task test:unit-race` scoped to the
+# package, plus a green advisory run in CI). A blocking gate must never be
+# able to fail on a pre-existing race the PR didn't touch.

.github/workflows/race_test.yml

@@ -0,0 +1,94 @@
+name: Race Detector Tests
+
+# Advisory-first race-detector job (CON-179 R3).
+#
+# Runs `go test -race` scoped to the internal/impl/<component> packages a PR
+# touches, mirroring the auto-scoping in integration_test.yml to keep the
+# 2-10x race-detector runtime cost bounded.
+#
+# Failures are advisory (a warning annotation, not a red check) unless the
+# failing package is listed in .github/race-blocking-packages.txt. Packages
+# are promoted to that list only after they hold a green -race baseline, so
+# this job can never fail a PR on a pre-existing race the PR didn't touch.
+
+on:
+  pull_request:
+  workflow_dispatch:
+    inputs:
+      filter:
+        description: 'Package filter (e.g. sql kafka). Required for manual runs.'
+        required: true
+        type: string
+
+jobs:
+  race-test:
+    runs-on: ubuntu-latest
+    timeout-minutes: 45
+    env:
+      # The Go race detector requires cgo.
+      CGO_ENABLED: 1
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+
+      - name: Detect changed packages
+        if: ${{ github.event_name == 'pull_request' }}
+        id: detect
+        run: |
+          FILTERS=$(
+            git diff --name-only "$(git merge-base HEAD origin/${{ github.base_ref }})"...HEAD \
+              | { grep '^internal/impl/' || true; } \
+              | cut -d/ -f3 \
+              | sort -u \
+              | tr '\n' ' '
+          )
+          echo "filters=${FILTERS}" >> "$GITHUB_OUTPUT"
+
+      - name: Install Go
+        if: ${{ steps.detect.outputs.filters != '' || github.event.inputs.filter != '' }}
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: 'go.mod'
+
+      - name: Run race-detector tests
+        if: ${{ steps.detect.outputs.filters != '' || github.event.inputs.filter != '' }}
+        env:
+          FILTER: ${{ steps.detect.outputs.filters || github.event.inputs.filter }}
+        run: |
+          BLOCKING_FILE=.github/race-blocking-packages.txt
+          PASSED=()
+          ADVISORY_FAILED=()
+          BLOCKING_FAILED=()
+
+          for pkg in ${FILTER}; do
+            [ -d "internal/impl/${pkg}" ] || continue
+            echo "::group::go test -race ./internal/impl/${pkg}/..."
+            if go test -count=1 -race -timeout 10m -shuffle=on "./internal/impl/${pkg}/..."; then
+              PASSED+=("${pkg}")
+            elif grep -qxF "${pkg}" "${BLOCKING_FILE}" 2>/dev/null; then
+              BLOCKING_FAILED+=("${pkg}")
+            else
+              ADVISORY_FAILED+=("${pkg}")
+            fi
+            echo "::endgroup::"
+          done
+
+          {
+            echo "## Race detector results"
+            echo ""
+            echo "| Package | Result |"
+            echo "| --- | --- |"
+            for pkg in "${PASSED[@]}"; do echo "| \`${pkg}\` | ✅ pass |"; done
+            for pkg in "${ADVISORY_FAILED[@]}"; do echo "| \`${pkg}\` | ⚠️ fail (advisory) |"; done
+            for pkg in "${BLOCKING_FAILED[@]}"; do echo "| \`${pkg}\` | ❌ fail (blocking) |"; done
+          } >> "$GITHUB_STEP_SUMMARY"
+
+          if [ ${#ADVISORY_FAILED[@]} -gt 0 ]; then
+            echo "::warning::Race detector failures (advisory, non-blocking): ${ADVISORY_FAILED[*]}. See the job log. Packages get promoted to blocking via .github/race-blocking-packages.txt once their -race baseline is green."
+          fi
+          if [ ${#BLOCKING_FAILED[@]} -gt 0 ]; then
+            echo "::error::Race detector failures in blocking packages: ${BLOCKING_FAILED[*]}. These packages have a green -race baseline; this failure was introduced by the PR."
+            exit 1
+          fi

CONTRIBUTING.md

@@ -169,6 +169,7 @@ The rule throughout is **conformance to the existing fleet**: mirror the shape t
 ## 6. Before You Open a PR
 
 - Run `task fmt`, `task lint`, and `task test` locally — all green.
+- Run `task test:unit-race` on the packages you touched (e.g. `go test -race ./internal/impl/<component>/...`). CI runs an advisory race-detector job on changed connector packages; packages listed in `.github/race-blocking-packages.txt` treat race failures as blocking.
 - Run `task docs` and commit the result: the generated component pages **and** the `internal/plugins/info.csv` row. CI fails on stale docs.
 - Every new component has an `internal/plugins/info.csv` entry with the correct distribution and cloud classification.
 - A license header on **every** new `.go` file (including test and benchmark helpers), matching the component's distribution.

go.mod

@@ -200,6 +200,7 @@ require (
 	go.opentelemetry.io/otel/sdk/metric v1.43.0
 	go.opentelemetry.io/otel/trace v1.44.0
 	go.starlark.net v0.0.0-20260210143700-b62fd896b91b
+	go.uber.org/goleak v1.3.0
 	go.uber.org/multierr v1.11.0
 	golang.org/x/crypto v0.54.0
 	golang.org/x/net v0.56.0

internal/impl/mysql/main_test.go

@@ -0,0 +1,26 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/v4/blob/main/licenses/rcl.md
+
+package mysql
+
+import (
+	"testing"
+
+	"go.uber.org/goleak"
+)
+
+// TestMain verifies that no goroutines are leaked by the tests in this
+// package (CON-179 R2).
+func TestMain(m *testing.M) {
+	goleak.VerifyTestMain(m,
+		goleak.IgnoreCurrent(),
+		// internal/license: InjectTestService starts an hourly expiry-metric
+		// loop whose cancel func is not reachable from tests.
+		goleak.IgnoreTopFunction("github.com/redpanda-data/connect/v4/internal/license.(*Service).updateExpiryMetricLoop"),
+	)
+}

internal/impl/postgresql/integration_test.go

@@ -297,6 +297,9 @@ pg_stream:
 
 	// Run 1: receive the snapshot rows but never acknowledge them, then simulate
 	// a crash by cancelling the run before the slot can be promoted.
+	run1Ctx, crash := context.WithCancel(context.Background())
+	defer crash()
+
 	received := make(chan struct{}, 1)
 	run1Builder := service.NewStreamBuilder()
 	require.NoError(t, run1Builder.SetLoggerYAML(`level: OFF`))
@@ -306,15 +309,21 @@ pg_stream:
 		case received <- struct{}{}:
 		default:
 		}
-		// Block without acking until the simulated crash cancels our context.
-		<-ctx.Done()
-		return ctx.Err()
+		// Block without acking until the simulated crash. Benthos invokes
+		// consumer funcs with context.Background(), so waiting on the passed
+		// ctx alone would block this goroutine forever and leak it (and the
+		// unacked stream behind it) past the end of the test.
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-run1Ctx.Done():
+			return run1Ctx.Err()
+		}
 	}))
 	run1, err := run1Builder.Build()
 	require.NoError(t, err)
 	license.InjectTestService(run1.Resources())
 
-	run1Ctx, crash := context.WithCancel(context.Background())
 	run1Done := make(chan struct{})
 	go func() {
 		defer close(run1Done)
@@ -335,6 +344,14 @@ pg_stream:
 	case <-time.After(30 * time.Second):
 		t.Fatal("run 1 did not stop after the simulated crash")
 	}
+	// Run returns as soon as its context is cancelled without tearing the
+	// stream down, so stop it explicitly here; otherwise the crashed stream
+	// (wedged on the never-acknowledged snapshot batch) leaks its goroutines
+	// past the end of the test. Errors are expected: the stream cannot stop
+	// gracefully by construction.
+	if err := run1.StopWithin(10 * time.Second); err != nil {
+		t.Log(err)
+	}
 
 	// The barrier must have prevented the temporary slot from being promoted to
 	// a permanent one, since the snapshot was never acknowledged. This is the

internal/impl/postgresql/main_test.go

@@ -0,0 +1,31 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/v4/blob/main/licenses/rcl.md
+
+package pgstream
+
+import (
+	"testing"
+
+	"go.uber.org/goleak"
+)
+
+// TestMain verifies that no goroutines are leaked by the tests in this
+// package (CON-179 R2).
+func TestMain(m *testing.M) {
+	goleak.VerifyTestMain(m,
+		goleak.IgnoreCurrent(),
+		// internal/license: InjectTestService starts an hourly expiry-metric
+		// loop whose cancel func is not reachable from tests.
+		goleak.IgnoreTopFunction("github.com/redpanda-data/connect/v4/internal/license.(*Service).updateExpiryMetricLoop"),
+		// input_pg_stream.go's snapshot ack barrier waits on snapshotAckWG in
+		// a goroutine that is documented to outlive a soft stop when snapshot
+		// batches are never acknowledged ("bounded by process lifetime").
+		// TestIntegrationPostgresSnapshotAckBarrier triggers this on purpose.
+		goleak.IgnoreAnyFunction("github.com/redpanda-data/connect/v4/internal/impl/postgresql.(*pgStreamInput).processStream.func3"),
+	)
+}

internal/impl/protobuf/main_test.go

@@ -0,0 +1,39 @@
+// Copyright 2026 Redpanda Data, 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 protobuf
+
+import (
+	"testing"
+
+	"go.uber.org/goleak"
+)
+
+// TestMain verifies that no goroutines are leaked by the tests in this
+// package (CON-179 R2).
+func TestMain(m *testing.M) {
+	goleak.VerifyTestMain(m,
+		goleak.IgnoreCurrent(),
+		// github.com/bufbuild/prototransform leaves its SchemaWatcher poll
+		// loop running: protobufProc.Close is a no-op and MultiModuleWatcher
+		// exposes no stop hook, so watchers started by BSR-backed tests live
+		// for the remainder of the process.
+		goleak.IgnoreTopFunction("github.com/bufbuild/prototransform.(*SchemaWatcher).start.func1"),
+		// net/http keepalive connections owned by the HTTP client of the
+		// unstoppable SchemaWatcher above; they linger until the transport is
+		// garbage collected.
+		goleak.IgnoreAnyFunction("net/http.(*persistConn).readLoop"),
+		goleak.IgnoreAnyFunction("net/http.(*persistConn).writeLoop"),
+	)
+}

internal/impl/protobuf/processor_protobuf_test.go

@@ -462,8 +462,13 @@ func runMockBSRServer(t *testing.T, importPath string) string {
 	mux := http.NewServeMux()
 	fileDescriptorSetServer := &fileDescriptorSetServer{fileDescriptorSet: files}
 	mux.Handle(reflectv1beta1connect.NewFileDescriptorSetServiceHandler(fileDescriptorSetServer))
+	server := &http.Server{Handler: h2c.NewHandler(mux, &http2.Server{})} //nolint:gosec,staticcheck // test server, no timeouts needed; h2c matches the pre-existing usage
+	// Close (rather than Shutdown) so that all connections are torn down,
+	// otherwise the package-level goroutine leak check trips on lingering
+	// HTTP connection goroutines.
+	t.Cleanup(func() { _ = server.Close() })
 	go func() {
-		if err := http.Serve(listener, h2c.NewHandler(mux, &http2.Server{})); err != nil && !errors.Is(err, http.ErrServerClosed) { //nolint:staticcheck
+		if err := server.Serve(listener); err != nil && !errors.Is(err, http.ErrServerClosed) {
 			require.NoError(t, err)
 		}
 	}()

internal/impl/sql/main_test.go

@@ -0,0 +1,29 @@
+// Copyright 2026 Redpanda Data, 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 sql
+
+import (
+	"testing"
+
+	"go.uber.org/goleak"
+)
+
+// TestMain verifies that no goroutines are leaked by the tests in this
+// package (CON-179 R2).
+func TestMain(m *testing.M) {
+	goleak.VerifyTestMain(m,
+		goleak.IgnoreCurrent(),
+	)
+}