From 226474b43426e1b70176488cbb2fd833a6d1e7af Mon Sep 17 00:00:00 2001
From: huang <youdang2000@gmail.com>
Date: Fri, 22 May 2026 16:21:46 +0800
Subject: [PATCH] =?UTF-8?q?=E8=BD=AE=E8=AE=AD=E6=9C=89=E9=97=AE=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 cmd/worker/main.go                         |  7 +++++--
 internal/polling/initializer.go            | 22 +++++++++++++++++++---
 internal/service/polling/config_service.go |  1 +
 3 files changed, 25 insertions(+), 5 deletions(-)

diff --git a/cmd/worker/main.go b/cmd/worker/main.go
index 1592d2d..741f057 100644
--- a/cmd/worker/main.go
+++ b/cmd/worker/main.go
@@ -314,10 +314,13 @@ func startPollingInitializer(ctx context.Context, runtime *workerRuntime, appLog
 	pollingInitializer.StartBackground(ctx)
 
 	runtime.pollingConfigMgr.WatchChanges(ctx, func(hadConfigs, hasConfigs bool) {
-		if !hadConfigs && hasConfigs {
-			appLogger.Info("轮询配置从空变为非空，触发队列重新初始化")
+		if hasConfigs {
+			appLogger.Info("轮询配置已变更，触发队列重新初始化",
+				zap.Bool("had_configs", hadConfigs))
 			pollingInitializer.Restart(ctx)
+			return
 		}
+		appLogger.Info("轮询配置已清空，跳过队列重新初始化")
 	})
 
 	return pollingInitializer
diff --git a/internal/polling/initializer.go b/internal/polling/initializer.go
index 61bcdd6..54a3b12 100644
--- a/internal/polling/initializer.go
+++ b/internal/polling/initializer.go
@@ -52,6 +52,7 @@ type PollingInitializer struct {
 
 	progress      initProgress
 	initCompleted atomic.Bool
+	restartQueued atomic.Bool
 
 	stopChan chan struct{}
 	wg       sync.WaitGroup
@@ -94,11 +95,12 @@ func (p *PollingInitializer) IsCompleted() bool {
 	return p.initCompleted.Load()
 }
 
-// Restart 重新执行初始化（当轮询配置从空变为非空时调用）
-// 使用 CAS 确保只有初始化已完成时才能重启，避免并发重入
+// Restart 重新执行初始化，用于配置变更后补齐新增任务类型的分片队列。
+// 若初始化正在进行中，则记录一次待重启请求，当前轮完成后再自动执行。
 func (p *PollingInitializer) Restart(ctx context.Context) {
 	if !p.initCompleted.CompareAndSwap(true, false) {
-		p.logger.Info("轮询初始化仍在进行中，跳过重启")
+		p.restartQueued.Store(true)
+		p.logger.Info("轮询初始化仍在进行中，已标记完成后重新初始化")
 		return
 	}
 	p.setStatus("pending", "")
@@ -200,6 +202,20 @@ func (p *PollingInitializer) run(ctx context.Context) {
 	p.logger.Info("分片渐进式初始化完成",
 		zap.Int64("total_loaded", snapshot.LoadedCards),
 		zap.Duration("duration", time.Since(snapshot.StartTime)))
+
+	if p.restartQueued.Swap(false) {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+		if p.initCompleted.CompareAndSwap(true, false) {
+			p.setStatus("pending", "")
+			p.wg.Add(1)
+			go p.run(ctx)
+			p.logger.Info("检测到初始化期间配置变更，完成后再次初始化")
+		}
+	}
 }
 
 // initBatch 使用 Pipeline 将一批卡写入分片队列和缓存
diff --git a/internal/service/polling/config_service.go b/internal/service/polling/config_service.go
index 3a68a8c..a9d03f3 100644
--- a/internal/service/polling/config_service.go
+++ b/internal/service/polling/config_service.go
@@ -155,6 +155,7 @@ func (s *ConfigService) Update(ctx context.Context, id uint, req *dto.UpdatePoll
 		return nil, errors.Wrap(errors.CodeInternalError, err, "更新轮询配置失败")
 	}
 
+	s.notifyConfigChanged(ctx, "updated")
 	return s.toResponse(config), nil
 }