From a3a212e810ab9286babcabbcd4fa06c2d8b6c20e Mon Sep 17 00:00:00 2001
From: prakhargarg105 <prakhar.garg@redpanda.com>
Date: Wed, 12 Aug 2026 09:25:00 -0700
Subject: [PATCH 01/10] test(CON-179 R2): goroutine-leak detection via goleak
 in starting-set packages
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Wire goleak.VerifyTestMain into internal/impl/{protobuf,sql,postgresql,mysql}
— the packages that shipped leak bugs (9caed49f7, 07c39cd94) plus the two
flagship CDC connector packages — with narrow, commented per-package ignore
lists, each verified green on a clean baseline (unit x3, -race x2,
postgres/mysql integration suites under Docker).

Two real test-side leaks surfaced and fixed rather than ignored:
- protobuf: runMockBSRServer used bare http.Serve with no shutdown, leaking
  the accept loop and connection goroutines; now an http.Server closed via
  t.Cleanup.
- postgresql: TestIntegrationPostgresSnapshotAckBarrier's consumer waited on
  a context benthos never cancels (runConsumerFunc passes Background), and
  run 1 was never stopped because Stream.Run returns on ctx cancel without
  tearing the stream down; now selects on the test-owned context and stops
  the crashed stream explicitly.

Also documents the "adding goleak to a package" recipe in the tester agent
guide, per the R2 acceptance criteria.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .claude/agents/tester.md                      | 23 +++++++++++
 go.mod                                        |  1 +
 internal/impl/mysql/main_test.go              | 26 +++++++++++++
 internal/impl/postgresql/integration_test.go  | 25 ++++++++++--
 internal/impl/postgresql/main_test.go         | 31 +++++++++++++++
 internal/impl/protobuf/main_test.go           | 39 +++++++++++++++++++
 .../impl/protobuf/processor_protobuf_test.go  |  7 +++-
 internal/impl/sql/main_test.go                | 29 ++++++++++++++
 8 files changed, 176 insertions(+), 5 deletions(-)
 create mode 100644 internal/impl/mysql/main_test.go
 create mode 100644 internal/impl/postgresql/main_test.go
 create mode 100644 internal/impl/protobuf/main_test.go
 create mode 100644 internal/impl/sql/main_test.go

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/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(),
+	)
+}

From 500a374d856b88565a5b0a70ac7c612a1a881d0c Mon Sep 17 00:00:00 2001
From: prakhargarg105 <prakhar.garg@redpanda.com>
Date: Wed, 12 Aug 2026 09:25:08 -0700
Subject: [PATCH 02/10] ci(CON-179 R3): advisory race-detector job scoped to
 changed connector packages
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add race_test.yml: on every PR it runs go test -race against only the
internal/impl/<component> packages the PR touches, using the same
change-detection as integration_test.yml to bound the 2-10x race-detector
cost. Failures surface as warning annotations and a step-summary table, not
a red check — unless the package is listed in
.github/race-blocking-packages.txt, the promotion list for packages that
have earned a green -race baseline. The list starts empty so the gate can
never fail on a pre-existing race a PR didn't touch.

Also reference task test:unit-race in the CONTRIBUTING §6 pre-PR checklist,
per the R3 acceptance criteria.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .github/race-blocking-packages.txt | 11 ++++
 .github/workflows/race_test.yml    | 94 ++++++++++++++++++++++++++++++
 CONTRIBUTING.md                    |  1 +
 3 files changed, 106 insertions(+)
 create mode 100644 .github/race-blocking-packages.txt
 create mode 100644 .github/workflows/race_test.yml

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/<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
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/<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.

