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/<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
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..0f7707daec
--- /dev/null
+++ b/.github/workflows/race_test.yml
@@ -0,0 +1,158 @@
+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
+    # Bounded comfortably above the worst case of the "Run race-detector
+    # tests" step: that step caps itself at MAX_PACKAGES=8 packages tested
+    # serially, each under its own `go test -timeout 10m` budget (80m worst
+    # case), plus headroom for per-package compilation, checkout with
+    # fetch-depth: 0, and the Go setup step. Packages beyond MAX_PACKAGES are
+    # skipped with a logged advisory warning rather than run, so the job
+    # should never actually approach this ceiling; it exists purely so a
+    # slow-but-not-hanging run can never be killed by the job timeout, which
+    # would produce a red check with no actual race (violating the
+    # advisory-first design).
+    timeout-minutes: 120
+    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
+        env:
+          BASE_REF: ${{ github.base_ref }}
+        run: |
+          set -euo pipefail
+          # Each of these runs as its own statement (rather than nested
+          # inside another command's argument list) so that `set -e` catches
+          # a git failure directly, instead of silently yielding an empty
+          # FILTERS that would be indistinguishable from "no packages
+          # touched".
+          MERGE_BASE=$(git merge-base HEAD "origin/${BASE_REF}")
+          CHANGED=$(git diff --name-only "${MERGE_BASE}"...HEAD)
+          # grep exits 1 when nothing matches "^internal/impl/" - that IS the
+          # legitimate "no connector packages touched" case, so only this
+          # command is allowed to fail non-fatally.
+          MATCHED=$(printf '%s\n' "${CHANGED}" | { grep '^internal/impl/' || true; })
+          # xargs (not tr) joins the package names: on empty/blank input it
+          # emits a genuinely empty string, whereas tr '\n' ' ' would emit a
+          # lone space that defeats the `filters != ''` step guards below and
+          # runs the job (checkout, Go install, empty summary table) on every
+          # PR that touches no connector packages.
+          FILTERS=$(printf '%s\n' "${MATCHED}" | cut -d/ -f3 | sort -u | xargs)
+          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: |
+          set -uo pipefail
+          BLOCKING_FILE=.github/race-blocking-packages.txt
+          # Cap the number of packages tested per run so the job's own
+          # worst-case duration is bounded (see the job-level timeout-minutes
+          # comment). Packages beyond the cap are skipped with a logged
+          # advisory warning rather than risking a job timeout kill, which
+          # would produce a red check with no actual race.
+          MAX_PACKAGES=8
+          PKG_TIMEOUT=10m
+
+          ALL_PKGS=()
+          for pkg in ${FILTER}; do
+            [ -d "internal/impl/${pkg}" ] && ALL_PKGS+=("${pkg}")
+          done
+
+          PKGS=("${ALL_PKGS[@]}")
+          if [ "${#ALL_PKGS[@]}" -gt "${MAX_PACKAGES}" ]; then
+            PKGS=("${ALL_PKGS[@]:0:${MAX_PACKAGES}}")
+            SKIPPED=("${ALL_PKGS[@]:${MAX_PACKAGES}}")
+            echo "::warning::Race detector job only tests the first ${MAX_PACKAGES} changed packages per run to bound job duration. Skipped this run (not tested, advisory only): ${SKIPPED[*]}"
+            echo "> ⚠️ Race detector: skipped ${#SKIPPED[@]} package(s) to bound job duration: \`${SKIPPED[*]}\`" >> "$GITHUB_STEP_SUMMARY"
+          fi
+
+          {
+            echo "## Race detector results"
+            echo ""
+            echo "| Package | Result |"
+            echo "| --- | --- |"
+          } >> "$GITHUB_STEP_SUMMARY"
+
+          BLOCKING_FAILED=()
+
+          for pkg in "${PKGS[@]}"; do
+            outfile=$(mktemp)
+            echo "::group::go test -race ./internal/impl/${pkg}/..."
+            # `set -e` is on by default for GitHub Actions run steps; the
+            # trailing `|| status=$?` keeps a non-zero go test exit from
+            # aborting the script so every package gets classified and its
+            # summary row written, instead of the loop dying on the first
+            # failure.
+            status=0
+            go test -count=1 -race -timeout "${PKG_TIMEOUT}" -shuffle=on "./internal/impl/${pkg}/..." >"${outfile}" 2>&1 || status=$?
+            cat "${outfile}"
+            echo "::endgroup::"
+
+            # Only an actual data race counts as a race for blocking
+            # purposes - build errors, flakes, and per-package timeouts must
+            # not be classified as races, and must never land in the
+            # blocking bucket even for a package on the blocking list (the
+            # ordinary unit-test workflow already gates plain failures).
+            is_race=false
+            grep -q "WARNING: DATA RACE" "${outfile}" && is_race=true
+            is_blocking=false
+            grep -qxF "${pkg}" "${BLOCKING_FILE}" 2>/dev/null && is_blocking=true
+
+            if [ "${status}" -eq 0 ]; then
+              result="✅ pass"
+            elif [ "${is_race}" = true ] && [ "${is_blocking}" = true ]; then
+              result="❌ RACE (blocking)"
+              BLOCKING_FAILED+=("${pkg}")
+            elif [ "${is_race}" = true ]; then
+              result="⚠️ RACE (advisory)"
+              echo "::warning::Data race detected in ${pkg} (advisory, non-blocking). See the job log. Packages get promoted to blocking via ${BLOCKING_FILE} once their -race baseline is green."
+            else
+              result="⚠️ failed (non-race)"
+              echo "::warning::go test failed in ${pkg} without a detected data race (build error, flake, or timeout). Not a race-detector finding, so this stays advisory-only; the ordinary unit-test workflow gates plain test failures."
+            fi
+
+            echo "| \`${pkg}\` | ${result} |" >> "$GITHUB_STEP_SUMMARY"
+            rm -f "${outfile}"
+          done
+
+          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..0fd78655ec 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` scoped to the packages you touched, e.g. `task test:unit-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.
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/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go
index 8ff0e62374..0b03bb1675 100644
--- a/internal/impl/postgresql/input_pg_stream.go
+++ b/internal/impl/postgresql/input_pg_stream.go
@@ -15,7 +15,7 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
-	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/Jeffail/checkpoint"
@@ -435,6 +435,8 @@ func newPgStreamInput(conf *service.ParsedConfig, mgr *service.Resources) (s ser
 		replicationLag:  replicationLag,
 		stopSig:         shutdown.NewSignaller(),
 
+		snapshotAckDrained: make(chan struct{}, 1),
+
 		iamAuthEnabled: iamAuthEnabled,
 	}
 
