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.
2026-06-02 14:50:25 +08:00
parent ba2f117265
commit 842782b3dd
7 changed files with 343 additions and 8 deletions
@@ -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 {
@@ -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