From 58255a01d0fd6361e209725a3f2f223528e5b88d Mon Sep 17 00:00:00 2001
From: prakhargarg105 <prakhar.garg@redpanda.com>
Date: Mon, 24 Aug 2026 10:25:14 -0700
Subject: [PATCH 03/10] protobuf: stop BSR schema watchers on processor Close,
 drop goleak ignores
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Review on #4721 flagged that the goleak ignores in main_test.go were
masking a real production leak: multiModuleWatcher created each
prototransform.SchemaWatcher with context.Background() and no stop
path, so every BSR-backed protobuf processor leaked a poll goroutine
per module on each construction. The original justification ("no stop
hook") was wrong — SchemaWatcher exposes Stop().

- multiModuleWatcher gains close(), which stops every watcher.
- protobufProc.Close calls it, giving the poll loops a shutdown path.
- BSR tests now Close their processors via t.Cleanup.
- All three goleak ignores (prototransform poll loop and the two
  open-ended net/http persistConn entries) are removed; goleak now
  guards against regressions of this leak instead of hiding it.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 internal/impl/protobuf/main_test.go               | 10 ----------
 internal/impl/protobuf/multimodule_watcher.go     |  9 +++++++++
 internal/impl/protobuf/processor_protobuf.go      |  5 ++++-
 internal/impl/protobuf/processor_protobuf_test.go |  5 +++++
 4 files changed, 18 insertions(+), 11 deletions(-)

diff --git a/internal/impl/protobuf/main_test.go b/internal/impl/protobuf/main_test.go
index 6e7544680f..f9225b8c46 100644
--- a/internal/impl/protobuf/main_test.go
+++ b/internal/impl/protobuf/main_test.go
@@ -25,15 +25,5 @@ import (
 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/multimodule_watcher.go b/internal/impl/protobuf/multimodule_watcher.go
index 299f37e228..73823400f5 100644
--- a/internal/impl/protobuf/multimodule_watcher.go
+++ b/internal/impl/protobuf/multimodule_watcher.go
@@ -208,3 +208,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..794b649c0c 100644
--- a/internal/impl/protobuf/processor_protobuf.go
+++ b/internal/impl/protobuf/processor_protobuf.go
@@ -557,6 +557,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 6d5aff2dec..613bcbe4b1 100644
--- a/internal/impl/protobuf/processor_protobuf_test.go
+++ b/internal/impl/protobuf/processor_protobuf_test.go
@@ -133,6 +133,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 +164,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 +280,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 +309,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 +372,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)

From 12a5b01d4a01a5bc501a8823da9ee5dc4ba6760d Mon Sep 17 00:00:00 2001
From: prakhargarg105 <prakhar.garg@redpanda.com>
Date: Mon, 24 Aug 2026 10:48:28 -0700
Subject: [PATCH 04/10] protobuf: stop BSR watchers on construction-failure
 paths
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Follow-up review on #4721: wiring close() into protobufProc.Close fixed
the happy path, but watcher poll goroutines still leaked when
construction failed after watchers were started:

- newProtobuf: newMultiModuleWatcher succeeds, then
  strToProtobufBSROperator fails (e.g. message not resolvable in any
  BSR module) — the framework never gets a processor to Close.
- newMultiModuleWatcher: an error on a later module dropped
  already-running watchers for earlier modules.

Both paths now stop the accumulated watchers before returning the
error. Adds TestProtobufBSRConstructionFailureStopsWatchers covering
the unresolvable-message path; the package's goleak TestMain is what
verifies the watchers actually stop (removing the newProtobuf close()
call makes goleak fail with the prototransform poll goroutine).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 internal/impl/protobuf/multimodule_watcher.go | 11 +++++++++
 internal/impl/protobuf/processor_protobuf.go  |  1 +
 .../impl/protobuf/processor_protobuf_test.go  | 24 +++++++++++++++++++
 3 files changed, 36 insertions(+)

diff --git a/internal/impl/protobuf/multimodule_watcher.go b/internal/impl/protobuf/multimodule_watcher.go
index 73823400f5..aeef4fac5e 100644
--- a/internal/impl/protobuf/multimodule_watcher.go
+++ b/internal/impl/protobuf/multimodule_watcher.go
@@ -69,6 +69,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)
@@ -98,6 +108,7 @@ func newMultiModuleWatcher(bsrModules []*service.ParsedConfig) (*multiModuleWatc
 		multiModuleWatcher.bsrClients[module] = watcher
 	}
 
+	ok = true
 	return multiModuleWatcher, nil
 }
 