@@ -486,11 +488,21 @@ type pgStreamInput struct {
 	controlSig      *postgresSignaller
 	stopSig         *shutdown.Signaller
 
-	// snapshotAckWG tracks in-flight snapshot batches: incremented when a
+	// snapshotAckPending tracks in-flight snapshot batches: incremented when a
 	// snapshot batch (nil LSN) is enqueued and decremented when it is
 	// acknowledged. The snapshot->stream handoff blocks until it drains so the
 	// replication slot is not promoted before snapshot rows are durable.
-	snapshotAckWG sync.WaitGroup
+	snapshotAckPending atomic.Int64
+	// snapshotAckDrained is signalled (best-effort, buffered capacity 1) each
+	// time snapshotAckPending reaches zero, so the handoff can wait on it
+	// without spawning a helper goroutine around WaitGroup.Wait.
+	snapshotAckDrained chan struct{}
+	// snapshotAckFailed is a sticky flag set when any snapshot batch settles
+	// via a nack (as opposed to an ack). The counter drains the same way for
+	// acks and nacks, so this flag is what tells the handoff apart: a drain
+	// caused by a nack must never promote the slot, since the corresponding
+	// rows were never durably delivered downstream.
+	snapshotAckFailed atomic.Bool
 
 	// IAM authentication fields
 	iamAuthEnabled bool
@@ -514,6 +526,12 @@ func (p *pgStreamInput) Connect(ctx context.Context) error {
 	}
 	// Reset our stop signal
 	p.stopSig = shutdown.NewSignaller()
+	// A new connect starts a fresh snapshot attempt, so any nack recorded
+	// against the previous epoch is no longer relevant: the snapshot rows are
+	// about to be re-read and re-emitted from scratch. Note that a late,
+	// cross-epoch nack racing in after this reset would simply re-set the
+	// flag and force another conservative re-run, which is safe.
+	p.snapshotAckFailed.Store(false)
 	go p.processStream(pgStream, batcher)
 	return err
 }
