fix:系统优雅退出

internal/cmd/cmd.go

@@ -302,7 +302,7 @@ var (
 
 			// Initialize active task count and start polling
 			task.InitActiveCount(ctx)
-			go task.StartAsyncPolling(ctx)
+			task.StartAsyncPolling(ctx)
 
 			// Start webhook dispatcher (event-driven delivery)
 			tenant.InitWebhookDispatcher(ctx)
@@ -312,6 +312,8 @@ var (
 
 			// Flush usage log writer on shutdown (s.Run blocks until server stops)
 			defer plugin.Shutdown(ctx)
+			defer tenant.ShutdownWebhookDispatcher()
+			defer task.StopAsyncPolling()
 			defer common.CloseChannelErrorWriter()
 			defer common.CloseUsageLogWriter()
 			defer response.CloseErrorLogWriter()

internal/logic/task/async_polling.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"io"
 	"net/http"
+	"sync"
 	"time"
 
 	"github.com/gogf/gf/v2/frame/g"
@@ -24,10 +25,14 @@ const (
 	timeoutBatchSize = 100
 )
 
+var pollingWg sync.WaitGroup
+
 // StartAsyncPolling 启动异步任务轮询 goroutine
 func StartAsyncPolling(ctx context.Context) {
 	g.Log().Info(ctx, "Starting async task polling...")
+	pollingWg.Add(1)
 	go func() {
+		defer pollingWg.Done()
 		ticker := time.NewTicker(pollInterval)
 		defer ticker.Stop()
 
@@ -43,6 +48,11 @@ func StartAsyncPolling(ctx context.Context) {
 	}()
 }
 
+// StopAsyncPolling 等待轮询 goroutine 完全退出，在服务停机时调用。
+func StopAsyncPolling() {
+	pollingWg.Wait()
+}
+
 // pollOnce 执行一次轮询
 func pollOnce(ctx context.Context) {
 	if !HasActiveTasks() {

internal/logic/tenant/webhook_dispatcher.go

@@ -2,6 +2,7 @@ package tenant
 
 import (
 	"context"
+	"sync"
 	"time"
 
 	"github.com/gogf/gf/v2/frame/g"
@@ -27,6 +28,7 @@ const (
 type WebhookDispatcher struct {
 	eventCh chan int64
 	ctx     context.Context
+	wg      sync.WaitGroup
 }
 
 var defaultDispatcher *WebhookDispatcher
@@ -40,6 +42,7 @@ func InitWebhookDispatcher(ctx context.Context) {
 	defaultDispatcher = d
 
 	for i := 0; i < dispatcherWorkerCount; i++ {
+		d.wg.Add(1)
 		go d.worker()
 	}
 	go d.sweepLoop()
@@ -65,6 +68,7 @@ func NotifyNewEvent(eventID int64) {
 
 // worker consumes event IDs from the channel and delivers them.
 func (d *WebhookDispatcher) worker() {
+	defer d.wg.Done()
 	for {
 		select {
 		case <-d.ctx.Done():
@@ -209,3 +213,16 @@ func (d *WebhookDispatcher) cleanup() {
 		g.Log().Infof(ctx, "webhook cleanup: deleted %d exhausted events older than %v", rows, failedCutoff.Format("Y-m-d"))
 	}
 }
+
+// Shutdown waits for all in-flight deliveries to complete.
+func (d *WebhookDispatcher) Shutdown() {
+	d.wg.Wait()
+}
+
+// ShutdownWebhookDispatcher waits for all in-flight webhook deliveries to complete.
+// Call this via defer after s.Run() returns.
+func ShutdownWebhookDispatcher() {
+	if defaultDispatcher != nil {
+		defaultDispatcher.Shutdown()
+	}
+}