feat(pool): honor context cancellation for blocking reads (BRPop) #3934
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| package redis_test | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/redis/go-redis/v9" | ||
| ) | ||
|
|
||
| // TestBRPopContextCancellation verifies that a blocking BRPop with infinite timeout | ||
| // respects ctx.Done() and returns promptly with context.Canceled. | ||
| func TestBRPopContextCancellation(t *testing.T) { | ||
| opt := redis.Options{ | ||
| Addr: ":6379", | ||
| ReadTimeout: -1, // block indefinitely for reads | ||
| WriteTimeout: -1, | ||
| ContextTimeoutEnabled: true, | ||
| } | ||
| rdb := redis.NewClient(&opt) | ||
| t.Cleanup(func() { _ = rdb.Close() }) | ||
|
|
||
| key := "brpop-cancel-key" | ||
| _ = rdb.Del(context.Background(), key).Err() | ||
|
|
||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| done := make(chan error, 1) | ||
| go func() { | ||
| _, err := rdb.BRPop(ctx, 0, key).Result() | ||
| done <- err | ||
| }() | ||
|
|
||
| // Ensure BRPop is blocked | ||
| time.Sleep(50 * time.Millisecond) | ||
| cancel() | ||
|
|
||
| select { | ||
| case err := <-done: | ||
| if err == nil || err != context.Canceled { | ||
| t.Fatalf("expected context.Canceled, got %v", err) | ||
| } | ||
| case <-time.After(2 * time.Second): | ||
| t.Fatal("BRPop did not return after context cancellation") | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -91,7 +91,7 @@ type Conn struct { | |
| // State machine for connection state management | ||
| // Replaces: usable, Inited, used | ||
| // Provides thread-safe state transitions with FIFO waiting queue | ||
| // States: CREATED → INITIALIZING → IDLE ↔ IN_USE | ||
| // ↓ | ||
| // UNUSABLE (handoff/reauth) | ||
| // ↓ | ||
|
|
@@ -320,7 +320,7 @@ func (cn *Conn) IsInited() bool { | |
| // This is the preferred method for acquiring a connection from the pool, as it | ||
| // ensures that only one goroutine marks the connection as used. | ||
| // | ||
| // Implementation: Uses state machine transitions IDLE ↔ IN_USE | ||
| // | ||
| // Returns true if the swap was successful (old value matched), false otherwise. | ||
| // Deprecated: Use GetStateMachine().TryTransition() directly for better state management. | ||
|
|
@@ -907,6 +907,14 @@ func (cn *Conn) RemoteAddr() net.Addr { | |
| func (cn *Conn) WithReader( | ||
| ctx context.Context, timeout time.Duration, fn func(rd *proto.Reader) error, | ||
| ) error { | ||
| // Fast cancellation path: if the context is done, abort before any socket ops. | ||
| if ctx != nil { | ||
| if err := ctx.Err(); err != nil { | ||
| return err | ||
| } | ||
|
Comment on lines
+910
to
+914
There was a problem hiding this comment.
When the context is canceled after the read has already started, as in the added Useful? React with 👍 / 👎. |
||
| } | ||
|
cursor[bot] marked this conversation as resolved.
|
||
|
|
||
| var cancelWatchDone chan struct{} | ||
| if timeout >= 0 { | ||
| // Use relaxed timeout if set, otherwise use provided timeout | ||
| effectiveTimeout := cn.getEffectiveReadTimeout(timeout) | ||
|
|
@@ -917,11 +925,33 @@ func (cn *Conn) WithReader( | |
| return errConnectionNotAvailable | ||
| } | ||
|
|
||
| // Compute and set initial read deadline | ||
| dl := cn.deadline(ctx, effectiveTimeout) | ||
| if err := netConn.SetReadDeadline(dl); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| // If we have no read deadline (e.g., BRPop(timeout=0) and no ctx deadline) | ||
| // but we do have a context, spawn a watcher to force an immediate deadline | ||
| // when ctx.Done() fires. This unblocks an in-flight Read without closing | ||
| // the socket and without affecting the common hot path where a deadline exists. | ||
| if ctx != nil && dl.Equal(noDeadline) { | ||
|
There was a problem hiding this comment.
This predicate only starts the cancellation watcher when the computed socket deadline is empty. If a caller uses Useful? React with 👍 / 👎. |
||
| cancelWatchDone = make(chan struct{}) | ||
| go func(nc net.Conn, done <-chan struct{}, c context.Context) { | ||
| select { | ||
| case <-c.Done(): | ||
| _ = nc.SetReadDeadline(time.Unix(0, getCachedTimeNs())) | ||
|
Comment on lines
+942
to
+943
There was a problem hiding this comment.
When the read finishes normally at about the same time the caller cancels the context, Useful? React with 👍 / 👎. |
||
| case <-done: | ||
| } | ||
| }(netConn, cancelWatchDone, ctx) | ||
| } | ||
| } | ||
|
|
||
| err := fn(cn.rd) | ||
| if cancelWatchDone != nil { | ||
| close(cancelWatchDone) | ||
|
There was a problem hiding this comment. Watcher goroutine may set stale deadline after completionLow Severity After   Reviewed by Cursor Bugbot for commit c459336. Configure here. |
||
| } | ||
| return err | ||
|
There was a problem hiding this comment. Context cancellation returns timeout error, not context.CanceledHigh Severity When the watcher goroutine fires on Additional Locations (1)Reviewed by Cursor Bugbot for commit c459336. Configure here. There was a problem hiding this comment.
When a context without a deadline is canceled while a blocking read is in progress, the watcher unblocks the socket by setting an immediate read deadline, so Useful? React with 👍 / 👎. |
||
| } | ||
|
|
||
| func (cn *Conn) WithWriter( | ||
|
|
@@ -1005,6 +1035,10 @@ func (cn *Conn) deadline(ctx context.Context, timeout time.Duration) time.Time { | |
| } | ||
|
|
||
| if ctx != nil { | ||
| // If context is already done, force immediate deadline to unblock socket ops. | ||
| if err := ctx.Err(); err != nil { | ||
| return time.Unix(0, nowNs) | ||
| } | ||
| deadline, ok := ctx.Deadline() | ||
| if ok { | ||
| if timeout == 0 { | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This workflow still declares a weekly
scheduletrigger, but the new job-level condition only permitspushand same-repopull_requestevents. On the cron eventgithub.event_nameisschedule, so the only job is skipped and the weekly vulnerability scan silently stops running; includeschedulein the condition or gate only the forked-PR case.Useful? React with 👍 / 👎.