From 842782b3dde36d5d233147b2178c47a3af467ff7 Mon Sep 17 00:00:00 2001 From: liangxu Date: Tue, 2 Jun 2026 14:50:25 +0800 Subject: [PATCH] feat(scheduler): random run-window scheduling and deduped manual triggers Support a random daily run window (random_window_start/end) where the next auto-run time is derived from a stable per-job hash, avoiding thundering-herd starts while keeping one run per day. Honors the last auto-run on initial scheduling so restarts don't double-fire. Manual triggers can opt into dedupe (dedupe_manual_triggers): an advisory-locked path collapses concurrent requests onto an existing pending/running job instead of queueing duplicates. Scheduler error messages are localized to Chinese. Register the media_supply_cache_sync job (dry-run aware) and add the ops migration that seeds it. --- server/cmd/scheduler/main.go | 13 +++ server/internal/ops/app/scheduler.go | 46 ++++++-- server/internal/ops/repository/scheduler.go | 109 ++++++++++++++++++ server/internal/scheduler/control_plane.go | 108 +++++++++++++++++ .../internal/scheduler/control_plane_test.go | 40 +++++++ ...a_supply_cache_sync_scheduler_job.down.sql | 8 ++ ...dia_supply_cache_sync_scheduler_job.up.sql | 27 +++++ 7 files changed, 343 insertions(+), 8 deletions(-) create mode 100644 server/migrations_ops/20260602130000_add_media_supply_cache_sync_scheduler_job.down.sql create mode 100644 server/migrations_ops/20260602130000_add_media_supply_cache_sync_scheduler_job.up.sql diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go index e255174..129f8c5 100644 --- a/server/cmd/scheduler/main.go +++ b/server/cmd/scheduler/main.go @@ -181,6 +181,19 @@ func main() { return schedulerRunRetentionWorker.RunOnce(runCtx, jobCtx) }, }, + { + Key: "media_supply_cache_sync", + Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { + modelID := internalscheduler.AsInt(jobCtx.Config, "model_id", 1) + if jobCtx.DryRun { + return map[string]any{ + "model_id": modelID, + "synced": 0, + }, nil + } + return app.MediaSupplyService.RunMediaResourceSyncOnce(runCtx, modelID) + }, + }, }) startSchedulerWorker(ctx, &workerWG, "control_plane", app.Logger.Sugar(), controlPlane.Run) diff --git a/server/internal/ops/app/scheduler.go b/server/internal/ops/app/scheduler.go index b7c1f4f..1199244 100644 --- a/server/internal/ops/app/scheduler.go +++ b/server/internal/ops/app/scheduler.go @@ -85,7 +85,7 @@ func (s *SchedulerService) ListJobs(ctx context.Context, in SchedulerJobListInpu Offset: (page - 1) * size, }) if err != nil { - return nil, response.ErrInternal(53001, "scheduler_jobs_query_failed", "failed to list scheduler jobs") + return nil, response.ErrInternal(53001, "scheduler_jobs_query_failed", "调度任务列表读取失败") } return &SchedulerJobListResult{Items: items, Total: total, Page: page, Size: size}, nil } @@ -140,22 +140,37 @@ func (s *SchedulerService) SetEnabled(ctx context.Context, actor *Actor, key str func (s *SchedulerService) Trigger(ctx context.Context, actor *Actor, key string, triggerType string, in SchedulerTriggerInput) (*domain.SchedulerTrigger, error) { key = normalizeSchedulerJobKey(key) - if _, err := s.repo.GetJobForRuntime(ctx, key); err != nil { + job, err := s.repo.GetJobForRuntime(ctx, key) + if err != nil { return nil, mapSchedulerJobError(err) } triggerType = strings.TrimSpace(triggerType) if triggerType != domain.SchedulerTriggerManual && triggerType != domain.SchedulerTriggerDryRun { return nil, response.ErrBadRequest(40083, "invalid_trigger_type", "trigger_type 不合法") } - trigger, err := s.repo.CreateTrigger(ctx, repository.SchedulerTriggerCreate{ + create := repository.SchedulerTriggerCreate{ JobKey: key, TriggerType: triggerType, RequestedBy: actorID(actor), Reason: strings.TrimSpace(in.Reason), ConfigOverride: sanitizeSchedulerConfig(in.ConfigOverride), - }) + } + if triggerType == domain.SchedulerTriggerManual && boolFromSchedulerConfig(job.Config, "dedupe_manual_triggers") { + trigger, created, triggerErr := s.repo.CreateDedupedManualTrigger(ctx, create) + if triggerErr != nil { + return nil, response.ErrInternal(53004, "scheduler_trigger_create_failed", "创建调度触发失败") + } + if created { + s.auditScheduler(ctx, actor, ActionSchedulerJobRun, key, map[string]any{ + "trigger_id": trigger.ID, + "reason": strings.TrimSpace(in.Reason), + }) + } + return trigger, nil + } + trigger, err := s.repo.CreateTrigger(ctx, create) if err != nil { - return nil, response.ErrInternal(53004, "scheduler_trigger_create_failed", "failed to create scheduler trigger") + return nil, response.ErrInternal(53004, "scheduler_trigger_create_failed", "创建调度触发失败") } action := ActionSchedulerJobRun if triggerType == domain.SchedulerTriggerDryRun { @@ -180,15 +195,30 @@ func (s *SchedulerService) ListRuns(ctx context.Context, key, status string, pag Offset: (page - 1) * size, }) if err != nil { - return nil, response.ErrInternal(53005, "scheduler_runs_query_failed", "failed to list scheduler runs") + return nil, response.ErrInternal(53005, "scheduler_runs_query_failed", "调度运行历史读取失败") } return &SchedulerRunListResult{Items: items, Total: total, Page: page, Size: size}, nil } +func boolFromSchedulerConfig(config map[string]any, key string) bool { + value, ok := config[key] + if !ok || value == nil { + return false + } + switch typed := value.(type) { + case bool: + return typed + case string: + return strings.EqualFold(strings.TrimSpace(typed), "true") + default: + return false + } +} + func (s *SchedulerService) ListInstances(ctx context.Context) ([]domain.SchedulerInstance, error) { items, err := s.repo.ListInstances(ctx) if err != nil { - return nil, response.ErrInternal(53006, "scheduler_instances_query_failed", "failed to list scheduler instances") + return nil, response.ErrInternal(53006, "scheduler_instances_query_failed", "调度实例读取失败") } return items, nil } @@ -255,7 +285,7 @@ func mapSchedulerJobError(err error) error { if errors.Is(err, repository.ErrSchedulerJobNotFound) { return response.ErrNotFound(40480, "scheduler_job_not_found", "调度任务不存在") } - return response.ErrInternal(53002, "scheduler_job_query_failed", "failed to load scheduler job") + return response.ErrInternal(53002, "scheduler_job_query_failed", "调度任务读取失败") } func actorID(actor *Actor) *int64 { diff --git a/server/internal/ops/repository/scheduler.go b/server/internal/ops/repository/scheduler.go index 174424f..88322f8 100644 --- a/server/internal/ops/repository/scheduler.go +++ b/server/internal/ops/repository/scheduler.go @@ -270,6 +270,115 @@ func (r *SchedulerRepository) CreateTrigger(ctx context.Context, in SchedulerTri `, in.JobKey, in.TriggerType, in.RequestedBy, reason, configJSON)) } +func (r *SchedulerRepository) CreateDedupedManualTrigger(ctx context.Context, in SchedulerTriggerCreate) (*domain.SchedulerTrigger, bool, error) { + if in.TriggerType != domain.SchedulerTriggerManual { + trigger, err := r.CreateTrigger(ctx, in) + return trigger, true, err + } + tx, err := r.pool.Begin(ctx) + if err != nil { + return nil, false, err + } + defer tx.Rollback(ctx) + + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))`, "scheduler_trigger", in.JobKey); err != nil { + return nil, false, err + } + if item, ok, err := findActiveTrigger(ctx, tx, in.JobKey, in.TriggerType); err != nil { + return nil, false, err + } else if ok { + return item, false, tx.Commit(ctx) + } + if _, ok, err := findRunningRun(ctx, tx, in.JobKey); err != nil { + return nil, false, err + } else if ok { + reason := in.Reason + if reason == "" { + reason = "已有调度任务正在执行，本次请求已合并" + } + item := &domain.SchedulerTrigger{ + ID: 0, + JobKey: in.JobKey, + TriggerType: in.TriggerType, + Status: domain.SchedulerRunSkipped, + RequestedBy: in.RequestedBy, + Reason: nullableString(reason), + ConfigOverride: decodeSchedulerJSONMap(nil), + RequestedAt: time.Now().UTC(), + } + if in.ConfigOverride != nil { + item.ConfigOverride = in.ConfigOverride + } + return item, false, tx.Commit(ctx) + } + + configJSON, err := marshalSchedulerJSONMap(in.ConfigOverride) + if err != nil { + return nil, false, err + } + trigger, err := scanSchedulerTrigger(tx.QueryRow(ctx, ` + INSERT INTO ops.scheduler_job_triggers (job_key, trigger_type, requested_by, reason, config_override) + VALUES ($1, $2, $3, $4, COALESCE($5::jsonb, '{}'::jsonb)) + RETURNING id, job_key, trigger_type, status, requested_by, reason, config_override, + scheduler_instance_id, run_id, requested_at, claimed_at, finished_at, error_message + `, in.JobKey, in.TriggerType, in.RequestedBy, nullableString(in.Reason), configJSON)) + if err != nil { + return nil, false, err + } + return trigger, true, tx.Commit(ctx) +} + +func (r *SchedulerRepository) FindActiveTrigger(ctx context.Context, jobKey, triggerType string) (*domain.SchedulerTrigger, bool, error) { + return findActiveTrigger(ctx, r.pool, jobKey, triggerType) +} + +func findActiveTrigger(ctx context.Context, q interface { + QueryRow(context.Context, string, ...any) pgx.Row +}, jobKey, triggerType string) (*domain.SchedulerTrigger, bool, error) { + item, err := scanSchedulerTrigger(q.QueryRow(ctx, ` + SELECT id, job_key, trigger_type, status, requested_by, reason, config_override, + scheduler_instance_id, run_id, requested_at, claimed_at, finished_at, error_message + FROM ops.scheduler_job_triggers + WHERE job_key = $1 + AND trigger_type = $2 + AND status IN ('pending', 'claimed') + ORDER BY requested_at ASC, id ASC + LIMIT 1 + `, jobKey, triggerType)) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, false, nil + } + return nil, false, err + } + return item, true, nil +} + +func (r *SchedulerRepository) FindRunningRun(ctx context.Context, jobKey string) (*domain.SchedulerRun, bool, error) { + return findRunningRun(ctx, r.pool, jobKey) +} + +func findRunningRun(ctx context.Context, q interface { + QueryRow(context.Context, string, ...any) pgx.Row +}, jobKey string) (*domain.SchedulerRun, bool, error) { + item, err := scanSchedulerRun(q.QueryRow(ctx, ` + SELECT id, job_key, trigger_id, scheduler_instance_id, trigger_type, status, + config_version, config_snapshot, stats, error_message, started_at, + finished_at, duration_ms + FROM ops.scheduler_job_runs + WHERE job_key = $1 AND status = 'running' + ORDER BY started_at DESC, id DESC + LIMIT 1 + `, jobKey)) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, false, nil + } + return nil, false, err + } + return item, true, nil +} + func (r *SchedulerRepository) ClaimPendingTrigger(ctx context.Context, jobKey, instanceID string) (*SchedulerTriggerClaim, bool, error) { var claim SchedulerTriggerClaim var configRaw []byte diff --git a/server/internal/scheduler/control_plane.go b/server/internal/scheduler/control_plane.go index 4ffc08c..d2e9dc9 100644 --- a/server/internal/scheduler/control_plane.go +++ b/server/internal/scheduler/control_plane.go @@ -222,6 +222,27 @@ func (p *ControlPlane) nextAutoAt(ctx context.Context, key string, now time.Time base := now.In(loc) return schedule.Next(base) } + if window, ok := parseRandomRunWindowLocal(job.Config); ok { + loc, locErr := time.LoadLocation(job.Timezone) + if locErr != nil { + loc = time.Local + } + var lastStartedAt *time.Time + if initial { + startedAt, ok, lastErr := p.repo.LastAutoRunStartedAt(ctx, key) + if lastErr != nil { + if p.logger != nil { + p.logger.Warn("scheduler last auto run lookup failed", + zap.String("job_key", key), + zap.Error(lastErr), + ) + } + } else if ok { + lastStartedAt = &startedAt + } + } + return nextRandomWindowAutoAt(key, now, loc, window, initial, lastStartedAt) + } if runAt := parseRunAtLocal(job.Config); runAt != "" { loc, locErr := time.LoadLocation(job.Timezone) if locErr != nil { @@ -629,6 +650,93 @@ func parseRunAtLocal(config map[string]any) string { return fmt.Sprint(value) } +type randomRunWindowLocal struct { + StartHour int + StartMinute int + EndHour int + EndMinute int +} + +func parseRandomRunWindowLocal(config map[string]any) (randomRunWindowLocal, bool) { + start := firstConfigString(config, "random_window_start", "random_window_start_local", "random_run_window_start") + end := firstConfigString(config, "random_window_end", "random_window_end_local", "random_run_window_end") + if start == "" || end == "" { + return randomRunWindowLocal{}, false + } + startHour, startMinute, startOK := parseHourMinute(start) + endHour, endMinute, endOK := parseHourMinute(end) + startTotal := startHour*60 + startMinute + endTotal := endHour*60 + endMinute + if !startOK || !endOK || endTotal <= startTotal { + return randomRunWindowLocal{}, false + } + return randomRunWindowLocal{ + StartHour: startHour, + StartMinute: startMinute, + EndHour: endHour, + EndMinute: endMinute, + }, true +} + +func nextRandomWindowAutoAt(key string, now time.Time, loc *time.Location, window randomRunWindowLocal, initial bool, lastStartedAt *time.Time) time.Time { + if loc == nil { + loc = time.Local + } + localNow := now.In(loc) + todayStart := time.Date(localNow.Year(), localNow.Month(), localNow.Day(), window.StartHour, window.StartMinute, 0, 0, loc) + todayEnd := time.Date(localNow.Year(), localNow.Month(), localNow.Day(), window.EndHour, window.EndMinute, 0, 0, loc) + if !todayEnd.After(todayStart) { + return now.Add(schedulerConfigRetryInterval) + } + todayCandidate := randomWindowCandidate(key, todayStart, todayEnd) + if !initial { + tomorrowStart := todayStart.AddDate(0, 0, 1) + return randomWindowCandidate(key, tomorrowStart, tomorrowStart.Add(todayEnd.Sub(todayStart))) + } + if lastStartedAt != nil && sameLocalDate((*lastStartedAt).In(loc), localNow) { + tomorrowStart := todayStart.AddDate(0, 0, 1) + return randomWindowCandidate(key, tomorrowStart, tomorrowStart.Add(todayEnd.Sub(todayStart))) + } + if localNow.Before(todayCandidate) { + return todayCandidate + } + if localNow.Before(todayEnd) { + return now + } + tomorrowStart := todayStart.AddDate(0, 0, 1) + return randomWindowCandidate(key, tomorrowStart, tomorrowStart.Add(todayEnd.Sub(todayStart))) +} + +func randomWindowCandidate(key string, start, end time.Time) time.Time { + seconds := int64(end.Sub(start).Seconds()) + if seconds <= 0 { + return start + } + hash := fnv.New64a() + _, _ = hash.Write([]byte("scheduler-random-window:" + key + ":" + start.Format("2006-01-02") + ":" + start.Format("15:04") + ":" + end.Format("15:04"))) + offset := int64(hash.Sum64() % uint64(seconds)) + return start.Add(time.Duration(offset) * time.Second) +} + +func sameLocalDate(a, b time.Time) bool { + ay, am, ad := a.Date() + by, bm, bd := b.Date() + return ay == by && am == bm && ad == bd +} + +func firstConfigString(config map[string]any, keys ...string) string { + for _, key := range keys { + value, ok := config[key] + if !ok || value == nil { + continue + } + if text := strings.TrimSpace(fmt.Sprint(value)); text != "" { + return text + } + } + return "" +} + func parseHourMinute(value string) (int, int, bool) { parts := strings.Split(strings.TrimSpace(value), ":") if len(parts) != 2 { diff --git a/server/internal/scheduler/control_plane_test.go b/server/internal/scheduler/control_plane_test.go index c04ab51..b2a8529 100644 --- a/server/internal/scheduler/control_plane_test.go +++ b/server/internal/scheduler/control_plane_test.go @@ -46,6 +46,46 @@ func TestNextIntervalAutoAtSchedulesAfterCurrentCompletion(t *testing.T) { } } +func TestNextRandomWindowAutoAtIsStableInsideWindow(t *testing.T) { + loc := time.FixedZone("CST", 8*60*60) + window, ok := parseRandomRunWindowLocal(map[string]any{ + "random_window_start": "01:00", + "random_window_end": "04:00", + }) + if !ok { + t.Fatal("expected random window config to parse") + } + now := time.Date(2026, 6, 2, 0, 30, 0, 0, loc) + + got := nextRandomWindowAutoAt("media_supply_cache_sync", now, loc, window, true, nil) + gotAgain := nextRandomWindowAutoAt("media_supply_cache_sync", now.Add(10*time.Minute), loc, window, true, nil) + start := time.Date(2026, 6, 2, 1, 0, 0, 0, loc) + end := time.Date(2026, 6, 2, 4, 0, 0, 0, loc) + if got.Before(start) || !got.Before(end) { + t.Fatalf("nextRandomWindowAutoAt() = %s, want inside [%s, %s)", got, start, end) + } + if !got.Equal(gotAgain) { + t.Fatalf("expected stable candidate, got %s and %s", got, gotAgain) + } +} + +func TestNextRandomWindowAutoAtSkipsTodayAfterAutoRun(t *testing.T) { + loc := time.FixedZone("CST", 8*60*60) + window, _ := parseRandomRunWindowLocal(map[string]any{ + "random_window_start": "01:00", + "random_window_end": "04:00", + }) + now := time.Date(2026, 6, 2, 2, 0, 0, 0, loc) + last := time.Date(2026, 6, 2, 1, 30, 0, 0, loc) + + got := nextRandomWindowAutoAt("media_supply_cache_sync", now, loc, window, true, &last) + wantStart := time.Date(2026, 6, 3, 1, 0, 0, 0, loc) + wantEnd := time.Date(2026, 6, 3, 4, 0, 0, 0, loc) + if got.Before(wantStart) || !got.Before(wantEnd) { + t.Fatalf("nextRandomWindowAutoAt() = %s, want inside tomorrow window [%s, %s)", got, wantStart, wantEnd) + } +} + func TestIsEmptySuccessRunRequiresConfiguredZeroMetrics(t *testing.T) { stats := map[string]any{ "hydrated_count": 0, diff --git a/server/migrations_ops/20260602130000_add_media_supply_cache_sync_scheduler_job.down.sql b/server/migrations_ops/20260602130000_add_media_supply_cache_sync_scheduler_job.down.sql new file mode 100644 index 0000000..643b0ca --- /dev/null +++ b/server/migrations_ops/20260602130000_add_media_supply_cache_sync_scheduler_job.down.sql @@ -0,0 +1,8 @@ +DELETE FROM ops.scheduler_job_triggers +WHERE job_key = 'media_supply_cache_sync'; + +DELETE FROM ops.scheduler_job_runs +WHERE job_key = 'media_supply_cache_sync'; + +DELETE FROM ops.scheduler_jobs +WHERE job_key = 'media_supply_cache_sync'; diff --git a/server/migrations_ops/20260602130000_add_media_supply_cache_sync_scheduler_job.up.sql b/server/migrations_ops/20260602130000_add_media_supply_cache_sync_scheduler_job.up.sql new file mode 100644 index 0000000..ce0d307 --- /dev/null +++ b/server/migrations_ops/20260602130000_add_media_supply_cache_sync_scheduler_job.up.sql @@ -0,0 +1,27 @@ +INSERT INTO ops.scheduler_jobs + (job_key, display_name, category, description, enabled, schedule_type, interval_seconds, timezone, timeout_seconds, batch_size, config) +VALUES + ( + 'media_supply_cache_sync', + '媒体资源缓存同步', + 'media_supply', + '每日低频同步权威媒体资源缓存，并在成本价上涨超过手动售价时清理手动改价。', + true, + 'interval', + 86400, + 'Asia/Shanghai', + 3600, + 1, + '{"model_id":1,"random_window_start":"01:00","random_window_end":"04:00","dedupe_manual_triggers":true}'::jsonb + ) +ON CONFLICT (job_key) DO UPDATE +SET display_name = EXCLUDED.display_name, + category = EXCLUDED.category, + description = EXCLUDED.description, + schedule_type = EXCLUDED.schedule_type, + interval_seconds = EXCLUDED.interval_seconds, + timezone = EXCLUDED.timezone, + timeout_seconds = EXCLUDED.timeout_seconds, + batch_size = EXCLUDED.batch_size, + config = ops.scheduler_jobs.config || EXCLUDED.config, + updated_at = NOW();