From 8b6004a7f72347d762fa8864db733fc8acb818f5 Mon Sep 17 00:00:00 2001 From: Bo-Yi Wu Date: Sat, 21 Mar 2026 14:54:30 +0800 Subject: [PATCH 1/5] fix(router): propagate HTTP request context into push notification tasks Previously, handleNotification accepted a context.Context parameter but ignored it (_ context.Context). The context passed from pushHandler (c.Request.Context()) was therefore never used when dispatching notifications, so client disconnection could not cancel in-flight push tasks. This commit: 1. Renames the ignored parameter to `ctx` so it is actually used. 2. Adds a withEitherCancel helper that merges the HTTP request context with the queue-task context, cancelling when either is done (client disconnects OR queue shuts down). 3. Threads the merged context through to notify.SendNotification, which already propagates it to PushToIOS / PushToAndroid / PushToHuawei and DispatchFeedback. This completes the fix originally intended by the goroutine pattern in issue #422, which was removed in #866 because the cancel() call was never reached in async mode. --- router/server.go | 32 +++++++++++++++++++++++++++----- 1 file changed, 27 insertions(+), 5 deletions(-) diff --git a/router/server.go b/router/server.go index d56e20ce..1ce51f67 100644 --- a/router/server.go +++ b/router/server.go @@ -280,9 +280,25 @@ func countNotificationTargets(notification *notify.PushNotification) int { return count } + +// withEitherCancel returns a context that is cancelled when either ctx1 or ctx2 is done. +// This is useful for merging an HTTP request context with a queue-task context so that +// a push notification is aborted when the caller disconnects OR when the queue shuts down. +func withEitherCancel(ctx1, ctx2 context.Context) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(ctx1) + go func() { + select { + case <-ctx2.Done(): + cancel() + case <-ctx.Done(): + } + }() + return ctx, cancel +} + // HandleNotification add notification to queue list. func handleNotification( - _ context.Context, + ctx context.Context, cfg *config.ConfYaml, req notify.RequestPush, q *queue.Queue, @@ -306,10 +322,16 @@ func handleNotification( } if isLocalSync { - func(msg *notify.PushNotification, cfg *config.ConfYaml) { - if err := q.QueueTask(func(ctx context.Context) error { + func(msg *notify.PushNotification, cfg *config.ConfYaml, reqCtx context.Context) { + if err := q.QueueTask(func(queueCtx context.Context) error { defer wg.Done() - resp, err := notify.SendNotification(ctx, msg, cfg) + // Merge the HTTP request context with the queue context so that + // the notification is cancelled when either the client disconnects + // or the queue shuts down. See: + // https://github.com/appleboy/gorush/issues/422 + mergedCtx, cancel := withEitherCancel(reqCtx, queueCtx) + defer cancel() + resp, err := notify.SendNotification(mergedCtx, msg, cfg) if err != nil { return err } @@ -318,7 +340,7 @@ func handleNotification( }); err != nil { logx.LogError.Error(err) } - }(notification, cfg) + }(notification, cfg, ctx) } else if err := q.Queue(notification); err != nil { resp := markFailedNotification(cfg, notification, "max capacity reached") logs = append(logs, resp...) From 6e1ec2cafd70e8f30a75dbec80e811b7926bc3d4 Mon Sep 17 00:00:00 2001 From: Bo-Yi Wu Date: Sat, 21 Mar 2026 15:30:14 +0800 Subject: [PATCH 2/5] fix(router): fix wg deadlock and logs race condition in handleNotification Address two issues raised in Copilot code review on PR #869: 1. wg deadlock: In local sync mode, wg.Add(1) is called before q.QueueTask(). If QueueTask returns an error (queue full/closed), the task closure never executes so wg.Done() is never called, causing wg.Wait() to block forever. Fix: call wg.Done() in the QueueTask error path when cfg.Core.Sync is true. 2. logs race condition: Multiple goroutines dispatched via QueueTask can concurrently append to the shared logs slice, causing a data race. Fix: protect all appends to logs inside QueueTask with a sync.Mutex. --- router/server.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/router/server.go b/router/server.go index 1ce51f67..87038ba1 100644 --- a/router/server.go +++ b/router/server.go @@ -313,6 +313,7 @@ func handleNotification( var ( count int wg sync.WaitGroup + mu sync.Mutex logs = make([]logx.LogPushEntry, 0) ) @@ -335,9 +336,14 @@ func handleNotification( if err != nil { return err } + mu.Lock() logs = append(logs, resp.Logs...) + mu.Unlock() return nil }); err != nil { + if cfg.Core.Sync { + wg.Done() + } logx.LogError.Error(err) } }(notification, cfg, ctx) From 0df607559a2c604183b1e03c3ccbf479f5668a12 Mon Sep 17 00:00:00 2001 From: Bo-Yi Wu Date: Sat, 21 Mar 2026 16:46:37 +0800 Subject: [PATCH 3/5] test(router): add unit tests for withEitherCancel Add 6 focused unit tests for withEitherCancel as suggested by Copilot code review: - TestWithEitherCancel_Ctx1Cancel: merged ctx cancels when ctx1 (HTTP request) is cancelled - TestWithEitherCancel_Ctx2Cancel: merged ctx cancels when ctx2 (queue task) is cancelled - TestWithEitherCancel_ExplicitCancel: calling the returned CancelFunc cancels merged ctx without affecting parents - TestWithEitherCancel_NoGoroutineLeak: internal goroutine exits after cancellation (no leak) - TestWithEitherCancel_AlreadyCancelledCtx1: merged ctx is immediately done when ctx1 is pre-cancelled - TestWithEitherCancel_AlreadyCancelledCtx2: merged ctx cancels promptly when ctx2 is pre-cancelled --- router/server_test.go | 167 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 167 insertions(+) diff --git a/router/server_test.go b/router/server_test.go index 59641fce..a2851875 100644 --- a/router/server_test.go +++ b/router/server_test.go @@ -853,3 +853,170 @@ func TestCountNotificationTargets(t *testing.T) { }) } } + +// TestWithEitherCancel_Ctx1Cancel verifies that the derived context is cancelled +// when ctx1 (the first parent, e.g. the HTTP request context) is cancelled. +func TestWithEitherCancel_Ctx1Cancel(t *testing.T) { + ctx1, cancel1 := context.WithCancel(context.Background()) + ctx2, cancel2 := context.WithCancel(context.Background()) + defer cancel2() + + merged, cancelMerged := withEitherCancel(ctx1, ctx2) + defer cancelMerged() + + // ctx1 has not been cancelled yet - merged should be alive + select { + case <-merged.Done(): + t.Fatal("merged context should not be done yet") + default: + } + + // Cancel ctx1 (simulates HTTP client disconnect) + cancel1() + + select { + case <-merged.Done(): + // expected + case <-time.After(100 * time.Millisecond): + t.Fatal("merged context should have been cancelled when ctx1 was cancelled") + } +} + +// TestWithEitherCancel_Ctx2Cancel verifies that the derived context is cancelled +// when ctx2 (the second parent, e.g. the queue-task context) is cancelled. +func TestWithEitherCancel_Ctx2Cancel(t *testing.T) { + ctx1, cancel1 := context.WithCancel(context.Background()) + ctx2, cancel2 := context.WithCancel(context.Background()) + defer cancel1() + + merged, cancelMerged := withEitherCancel(ctx1, ctx2) + defer cancelMerged() + + // Neither parent cancelled yet - merged should be alive + select { + case <-merged.Done(): + t.Fatal("merged context should not be done yet") + default: + } + + // Cancel ctx2 (simulates queue shutdown) + cancel2() + + select { + case <-merged.Done(): + // expected + case <-time.After(100 * time.Millisecond): + t.Fatal("merged context should have been cancelled when ctx2 was cancelled") + } +} + +// TestWithEitherCancel_ExplicitCancel verifies that calling the returned +// CancelFunc directly cancels the merged context without affecting either parent. +func TestWithEitherCancel_ExplicitCancel(t *testing.T) { + ctx1, cancel1 := context.WithCancel(context.Background()) + ctx2, cancel2 := context.WithCancel(context.Background()) + defer cancel1() + defer cancel2() + + merged, cancelMerged := withEitherCancel(ctx1, ctx2) + + // Explicitly cancel the merged context + cancelMerged() + + select { + case <-merged.Done(): + // expected + case <-time.After(100 * time.Millisecond): + t.Fatal("merged context should have been cancelled by explicit cancelMerged()") + } + + // Parents should still be alive + select { + case <-ctx1.Done(): + t.Fatal("ctx1 should not be cancelled") + default: + } + select { + case <-ctx2.Done(): + t.Fatal("ctx2 should not be cancelled") + default: + } +} + +// TestWithEitherCancel_NoGoroutineLeak verifies that the internal goroutine +// spawned by withEitherCancel exits when the merged context is cancelled, +// preventing goroutine leaks. +func TestWithEitherCancel_NoGoroutineLeak(t *testing.T) { + ctx1, cancel1 := context.WithCancel(context.Background()) + ctx2, cancel2 := context.WithCancel(context.Background()) + defer cancel2() + + goroutinesBefore := runtime.NumGoroutine() + + merged, cancelMerged := withEitherCancel(ctx1, ctx2) + + // Give the internal goroutine time to start + time.Sleep(10 * time.Millisecond) + goroutinesDuring := runtime.NumGoroutine() + assert.GreaterOrEqual(t, goroutinesDuring, goroutinesBefore, + "at least one new goroutine should exist while merged context is live") + + // Cancel via ctx1 - this should trigger the internal goroutine to exit + cancel1() + cancelMerged() // also call cancelMerged to release resources + + // Give the goroutine time to clean up + time.Sleep(50 * time.Millisecond) + goroutinesAfter := runtime.NumGoroutine() + + assert.LessOrEqual(t, goroutinesAfter, goroutinesBefore+1, + "goroutine count should return to baseline after cancellation (allow ±1 for runtime variance)") + + // merged context must be done + select { + case <-merged.Done(): + // expected + default: + t.Fatal("merged context should be done after cancel1() and cancelMerged()") + } +} + +// TestWithEitherCancel_AlreadyCancelledCtx1 verifies that if ctx1 is already +// cancelled before withEitherCancel is called, the merged context is immediately done. +func TestWithEitherCancel_AlreadyCancelledCtx1(t *testing.T) { + ctx1, cancel1 := context.WithCancel(context.Background()) + cancel1() // already cancelled + + ctx2, cancel2 := context.WithCancel(context.Background()) + defer cancel2() + + merged, cancelMerged := withEitherCancel(ctx1, ctx2) + defer cancelMerged() + + select { + case <-merged.Done(): + // expected - ctx1 was already done, so merged should be immediately done + case <-time.After(100 * time.Millisecond): + t.Fatal("merged context should be immediately done when ctx1 is already cancelled") + } +} + +// TestWithEitherCancel_AlreadyCancelledCtx2 verifies that if ctx2 is already +// cancelled before withEitherCancel is called, the merged context is cancelled promptly. +func TestWithEitherCancel_AlreadyCancelledCtx2(t *testing.T) { + ctx1, cancel1 := context.WithCancel(context.Background()) + defer cancel1() + + ctx2, cancel2 := context.WithCancel(context.Background()) + cancel2() // already cancelled + + merged, cancelMerged := withEitherCancel(ctx1, ctx2) + defer cancelMerged() + + select { + case <-merged.Done(): + // expected + case <-time.After(100 * time.Millisecond): + t.Fatal("merged context should be cancelled promptly when ctx2 is already cancelled") + } +} From 75f33c76e62b73486ebd77fa6435e846ae51f40b Mon Sep 17 00:00:00 2001 From: Bo-Yi Wu Date: Sat, 21 Mar 2026 17:35:53 +0800 Subject: [PATCH 4/5] style(router): fix gofmt double blank line before withEitherCancel Remove extra blank line between countNotificationTargets and withEitherCancel. gofmt does not allow two consecutive blank lines between top-level declarations. --- router/server.go | 1 - 1 file changed, 1 deletion(-) diff --git a/router/server.go b/router/server.go index 87038ba1..a1c79cf0 100644 --- a/router/server.go +++ b/router/server.go @@ -280,7 +280,6 @@ func countNotificationTargets(notification *notify.PushNotification) int { return count } - // withEitherCancel returns a context that is cancelled when either ctx1 or ctx2 is done. // This is useful for merging an HTTP request context with a queue-task context so that // a push notification is aborted when the caller disconnects OR when the queue shuts down. From adb7be4bf5246550f09e56f7cd3410d81b624457 Mon Sep 17 00:00:00 2001 From: Bo-Yi Wu Date: Sat, 21 Mar 2026 17:39:54 +0800 Subject: [PATCH 5/5] style(router): fix golangci-lint issues in withEitherCancel tests Fix 5 golangci-lint issues in server_test.go: modernize (x4): Replace context.WithCancel(context.Background()) + defer cancel() with t.Context() in test helpers where the context is never explicitly cancelled. t.Context() is automatically cancelled when the test ends, making the code cleaner. golines (x1): Shorten the too-long assert.LessOrEqual message in TestWithEitherCancel_NoGoroutineLeak to fit within the line length limit. --- router/server_test.go | 18 +++++++----------- 1 file changed, 7 insertions(+), 11 deletions(-) diff --git a/router/server_test.go b/router/server_test.go index a2851875..c9068d77 100644 --- a/router/server_test.go +++ b/router/server_test.go @@ -858,8 +858,7 @@ func TestCountNotificationTargets(t *testing.T) { // when ctx1 (the first parent, e.g. the HTTP request context) is cancelled. func TestWithEitherCancel_Ctx1Cancel(t *testing.T) { ctx1, cancel1 := context.WithCancel(context.Background()) - ctx2, cancel2 := context.WithCancel(context.Background()) - defer cancel2() + ctx2 := t.Context() merged, cancelMerged := withEitherCancel(ctx1, ctx2) defer cancelMerged() @@ -885,9 +884,8 @@ func TestWithEitherCancel_Ctx1Cancel(t *testing.T) { // TestWithEitherCancel_Ctx2Cancel verifies that the derived context is cancelled // when ctx2 (the second parent, e.g. the queue-task context) is cancelled. func TestWithEitherCancel_Ctx2Cancel(t *testing.T) { - ctx1, cancel1 := context.WithCancel(context.Background()) + ctx1 := t.Context() ctx2, cancel2 := context.WithCancel(context.Background()) - defer cancel1() merged, cancelMerged := withEitherCancel(ctx1, ctx2) defer cancelMerged() @@ -948,8 +946,7 @@ func TestWithEitherCancel_ExplicitCancel(t *testing.T) { // preventing goroutine leaks. func TestWithEitherCancel_NoGoroutineLeak(t *testing.T) { ctx1, cancel1 := context.WithCancel(context.Background()) - ctx2, cancel2 := context.WithCancel(context.Background()) - defer cancel2() + ctx2 := t.Context() goroutinesBefore := runtime.NumGoroutine() @@ -969,8 +966,9 @@ func TestWithEitherCancel_NoGoroutineLeak(t *testing.T) { time.Sleep(50 * time.Millisecond) goroutinesAfter := runtime.NumGoroutine() + // Allow a ±1 goroutine variance due to runtime scheduling assert.LessOrEqual(t, goroutinesAfter, goroutinesBefore+1, - "goroutine count should return to baseline after cancellation (allow ±1 for runtime variance)") + "goroutine count should be near baseline after cancellation") // merged context must be done select { @@ -987,8 +985,7 @@ func TestWithEitherCancel_AlreadyCancelledCtx1(t *testing.T) { ctx1, cancel1 := context.WithCancel(context.Background()) cancel1() // already cancelled - ctx2, cancel2 := context.WithCancel(context.Background()) - defer cancel2() + ctx2 := t.Context() merged, cancelMerged := withEitherCancel(ctx1, ctx2) defer cancelMerged() @@ -1004,8 +1001,7 @@ func TestWithEitherCancel_AlreadyCancelledCtx1(t *testing.T) { // TestWithEitherCancel_AlreadyCancelledCtx2 verifies that if ctx2 is already // cancelled before withEitherCancel is called, the merged context is cancelled promptly. func TestWithEitherCancel_AlreadyCancelledCtx2(t *testing.T) { - ctx1, cancel1 := context.WithCancel(context.Background()) - defer cancel1() + ctx1 := t.Context() ctx2, cancel2 := context.WithCancel(context.Background()) cancel2() // already cancelled