diff --git a/internal/impl/protobuf/processor_protobuf.go b/internal/impl/protobuf/processor_protobuf.go
index 794b649c0c..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 {
diff --git a/internal/impl/protobuf/processor_protobuf_test.go b/internal/impl/protobuf/processor_protobuf_test.go
index 613bcbe4b1..6fd9dc5fa8 100644
--- a/internal/impl/protobuf/processor_protobuf_test.go
+++ b/internal/impl/protobuf/processor_protobuf_test.go
@@ -480,3 +480,27 @@ func runMockBSRServer(t *testing.T, importPath string) string {
 
 	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)
+}

From 8aae4ce30e32ff0de36e8b18c684a2c258f377ac Mon Sep 17 00:00:00 2001
From: prakhargarg105 <prakhar.garg@redpanda.com>
Date: Mon, 24 Aug 2026 11:05:57 -0700
Subject: [PATCH 05/10] protobuf, postgres_cdc: fix remaining goroutine leaks
 flagged by goleak review
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two more leak paths from review on #4721:

protobuf: prototransform.NewSchemaWatcher starts its poll loop
immediately, so when AwaitReady failed (unreachable BSR URL, rejected
API key) newSchemaWatcher returned without Stop() and the watcher never
reached bsrClients — no cleanup path could touch it. Stop the watcher
before returning the error.

postgres_cdc: the snapshot->stream handoff parked a goroutine on
snapshotAckWG.Wait(); a soft stop with unacked snapshot batches
abandoned it for the life of the process (WaitGroup.Wait cannot be
interrupted). Replace the WaitGroup with an atomic pending counter plus
a buffered drain-notification channel so the handoff waits directly on
drain-or-soft-stop with no helper goroutine at all. Semantics are
unchanged: block until acks drain or soft stop, and only promote the
replication slot when actually drained.

The processStream.func3 goleak ignore is removed from the postgresql
TestMain; TestIntegrationPostgresSnapshotAckBarrier (which triggers
exactly the previously-leaking path) passes under the stricter check.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 internal/impl/postgresql/input_pg_stream.go   | 53 ++++++++++++-------
 internal/impl/postgresql/main_test.go         |  5 --
 internal/impl/protobuf/multimodule_watcher.go |  1 +
 3 files changed, 36 insertions(+), 23 deletions(-)

diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go
index 8ff0e62374..6e63e462b8 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,15 @@ 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{}
 
 	// IAM authentication fields
 	iamAuthEnabled bool
@@ -572,8 +578,9 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher
 				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 +589,16 @@ 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:
+			drainWait:
+				for p.snapshotAckPending.Load() > 0 {
+					select {
+					case <-p.snapshotAckDrained:
+					case <-p.stopSig.SoftStopChan():
+						break drainWait
+					}
+				}
+				if p.snapshotAckPending.Load() == 0 {
 					pgStream.MarkSnapshotAcknowledged()
-				case <-p.stopSig.SoftStopChan():
 				}
 				break
 			}
