From 72c85c9c8d0d20373a1ac93c47c08ca114bf2eb0 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Philip=20Dub=C3=A9?=
 <159546+serprex@users.noreply.github.com>
Date: Tue, 21 Apr 2026 16:52:05 +0000
Subject: [PATCH] replication: fix BinlogSyncer.Close hang on SSH tunnels

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")
---
 replication/binlogsyncer.go      | 14 ++++--
 replication/binlogsyncer_test.go | 76 ++++++++++++++++++++++++++++++++
 2 files changed, 87 insertions(+), 3 deletions(-)

diff --git a/replication/binlogsyncer.go b/replication/binlogsyncer.go
index 28715cd44..ad4ab0fd8 100644
--- a/replication/binlogsyncer.go
+++ b/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))
+			}
 		}
 	}
 
diff --git a/replication/binlogsyncer_test.go b/replication/binlogsyncer_test.go
index 9895673e3..deefba75a 100644
--- a/replication/binlogsyncer_test.go
+++ b/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")
+	}
+}