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")

Author: serprex

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,70 @@ 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")
+	}
+}