From 798d83b1c3e6ec402eaf5d35bb542fbbc6b0db6f Mon Sep 17 00:00:00 2001 From: Joy Gao <17896160+jgao54@users.noreply.github.com> Date: Fri, 17 Apr 2026 12:37:54 -1000 Subject: [PATCH 1/4] fix mysql stuck on syncer.Close() --- flow/connectors/mysql/cdc.go | 23 ++++++++- flow/connectors/mysql/ssh_keepalive_test.go | 57 +++++++++++++++++++-- 2 files changed, 74 insertions(+), 6 deletions(-) diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index e31b1049a..eb30fed98 100644 --- a/flow/connectors/mysql/cdc.go +++ b/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 diff --git a/flow/connectors/mysql/ssh_keepalive_test.go b/flow/connectors/mysql/ssh_keepalive_test.go index aed8712a7..0c4cfe9be 100644 --- a/flow/connectors/mysql/ssh_keepalive_test.go +++ b/flow/connectors/mysql/ssh_keepalive_test.go @@ -21,11 +21,12 @@ 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, @@ -270,3 +271,49 @@ 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, sshProxy := setupMySQLConnectorWithProxy( + 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) + require.NoError(t, err) + + // Let CDC streaming establish before blocking the tunnel. + time.Sleep(2 * time.Second) + + // Black-hole the SSH transport so syncer.Close() hangs + _, err = sshProxy.AddToxic("latency", "latency", "", 1.0, toxiproxy.Attributes{ + "latency": 120000, + }) + require.NoError(t, err) + + // Must be < SSHKeepaliveInterval so the timing bounds below prove the syncCloseTimeout + // (not SSH keepalive's own force-close) is what unblocked syncer.Close. + syncerCloseTimeout := 2 * time.Second + require.Less(t, syncerCloseTimeout, utils.SSHKeepaliveInterval) + 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") + } +} From 7006b0e0827d0f35cbfcbb41307a581fb3cafaa7 Mon Sep 17 00:00:00 2001 From: Joy Gao <17896160+jgao54@users.noreply.github.com> Date: Fri, 17 Apr 2026 15:24:10 -1000 Subject: [PATCH 2/4] review --- flow/connectors/mysql/cdc.go | 18 +++---- flow/connectors/mysql/ssh_keepalive_test.go | 57 ++------------------- flow/shared/timeout/close.go | 19 +++++++ flow/shared/timeout/close_test.go | 38 ++++++++++++++ 4 files changed, 68 insertions(+), 64 deletions(-) create mode 100644 flow/shared/timeout/close.go create mode 100644 flow/shared/timeout/close_test.go diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index eb30fed98..9ed6ab91b 100644 --- a/flow/connectors/mysql/cdc.go +++ b/flow/connectors/mysql/cdc.go @@ -29,6 +29,7 @@ import ( "github.com/PeerDB-io/peerdb/flow/shared" "github.com/PeerDB-io/peerdb/flow/shared/datatypes" "github.com/PeerDB-io/peerdb/flow/shared/exceptions" + "github.com/PeerDB-io/peerdb/flow/shared/timeout" "github.com/PeerDB-io/peerdb/flow/shared/types" ) @@ -324,18 +325,11 @@ func (c *MySqlConnector) startCdcStreamingGtid( // 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") +func (c *MySqlConnector) closeSyncer(syncer *replication.BinlogSyncer, deadline time.Duration) { + timeout.CloseWithTimeout(syncer.Close, deadline, func() { + c.logger.Error("[mysql] syncer.Close hung, force-closing SSH tunnel") _ = c.ssh.Close() - } + }) } func (c *MySqlConnector) ReplPing(context.Context) error { @@ -373,7 +367,7 @@ func (c *MySqlConnector) PullRecords( if err != nil { return err } - defer c.closeSyncerWithTimeout(syncer, 10*time.Second) + defer c.closeSyncer(syncer, 10*time.Second) c.logger.Info("[mysql] PullRecords started streaming") diff --git a/flow/connectors/mysql/ssh_keepalive_test.go b/flow/connectors/mysql/ssh_keepalive_test.go index 0c4cfe9be..aed8712a7 100644 --- a/flow/connectors/mysql/ssh_keepalive_test.go +++ b/flow/connectors/mysql/ssh_keepalive_test.go @@ -21,12 +21,11 @@ import ( ) const ( - toxiproxyDownProxyPort = 42001 - toxiproxyLatencyProxyPort = 42002 - toxiproxyResetProxyPort = 42003 - toxiproxyCDCHangProxyPort = 42004 - toxiproxyCDCCloseHangProxyPort = 42005 - toxiproxyCloseSyncerWithTimeoutPort = 42006 + toxiproxyDownProxyPort = 42001 + toxiproxyLatencyProxyPort = 42002 + toxiproxyResetProxyPort = 42003 + toxiproxyCDCHangProxyPort = 42004 + toxiproxyCDCCloseHangProxyPort = 42005 ) func setupMySQLConnectorWithProxy(ctx context.Context, t *testing.T, proxyName string, proxyPort int, @@ -271,49 +270,3 @@ 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, sshProxy := setupMySQLConnectorWithProxy( - 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) - require.NoError(t, err) - - // Let CDC streaming establish before blocking the tunnel. - time.Sleep(2 * time.Second) - - // Black-hole the SSH transport so syncer.Close() hangs - _, err = sshProxy.AddToxic("latency", "latency", "", 1.0, toxiproxy.Attributes{ - "latency": 120000, - }) - require.NoError(t, err) - - // Must be < SSHKeepaliveInterval so the timing bounds below prove the syncCloseTimeout - // (not SSH keepalive's own force-close) is what unblocked syncer.Close. - syncerCloseTimeout := 2 * time.Second - require.Less(t, syncerCloseTimeout, utils.SSHKeepaliveInterval) - 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") - } -} diff --git a/flow/shared/timeout/close.go b/flow/shared/timeout/close.go new file mode 100644 index 000000000..65c3ee36d --- /dev/null +++ b/flow/shared/timeout/close.go @@ -0,0 +1,19 @@ +package timeout + +import "time" + +// CloseWithTimeout runs closer in a goroutine and waits up to timeout for it +// to return. If it doesn't, onTimeout is invoked and CloseWithTimeout returns +// without waiting further. +func CloseWithTimeout(closer func(), timeout time.Duration, onTimeout func()) { + done := make(chan struct{}) + go func() { + closer() + close(done) + }() + select { + case <-done: + case <-time.After(timeout): + onTimeout() + } +} diff --git a/flow/shared/timeout/close_test.go b/flow/shared/timeout/close_test.go new file mode 100644 index 000000000..ca96f16c6 --- /dev/null +++ b/flow/shared/timeout/close_test.go @@ -0,0 +1,38 @@ +package timeout + +import ( + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestCloseWithTimeout_FastClose(t *testing.T) { + var onTimeoutCalls atomic.Int32 + CloseWithTimeout( + func() {}, // returns immediately + time.Second, + func() { onTimeoutCalls.Add(1) }, + ) + require.Zero(t, onTimeoutCalls.Load()) +} + +func TestCloseWithTimeout_ClosureHangs(t *testing.T) { + var onTimeoutCalls atomic.Int32 + release := make(chan struct{}) + defer close(release) + + timeout := 50 * time.Millisecond + start := time.Now() + CloseWithTimeout( + func() { <-release }, // blocks + timeout, + func() { onTimeoutCalls.Add(1) }, + ) + elapsed := time.Since(start) + + require.Equal(t, int32(1), onTimeoutCalls.Load()) + require.GreaterOrEqual(t, elapsed, timeout) + require.Less(t, elapsed, timeout+time.Second) +} From 358f9cbb15241db25cf19e7fc335ca5fd960cd58 Mon Sep 17 00:00:00 2001 From: Ilia Demianenko Date: Tue, 21 Apr 2026 15:33:46 -0700 Subject: [PATCH 3/4] Toxi infra for mysql stuck on syncer.Close() (#4207) --- flow/connectors/mysql/cdc.go | 18 ++- flow/connectors/mysql/ssh_keepalive_test.go | 126 +++++++++++++++--- .../utils/ssh_keepalive_test_helper.go | 12 +- flow/shared/timeout/close.go | 19 --- flow/shared/timeout/close_test.go | 38 ------ 5 files changed, 130 insertions(+), 83 deletions(-) delete mode 100644 flow/shared/timeout/close.go delete mode 100644 flow/shared/timeout/close_test.go diff --git a/flow/connectors/mysql/cdc.go b/flow/connectors/mysql/cdc.go index 9ed6ab91b..eb30fed98 100644 --- a/flow/connectors/mysql/cdc.go +++ b/flow/connectors/mysql/cdc.go @@ -29,7 +29,6 @@ import ( "github.com/PeerDB-io/peerdb/flow/shared" "github.com/PeerDB-io/peerdb/flow/shared/datatypes" "github.com/PeerDB-io/peerdb/flow/shared/exceptions" - "github.com/PeerDB-io/peerdb/flow/shared/timeout" "github.com/PeerDB-io/peerdb/flow/shared/types" ) @@ -325,11 +324,18 @@ func (c *MySqlConnector) startCdcStreamingGtid( // 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) closeSyncer(syncer *replication.BinlogSyncer, deadline time.Duration) { - timeout.CloseWithTimeout(syncer.Close, deadline, func() { - c.logger.Error("[mysql] syncer.Close hung, force-closing SSH tunnel") +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 { @@ -367,7 +373,7 @@ func (c *MySqlConnector) PullRecords( if err != nil { return err } - defer c.closeSyncer(syncer, 10*time.Second) + defer c.closeSyncerWithTimeout(syncer, 10*time.Second) c.logger.Info("[mysql] PullRecords started streaming") diff --git a/flow/connectors/mysql/ssh_keepalive_test.go b/flow/connectors/mysql/ssh_keepalive_test.go index aed8712a7..82372b710 100644 --- a/flow/connectors/mysql/ssh_keepalive_test.go +++ b/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) + 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") + } +} diff --git a/flow/connectors/utils/ssh_keepalive_test_helper.go b/flow/connectors/utils/ssh_keepalive_test_helper.go index 932ed9e67..fb26c27dd 100644 --- a/flow/connectors/utils/ssh_keepalive_test_helper.go +++ b/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() diff --git a/flow/shared/timeout/close.go b/flow/shared/timeout/close.go deleted file mode 100644 index 65c3ee36d..000000000 --- a/flow/shared/timeout/close.go +++ /dev/null @@ -1,19 +0,0 @@ -package timeout - -import "time" - -// CloseWithTimeout runs closer in a goroutine and waits up to timeout for it -// to return. If it doesn't, onTimeout is invoked and CloseWithTimeout returns -// without waiting further. -func CloseWithTimeout(closer func(), timeout time.Duration, onTimeout func()) { - done := make(chan struct{}) - go func() { - closer() - close(done) - }() - select { - case <-done: - case <-time.After(timeout): - onTimeout() - } -} diff --git a/flow/shared/timeout/close_test.go b/flow/shared/timeout/close_test.go deleted file mode 100644 index ca96f16c6..000000000 --- a/flow/shared/timeout/close_test.go +++ /dev/null @@ -1,38 +0,0 @@ -package timeout - -import ( - "sync/atomic" - "testing" - "time" - - "github.com/stretchr/testify/require" -) - -func TestCloseWithTimeout_FastClose(t *testing.T) { - var onTimeoutCalls atomic.Int32 - CloseWithTimeout( - func() {}, // returns immediately - time.Second, - func() { onTimeoutCalls.Add(1) }, - ) - require.Zero(t, onTimeoutCalls.Load()) -} - -func TestCloseWithTimeout_ClosureHangs(t *testing.T) { - var onTimeoutCalls atomic.Int32 - release := make(chan struct{}) - defer close(release) - - timeout := 50 * time.Millisecond - start := time.Now() - CloseWithTimeout( - func() { <-release }, // blocks - timeout, - func() { onTimeoutCalls.Add(1) }, - ) - elapsed := time.Since(start) - - require.Equal(t, int32(1), onTimeoutCalls.Load()) - require.GreaterOrEqual(t, elapsed, timeout) - require.Less(t, elapsed, timeout+time.Second) -} From 17d74a00abc47fd4714d2ad8ee02f2a9a24ba05e Mon Sep 17 00:00:00 2001 From: Joy Gao <17896160+jgao54@users.noreply.github.com> Date: Tue, 21 Apr 2026 17:04:51 -1000 Subject: [PATCH 4/4] lint --- flow/connectors/mysql/ssh_keepalive_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flow/connectors/mysql/ssh_keepalive_test.go b/flow/connectors/mysql/ssh_keepalive_test.go index 82372b710..0cd4ab278 100644 --- a/flow/connectors/mysql/ssh_keepalive_test.go +++ b/flow/connectors/mysql/ssh_keepalive_test.go @@ -331,7 +331,7 @@ func TestMySQLCloseSyncerWithTimeout(t *testing.T) { pos, err := connector.GetMasterPos(ctx) require.NoError(t, err) - syncer, _, _, _, err := connector.startCdcStreamingFilePos(ctx, pos) + syncer, _, _, _, err := connector.startCdcStreamingFilePos(ctx, pos) //nolint:dogsled require.NoError(t, err) // Let CDC streaming establish before blocking the MySQL server.