fix(server): 后台 ticker 纳入 Shutdown;vault Drain 告警;bizdev 心跳超时(R-6)
reminder/vault 可停止;Drain 失败 slog.Warn;心跳/释放 lease 使用 10s 超时。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -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,
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user