@@ -687,7 +693,7 @@ func (p *pgStreamInput) flushBatch(
 
 	ackFn := func(ctx context.Context, _ error) error {
 		if isSnapshot {
-			defer p.snapshotAckWG.Done()
+			defer p.snapshotAckDone()
 		}
 		maxOffset := resolveFn()
 		if maxOffset == nil {
@@ -703,19 +709,30 @@ 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.
+func (p *pgStreamInput) snapshotAckDone() {
+	if p.snapshotAckPending.Add(-1) == 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/main_test.go b/internal/impl/postgresql/main_test.go
index fa5d68df20..5009fafea4 100644
--- a/internal/impl/postgresql/main_test.go
+++ b/internal/impl/postgresql/main_test.go
@@ -22,10 +22,5 @@ func TestMain(m *testing.M) {
 		// 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/multimodule_watcher.go b/internal/impl/protobuf/multimodule_watcher.go
index aeef4fac5e..fa848d315e 100644
--- a/internal/impl/protobuf/multimodule_watcher.go
+++ b/internal/impl/protobuf/multimodule_watcher.go
@@ -144,6 +144,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)
 	}
 

From 5614d2804094472df8652ed00067055d1df196b9 Mon Sep 17 00:00:00 2001
From: prakhargarg105 <prakhar.garg@redpanda.com>
Date: Mon, 24 Aug 2026 11:41:44 -0700
Subject: [PATCH 06/10] protobuf: add regression tests for watcher
 construction-failure leak paths
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Review on #4721 noted two of the leak fixes had no test exercising
them, so the package goleak gate could not catch a regression:

- TestProtobufBSRPartialModuleFailureStopsWatchers: two-module config
  where the first module's watcher starts against the mock BSR server
  and the second fails module parsing, covering the deferred close()
  guard in newMultiModuleWatcher.
- TestProtobufBSRAwaitReadyFailureStopsWatcher: unreachable BSR
  endpoint, covering the watcher.Stop() in newSchemaWatcher's
  AwaitReady error branch. watcherTimeout becomes a var so the test
  can shorten it to 250ms — prototransform retries until the
  constructor context dies, so with the const the failure always took
  the full 10s.

Both tests verified negatively: removing either fix makes the package
run fail with the corresponding leaked prototransform goroutine.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 internal/impl/protobuf/multimodule_watcher.go |  6 +-
 .../impl/protobuf/processor_protobuf_test.go  | 56 +++++++++++++++++++
 2 files changed, 61 insertions(+), 1 deletion(-)

diff --git a/internal/impl/protobuf/multimodule_watcher.go b/internal/impl/protobuf/multimodule_watcher.go
index fa848d315e..da8dac3f7a 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
diff --git a/internal/impl/protobuf/processor_protobuf_test.go b/internal/impl/protobuf/processor_protobuf_test.go
index 6fd9dc5fa8..b41ddcb222 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"
@@ -504,3 +505,58 @@ bsr:
 	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)
+}
+
+// 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)
+}

From e3b9308fe4ddb81a6835d20b5672178a53665082 Mon Sep 17 00:00:00 2001
From: prakhargarg105 <prakhar.garg@redpanda.com>
Date: Mon, 24 Aug 2026 12:39:41 -0700
Subject: [PATCH 07/10] postgres_cdc, protobuf: fix slot promotion on abandoned
 handoff and dup-module watcher orphaning

From a high-effort review pass over this branch:

postgres_cdc: after the snapshot handoff was abandoned on soft stop,
the trailing counter re-check could still call MarkSnapshotAcknowledged
when teardown nacks drained the counter (reachable with
auto_replay_nacks: false, where nacks hit the raw ackFn), promoting the
replication slot over rows never durably delivered and skipping the
snapshot on restart. Mark is now gated on a drain observed via the
drain channel, never the soft-stop exit. Also restore the WaitGroup's
lost double-settle safety net: snapshotAckDone panics if the counter
goes negative instead of silently disabling the drain signal forever.

protobuf: two bsr entries naming the same module silently overwrote the
first watcher in bsrClients, leaving its running poll goroutine
unreachable by close(). Duplicate modules are now a config error (the
construction-failure guard stops the earlier watchers); covered by
TestProtobufBSRDuplicateModuleStopsWatchers, negatively verified
against the goleak gate.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 internal/impl/postgresql/input_pg_stream.go   | 26 ++++++++++++++----
 internal/impl/protobuf/multimodule_watcher.go |  7 +++++
 .../impl/protobuf/processor_protobuf_test.go  | 27 +++++++++++++++++++
 3 files changed, 55 insertions(+), 5 deletions(-)

diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go
index 6e63e462b8..2d5ceea0d7 100644
--- a/internal/impl/postgresql/input_pg_stream.go
+++ b/internal/impl/postgresql/input_pg_stream.go
@@ -572,7 +572,10 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher
 				// 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 acks drain or soft-stop (no timeout, by design). If
+				// soft-stop wins the race, the slot is left unpromoted since
+				// batches settling during teardown are nacks, not durable
+				// delivery, so the snapshot re-runs on restart.
 				nextTimedBatchChan = nil
 				flushedBatch, err := batcher.Flush(ctx)
 				if err != nil {
@@ -589,15 +592,21 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher
 					p.stopSig.TriggerSoftStop()
 					break
 				}
+				drained := p.snapshotAckPending.Load() == 0
 			drainWait:
-				for p.snapshotAckPending.Load() > 0 {
+				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
 					}
 				}
