diff --git a/.claude/agents/tester.md b/.claude/agents/tester.md index 8796dad71c..3d9b41b275 100644 --- a/.claude/agents/tester.md +++ b/.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//...` 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 diff --git a/.github/race-blocking-packages.txt b/.github/race-blocking-packages.txt new file mode 100644 index 0000000000..ccc82889f1 --- /dev/null +++ b/.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. diff --git a/.github/workflows/race_test.yml b/.github/workflows/race_test.yml new file mode 100644 index 0000000000..84a772e40e --- /dev/null +++ b/.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/ 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 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 36a15b5e9a..eb368456f2 100644 --- a/CONTRIBUTING.md +++ b/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//...`). 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. diff --git a/go.mod b/go.mod index d6aee2c1cb..4a531e4dff 100644 --- a/go.mod +++ b/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 diff --git a/internal/impl/mysql/main_test.go b/internal/impl/mysql/main_test.go new file mode 100644 index 0000000000..956e907988 --- /dev/null +++ b/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"), + ) +} diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index bcb67946a1..4c23e632a4 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/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 diff --git a/internal/impl/postgresql/main_test.go b/internal/impl/postgresql/main_test.go new file mode 100644 index 0000000000..fa5d68df20 --- /dev/null +++ b/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"), + ) +} diff --git a/internal/impl/protobuf/main_test.go b/internal/impl/protobuf/main_test.go new file mode 100644 index 0000000000..6e7544680f --- /dev/null +++ b/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"), + ) +} diff --git a/internal/impl/protobuf/processor_protobuf_test.go b/internal/impl/protobuf/processor_protobuf_test.go index d7f6fd01d9..6d5aff2dec 100644 --- a/internal/impl/protobuf/processor_protobuf_test.go +++ b/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) } }() diff --git a/internal/impl/sql/main_test.go b/internal/impl/sql/main_test.go new file mode 100644 index 0000000000..a880643b7c --- /dev/null +++ b/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(), + ) +}