@@ -564,16 +582,23 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher
 		case batch := <-pgStream.Messages():
 			if len(batch) == 1 && batch[0].Operation == pglogicalstream.SnapshotCompleteOpType {
 				// Snapshot fully emitted. Flush any buffered rows, then block
-				// until every snapshot batch is acknowledged downstream before
-				// signalling the stream to promote the replication slot. Blocks
-				// until acks drain or soft-stop (no timeout, by design).
+				// until every snapshot batch has settled downstream before
+				// deciding whether to promote the replication slot. Blocks
+				// until settlement drains or soft-stop (no timeout, by
+				// design). The counter alone can't tell an ack from a nack -
+				// it drains either way - so snapshotAckFailed is consulted
+				// once drained to distinguish "durably delivered" from
+				// "rejected downstream". If soft-stop wins the race instead,
+				// the slot is likewise left unpromoted: batches settling
+				// during teardown are nacks, not durable delivery.
 				nextTimedBatchChan = nil
 				flushedBatch, err := batcher.Flush(ctx)
 				if err != nil {
 					p.logger.Debugf("error flushing snapshot completion batch: %s", err)
 					// The sentinel is a one-shot signal; if we bail here without
-					// acking, the barrier's snapshot goroutine blocks on
-					// snapshotAcked forever. Trigger a restart instead of stalling.
+					// acking, the drain wait below would block forever waiting for
+					// snapshotAckPending to reach zero. Trigger a restart instead
+					// of stalling.
 					p.stopSig.TriggerSoftStop()
 					break
 				}
@@ -582,17 +607,33 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher
 					p.stopSig.TriggerSoftStop()
 					break
 				}
-				drained := make(chan struct{})
-				go func() {
-					// May outlive the select below if soft-stop fires while the
-					// downstream is stalled; bounded by process lifetime.
-					p.snapshotAckWG.Wait()
-					close(drained)
-				}()
-				select {
-				case <-drained:
+				drained := p.snapshotAckPending.Load() == 0
+			drainWait:
+				for !drained {
+					select {
+					case <-p.snapshotAckDrained:
+						drained = p.snapshotAckPending.Load() == 0
+					case <-p.stopSig.SoftStopChan():
+						// Abandon the handoff without promoting the slot:
+						// batches settling during teardown are nacks, not
+						// durable delivery, so the snapshot must re-run on
+						// restart.
+						break drainWait
+					}
+				}
+				switch {
+				case drained && !p.snapshotAckFailed.Load():
 					pgStream.MarkSnapshotAcknowledged()
-				case <-p.stopSig.SoftStopChan():
+				case drained:
+					// The counter reached zero because every outstanding
+					// snapshot batch settled, but at least one of them was
+					// rejected downstream (a nack), not durably delivered.
+					// Promoting the slot here would let those rows be lost
+					// forever, so leave it unpromoted and restart the stream:
+					// the snapshot will re-read and re-emit every row on the
+					// next connect.
+					p.logger.Errorf("a snapshot batch was rejected downstream (nacked); leaving the replication slot unpromoted so the snapshot re-runs on restart")
+					p.stopSig.TriggerSoftStop()
 				}
 				break
 			}
