handle ssh dial error

Author: jgao54

flow/alerting/classifier.go

@@ -428,6 +428,17 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	if sshDialErr, ok := errors.AsType[*exceptions.SSHTunnelDialError](err); ok {
+		errInfo := ErrorInfo{
+			Source: ErrorSourceSSH,
+			Code:   "TUNNEL_DIAL_ERROR",
+		}
+		if sshDialErr.Retryable {
+			return ErrorRetryRecoverable, errInfo
+		}
+		return ErrorOther, errInfo
+	}
+
 	// 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{

flow/alerting/classifier_test.go

@@ -72,6 +72,21 @@ func TestSSHTunnelConnectionErrorShouldBeConnectivity(t *testing.T) {
 	}, errInfo, "Unexpected error info")
 }
 
+func TestSSHTunnelRecoverableDialError(t *testing.T) {
+	t.Parallel()
+
+	err := fmt.Errorf("failed to sync records: %w",
+		fmt.Errorf("failed to get schema for watermark table xxx.xxx: %w",
+			exceptions.NewMySQLExecuteError(exceptions.NewSSHTunnelDialError(
+				errors.New("ssh: unexpected packet in response to channel open: <nil>")))))
+	errorClass, errInfo := GetErrorClass(t.Context(), err)
+	assert.Equal(t, ErrorRetryRecoverable, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceSSH,
+		Code:   "TUNNEL_DIAL_ERROR",
+	}, errInfo)
+}
+
 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/utils/ssh.go

@@ -110,7 +110,7 @@ func (tunnel *SSHTunnel) IsBad() bool {
 func (tunnel *SSHTunnel) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
 	conn, err := tunnel.Client.DialContext(ctx, network, address)
 	if err != nil {
-		return nil, err
+		return nil, exceptions.NewSSHTunnelDialError(err)
 	}
 	return NewDeadlineCapableConn(conn), nil
 }

flow/shared/exceptions/ssh.go

@@ -1,5 +1,7 @@
 package exceptions
 
+import "strings"
+
 // SSHTunnelSetupError represents errors during SSH Tunnel Setup, the SSH library we use does not provide a common error type,
 // just does fmt.Errorf
 type SSHTunnelSetupError struct {
@@ -18,6 +20,29 @@ func (e *SSHTunnelSetupError) Unwrap() error {
 	return e.error
 }
 
+// SSHTunnelDialError represents a failure to open a new connection through an established SSH tunnel.
+type SSHTunnelDialError struct {
+	error
+	Retryable bool
+}
+
+func NewSSHTunnelDialError(err error) *SSHTunnelDialError {
+	// x/crypto/ssh's mux returns this untyped error when the SSH connection
+	// is torn down while a channel open is in flight.
+	if strings.Contains(err.Error(), "unexpected packet in response to channel open") {
+		return &SSHTunnelDialError{err, true}
+	}
+	return &SSHTunnelDialError{err, false}
+}
+
+func (e *SSHTunnelDialError) Error() string {
+	return "SSH Tunnel Dial Error: " + e.error.Error()
+}
+
+func (e *SSHTunnelDialError) 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 {