-				if p.snapshotAckPending.Load() == 0 {
+				if drained {
 					pgStream.MarkSnapshotAcknowledged()
 				}
 				break
@@ -723,9 +732,16 @@ func (p *pgStreamInput) flushBatch(
 }
 
 // snapshotAckDone marks one in-flight snapshot batch as settled and signals
-// the drain channel when the count reaches zero.
+// 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() {
-	if p.snapshotAckPending.Add(-1) == 0 {
+	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:
diff --git a/internal/impl/protobuf/multimodule_watcher.go b/internal/impl/protobuf/multimodule_watcher.go
index da8dac3f7a..906a287da7 100644
--- a/internal/impl/protobuf/multimodule_watcher.go
+++ b/internal/impl/protobuf/multimodule_watcher.go
@@ -105,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
diff --git a/internal/impl/protobuf/processor_protobuf_test.go b/internal/impl/protobuf/processor_protobuf_test.go
index b41ddcb222..dad05c45c2 100644
--- a/internal/impl/protobuf/processor_protobuf_test.go
+++ b/internal/impl/protobuf/processor_protobuf_test.go
@@ -535,6 +535,33 @@ bsr:
 	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

From a96ac3a98d3c96f82d7ddc80f60dd8ef857df0ef Mon Sep 17 00:00:00 2001
From: prakhargarg105 <prakhar.garg@redpanda.com>
Date: Mon, 24 Aug 2026 12:39:52 -0700
Subject: [PATCH 08/10] ci: harden race-detector job and make scoped unit-race
 task real

Review findings against the new advisory race job:

- Classify failures by grepping output for 'WARNING: DATA RACE': only
  an actual race in a blocking-listed package can exit 1. Build errors,
  flakes, and per-package timeouts are advisory warnings even for
  blocking packages (the ordinary unit-test workflow gates those).
- Write each package's summary row incrementally and cap the run at
  MAX_PACKAGES=8 with a logged advisory truncation, with job
  timeout-minutes raised above the composed worst case, so a job-level
  kill can never turn a wide PR red without a race.
- Pass github.base_ref via env and quote it (git ref names may contain
  shell metacharacters), and split merge-base/diff into checked
  statements under pipefail so git failures fail the step loudly
  instead of silently scoping to zero packages. integration_test.yml
  carries the same two inherited flaws; fixing it (or extracting a
  shared composite action) is left as a follow-up to keep this PR from
  touching working CI.
- test:unit-race now accepts package paths (task test:unit-race --
  ./internal/impl/foo/) and mirrors the CI flags (-count=1 -race
  -shuffle=on -timeout 10m) so the documented promotion baseline is
  actually executable and immune to the test cache; CONTRIBUTING.md
  shows the scoped invocation.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .github/workflows/race_test.yml | 115 ++++++++++++++++++++++++--------
 CONTRIBUTING.md                 |   2 +-
 taskfiles/test.yml              |   4 +-
 3 files changed, 90 insertions(+), 31 deletions(-)

diff --git a/.github/workflows/race_test.yml b/.github/workflows/race_test.yml
index 84a772e40e..1022bdeffd 100644
--- a/.github/workflows/race_test.yml
+++ b/.github/workflows/race_test.yml
@@ -23,7 +23,17 @@ on:
 jobs:
   race-test:
     runs-on: ubuntu-latest
-    timeout-minutes: 45
+    # 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
@@ -36,14 +46,22 @@ jobs:
       - name: Detect changed packages
         if: ${{ github.event_name == 'pull_request' }}
         id: detect
+        env:
+          BASE_REF: ${{ github.base_ref }}
         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' ' '
-          )
+          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; })
+          FILTERS=$(printf '%s\n' "${MATCHED}" | cut -d/ -f3 | sort -u | tr '\n' ' ')
           echo "filters=${FILTERS}" >> "$GITHUB_OUTPUT"
 
       - name: Install Go
@@ -57,38 +75,79 @@ jobs:
         env:
           FILTER: ${{ steps.detect.outputs.filters || github.event.inputs.filter }}
         run: |
+          set -uo pipefail
           BLOCKING_FILE=.github/race-blocking-packages.txt