@@ -685,9 +726,16 @@ func (p *pgStreamInput) flushBatch(
 	// in the read loop).
 	isSnapshot := lsn == nil
 
-	ackFn := func(ctx context.Context, _ error) error {
+	ackFn := func(ctx context.Context, ackErr error) error {
 		if isSnapshot {
-			defer p.snapshotAckWG.Done()
+			// Record whether this batch settled via nack *before* the
+			// deferred snapshotAckDone runs, so the flag is guaranteed
+			// visible to the handoff by the time the drain signal (which
+			// snapshotAckDone sends) wakes it up.
+			if ackErr != nil {
+				p.snapshotAckFailed.Store(true)
+			}
+			defer p.snapshotAckDone()
 		}
 		maxOffset := resolveFn()
 		if maxOffset == nil {
@@ -703,19 +751,37 @@ func (p *pgStreamInput) flushBatch(
 		return nil
 	}
 	if isSnapshot {
-		p.snapshotAckWG.Add(1)
+		p.snapshotAckPending.Add(1)
 	}
 	select {
 	case p.msgChan <- asyncMessage{msg: batch, ackFn: ackFn}:
 	case <-ctx.Done():
 		if isSnapshot {
-			p.snapshotAckWG.Done()
+			p.snapshotAckDone()
 		}
 		return ctx.Err()
 	}
 	return nil
 }
 
+// snapshotAckDone marks one in-flight snapshot batch as settled and signals
+// the drain channel when the count reaches zero. A negative result means a
+// batch settled twice (e.g. both acked and nacked), which would otherwise
+// silently prevent the drain from ever completing, so we panic loudly
+// instead of promoting the slot on a corrupted counter.
+func (p *pgStreamInput) snapshotAckDone() {
+	n := p.snapshotAckPending.Add(-1)
+	if n < 0 {
+		panic("postgres cdc: snapshot ack counter went negative (batch settled twice)")
+	}
+	if n == 0 {
+		select {
+		case p.snapshotAckDrained <- struct{}{}:
+		default:
+		}
+	}
+}
+
 func (p *pgStreamInput) ReadBatch(ctx context.Context) (service.MessageBatch, service.AckFunc, error) {
 	select {
 	case m := <-p.msgChan:
diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go
index bcb67946a1..0b33654a70 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
@@ -398,6 +415,150 @@ pg_stream:
 	require.NoError(t, run2.StopWithin(10*time.Second))
 }
 
+// TestIntegrationPostgresSnapshotNackBarrier verifies that the snapshot->stream
+// handoff distinguishes an ack from a nack: settling the last in-flight
+// snapshot batch via a nack must not promote the replication slot, even though
+// the same "pending count reaches zero" condition fires either way. With
+// auto_replay_nacks disabled the raw nack reaches the input's ack func
+// directly (instead of being retried internally forever), which is what
+// exercises the bug: the handoff previously only checked whether the pending
+// count drained, not whether it drained because rows were durably delivered
+// or because they were rejected downstream. See CON-179.
+func TestIntegrationPostgresSnapshotNackBarrier(t *testing.T) {
+	integration.CheckSkip(t)
+	databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16")
+	require.NoError(t, err)
+
+	const rowCount = 5
+	for i := range rowCount {
+		f := pgtest.GetFakeFlightRecord()
+		_, err = db.Exec(`INSERT INTO "FlightsCompositePK" ("Seq", "Name", "CreatedAt") VALUES ($1, $2, $3);`, i, f.RealAddress.City, time.Unix(f.CreatedAt, 0).Format(time.RFC3339))
+		require.NoError(t, err)
+	}
+
+	// batching.count == rowCount forces all snapshot rows into a single output
+	// batch, so nacking that one batch nacks the *entire* snapshot in a single
+	// settle - exactly the "last in-flight snapshot batch settles via nack"
+	// scenario the fix targets. auto_replay_nacks is disabled so the nack
+	// reaches pgStreamInput's ack func as a real error instead of being
+	// silently retried forever by the framework.
+	template := fmt.Sprintf(`
+pg_stream:
+    dsn: %s
+    slot_name: test_slot_snapshot_nack_barrier
+    stream_snapshot: true
+    snapshot_batch_size: 1000
+    schema: public
+    tables:
+       - '"FlightsCompositePK"'
+    batching:
+      count: %d
+      period: 1h
+    auto_replay_nacks: false
+`, databaseURL, rowCount)
+
+	// Run 1: every batch delivered downstream is permanently rejected
+	// (nacked), including the lone snapshot batch. The snapshotAckPending
+	// counter drains to zero via that nack alone - no crash or teardown race
+	// required to reach the buggy state.
+	simulatedNackErr := errors.New("simulated downstream rejection")
+	received := make(chan struct{}, 1)
+	run1Builder := service.NewStreamBuilder()
+	require.NoError(t, run1Builder.SetLoggerYAML(`level: OFF`))
+	require.NoError(t, run1Builder.AddInputYAML(template))
+	require.NoError(t, run1Builder.AddBatchConsumerFunc(func(_ context.Context, _ service.MessageBatch) error {
+		select {
+		case received <- struct{}{}:
+		default:
+		}
+		return simulatedNackErr
+	}))
+	run1, err := run1Builder.Build()
+	require.NoError(t, err)
+	license.InjectTestService(run1.Resources())
+
+	run1Ctx, cancelRun1 := context.WithCancel(context.Background())
+	defer cancelRun1()
+	run1Done := make(chan struct{})
+	go func() {
+		defer close(run1Done)
+		_ = run1.Run(run1Ctx)
+	}()
+
+	select {
+	case <-received:
+	case <-time.After(30 * time.Second):
+		t.Fatal("snapshot rows were never delivered to the run-1 output")
+	}
+	// Give the handoff time to have acted on the nack - and, in the buggy
+	// version, to have mistaken the drain for a successful delivery and
+	// promoted the slot - before we inspect it.
+	time.Sleep(2 * time.Second)
+
+	// The nack must have prevented the temporary slot from being promoted to
+	// a permanent one, since the snapshot batch was rejected, not durably
+	// delivered. This is the core guarantee: without the fix, the permanent
+	// slot would exist here even though every row was nacked.
+	var permanentSlots int
+	require.NoError(t, db.QueryRow(`SELECT count(*) FROM pg_replication_slots WHERE slot_name = 'test_slot_snapshot_nack_barrier'`).Scan(&permanentSlots))
+	require.Zero(t, permanentSlots, "replication slot must not be promoted when the snapshot batch was nacked, not acked")
+
+	// Stop run 1. Unlike a crash, the nack path tears the stream down
+	// gracefully (the fix triggers a soft stop and the normal Close path
+	// runs), so no backend-termination trick should be needed here - but we
+	// still poll for the temporary slot's release before restarting, the same
+	// way the ack-barrier test does, since a graceful close racing the
+	// database's own bookkeeping is still asynchronous from this test's point
+	// of view.
+	cancelRun1()
+	select {
+	case <-run1Done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("run 1 did not stop after being cancelled")
+	}
+	if err := run1.StopWithin(10 * time.Second); err != nil {
+		t.Log(err)
+	}
+
+	require.Eventually(t, func() bool {
+		_, _ = db.Exec(`SELECT pg_terminate_backend(active_pid) FROM pg_replication_slots WHERE slot_name = 'test_slot_snapshot_nack_barrier_tmp' AND active_pid IS NOT NULL`)
+		var tmpSlots int
+		if err := db.QueryRow(`SELECT count(*) FROM pg_replication_slots WHERE slot_name = 'test_slot_snapshot_nack_barrier_tmp'`).Scan(&tmpSlots); err != nil {
+			return false
+		}
+		return tmpSlots == 0
+	}, 30*time.Second, 500*time.Millisecond, "temporary snapshot slot from run 1 was not released")
+
+	// Run 2: restart against the same slot. Since run 1's snapshot batch was
+	// nacked rather than acked, the slot must not have been promoted, so the
+	// snapshot re-runs and every row is delivered again.
+	var mu sync.Mutex
+	var reads int
+	run2Builder := service.NewStreamBuilder()
+	require.NoError(t, run2Builder.SetLoggerYAML(`level: OFF`))
+	require.NoError(t, run2Builder.AddInputYAML(template))
+	require.NoError(t, run2Builder.AddConsumerFunc(func(_ context.Context, m *service.Message) error {
+		if op, _ := m.MetaGet("operation"); op == "read" { // ReadOpType: snapshot row
+			mu.Lock()
+			reads++
+			mu.Unlock()
+		}
+		return nil
+	}))
+	run2, err := run2Builder.Build()
+	require.NoError(t, err)
+	license.InjectTestService(run2.Resources())
+	go func() { _ = run2.Run(t.Context()) }()
+
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		mu.Lock()
+		defer mu.Unlock()
+		assert.Equal(c, rowCount, reads, "snapshot should have re-run and re-delivered every row since run 1's snapshot batch was nacked, not acked")
+	}, 30*time.Second, 100*time.Millisecond)
+
+	require.NoError(t, run2.StopWithin(10*time.Second))
+}
+
 func TestIntegrationPgStreamingFromRemoteDB(t *testing.T) {
 	t.Skip("This test requires a remote database to run. Aimed to test remote databases")
 
diff --git a/internal/impl/postgresql/main_test.go b/internal/impl/postgresql/main_test.go
new file mode 100644
index 0000000000..5009fafea4
--- /dev/null
+++ b/internal/impl/postgresql/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 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"),
+	)
+}
diff --git a/internal/impl/protobuf/main_test.go b/internal/impl/protobuf/main_test.go
new file mode 100644
index 0000000000..f9225b8c46
--- /dev/null
+++ b/internal/impl/protobuf/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 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(),
+	)
+}
diff --git a/internal/impl/protobuf/multimodule_watcher.go b/internal/impl/protobuf/multimodule_watcher.go
index 299f37e228..906a287da7 100644
--- a/internal/impl/protobuf/multimodule_watcher.go
+++ b/internal/impl/protobuf/multimodule_watcher.go
@@ -53,7 +53,11 @@ import (
 	"github.com/redpanda-data/benthos/v4/public/service"
 )
 
