From 3f26cb4475810c8c03e0a627bb62d20e2a4f0d58 Mon Sep 17 00:00:00 2001
From: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Wed, 21 Oct 2026 15:07:36 -0700
Subject: [PATCH] notification: cancel the webhook queue only after all workers
 stop

Each target's router goroutine deferred the shared cancel, so the first
worker to exit tore down the queue context for every other target and
the dead letter logger.
---
 weed/notification/webhook/webhook_queue.go | 15 +++++++++++----
 1 file changed, 11 insertions(+), 4 deletions(-)

diff --git a/weed/notification/webhook/webhook_queue.go b/weed/notification/webhook/webhook_queue.go
index 1c5626c8f9..a5aa9b7d25 100644
--- a/weed/notification/webhook/webhook_queue.go
+++ b/weed/notification/webhook/webhook_queue.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/ThreeDotsLabs/watermill"
@@ -211,6 +212,7 @@ func (w *Queue) setupWatermillQueue(targets []*target) error {
 	}
 	w.queueChannel = gochannel.NewGoChannel(pubSubConfig, logger)
 
+	var wg sync.WaitGroup
 	for _, t := range targets {
 		// One router per target so each keeps its own retry/backoff settings.
 		// Handlers subscribe to the target's own topic, so a message is
@@ -251,18 +253,23 @@ func (w *Queue) setupWatermillQueue(targets []*target) error {
 			t.handleWebhook,
 		)
 
+		wg.Add(1)
 		go func() {
-			// cancels the queue context so the dead letter logger exists in case context not canceled by the shutdown signal already
-			defer w.cancel()
-
+			defer wg.Done()
 			if err := router.Run(w.ctx); err != nil && !errors.Is(err, context.Canceled) {
 				glog.Errorf("webhook pubsub worker stopped with error: %v", err)
 			}
-
 			glog.Info("webhook pubsub worker stopped")
 		}()
 	}
 
+	// the queue context lives until every target's worker has stopped so the
+	// dead letter logger keeps running while any of them is still up
+	go func() {
+		wg.Wait()
+		w.cancel()
+	}()
+
 	return nil
 }