fix mysql stuck on syncer.Close()

flow/connectors/mysql/cdc.go

@@ -318,6 +318,26 @@ func (c *MySqlConnector) startCdcStreamingGtid(
 	return syncer, stream, gset, mysql.Position{}, nil
 }
 
+// closeSyncerWithTimeout closes the syncer with a 10s timeout and, on timeout,
+// force-closes the SSH tunnel if SSH tunnel exists. go-mysql's BinlogSyncer.close()
+// can hang indefinitely when its two unblock mechanisms fail (SetReadDeadline fails
+// for ssh tunnel, and killing a connection that has already been reaped by the server
+// but is not propagated to the client). This can lead to syncer.Close() stuck indefinitely.
+// TODO: better to fix properly upstream
+func (c *MySqlConnector) closeSyncerWithTimeout(syncer *replication.BinlogSyncer, timeout time.Duration) {
+	done := make(chan struct{})
+	go func() {
+		syncer.Close()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(timeout):
+		c.logger.Error("[mysql] syncer.Close hung, force-closing SSH tunnel to unblock")
+		_ = c.ssh.Close()
+	}
+}
+
 func (c *MySqlConnector) ReplPing(context.Context) error {
 	return nil
 }
@@ -353,7 +373,8 @@ func (c *MySqlConnector) PullRecords(
 	if err != nil {
 		return err
 	}
-	defer syncer.Close()
+	defer c.closeSyncerWithTimeout(syncer, 10*time.Second)
+
 	c.logger.Info("[mysql] PullRecords started streaming")
 
 	var skewLossReported bool

flow/connectors/mysql/ssh_keepalive_test.go

@@ -21,20 +21,17 @@ import (
 )
 
 const (
-	toxiproxyDownProxyPort         = 42001
-	toxiproxyLatencyProxyPort      = 42002
-	toxiproxyResetProxyPort        = 42003
-	toxiproxyCDCHangProxyPort      = 42004
-	toxiproxyCDCCloseHangProxyPort = 42005
+	toxiproxyDownProxyPort              = 42001
+	toxiproxyLatencyProxyPort           = 42002
+	toxiproxyResetProxyPort             = 42003
+	toxiproxyCDCHangProxyPort           = 42004
+	toxiproxyCDCCloseHangProxyPort      = 42005
+	toxiproxyCloseSyncerWithTimeoutPort = 42006
 )
 
-func setupMySQLConnectorWithProxy(ctx context.Context, t *testing.T, proxyName string, proxyPort int,
-) (*MySqlConnector, *toxiproxy.Proxy) {
+func resolveMySQL(t *testing.T) (string, uint32, string) {
 	t.Helper()
 
-	toxiproxyClient := utils.NewToxiproxyClient(t)
-	sshProxy := utils.CreateSSHProxy(t, toxiproxyClient, proxyName, proxyPort)
-
 	mysqlHost := "mysql"
 	if envHost := os.Getenv("CI_MYSQL_HOST"); envHost != "" {
 		mysqlHost = envHost
@@ -52,6 +49,19 @@ func setupMySQLConnectorWithProxy(ctx context.Context, t *testing.T, proxyName s
 		mysqlRootPass = envPass
 	}
 
+	return mysqlHost, mysqlPort, mysqlRootPass
+}
+
+// Connector -> Toxi -> SSH -> MySQL
+func setupMySQLConnectorWithSSHProxy(ctx context.Context, t *testing.T, proxyName string, proxyPort int,
+) (*MySqlConnector, *toxiproxy.Proxy) {
+	t.Helper()
+
+	toxiproxyClient := utils.NewToxiproxyClient(t)
+	sshProxy := utils.CreateSSHProxy(t, toxiproxyClient, proxyName, proxyPort)
+
+	mysqlHost, mysqlPort, mysqlRootPass := resolveMySQL(t)
+
 	connector, err := NewMySqlConnector(ctx, &protos.MySqlConfig{
 		Host:     mysqlHost,
 		Port:     mysqlPort,
@@ -75,11 +85,48 @@ func setupMySQLConnectorWithProxy(ctx context.Context, t *testing.T, proxyName s
 	return connector, sshProxy
 }
 
-func setupMySQLConnectorWithSSH(ctx context.Context, t *testing.T, proxyName string, proxyPort int,
+// Connector -> SSH -> Toxi -> MySQL
+func setupMySQLConnectorWithMySQLProxy(
+	ctx context.Context, t *testing.T, proxyName string, proxyPort int,
+) (*MySqlConnector, *toxiproxy.Proxy) {
+	t.Helper()
+
+	toxiproxyClient := utils.NewToxiproxyClient(t)
+	mysqlHost, mysqlPort, mysqlRootPass := resolveMySQL(t)
+	upstream := mysqlHost + ":" + strconv.FormatUint(uint64(mysqlPort), 10)
+	mysqlProxy := utils.CreateToxiproxyForward(t, toxiproxyClient, proxyName, proxyPort, upstream)
+
+	sshPortStr := utils.SSHServerPort
+	sshPort, err := strconv.ParseUint(sshPortStr, 10, 32)
+	require.NoError(t, err)
+
+	connector, err := NewMySqlConnector(ctx, &protos.MySqlConfig{
+		Host:     utils.MySQLProxyHost,
+		Port:     uint32(proxyPort),
+		User:     "root",
+		Password: mysqlRootPass,
+		Database: "mysql",
+		SshConfig: &protos.SSHConfig{
+			Host:     "localhost",
+			Port:     uint32(sshPort),
+			User:     "testuser",
+			Password: "testpass",
+		},
+		DisableTls: true,
+	})
+	require.NoError(t, err)
+
+	err = connector.ConnectionActive(ctx)
+	require.NoError(t, err, "Initial connection should work")
+
+	return connector, mysqlProxy
+}
+
+func setupMySQLSSHKeepaliveHarness(ctx context.Context, t *testing.T, proxyName string, proxyPort int,
 ) (*MySqlConnector, utils.SSHKeepaliveTestConfig) {
 	t.Helper()
 
-	connector, sshProxy := setupMySQLConnectorWithProxy(ctx, t, proxyName, proxyPort)
+	connector, sshProxy := setupMySQLConnectorWithSSHProxy(ctx, t, proxyName, proxyPort)
 	keepaliveChan := connector.ssh.GetKeepaliveChan(ctx)
 
 	return connector, utils.SSHKeepaliveTestConfig{
@@ -92,12 +139,12 @@ func setupMySQLConnectorWithSSH(ctx context.Context, t *testing.T, proxyName str
 	}
 }
 
-func TestMySQLSSHKeepaliveWithToxiproxy(t *testing.T) {
+func TestMySQLSSHKeepaliveTunnelDown(t *testing.T) {
 	t.Parallel()
 	if os.Getenv("CI_MYSQL_VERSION") == "maria" {
 		t.Skip("Skipping SSH keepalive test for MariaDB")
 	}
-	connector, cfg := setupMySQLConnectorWithSSH(t.Context(), t, "my-ssh-keepalive-test", toxiproxyDownProxyPort)
+	connector, cfg := setupMySQLSSHKeepaliveHarness(t.Context(), t, "my-ssh-keepalive-test", toxiproxyDownProxyPort)
 	defer connector.Close()
 	utils.RunSSHKeepaliveDownTest(t, cfg)
 }
@@ -107,7 +154,7 @@ func TestMySQLSSHKeepaliveLatency(t *testing.T) {
 	if os.Getenv("CI_MYSQL_VERSION") == "maria" {
 		t.Skip("Skipping SSH keepalive test for MariaDB")
 	}
-	connector, cfg := setupMySQLConnectorWithSSH(t.Context(), t, "my-ssh-latency-test", toxiproxyLatencyProxyPort)
+	connector, cfg := setupMySQLSSHKeepaliveHarness(t.Context(), t, "my-ssh-latency-test", toxiproxyLatencyProxyPort)
 	defer connector.Close()
 	utils.RunSSHKeepaliveLatencyTest(t, cfg)
 }
@@ -117,7 +164,7 @@ func TestMySQLSSHResetPeer(t *testing.T) {
 	if os.Getenv("CI_MYSQL_VERSION") == "maria" {
 		t.Skip("Skipping SSH keepalive test for MariaDB")
 	}
-	connector, cfg := setupMySQLConnectorWithSSH(t.Context(), t, "my-ssh-reset-peer-test", toxiproxyResetProxyPort)
+	connector, cfg := setupMySQLSSHKeepaliveHarness(t.Context(), t, "my-ssh-reset-peer-test", toxiproxyResetProxyPort)
 	defer connector.Close()
 	utils.RunSSHResetPeerTest(t, cfg)
 }
@@ -165,7 +212,7 @@ func TestMySQLSSHKeepaliveCDCHang(t *testing.T) {
 	}
 	ctx := t.Context()
 
-	connector, sshProxy := setupMySQLConnectorWithProxy(ctx, t, "my-ssh-cdc-down-test", toxiproxyCDCHangProxyPort)
+	connector, sshProxy := setupMySQLConnectorWithSSHProxy(ctx, t, "my-ssh-cdc-down-test", toxiproxyCDCHangProxyPort)
 	defer connector.Close()
 
 	keepaliveChan := connector.ssh.GetKeepaliveChan(ctx)
@@ -219,7 +266,7 @@ func TestMySQLSSHKeepaliveCDCCloseHang(t *testing.T) {
 
 	ctx := t.Context()
 
-	connector, sshProxy := setupMySQLConnectorWithProxy(ctx, t, "my-ssh-cdc-latency-test", toxiproxyCDCCloseHangProxyPort)
+	connector, sshProxy := setupMySQLConnectorWithSSHProxy(ctx, t, "my-ssh-cdc-latency-test", toxiproxyCDCCloseHangProxyPort)
 	defer connector.Close()
 
 	keepaliveChan := connector.ssh.GetKeepaliveChan(ctx)
@@ -270,3 +317,46 @@ func TestMySQLSSHKeepaliveCDCCloseHang(t *testing.T) {
 		t.Fatal("PullRecords did not return after SSH keepalive closed the connection")
 	}
 }
+
+func TestMySQLCloseSyncerWithTimeout(t *testing.T) {
+	t.Parallel()
+	if os.Getenv("CI_MYSQL_VERSION") == "maria" {
+		t.Skip("Skipping for MariaDB")
+	}
+
+	ctx := t.Context()
+	connector, mysqlProxy := setupMySQLConnectorWithMySQLProxy(
+		ctx, t, "my-close-syncer-timeout-test", toxiproxyCloseSyncerWithTimeoutPort)
+	defer connector.Close()
+
+	pos, err := connector.GetMasterPos(ctx)
+	require.NoError(t, err)
+	syncer, _, _, _, err := connector.startCdcStreamingFilePos(ctx, pos) //nolint:dogsled
+	require.NoError(t, err)
+
+	// Let CDC streaming establish before blocking the MySQL server.
+	time.Sleep(2 * time.Second)
+
+	// Black-hole the MySQL server so syncer.Close() hangs
+	_, err = mysqlProxy.AddToxic("latency", "latency", "", 1.0, toxiproxy.Attributes{
+		"latency": 120000,
+	})
+	require.NoError(t, err)
+
+	syncerCloseTimeout := 2 * time.Second
+	done := make(chan struct{})
+	start := time.Now()
+	go func() {
+		connector.closeSyncerWithTimeout(syncer, syncerCloseTimeout)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		elapsed := time.Since(start)
+		require.GreaterOrEqual(t, elapsed, syncerCloseTimeout)
+		require.Less(t, elapsed, syncerCloseTimeout+time.Second)
+	case <-time.After(10 * time.Second):
+		t.Fatal("closeSyncerWithTimeout did not return on timeout")
+	}
+}

flow/connectors/utils/ssh_keepalive_test_helper.go

@@ -19,6 +19,7 @@ const (
 	SSHServerPort    = "2222"
 	ToxiproxyHost    = "localhost"
 	SSHServerHost    = "openssh"
+	MySQLProxyHost   = "toxiproxy"
 )
 
 // Callbacks avoid adding test-only methods to the connector interfaces in core.go.
@@ -36,9 +37,11 @@ func NewToxiproxyClient(t *testing.T) *toxiproxy.Client {
 	return client
 }
 
-func CreateSSHProxy(t *testing.T, client *toxiproxy.Client, name string, port int) *toxiproxy.Proxy {
+func CreateToxiproxyForward(
+	t *testing.T, client *toxiproxy.Client, name string, listenPort int, upstream string,
+) *toxiproxy.Proxy {
 	t.Helper()
-	proxy, err := client.CreateProxy(name, "0.0.0.0:"+strconv.Itoa(port), SSHServerHost+":"+SSHServerPort)
+	proxy, err := client.CreateProxy(name, "0.0.0.0:"+strconv.Itoa(listenPort), upstream)
 	require.NoError(t, err)
 	t.Cleanup(func() {
 		if err := proxy.Delete(); err != nil {
@@ -48,6 +51,11 @@ func CreateSSHProxy(t *testing.T, client *toxiproxy.Client, name string, port in
 	return proxy
 }
 
+func CreateSSHProxy(t *testing.T, client *toxiproxy.Client, name string, port int) *toxiproxy.Proxy {
+	t.Helper()
+	return CreateToxiproxyForward(t, client, name, port, SSHServerHost+":"+SSHServerPort)
+}
+
 func RunSSHKeepaliveDownTest(t *testing.T, cfg SSHKeepaliveTestConfig) {
 	t.Helper()
 	ctx := t.Context()