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)