From ce7d1b8ac5b4d4546b44b15dc1f8a49e6f20603c Mon Sep 17 00:00:00 2001
From: weichen <weichen@example.com>
Date: Fri, 18 Sep 2026 03:44:08 +0000
Subject: [PATCH] =?UTF-8?q?fix(server):=20=E5=90=8E=E5=8F=B0=20ticker=20?=
 =?UTF-8?q?=E7=BA=B3=E5=85=A5=20Shutdown=EF=BC=9Bvault=20Drain=20=E5=91=8A?=
 =?UTF-8?q?=E8=AD=A6=EF=BC=9Bbizdev=20=E5=BF=83=E8=B7=B3=E8=B6=85=E6=97=B6?=
 =?UTF-8?q?=EF=BC=88R-6=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

reminder/vault 可停止；Drain 失败 slog.Warn；心跳/释放 lease 使用 10s 超时。

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 internal/bizdev/service.go                    |  8 ++++++--
 internal/server/notification_reminder_loop.go | 10 ++++++++--
 internal/server/server.go                     | 10 ++++++++++
 internal/server/vault.go                      | 13 +++++++++++--
 4 files changed, 35 insertions(+), 6 deletions(-)

diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go
index 6b6ddf8f..138d43fa 100644
--- a/internal/bizdev/service.go
+++ b/internal/bizdev/service.go
@@ -825,13 +825,17 @@ func (s *Service) ExecuteTask(ctx context.Context, req TaskRunRequest) (*TaskRun
 				case <-stopHB:
 					return
 				case <-ticker.C:
-					_, _ = s.docGov.HeartbeatLease(context.Background(), docID, string(docgov.LeaseHolderSquad), leaseHolderID, leaseTTL)
+					hbCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+					_, _ = s.docGov.HeartbeatLease(hbCtx, docID, string(docgov.LeaseHolderSquad), leaseHolderID, leaseTTL)
+					cancel()
 				}
 			}
 		}()
 		defer func() {
 			if leaseAcquired {
-				_, _ = s.docGov.ReleaseLease(context.Background(), docID, string(docgov.LeaseHolderSquad), leaseHolderID)
+				relCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+				_, _ = s.docGov.ReleaseLease(relCtx, docID, string(docgov.LeaseHolderSquad), leaseHolderID)
+				cancel()
 				if streamer := stream.FromContext(ctx); streamer != nil {
 					streamer.EmitGovernance(stream.GovernancePayload{
 						DocID:             docID,
diff --git a/internal/server/notification_reminder_loop.go b/internal/server/notification_reminder_loop.go
index 7210756d..fef61fae 100644
--- a/internal/server/notification_reminder_loop.go
+++ b/internal/server/notification_reminder_loop.go
@@ -12,12 +12,18 @@ func (s *Server) startNotificationReminderLoop() {
 	if s == nil || s.notifySvc == nil || !s.notifySvc.Available() {
 		return
 	}
+	s.reminderStop = make(chan struct{})
 	go func() {
 		ticker := time.NewTicker(defaultReminderPollInterval)
 		defer ticker.Stop()
 		s.runStageAdvanceReminders(context.Background())
-		for range ticker.C {
-			s.runStageAdvanceReminders(context.Background())
+		for {
+			select {
+			case <-s.reminderStop:
+				return
+			case <-ticker.C:
+				s.runStageAdvanceReminders(context.Background())
+			}
 		}
 	}()
 }
diff --git a/internal/server/server.go b/internal/server/server.go
index ba3e5fc3..9d92073d 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -77,6 +77,8 @@ type Server struct {
 	vaultStore            *artifactvault.FilesystemStore
 	vaultPub              *artifactvault.Publisher
 	vaultPublisherStarted int32
+	vaultStop             chan struct{}
+	reminderStop          chan struct{}
 }
 
 // NewFromRuntime attaches the HTTP adapter to an already-wired composition root.
@@ -357,6 +359,14 @@ func (s *Server) Start() error {
 
 // Shutdown 优雅停机
 func (s *Server) Shutdown(ctx context.Context) error {
+	if s.reminderStop != nil {
+		close(s.reminderStop)
+		s.reminderStop = nil
+	}
+	if s.vaultStop != nil {
+		close(s.vaultStop)
+		s.vaultStop = nil
+	}
 	if s.taskRuns != nil {
 		_ = s.taskRuns.WaitIdle(ctx)
 	}
diff --git a/internal/server/vault.go b/internal/server/vault.go
index d9c1d3ee..8e4b643b 100644
--- a/internal/server/vault.go
+++ b/internal/server/vault.go
@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"net/http"
 	"sync/atomic"
 	"time"
@@ -56,11 +57,19 @@ func (s *Server) startVaultPublisher() {
 	if s == nil || s.vaultPub == nil || s.cfg == nil || !s.cfg.FAVPublisherEnabled {
 		return
 	}
+	s.vaultStop = make(chan struct{})
 	go func() {
 		t := time.NewTicker(2 * time.Second)
 		defer t.Stop()
-		for range t.C {
-			_ = s.vaultPub.Drain(context.Background())
+		for {
+			select {
+			case <-s.vaultStop:
+				return
+			case <-t.C:
+				if err := s.vaultPub.Drain(context.Background()); err != nil {
+					slog.Warn("vault_drain_failed", "error", err)
+				}
+			}
 		}
 	}()
 	atomic.StoreInt32(&s.vaultPublisherStarted, 1)