Classify mid-CDC SSH tunnel failures as connectivity errors (PeerDB-io#4443)

## What

When an SSH tunnel's keepalive fails mid-CDC, the Postgres connector
proactively closes the replication connection. That surfaces in the CDC
loop as a generic `ReceiveMessage failed` error which — depending on the
underlying type — could fall through to the unclassified `OTHER` bucket
(`NotifyTelemetry`) and **page us**, even though tunnel failures are
retriable and almost always customer-side.

SSH *setup* failures were already classified as `NOTIFY_CONNECTIVITY`;
this closes the gap for failures that happen *during* CDC while the
tunnel is open.

## Changes

- New `SSHTunnelConnectionError` exception type.
- CDC `PullCdcRecords` wraps the read error in it when `p.ssh.IsBad()`.
- Classifier maps it to `NOTIFY_CONNECTIVITY` (notify user, no page),
placed before the generic `net.*` matchers so the tunnel context isn't
lost.
- `SSHTunnel.badTunnel` is now `atomic.Bool` (read from the CDC loop) +
nil-safe `IsBad()` accessor.
- Added classifier unit test.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: dtunikov

flow/alerting/classifier.go

@@ -418,6 +418,14 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	// An SSH-tunneled connection that goes bad (e.g. keepalive failure) is wrapped explicitly.
+	if _, ok := errors.AsType[*exceptions.SSHTunnelConnectionError](err); ok {
+		return ErrorNotifyConnectivity, ErrorInfo{
+			Source: ErrorSourceSSH,
+			Code:   "TUNNEL_CONNECTION_LOST",
+		}
+	}
+
 	// Other connection reset errors can mostly be ignored
 	if errors.Is(err, syscall.ECONNRESET) {
 		return ErrorIgnoreConnTemporary, ErrorInfo{

flow/alerting/classifier_test.go

@@ -57,6 +57,20 @@ func TestOtherDNSErrorsShouldBeConnectivity(t *testing.T) {
 	assert.Regexp(t, "^lookup "+hostName+"( on [\\w\\d\\.:]*)?: no such host$", errInfo.Code, "Unexpected error code")
 }
 
+func TestSSHTunnelConnectionErrorShouldBeConnectivity(t *testing.T) {
+	t.Parallel()
+
+	// Mirrors how the Postgres CDC loop wraps a read failure once the SSH tunnel has gone bad.
+	err := fmt.Errorf("error in PullRecords: %w",
+		exceptions.NewSSHTunnelConnectionError(fmt.Errorf("ReceiveMessage failed: %w", net.ErrClosed)))
+	errorClass, errInfo := GetErrorClass(t.Context(), err)
+	assert.Equal(t, ErrorNotifyConnectivity, errorClass, "Unexpected error class")
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceSSH,
+		Code:   "TUNNEL_CONNECTION_LOST",
+	}, errInfo, "Unexpected error info")
+}
+
 func TestNeonConnectivityErrorShouldBeConnectivity(t *testing.T) {
 	t.Skip("Not a good idea to run this test in CI as it goes to Neon, maybe we need a better mock")
 	config, err := pgx.ParseConfig("postgres://random-endpoint-id-here.us-east-2.aws.neon.tech:5432/db?options=endpoint%3Dtest_endpoint")

flow/connectors/postgres/cdc.go

@@ -722,6 +722,10 @@ func PullCdcRecords[Items model.Items](
 		}
 
 		if err != nil {
+			// If the SSH tunnel went bad (e.g. keepalive failure), we proactively close the replication connection.
+			if p.ssh.IsBad() {
+				return exceptions.NewSSHTunnelConnectionError(fmt.Errorf("ReceiveMessage failed: %w", err))
+			}
 			return fmt.Errorf("ReceiveMessage failed: %w", err)
 		}
 

flow/connectors/utils/ssh.go

@@ -23,7 +23,7 @@ type SSHTunnel struct {
 	*ssh.Client
 	logger        *slog.Logger
 	keepaliveChan atomic.Pointer[chan struct{}]
-	badTunnel     bool
+	badTunnel     atomic.Bool
 }
 
 // GetSSHClientConfig returns an *ssh.ClientConfig based on provided credentials.
@@ -101,6 +101,12 @@ func NewSSHTunnel(
 	return nil, nil
 }
 
+// IsBad reports whether the tunnel has been marked unusable (keepalive failure or explicit close).
+// Safe to call on a nil tunnel, in which case it returns false.
+func (tunnel *SSHTunnel) IsBad() bool {
+	return tunnel != nil && tunnel.badTunnel.Load()
+}
+
 func (tunnel *SSHTunnel) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
 	conn, err := tunnel.Client.DialContext(ctx, network, address)
 	if err != nil {
@@ -114,7 +120,7 @@ func (tunnel *SSHTunnel) Close() error {
 		if keepaliveChan := tunnel.keepaliveChan.Swap(nil); keepaliveChan != nil {
 			close(*keepaliveChan)
 		}
-		tunnel.badTunnel = true
+		tunnel.badTunnel.Store(true)
 		return tunnel.Client.Close()
 	}
 	return nil
@@ -141,7 +147,7 @@ func (tunnel *SSHTunnel) runKeepaliveLoop(
 				if keepaliveChan := tunnel.keepaliveChan.Swap(nil); keepaliveChan != nil {
 					close(*keepaliveChan)
 				}
-				tunnel.badTunnel = true
+				tunnel.badTunnel.Store(true)
 				if onFailure != nil {
 					onFailure()
 				}
@@ -169,7 +175,7 @@ func (tunnel *SSHTunnel) runKeepaliveLoop(
 			if keepaliveChan := tunnel.keepaliveChan.Swap(nil); keepaliveChan != nil {
 				close(*keepaliveChan)
 			}
-			tunnel.badTunnel = true
+			tunnel.badTunnel.Store(true)
 			if onFailure != nil {
 				onFailure()
 			}
@@ -179,7 +185,7 @@ func (tunnel *SSHTunnel) runKeepaliveLoop(
 }
 
 func (tunnel *SSHTunnel) StartKeepalive(ctx context.Context, onFailure func()) {
-	if tunnel == nil || tunnel.Client == nil || tunnel.badTunnel {
+	if tunnel == nil || tunnel.Client == nil || tunnel.badTunnel.Load() {
 		return
 	}
 	if tunnel.keepaliveChan.Load() != nil {
@@ -195,7 +201,7 @@ func (tunnel *SSHTunnel) StartKeepalive(ctx context.Context, onFailure func()) {
 // returns a channel that is closed if the SSH keepalive fails,
 // or nil if no SSH tunnel is configured
 func (tunnel *SSHTunnel) GetKeepaliveChan(ctx context.Context) <-chan struct{} {
-	if tunnel == nil || tunnel.Client == nil || tunnel.badTunnel {
+	if tunnel == nil || tunnel.Client == nil || tunnel.badTunnel.Load() {
 		// nil channel would be of no consequence in a select
 		// UNLESS it's the only branch in a select, in which case it would block forever
 		return nil

flow/connectors/utils/ssh_test.go

@@ -23,7 +23,8 @@ func TestSSHTunnel_GetKeepaliveChan_NilCases(t *testing.T) {
 	require.Nil(t, tunnel.GetKeepaliveChan(context.Background()), "Tunnel with nil client should return nil channel")
 
 	// Bad tunnel
-	tunnel = &SSHTunnel{Client: nil, badTunnel: true}
+	tunnel = &SSHTunnel{Client: nil}
+	tunnel.badTunnel.Store(true)
 	require.Nil(t, tunnel.GetKeepaliveChan(context.Background()), "Bad tunnel should return nil channel")
 }
 
@@ -39,7 +40,8 @@ func TestSSHTunnel_StartKeepalive_NilCases(t *testing.T) {
 	tunnel.StartKeepalive(context.Background(), onFailure)
 	require.False(t, called, "Tunnel with nil client should not call onFailure")
 
-	tunnel = &SSHTunnel{Client: nil, badTunnel: true}
+	tunnel = &SSHTunnel{Client: nil}
+	tunnel.badTunnel.Store(true)
 	tunnel.StartKeepalive(context.Background(), onFailure)
 	require.False(t, called, "Bad tunnel should not call onFailure")
 }

flow/shared/exceptions/ssh.go

@@ -17,3 +17,21 @@ func (e *SSHTunnelSetupError) Error() string {
 func (e *SSHTunnelSetupError) Unwrap() error {
 	return e.error
 }
+
+// SSHTunnelConnectionError represents errors that surface on a connection tunneled over SSH after the
+// SSH tunnel has gone bad (e.g. keepalive failure causes us to close the underlying connections mid-CDC).
+type SSHTunnelConnectionError struct {
+	error
+}
+
+func NewSSHTunnelConnectionError(err error) *SSHTunnelConnectionError {
+	return &SSHTunnelConnectionError{err}
+}
+
+func (e *SSHTunnelConnectionError) Error() string {
+	return "SSH Tunnel Connection Error: " + e.error.Error()
+}
+
+func (e *SSHTunnelConnectionError) Unwrap() error {
+	return e.error
+}