diff --git a/pipe.go b/pipe.go index 8e9a40e6..4344a0bd 100644 --- a/pipe.go +++ b/pipe.go @@ -1910,15 +1910,25 @@ func (p *pipe) Close() { } if block == 1 && (stopping1 || stopping2) { // make sure there is no block cmd p.incrWaits() + // The timer closes the connection after one second. That is the + // only way to unblock the two steps below when the peer is silent: + // PutOne waits for a free slot, <-ch waits for the reply. A closed + // connection makes the background loops fail and drain the queue, + // which frees a slot and answers the PING with an error. The timer + // costs a goroutine only when it fires. + // + // The drain runs in the background worker. There is always one + // here: it is already running, or the branch above started it, or + // syncDo starts it when the closed connection breaks its read. + var escape *time.Timer + if p.conn != nil { + escape = time.AfterFunc(time.Second, func() { p.conn.Close() }) + } ch, _ := p.queue.PutOne(context.Background(), cmds.PingCmd) - select { - case <-ch: - p.decrWaits() - case <-time.After(time.Second): - go func(ch chan RedisResult) { - <-ch - p.decrWaits() - }(ch) + <-ch + p.decrWaits() + if escape != nil { + escape.Stop() } } } diff --git a/pipe_close_test.go b/pipe_close_test.go new file mode 100644 index 00000000..ad81b35e --- /dev/null +++ b/pipe_close_test.go @@ -0,0 +1,168 @@ +package rueidis + +import ( + "context" + "net" + "sync" + "testing" + "time" + + "github.com/redis/rueidis/internal/cmds" +) + +// TestCloseReturnsOnFullRing: pipe.Close() must return in bounded time even +// when the command ring is full and the connection has stopped answering. +// +// Close() enqueues a final PING through the ring and waits for its reply with +// a one-second escape. The escape used to cover only the reply: the enqueue +// itself (queue.PutOne) blocks while every ring slot is occupied, and at +// Close() time nothing can free a slot when the peer has stopped replying but +// the TCP connection is still open: +// +// - the keepalive ping would normally detect the stall and error the +// connection, which drains the ring (see TestExitOnRingFullAndPingTimeout) +// — but Close() increments blcksig, which makes backgroundPing skip its +// check, and stores errClosing, which stops the ping timer from +// re-arming; +// - Close() closes the network connection only AFTER the enqueue, so the +// background loops keep waiting and the drain never starts. +// +// So Close() blocked forever. That turns into a goroutine calling +// client.Close() that never returns (a hung shutdown), or — worse — the +// sentinel client closing a replaced connection while holding its mutex, +// wedging all of its topology handling. +// +// It takes a connection that stalls without erroring (network partition, +// frozen server), enough in-flight commands to fill the ring, and a Close() +// arriving before the keepalive ping notices the stall. +func TestCloseReturnsOnFullRing(t *testing.T) { + defer ShouldNotLeak(SetupLeakDetection()) + p, mock, _, closeConn := setup(t, ClientOption{ + RingScaleEachConn: 1, // 2-slot ring, cheap to fill + ConnWriteTimeout: 500 * time.Millisecond, + Dialer: net.Dialer{KeepAlive: 500 * time.Millisecond}, + }) + p.background() + + // Fill the ring: the writer sends the commands into the socket, the mock + // reads them and never replies, so every slot stays occupied waiting for + // a reply. + ringLen := len(p.queue.(*ring).store) + for i := 0; i < ringLen; i++ { + go func() { _ = p.Do(context.Background(), cmds.NewCompleted([]string{"GET", "a"})).Error() }() + } + for i := 0; i < ringLen; i++ { + mock.Expect("GET", "a") + } + time.Sleep(50 * time.Millisecond) + + done := make(chan struct{}) + go func() { p.Close(); close(done) }() + + // Close's internal escape is one second; three is a deadlock. + select { + case <-done: + case <-time.After(3 * time.Second): + closeConn() // unpark the wedged Close so the leak detector can finish + <-done + t.Fatal("pipe.Close() did not return within 3s with a full ring and a silent connection") + } +} + +// TestCloseAfterConnErrorOnFullRing pins the behavior that keeps the common +// failure path safe: when the connection ERRORS instead of stalling silently +// — the usual case when a server process dies and the OS resets the +// connection — the background loops exit and drain the ring, so a Close() +// that follows returns promptly. The silent-stall variant, where nothing +// drains the ring, is covered by TestCloseReturnsOnFullRing. +func TestCloseAfterConnErrorOnFullRing(t *testing.T) { + defer ShouldNotLeak(SetupLeakDetection()) + p, mock, _, closeConn := setup(t, ClientOption{ + RingScaleEachConn: 1, + ConnWriteTimeout: 500 * time.Millisecond, + Dialer: net.Dialer{KeepAlive: 500 * time.Millisecond}, + }) + p.background() + + ringLen := len(p.queue.(*ring).store) + for i := 0; i < ringLen; i++ { + go func() { _ = p.Do(context.Background(), cmds.NewCompleted([]string{"GET", "a"})).Error() }() + } + for i := 0; i < ringLen; i++ { + mock.Expect("GET", "a") + } + time.Sleep(50 * time.Millisecond) + + // The connection errors first (like a killed server), then Close runs. + closeConn() + + done := make(chan struct{}) + go func() { p.Close(); close(done) }() + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("pipe.Close() did not return although the connection had already errored") + } +} + +// TestCloseReturnsOnFullRingBeforeBackground pins the link Close relies on to +// escape a stuck enqueue: cutting the connection only unparks queue.PutOne +// because the background loops then error out and drain the queue. That +// requires a background worker, and this is the one path to the enqueue where +// none is running yet. +// +// The pipe is still in sync mode: one caller owns the connection in syncDo, +// the others have queued behind it and filled the ring, and the peer never +// answers. Nothing here has started _background, and Close does not start it +// either — waits != 1 tells it a sync read may be in flight, and racing a +// second reader onto the same connection is what 8503b21 fixed. So the drain +// hangs on the sync read breaking first: the connection Close cuts is what +// breaks it, and its error path (pipe.go, syncDo) is what finally starts the +// worker that drains the ring. +func TestCloseReturnsOnFullRingBeforeBackground(t *testing.T) { + defer ShouldNotLeak(SetupLeakDetection()) + p, mock, _, closeConn := setup(t, ClientOption{ + RingScaleEachConn: 1, // 2-slot ring, cheap to fill + }) + // No p.background() here: the pipe must stay in sync mode. + + var callers sync.WaitGroup + + // context.Background() carries neither a deadline nor a Done channel, so + // this one takes the syncDo path and owns the connection. ConnWriteTimeout + // is unset, so syncDo clears the connection deadline and its read has no + // bound of its own. + callers.Add(1) + go func() { + defer callers.Done() + _ = p.Do(context.Background(), cmds.NewCompleted([]string{"GET", "a"})).Error() + }() + mock.Expect("GET", "a") // reaches the peer, which never answers + time.Sleep(50 * time.Millisecond) + + // Fill the ring behind the sync command. These never reach the socket: + // with no background writer they just sit in their slots. + ringLen := len(p.queue.(*ring).store) + callers.Add(ringLen) + for i := 0; i < ringLen; i++ { + go func() { + defer callers.Done() + _ = p.Do(context.Background(), cmds.NewCompleted([]string{"GET", "b"})).Error() + }() + } + time.Sleep(50 * time.Millisecond) + + done := make(chan struct{}) + go func() { p.Close(); close(done) }() + + // Close's internal escape is one second; three is a deadlock. + select { + case <-done: + case <-time.After(3 * time.Second): + closeConn() // unpark the wedged Close so the leak detector can finish + <-done + t.Fatal("pipe.Close() did not return with a full ring and no background worker running") + } + callers.Wait() +}