replication: fix BinlogSyncer.Close hang on SSH tunnels

replication/binlogsyncer.go

@@ -240,9 +240,17 @@ func (b *BinlogSyncer) close() {
 	b.cancel()
 
 	if b.c != nil {
-		err := b.c.SetReadDeadline(utils.Now().Add(100 * time.Millisecond))
-		if err != nil {
-			b.cfg.Logger.Warn("could not set read deadline", slog.Any("error", err))
+		// SetReadDeadline unblocks ReadPacket so onStream notices ctx cancellation.
+		// Some transports (notably SSH tunnels) refuse deadlines with an error, in
+		// which case close the underlying connection to force ReadPacket to return.
+		// Otherwise wg.Wait below parks forever when KILL also fails to reach server
+		// (e.g. thread already reaped, "ERROR 1094 unknown thread id")
+		if err := b.c.SetReadDeadline(utils.Now().Add(100 * time.Millisecond)); err != nil {
+			b.cfg.Logger.Warn("could not set read deadline, closing connection to unblock reader",
+				slog.Any("error", err))
+			if err := b.c.Close(); err != nil {
+				b.cfg.Logger.Warn("could not close connection", slog.Any("error", err))
+			}
 		}
 	}
 

replication/binlogsyncer_test.go

@@ -1,11 +1,19 @@
 package replication
 
 import (
+	"errors"
+	"io"
+	"net"
 	"os"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
+
+	"github.com/go-mysql-org/go-mysql/client"
+	"github.com/go-mysql-org/go-mysql/packet"
 )
 
 func TestLocalHostname(t *testing.T) {
@@ -48,3 +56,71 @@ func TestLocalHostname_os(t *testing.T) {
 	h, _ := os.Hostname()
 	require.Equal(t, h, b.localHostname())
 }
+
+// deadlinelessConn mimics the net.Conn surface of an ssh tunneled channel:
+// Read blocks until Close, and SetReadDeadline refuses with an error rather
+// than interrupting the parked Read.
+type deadlinelessConn struct {
+	closeOnce sync.Once
+	closed    chan struct{}
+}
+
+func newDeadlinelessConn() *deadlinelessConn {
+	return &deadlinelessConn{closed: make(chan struct{})}
+}
+
+func (c *deadlinelessConn) Read(b []byte) (int, error) {
+	<-c.closed
+	return 0, io.EOF
+}
+
+func (c *deadlinelessConn) Write(b []byte) (int, error) { return len(b), nil }
+
+func (c *deadlinelessConn) Close() error {
+	c.closeOnce.Do(func() { close(c.closed) })
+	return nil
+}
+
+func (*deadlinelessConn) LocalAddr() net.Addr         { return &net.TCPAddr{} }
+func (*deadlinelessConn) RemoteAddr() net.Addr        { return &net.TCPAddr{} }
+func (*deadlinelessConn) SetDeadline(time.Time) error { return errors.New("deadline not supported") }
+func (*deadlinelessConn) SetReadDeadline(time.Time) error {
+	return errors.New("deadline not supported")
+}
+
+func (*deadlinelessConn) SetWriteDeadline(time.Time) error {
+	return errors.New("deadline not supported")
+}
+
+// TestCloseUnblocksWhenSetReadDeadlineFails exercises the deadlock path where
+// SetReadDeadline cannot unblock the binlog reader (e.g. ssh tunnel) and KILL
+// also fails to reach the server (thread already reaped). Under the previous
+// behaviour close() parked indefinitely on wg.Wait.
+func TestCloseUnblocksWhenSetReadDeadlineFails(t *testing.T) {
+	b := NewBinlogSyncer(BinlogSyncerConfig{ServerID: 1})
+
+	fake := newDeadlinelessConn()
+	b.c = &client.Conn{Conn: packet.NewConn(fake)}
+	b.running = true
+
+	// Mimic onStream's parked ReadPacket: block until the underlying conn is
+	// closed, then honour ctx cancellation before signalling wg.Done.
+	b.wg.Add(1)
+	go func() {
+		defer b.wg.Done()
+		_, _ = b.c.ReadPacket()
+		<-b.ctx.Done()
+	}()
+
+	done := make(chan struct{})
+	go func() {
+		b.Close()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Close hung when SetReadDeadline refused a deadline")
+	}
+}