-          PASSED=()
-          ADVISORY_FAILED=()
-          BLOCKING_FAILED=()
+          # 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}" ] || 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::"
+            [ -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 "| --- | --- |"
-            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
+          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 eb368456f2..0fd78655ec 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -169,7 +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 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/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"

From f3246dcab7f39e915bf82d86189745b825c60d15 Mon Sep 17 00:00:00 2001
From: prakhargarg105 <prakhar.garg@redpanda.com>
Date: Mon, 24 Aug 2026 12:55:06 -0700
Subject: [PATCH 09/10] ci: emit truly empty filters when no connector packages
 changed
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Review on #4721: tr '\n' ' ' turned the blank-input case into a lone
space, which survives command substitution, so the 'filters != empty'
guards never skipped — every docs-only PR still ran checkout with
fetch-depth 0, installed Go, and posted an empty results table. Join
the package list with xargs instead, which emits an empty string on
blank input (verified locally for both the empty and real-diff cases).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .github/workflows/race_test.yml | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/.github/workflows/race_test.yml b/.github/workflows/race_test.yml
index 1022bdeffd..0f7707daec 100644
--- a/.github/workflows/race_test.yml
+++ b/.github/workflows/race_test.yml
@@ -61,7 +61,12 @@ jobs:
           # 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; })
-          FILTERS=$(printf '%s\n' "${MATCHED}" | cut -d/ -f3 | sort -u | tr '\n' ' ')
+          # 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

From 4777e0073759de523588965143c2789cad7f19b8 Mon Sep 17 00:00:00 2001
From: prakhargarg105 <prakhar.garg@redpanda.com>
Date: Mon, 24 Aug 2026 13:12:19 -0700
Subject: [PATCH 10/10] postgres_cdc: never promote the replication slot off a
 nack-drained snapshot barrier
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Follow-up review on #4721: gating MarkSnapshotAcknowledged on the drain
channel still conflated acks with nacks — ackFn discarded its error, so
with auto_replay_nacks: false a downstream that permanently rejects the
final snapshot batch drained the counter, fired the drain, and promoted
the slot over rows never durably delivered (deterministically, no soft
stop needed). The soft-stop-vs-drain select race had the same hole.

ackFn now records a sticky snapshotAckFailed flag before the counter
decrement, and the handoff promotes only on a clean drain; a nack drain
logs and soft-stops so the stream restarts and re-runs the snapshot,
matching the neighboring flush-failure paths. The flag resets on
Connect (a fresh snapshot re-reads the rows); the pending counter
deliberately does not (late cross-epoch settles still decrement it).

Adds TestIntegrationPostgresSnapshotNackBarrier: auto_replay_nacks
false, the whole snapshot in one batch, consumer nacks it — asserts the
slot is never promoted and the snapshot re-runs on restart. Negatively
verified: with the fix reverted the test fails with the slot promoted.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 internal/impl/postgresql/input_pg_stream.go  |  49 +++++--
 internal/impl/postgresql/integration_test.go | 144 +++++++++++++++++++
 2 files changed, 185 insertions(+), 8 deletions(-)

diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go
index 2d5ceea0d7..0b03bb1675 100644
--- a/internal/impl/postgresql/input_pg_stream.go
+++ b/internal/impl/postgresql/input_pg_stream.go
@@ -497,6 +497,12 @@ type pgStreamInput struct {
 	// 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
@@ -520,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
 }
@@ -570,12 +582,15 @@ 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). If
-				// soft-stop wins the race, the slot is left unpromoted since
-				// batches settling during teardown are nacks, not durable
-				// delivery, so the snapshot re-runs on restart.
+				// 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 {
@@ -606,8 +621,19 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher
 						break drainWait
 					}
 				}
-				if drained {
+				switch {
+				case drained && !p.snapshotAckFailed.Load():
 					pgStream.MarkSnapshotAcknowledged()
+				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
 			}
@@ -700,8 +726,15 @@ 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 {
+			// 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()
diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go
index 4c23e632a4..0b33654a70 100644
--- a/internal/impl/postgresql/integration_test.go
+++ b/internal/impl/postgresql/integration_test.go
@@ -415,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")