-const watcherTimeout = 10 * time.Second
+// watcherTimeout bounds how long newSchemaWatcher waits for a freshly created
+// prototransform.SchemaWatcher to become ready. It is a variable (rather than
+// a const) so that tests can shorten it to make AwaitReady failures fast to
+// exercise.
+var watcherTimeout = 10 * time.Second
 
 type multiModuleWatcher struct {
 	bsrClients map[string]*prototransform.SchemaWatcher
@@ -69,6 +73,16 @@ func newMultiModuleWatcher(bsrModules []*service.ParsedConfig) (*multiModuleWatc
 
 	// Initialise one client for each module
 	multiModuleWatcher.bsrClients = make(map[string]*prototransform.SchemaWatcher)
+
+	// Any error return below drops out of the loop before every module has a
+	// watcher, so guard against leaking the watchers that already started.
+	ok := false
+	defer func() {
+		if !ok {
+			multiModuleWatcher.close()
+		}
+	}()
+
 	for _, bsrModule := range bsrModules {
 		var bsrURL string
 		bsrURL, err := bsrModule.FieldString(fieldBSRUrl)
@@ -91,6 +105,13 @@ func newMultiModuleWatcher(bsrModules []*service.ParsedConfig) (*multiModuleWatc
 			return nil, err
 		}
 
+		if _, exists := multiModuleWatcher.bsrClients[module]; exists {
+			// Without this check the earlier watcher's map entry would be
+			// silently overwritten below, leaving its running poll
+			// goroutine unreachable by close().
+			return nil, fmt.Errorf("duplicate BSR module %q", module)
+		}
+
 		watcher, err := newSchemaWatcher(context.Background(), bsrURL, bsrAPIKey, module, version)
 		if err != nil {
 			return nil, err
@@ -98,6 +119,7 @@ func newMultiModuleWatcher(bsrModules []*service.ParsedConfig) (*multiModuleWatc
 		multiModuleWatcher.bsrClients[module] = watcher
 	}
 
+	ok = true
 	return multiModuleWatcher, nil
 }
 
@@ -133,6 +155,7 @@ func newSchemaWatcher(ctx context.Context, bsrURL, bsrAPIKey, module, version st
 	ctxWithTimeout, cancel := context.WithTimeout(ctx, watcherTimeout)
 	defer cancel()
 	if err = watcher.AwaitReady(ctxWithTimeout); err != nil {
+		watcher.Stop()
 		return nil, fmt.Errorf("schema watcher never became ready: %w", err)
 	}
 
@@ -208,3 +231,12 @@ func (w *multiModuleWatcher) FindEnumByName(enum protoreflect.FullName) (protore
 	}
 	return nil, fmt.Errorf("could not find %s in any loaded modules", enum)
 }
+
+// close stops every schema watcher owned by w, cancelling their background
+// polling goroutines. It is safe to call multiple times as Stop() is
+// idempotent.
+func (w *multiModuleWatcher) close() {
+	for _, schemaWatcher := range w.bsrClients {
+		schemaWatcher.Stop()
+	}
+}
diff --git a/internal/impl/protobuf/processor_protobuf.go b/internal/impl/protobuf/processor_protobuf.go
index 3a74ac56d9..170f6cc8cf 100644
--- a/internal/impl/protobuf/processor_protobuf.go
+++ b/internal/impl/protobuf/processor_protobuf.go
@@ -534,6 +534,7 @@ func newProtobuf(conf *service.ParsedConfig, mgr *service.Resources) (*protobufP
 			return nil, fmt.Errorf("creating multiModuleWatcher: %w", err)
 		}
 		if p.operator, err = strToProtobufBSROperator(p.multiModuleWatcher, operatorStr, message, opts); err != nil {
+			p.multiModuleWatcher.close()
 			return nil, err
 		}
 	} else {
@@ -557,6 +558,9 @@ func (p *protobufProc) Process(_ context.Context, msg *service.Message) (service
 	return service.MessageBatch{msg}, nil
 }
 
-func (*protobufProc) Close(context.Context) error {
+func (p *protobufProc) Close(context.Context) error {
+	if p.multiModuleWatcher != nil {
+		p.multiModuleWatcher.close()
+	}
 	return nil
 }
diff --git a/internal/impl/protobuf/processor_protobuf_test.go b/internal/impl/protobuf/processor_protobuf_test.go
index d7f6fd01d9..dad05c45c2 100644
--- a/internal/impl/protobuf/processor_protobuf_test.go
+++ b/internal/impl/protobuf/processor_protobuf_test.go
@@ -44,6 +44,7 @@ import (
 	"net/http"
 	"strconv"
 	"testing"
+	"time"
 
 	"buf.build/gen/go/bufbuild/reflect/connectrpc/go/buf/reflect/v1beta1/reflectv1beta1connect"
 	v1beta1 "buf.build/gen/go/bufbuild/reflect/protocolbuffers/go/buf/reflect/v1beta1"
@@ -133,6 +134,7 @@ discard_unknown: %t
 
 			proc, err := newProtobuf(conf, service.MockResources())
 			require.NoError(t, err)
+			t.Cleanup(func() { require.NoError(t, proc.Close(context.Background())) })
 
 			msgs, res := proc.Process(t.Context(), service.NewMessage([]byte(test.input)))
 			require.NoError(t, res)
@@ -163,6 +165,7 @@ discard_unknown: %t
 
 			proc, err := newProtobuf(conf, service.MockResources())
 			require.NoError(t, err)
+			t.Cleanup(func() { require.NoError(t, proc.Close(context.Background())) })
 
 			msgs, res := proc.Process(t.Context(), service.NewMessage([]byte(test.input)))
 			require.NoError(t, res)
@@ -278,6 +281,7 @@ use_enum_numbers: %t
 
 			proc, err := newProtobuf(conf, service.MockResources())
 			require.NoError(t, err)
+			t.Cleanup(func() { require.NoError(t, proc.Close(context.Background())) })
 
 			msgs, res := proc.Process(t.Context(), service.NewMessage(test.input))
 			require.NoError(t, res)
@@ -306,6 +310,7 @@ use_enum_numbers: %t
 
 			proc, err := newProtobuf(conf, service.MockResources())
 			require.NoError(t, err)
+			t.Cleanup(func() { require.NoError(t, proc.Close(context.Background())) })
 
 			msgs, res := proc.Process(t.Context(), service.NewMessage(test.input))
 			require.NoError(t, res)
@@ -368,6 +373,7 @@ import_paths: [ %v ]
 
 			proc, err := newProtobuf(conf, service.MockResources())
 			require.NoError(t, err)
+			t.Cleanup(func() { require.NoError(t, proc.Close(context.Background())) })
 
 			_, err = proc.Process(t.Context(), service.NewMessage([]byte(test.input)))
 			require.Error(t, err)
@@ -462,11 +468,122 @@ 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)
 		}
 	}()
 
 	return listener.Addr().String()
 }
+
+// TestProtobufBSRConstructionFailureStopsWatchers ensures that a
+// construction-time failure after the BSR watchers have started (in this
+// case, an operator that references a message the loaded module doesn't
+// resolve) still stops those watchers rather than leaking their background
+// polling goroutines. newProtobuf returns nil on error so there's no
+// processor to Close here; the package's goleak TestMain is what proves the
+// watchers were actually stopped.
+func TestProtobufBSRConstructionFailureStopsWatchers(t *testing.T) {
+	mockBSRServerAddress := runMockBSRServer(t, "../../../config/test/protobuf/schema")
+
+	conf, err := protobufProcessorSpec().ParseYAML(fmt.Sprintf(`
+operator: from_json
+message: does.not.Exist
+bsr:
+  - module: "testing"
+    url: %s
+`, "http://"+mockBSRServerAddress), nil)
+	require.NoError(t, err)
+
+	proc, err := newProtobuf(conf, service.MockResources())
+	require.Error(t, err)
+	require.Nil(t, proc)
+}
+
+// TestProtobufBSRPartialModuleFailureStopsWatchers ensures that when
+// newMultiModuleWatcher fails partway through constructing watchers for
+// multiple `bsr` modules, any watcher that had already started for an
+// earlier module is stopped rather than leaked. The second module here
+// ("invalid") has no `url` and fails to even parse in newSchemaWatcher
+// (before a watcher for it is created), which triggers the `ok`/deferred
+// close() guard in newMultiModuleWatcher for the first module's
+// already-running watcher. newProtobuf returns nil on error so there's no
+// processor to Close here; the package's goleak TestMain is what proves the
+// first module's watcher was actually stopped.
+func TestProtobufBSRPartialModuleFailureStopsWatchers(t *testing.T) {
+	mockBSRServerAddress := runMockBSRServer(t, "../../../config/test/protobuf/schema")
+
+	conf, err := protobufProcessorSpec().ParseYAML(fmt.Sprintf(`
+operator: from_json
+message: testing.Person
+bsr:
+  - module: "testing"
+    url: %s
+  - module: "invalid"
+`, "http://"+mockBSRServerAddress), nil)
+	require.NoError(t, err)
+
+	proc, err := newProtobuf(conf, service.MockResources())
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "expected three segments")
+	require.Nil(t, proc)
+}
+
+// TestProtobufBSRDuplicateModuleStopsWatchers ensures that when the `bsr`
+// config lists the same module twice, newMultiModuleWatcher rejects the
+// config instead of silently overwriting the first module's map entry with
+// the second, which would otherwise leave the first watcher's background
+// polling goroutine unreachable by close(). newProtobuf returns nil on error
+// so there's no processor to Close here; the package's goleak TestMain is
+// what proves the first watcher was actually stopped.
+func TestProtobufBSRDuplicateModuleStopsWatchers(t *testing.T) {
+	mockBSRServerAddress := runMockBSRServer(t, "../../../config/test/protobuf/schema")
+
+	conf, err := protobufProcessorSpec().ParseYAML(fmt.Sprintf(`
+operator: from_json
+message: testing.Person
+bsr:
+  - module: "testing"
+    url: %[1]s
+  - module: "testing"
+    url: %[1]s
+`, "http://"+mockBSRServerAddress), nil)
+	require.NoError(t, err)
+
+	proc, err := newProtobuf(conf, service.MockResources())
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "duplicate BSR module")
+	require.Nil(t, proc)
+}
+
+// TestProtobufBSRAwaitReadyFailureStopsWatcher ensures that when a schema
+// watcher's AwaitReady call fails (e.g. because the BSR endpoint is
+// unreachable), newSchemaWatcher stops the watcher before returning an error
+// rather than leaking its background polling goroutine. newProtobuf returns
+// nil on error so there's no processor to Close here; the package's goleak
+// TestMain is what proves the watcher was actually stopped.
+func TestProtobufBSRAwaitReadyFailureStopsWatcher(t *testing.T) {
+	original := watcherTimeout
+	watcherTimeout = 250 * time.Millisecond
+	t.Cleanup(func() { watcherTimeout = original })
+
+	conf, err := protobufProcessorSpec().ParseYAML(`
+operator: from_json
+message: testing.Person
+bsr:
+  - module: "testing"
+    url: http://127.0.0.1:1
+`, nil)
+	require.NoError(t, err)
+
+	proc, err := newProtobuf(conf, service.MockResources())
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "schema watcher never became ready")
+	require.Nil(t, proc)
+}
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(),
+	)
+}
diff --git a/taskfiles/test.yml b/taskfiles/test.yml
index 8cc1cc36bc..73d7eb01ee 100644
--- a/taskfiles/test.yml
+++ b/taskfiles/test.yml
@@ -11,11 +11,11 @@ tasks:
       - go test {{.GO_FLAGS}} -timeout {{.TIMEOUT}} -shuffle=on {{if .CI}}{{else}}-v{{end}} ./...
 
   unit-race:
-    desc: Run unit tests with race detection
+    desc: "Run unit tests with race detection, optionally scoped to a package (e.g. task test:unit-race -- ./internal/impl/kafka/). Flags mirror the race-detector CI job (race_test.yml) so local results can't diverge via the test cache."
     aliases:
       - ut-race
     cmds:
-      - go test {{.GO_FLAGS}} -timeout 3m -shuffle=on -race {{if .CI}}{{else}}-v{{end}} ./...
+      - go test {{.GO_FLAGS}} -count=1 -timeout 10m -shuffle=on -race {{if .CI}}{{else}}-v{{end}} {{.CLI_ARGS | default "./..."}}
 
   integration:
     desc: "Run integration tests (e.g. task test:integration -- aws kafka). See cmd/tools/integration/README.md"