From c9fdd0fb44454e0b5b63846ebd617b7876454d8d Mon Sep 17 00:00:00 2001 From: FZambia Date: Thu, 23 Jul 2026 19:51:49 +0300 Subject: [PATCH 1/3] fix(pipe): don't let Close hang when the command ring is full MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Close enqueues a final PING through the ring and waits for its reply with a one-second escape. The escape covered 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 — 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. In practice that is a client.Close() that never returns during a silent connection stall (network partition, frozen server) — a hung shutdown — or the sentinel client closing a replaced connection while holding its mutex, wedging all of its further topology handling. Run the enqueue-and-wait in a goroutine so the existing one-second escape bounds both. If the enqueue is stuck, Close proceeds and closes the network connection, which errors the background loops and drains the ring — unparking that goroutine, so nothing leaks. The regression test fails against the previous Close. A companion test pins the already-safe path: when the connection errors first, the ring is drained and Close returns promptly. --- pipe.go | 23 +++++++--- pipe_close_test.go | 106 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 122 insertions(+), 7 deletions(-) create mode 100644 pipe_close_test.go diff --git a/pipe.go b/pipe.go index 8e9a40e6..64632b88 100644 --- a/pipe.go +++ b/pipe.go @@ -1910,15 +1910,24 @@ func (p *pipe) Close() { } if block == 1 && (stopping1 || stopping2) { // make sure there is no block cmd p.incrWaits() - ch, _ := p.queue.PutOne(context.Background(), cmds.PingCmd) - select { - case <-ch: + // Run the enqueue in a goroutine so that the one-second bound below + // covers it too. queue.PutOne blocks while every ring slot is + // occupied, and at Close time nothing else can free a slot: the + // keepalive ping is disabled (blcksig is non-zero and the error is + // already set), and the connection is closed only further down. If + // the enqueue blocks, Close proceeds after the timeout and closes + // the connection, which makes the background loops drain the ring + // and unpark this goroutine — so it does not leak. + answered := make(chan struct{}) + go func() { + ch, _ := p.queue.PutOne(context.Background(), cmds.PingCmd) + <-ch p.decrWaits() + close(answered) + }() + select { + case <-answered: case <-time.After(time.Second): - go func(ch chan RedisResult) { - <-ch - p.decrWaits() - }(ch) } } } diff --git a/pipe_close_test.go b/pipe_close_test.go new file mode 100644 index 00000000..31473666 --- /dev/null +++ b/pipe_close_test.go @@ -0,0 +1,106 @@ +package rueidis + +import ( + "context" + "net" + "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") + } +} From 3671b89e3c09eefcad06322010b4837e0976a769 Mon Sep 17 00:00:00 2001 From: FZambia Date: Mon, 27 Jul 2026 18:56:45 +0300 Subject: [PATCH 2/3] fix(pipe): bound the Close enqueue with a timer instead of a goroutine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit moved the enqueue-and-wait into a goroutine so the existing one-second escape could cover both. That costs a goroutine on every graceful close, on a termination path that already has enough of them. Bound the enqueue directly instead: arm a timer that cuts the connection after one second, and stop it as soon as queue.PutOne returns. Cutting the connection is what unparks a stuck enqueue anyway — the background loops error out and drain the queue — and time.AfterFunc costs a goroutine only when it fires, which happens only when the enqueue is stuck. A normal close now allocates a timer and stops it, nothing more. The escape does nothing Close does not already do a few lines further down, and syncDo closes the same connection on its own error path. A deadline on the connection would be cheaper still, but it is shared state that _background and the syncDo family overwrite, with no way to read it back, so it cannot be relied on. Cutting the connection also breaks a sync read parked on it, which the goroutine variant did not do. That covers the remaining case: a pipe still in sync mode, with callers queued behind the sync command filling the ring. Close does not start the background worker there, because waits != 1 tells it a sync read may be in flight, so the drain waits on the sync read failing first. TestCloseReturnsOnFullRingBeforeBackground pins it; like TestCloseReturnsOnFullRing it deadlocks without the fix. --- pipe.go | 42 ++++++++++++++++++------------- pipe_close_test.go | 62 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 17 deletions(-) diff --git a/pipe.go b/pipe.go index 64632b88..8f7f542f 100644 --- a/pipe.go +++ b/pipe.go @@ -1910,24 +1910,32 @@ func (p *pipe) Close() { } if block == 1 && (stopping1 || stopping2) { // make sure there is no block cmd p.incrWaits() - // Run the enqueue in a goroutine so that the one-second bound below - // covers it too. queue.PutOne blocks while every ring slot is - // occupied, and at Close time nothing else can free a slot: the - // keepalive ping is disabled (blcksig is non-zero and the error is - // already set), and the connection is closed only further down. If - // the enqueue blocks, Close proceeds after the timeout and closes - // the connection, which makes the background loops drain the ring - // and unpark this goroutine — so it does not leak. - answered := make(chan struct{}) - go func() { - ch, _ := p.queue.PutOne(context.Background(), cmds.PingCmd) - <-ch - p.decrWaits() - close(answered) - }() + // The one-second bound of this graceful shutdown must cover the + // enqueue as well: queue.PutOne blocks while the slot it claims is + // still occupied, and at Close time nothing else can free it, + // because the keepalive ping is disabled (blcksig is non-zero and + // the error is already set) and the connection is closed only + // further down. Cutting the connection is what unparks it: the + // background loops then error out and drain the queue. The timer + // costs a goroutine only if it fires, which happens only when the + // enqueue is stuck. + deadline := time.Now().Add(time.Second) + 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) + if escape != nil { + escape.Stop() + } select { - case <-answered: - case <-time.After(time.Second): + case <-ch: + p.decrWaits() + case <-time.After(time.Until(deadline)): // the remaining budget, so the whole shutdown stays within one second + go func(ch chan RedisResult) { + <-ch + p.decrWaits() + }(ch) } } } diff --git a/pipe_close_test.go b/pipe_close_test.go index 31473666..ad81b35e 100644 --- a/pipe_close_test.go +++ b/pipe_close_test.go @@ -3,6 +3,7 @@ package rueidis import ( "context" "net" + "sync" "testing" "time" @@ -104,3 +105,64 @@ func TestCloseAfterConnErrorOnFullRing(t *testing.T) { 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() +} From b7bf06a31bb2aa9f73dcd4ea1694982c48d09f82 Mon Sep 17 00:00:00 2001 From: FZambia Date: Tue, 28 Jul 2026 08:13:34 +0300 Subject: [PATCH 3/3] fix(pipe): let the escape bound the reply wait too The timer was stopped as soon as queue.PutOne returned, so it covered only the enqueue and the reply still needed its own one-second escape. Stopping it after the wait covers both: a silent peer leaves the timer to close the connection, the background loops fail and drain the queue, and the drain answers the PING with an error, which ends the wait. That removes the select and the goroutine it spawned on timeout, so a stuck shutdown now costs the timer callback alone, and a healthy one costs nothing. TestCloseWithGracefulPeriodExceeded covers the case this rests on, a peer that never answers a PING that did get a slot. It still returns after a second. --- pipe.go | 31 ++++++++++++------------------- 1 file changed, 12 insertions(+), 19 deletions(-) diff --git a/pipe.go b/pipe.go index 8f7f542f..4344a0bd 100644 --- a/pipe.go +++ b/pipe.go @@ -1910,33 +1910,26 @@ func (p *pipe) Close() { } if block == 1 && (stopping1 || stopping2) { // make sure there is no block cmd p.incrWaits() - // The one-second bound of this graceful shutdown must cover the - // enqueue as well: queue.PutOne blocks while the slot it claims is - // still occupied, and at Close time nothing else can free it, - // because the keepalive ping is disabled (blcksig is non-zero and - // the error is already set) and the connection is closed only - // further down. Cutting the connection is what unparks it: the - // background loops then error out and drain the queue. The timer - // costs a goroutine only if it fires, which happens only when the - // enqueue is stuck. - deadline := time.Now().Add(time.Second) + // 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) + <-ch + p.decrWaits() if escape != nil { escape.Stop() } - select { - case <-ch: - p.decrWaits() - case <-time.After(time.Until(deadline)): // the remaining budget, so the whole shutdown stays within one second - go func(ch chan RedisResult) { - <-ch - p.decrWaits() - }(ch) - } } } p.decrWaits()