root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(server): hot-reload config store and reloadable infra clients

Browse Source 
- Replace single Load() with a watching Store that re-reads config files
  (and config.local.yaml) and fans out a ReloadEvent with a per-field diff
  so consumers can decide whether the change is hot-applicable or requires
  a process restart.
- Wrap llm, retrieval, vector store, and object storage clients in
  Reloadable* shells so the bootstrap can swap their underlying impls when
  config changes without re-instantiating handlers.
- Make jwt.Manager and ops TokenIssuer mutable under a lock so secrets and
  TTLs can be rotated live; thread default plan code through a setter on
  the ops AdminUserService.
- Wire ConfigStore through bootstrap and every cmd/main.go, scheduler /
  worker / tenant-api / ops-api start the watcher; services and handlers
  take a config.Provider so they always read current values for things
  like generation.stream_enabled, scheduler dispatch, retrieval, etc.
- Switch shared/config decoding off viper to a Kratos-derived runtime
  package so env placeholders (\${VAR:default}) resolve consistently and
  the same source machinery powers both the loader and the watcher.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
liangxu
2026-05-01 16:01:23 +08:00
parent ce2d8a2907
commit 618399f86d
61 changed files with 3186 additions and 496 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/scheduler/schedule_dispatch_worker.go
+62 -26
View File
@@ -30,11 +30,8 @@ type ScheduleDispatchWorker struct {
promptRuleService *tenantapp.PromptRuleGenerationService
cache sharedcache.Cache
logger *zap.Logger
interval time.Duration
timeout time.Duration
batchSize int
dispatchWorkers int
queueBackpressure int
cfg scheduleDispatchRuntimeConfig
configProvider config.Provider
}
type dueScheduleTask struct {
@@ -62,6 +59,14 @@ type scheduleTaskCacheUpdate struct {
tenantID int64
}
type scheduleDispatchRuntimeConfig struct {
Interval time.Duration
Timeout time.Duration
BatchSize int
DispatchWorkers int
QueueBackpressure int
}
func NewScheduleDispatchWorker(
pool *pgxpool.Pool,
rabbitMQClient *rabbitmq.Client,
@@ -76,14 +81,27 @@ func NewScheduleDispatchWorker(
promptRuleService: promptRuleService,
cache: cache,
logger: logger,
interval: schedulerDispatchInterval(cfg),
timeout: schedulerDispatchTimeout(cfg),
batchSize: schedulerDispatchBatchSize(cfg),
dispatchWorkers: schedulerDispatchConcurrency(cfg),
queueBackpressure: cfg.GenerationQueueBackpressureLimit,
cfg: scheduleDispatchRuntimeFromConfig(cfg),
}
}
func (w *ScheduleDispatchWorker) WithConfigProvider(provider config.Provider) *ScheduleDispatchWorker {
w.configProvider = provider
return w
}
func (w *ScheduleDispatchWorker) runtimeConfig() scheduleDispatchRuntimeConfig {
if w != nil && w.configProvider != nil {
if cfg := w.configProvider.Current(); cfg != nil {
return scheduleDispatchRuntimeFromConfig(cfg.Scheduler)
}
}
if w == nil {
return scheduleDispatchRuntimeFromConfig(config.SchedulerConfig{})
}
return w.cfg
}
func (w *ScheduleDispatchWorker) Start(ctx context.Context) {
go w.Run(ctx)
}
@@ -98,14 +116,14 @@ func (w *ScheduleDispatchWorker) Run(ctx context.Context) {
func (w *ScheduleDispatchWorker) run(ctx context.Context) {
w.runOnce(context.Background())
ticker := time.NewTicker(w.interval)
defer ticker.Stop()
for {
interval := w.runtimeConfig().Interval
timer := time.NewTimer(interval)
select {
case <-ctx.Done():
timer.Stop()
return
case <-ticker.C:
case <-timer.C:
w.runOnce(context.Background())
}
}
@@ -132,10 +150,11 @@ func (w *ScheduleDispatchWorker) runOnce(parent context.Context) {
}
} else if paused {
if w.logger != nil {
runtimeCfg := w.runtimeConfig()
w.logger.Info("schedule dispatch paused by generation queue backpressure",
zap.Int("queue_depth", stats.Messages),
zap.Int("queue_consumers", stats.Consumers),
zap.Int("queue_limit", w.queueBackpressure),
zap.Int("queue_limit", runtimeCfg.QueueBackpressure),
)
}
return
@@ -155,6 +174,7 @@ func (w *ScheduleDispatchWorker) runOnce(parent context.Context) {
}
func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context) (int, error) {
runtimeCfg := w.runtimeConfig()
tx, err := w.pool.Begin(ctx)
if err != nil {
return 0, err
@@ -170,7 +190,7 @@ func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context) (in
ORDER BY created_at ASC, id ASC
FOR UPDATE SKIP LOCKED
LIMIT $1
`, w.batchSize)
`, runtimeCfg.BatchSize)
if err != nil {
return 0, err
}
@@ -180,7 +200,7 @@ func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context) (in
updates := make([]struct {
cacheUpdate scheduleTaskCacheUpdate
nextRunAt *time.Time
}, 0, w.batchSize)
}, 0, runtimeCfg.BatchSize)
for rows.Next() {
var (
id int64
@@ -248,6 +268,7 @@ func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context) (in
}
func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context) ([]dueScheduleTask, error) {
runtimeCfg := w.runtimeConfig()
tx, err := w.pool.Begin(ctx)
if err != nil {
return nil, err
@@ -266,17 +287,17 @@ func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context) ([]dueSc
ORDER BY next_run_at ASC, id ASC
FOR UPDATE SKIP LOCKED
LIMIT $2
`, time.Now(), w.batchSize)
`, time.Now(), runtimeCfg.BatchSize)
if err != nil {
return nil, err
}
now := time.Now()
tasks := make([]dueScheduleTask, 0, w.batchSize)
tasks := make([]dueScheduleTask, 0, runtimeCfg.BatchSize)
updates := make([]struct {
cacheUpdate scheduleTaskCacheUpdate
nextRun *time.Time
}, 0, w.batchSize)
}, 0, runtimeCfg.BatchSize)
for rows.Next() {
var (
task dueScheduleTask
@@ -386,7 +407,7 @@ func (w *ScheduleDispatchWorker) dispatch(parent context.Context, task dueSchedu
}
for idx := 0; idx < generateCount; idx++ {
ctx, cancel := context.WithTimeout(parent, w.timeout)
ctx, cancel := w.stageContext(parent)
resp, err := w.promptRuleService.EnqueueScheduledGeneration(ctx, tenantapp.SchedulePromptRuleGenerationInput{
ScheduleTaskID: task.ID,
OperatorID: task.OperatorID,
@@ -438,7 +459,8 @@ func (w *ScheduleDispatchWorker) dispatchBatch(parent context.Context, tasks []d
if len(tasks) == 0 {
return
}
if w.dispatchWorkers <= 1 {
runtimeCfg := w.runtimeConfig()
if runtimeCfg.DispatchWorkers <= 1 {
for _, task := range tasks {
if parent.Err() != nil {
return
@@ -448,7 +470,7 @@ func (w *ScheduleDispatchWorker) dispatchBatch(parent context.Context, tasks []d
return
}
sem := make(chan struct{}, w.dispatchWorkers)
sem := make(chan struct{}, runtimeCfg.DispatchWorkers)
var wg sync.WaitGroup
for _, task := range tasks {
if parent.Err() != nil {
@@ -467,7 +489,11 @@ func (w *ScheduleDispatchWorker) dispatchBatch(parent context.Context, tasks []d
}
func (w *ScheduleDispatchWorker) shouldPauseDispatch(ctx context.Context) (bool, rabbitmq.QueueStats, error) {
if w == nil || w.rabbitMQ == nil || w.queueBackpressure <= 0 {
if w == nil || w.rabbitMQ == nil {
return false, rabbitmq.QueueStats{}, nil
}
runtimeCfg := w.runtimeConfig()
if runtimeCfg.QueueBackpressure <= 0 {
return false, rabbitmq.QueueStats{}, nil
}
@@ -475,7 +501,17 @@ func (w *ScheduleDispatchWorker) shouldPauseDispatch(ctx context.Context) (bool,
if err != nil {
return false, rabbitmq.QueueStats{}, err
}
return stats.Messages >= w.queueBackpressure, stats, nil
return stats.Messages >= runtimeCfg.QueueBackpressure, stats, nil
}
func scheduleDispatchRuntimeFromConfig(cfg config.SchedulerConfig) scheduleDispatchRuntimeConfig {
return scheduleDispatchRuntimeConfig{
Interval: schedulerDispatchInterval(cfg),
Timeout: schedulerDispatchTimeout(cfg),
BatchSize: schedulerDispatchBatchSize(cfg),
DispatchWorkers: schedulerDispatchConcurrency(cfg),
QueueBackpressure: cfg.GenerationQueueBackpressureLimit,
}
}
func schedulerDispatchInterval(cfg config.SchedulerConfig) time.Duration {
@@ -510,7 +546,7 @@ func (w *ScheduleDispatchWorker) stageContext(parent context.Context) (context.C
if parent == nil {
parent = context.Background()
}
return context.WithTimeout(parent, w.timeout)
return context.WithTimeout(parent, w.runtimeConfig().Timeout)
}
func (w *ScheduleDispatchWorker) invalidateScheduleTaskCaches(updates []scheduleTaskCacheUpdate) {