diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go index 2db4981..106b282 100644 --- a/server/cmd/scheduler/main.go +++ b/server/cmd/scheduler/main.go @@ -96,6 +96,7 @@ func main() { WithCache(app.Cache) monitoringRetentionWorker := internalscheduler.NewMonitoringRetentionWorker(app.MonitoringDB, app.Logger) aiPointUsageCleanupWorker := internalscheduler.NewAIPointUsageCleanupWorker(app.DB, app.Logger) + schedulerRunRetentionWorker := internalscheduler.NewSchedulerRunRetentionWorker(app.DB, app.Logger) controlPlane := internalscheduler.NewControlPlane(app.DB, app.Logger, []internalscheduler.ControlledJob{ { @@ -152,6 +153,12 @@ func main() { return aiPointUsageCleanupWorker.RunOnce(runCtx, jobCtx) }, }, + { + Key: "scheduler_run_retention", + Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { + return schedulerRunRetentionWorker.RunOnce(runCtx, jobCtx) + }, + }, }) startSchedulerWorker(ctx, &workerWG, "control_plane", app.Logger.Sugar(), controlPlane.Run) diff --git a/server/internal/ops/repository/scheduler.go b/server/internal/ops/repository/scheduler.go index 26384cb..174424f 100644 --- a/server/internal/ops/repository/scheduler.go +++ b/server/internal/ops/repository/scheduler.go @@ -67,6 +67,20 @@ type SchedulerRunFinish struct { ErrorMessage *string } +type SchedulerRunRecord struct { + JobKey string + TriggerID *int64 + SchedulerInstanceID string + TriggerType string + ConfigVersion int + ConfigSnapshot map[string]any + Status string + Stats map[string]any + ErrorMessage *string + StartedAt time.Time + FinishedAt time.Time +} + type SchedulerTriggerCreate struct { JobKey string TriggerType string @@ -332,6 +346,35 @@ func (r *SchedulerRepository) FinishRun(ctx context.Context, in SchedulerRunFini `, in.RunID, in.Status, statsJSON, in.ErrorMessage)) } +func (r *SchedulerRepository) RecordRun(ctx context.Context, in SchedulerRunRecord) (*domain.SchedulerRun, error) { + configJSON, err := marshalSchedulerJSONMap(in.ConfigSnapshot) + if err != nil { + return nil, err + } + statsJSON, err := marshalSchedulerJSONMap(in.Stats) + if err != nil { + return nil, err + } + duration := in.FinishedAt.Sub(in.StartedAt).Milliseconds() + if duration < 0 { + duration = 0 + } + return scanSchedulerRun(r.pool.QueryRow(ctx, ` + INSERT INTO ops.scheduler_job_runs ( + job_key, trigger_id, scheduler_instance_id, trigger_type, + status, config_version, config_snapshot, stats, error_message, + started_at, finished_at, duration_ms + ) + VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7::jsonb, '{}'::jsonb), + COALESCE($8::jsonb, '{}'::jsonb), $9, $10, $11, $12) + RETURNING id, job_key, trigger_id, scheduler_instance_id, trigger_type, status, + config_version, config_snapshot, stats, error_message, started_at, + finished_at, duration_ms + `, in.JobKey, in.TriggerID, nullableString(in.SchedulerInstanceID), in.TriggerType, + in.Status, in.ConfigVersion, configJSON, statsJSON, in.ErrorMessage, + in.StartedAt, in.FinishedAt, duration)) +} + func (r *SchedulerRepository) ListRuns(ctx context.Context, f SchedulerRunFilter) ([]domain.SchedulerRun, int64, error) { args := []any{} where := "1=1" diff --git a/server/internal/scheduler/control_plane.go b/server/internal/scheduler/control_plane.go index 1b217a8..4ffc08c 100644 --- a/server/internal/scheduler/control_plane.go +++ b/server/internal/scheduler/control_plane.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "hash/fnv" + "math" "os" "strconv" "strings" @@ -281,6 +282,12 @@ func (p *ControlPlane) runJobOnce(parent context.Context, job ControlledJob, tri if dryRun { mergedConfig["dry_run"] = true } + // Empty-success suppression is only for high-frequency automatic polling jobs. + // Suppressed runs intentionally do not create running rows and do not advance + // LastAutoRunStartedAt, so process restarts may run them immediately once as + // overdue. Manual, dry-run, failed, and non-empty executions still get full + // audit rows below. + suppressEmptySuccessRun := shouldSuppressEmptySuccessRun(triggerType, dryRun, mergedConfig) lockKey := advisoryLockKey(job.Key) lockConn, locked := p.tryAdvisoryLock(parent, lockKey) @@ -295,19 +302,25 @@ func (p *ControlPlane) runJobOnce(parent context.Context, job ControlledJob, tri } defer p.advisoryUnlock(context.Background(), lockConn, lockKey) - run, err := p.repo.StartRun(parent, opsrepo.SchedulerRunStart{ - JobKey: job.Key, - TriggerID: triggerID, - SchedulerInstanceID: p.instanceID, - TriggerType: triggerType, - ConfigVersion: cfg.Version, - ConfigSnapshot: schedulerConfigSnapshot(cfg, mergedConfig), - }) - if err != nil { - if p.logger != nil { - p.logger.Warn("scheduler run start failed", zap.String("job_key", job.Key), zap.Error(err)) + configSnapshot := schedulerConfigSnapshot(cfg, mergedConfig) + startedAt := time.Now() + var run *opsdomain.SchedulerRun + if !suppressEmptySuccessRun { + run, err = p.repo.StartRun(parent, opsrepo.SchedulerRunStart{ + JobKey: job.Key, + TriggerID: triggerID, + SchedulerInstanceID: p.instanceID, + TriggerType: triggerType, + ConfigVersion: cfg.Version, + ConfigSnapshot: configSnapshot, + }) + if err != nil { + if p.logger != nil { + p.logger.Warn("scheduler run start failed", zap.String("job_key", job.Key), zap.Error(err)) + } + return } - return + startedAt = run.StartedAt } timeout := time.Duration(cfg.TimeoutSeconds) * time.Second @@ -335,25 +348,61 @@ func (p *ControlPlane) runJobOnce(parent context.Context, job ControlledJob, tri stats = map[string]any{} } stats["dry_run"] = dryRun + finishedAt := time.Now() + if suppressEmptySuccessRun && runErr == nil && isEmptySuccessRun(stats, mergedConfig) { + if p.logger != nil { + p.logger.Info("scheduler empty success run suppressed", zap.String("job_key", job.Key)) + } + return + } finishCtx, finishCancel := context.WithTimeout(context.Background(), schedulerFinishTimeout) defer finishCancel() - finished, finishErr := p.repo.FinishRun(finishCtx, opsrepo.SchedulerRunFinish{ - RunID: run.ID, - Status: status, - Stats: stats, - ErrorMessage: errMsg, - }) + var finished *opsdomain.SchedulerRun + var finishErr error + if run == nil { + finished, finishErr = p.repo.RecordRun(finishCtx, opsrepo.SchedulerRunRecord{ + JobKey: job.Key, + TriggerID: triggerID, + SchedulerInstanceID: p.instanceID, + TriggerType: triggerType, + ConfigVersion: cfg.Version, + ConfigSnapshot: configSnapshot, + Status: status, + Stats: stats, + ErrorMessage: errMsg, + StartedAt: startedAt, + FinishedAt: finishedAt, + }) + } else { + finished, finishErr = p.repo.FinishRun(finishCtx, opsrepo.SchedulerRunFinish{ + RunID: run.ID, + Status: status, + Stats: stats, + ErrorMessage: errMsg, + }) + } if finishErr != nil && p.logger != nil { - p.logger.Warn("scheduler run finish failed", zap.String("job_key", job.Key), zap.Int64("run_id", run.ID), zap.Error(finishErr)) + fields := []zap.Field{zap.String("job_key", job.Key), zap.Error(finishErr)} + if run != nil { + fields = append(fields, zap.Int64("run_id", run.ID)) + } + if finished != nil { + fields = append(fields, zap.Int64("finished_run_id", finished.ID)) + } + p.logger.Warn("scheduler run finish failed", fields...) } if triggerID != nil { triggerStatus := status if finishErr != nil { triggerStatus = opsdomain.SchedulerRunFailed } - runID := run.ID - _ = p.repo.FinishTrigger(finishCtx, *triggerID, &runID, triggerStatus, errMsg) + var runID *int64 + if finished != nil { + value := finished.ID + runID = &value + } + _ = p.repo.FinishTrigger(finishCtx, *triggerID, runID, triggerStatus, errMsg) } if runErr != nil { if p.logger != nil { @@ -477,6 +526,101 @@ func boolFromMap(in map[string]any, key string) bool { } } +func shouldSuppressEmptySuccessRun(triggerType string, dryRun bool, config map[string]any) bool { + if dryRun || triggerType != opsdomain.SchedulerTriggerAuto { + return false + } + return boolFromMap(config, "suppress_empty_success_runs") +} + +func isEmptySuccessRun(stats map[string]any, config map[string]any) bool { + if stats == nil || !boolFromMap(config, "suppress_empty_success_runs") { + return false + } + keys := emptySuccessMetricKeys(config) + if len(keys) == 0 { + return false + } + for _, key := range keys { + value, ok := stats[key] + if !ok { + return false + } + if numericValue(value) != 0 { + return false + } + } + return true +} + +func emptySuccessMetricKeys(config map[string]any) []string { + value, ok := config["empty_success_metric_keys"] + if !ok || value == nil { + return nil + } + switch typed := value.(type) { + case []string: + return typed + case []any: + out := make([]string, 0, len(typed)) + for _, item := range typed { + if key := strings.TrimSpace(fmt.Sprint(item)); key != "" { + out = append(out, key) + } + } + return out + case string: + parts := strings.Split(typed, ",") + out := make([]string, 0, len(parts)) + for _, part := range parts { + if key := strings.TrimSpace(part); key != "" { + out = append(out, key) + } + } + return out + default: + return nil + } +} + +func numericValue(value any) float64 { + switch typed := value.(type) { + case int: + return float64(typed) + case int8: + return float64(typed) + case int16: + return float64(typed) + case int32: + return float64(typed) + case int64: + return float64(typed) + case uint: + return float64(typed) + case uint8: + return float64(typed) + case uint16: + return float64(typed) + case uint32: + return float64(typed) + case uint64: + return float64(typed) + case float32: + return float64(typed) + case float64: + if math.IsNaN(typed) { + return 0 + } + return typed + case string: + parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64) + if err == nil { + return parsed + } + } + return 0 +} + func parseRunAtLocal(config map[string]any) string { value, ok := config["run_at_local"] if !ok || value == nil { diff --git a/server/internal/scheduler/control_plane_test.go b/server/internal/scheduler/control_plane_test.go index dc284e8..c04ab51 100644 --- a/server/internal/scheduler/control_plane_test.go +++ b/server/internal/scheduler/control_plane_test.go @@ -45,3 +45,47 @@ func TestNextIntervalAutoAtSchedulesAfterCurrentCompletion(t *testing.T) { t.Fatalf("nextIntervalAutoAt() = %s, want %s", got, want) } } + +func TestIsEmptySuccessRunRequiresConfiguredZeroMetrics(t *testing.T) { + stats := map[string]any{ + "hydrated_count": 0, + "claimed_count": 0, + } + config := map[string]any{ + "suppress_empty_success_runs": true, + "empty_success_metric_keys": []any{"hydrated_count", "claimed_count"}, + } + + if !isEmptySuccessRun(stats, config) { + t.Fatal("expected empty success run to be suppressed") + } +} + +func TestIsEmptySuccessRunKeepsRunsWithOutput(t *testing.T) { + stats := map[string]any{ + "recoverable_count": 1, + "republished_count": 0, + } + config := map[string]any{ + "suppress_empty_success_runs": true, + "empty_success_metric_keys": []any{"recoverable_count", "republished_count"}, + } + + if isEmptySuccessRun(stats, config) { + t.Fatal("expected non-empty success run to be retained") + } +} + +func TestShouldSuppressEmptySuccessRunOnlyForAutoNonDryRun(t *testing.T) { + config := map[string]any{"suppress_empty_success_runs": true} + + if !shouldSuppressEmptySuccessRun("auto", false, config) { + t.Fatal("expected auto non-dry-run to allow suppression") + } + if shouldSuppressEmptySuccessRun("manual", false, config) { + t.Fatal("manual runs must not be suppressed") + } + if shouldSuppressEmptySuccessRun("auto", true, config) { + t.Fatal("dry runs must not be suppressed") + } +} diff --git a/server/internal/scheduler/scheduler_run_retention_worker.go b/server/internal/scheduler/scheduler_run_retention_worker.go new file mode 100644 index 0000000..d672e60 --- /dev/null +++ b/server/internal/scheduler/scheduler_run_retention_worker.go @@ -0,0 +1,248 @@ +package scheduler + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" +) + +const ( + defaultSchedulerRunRetentionDays = 15 + defaultSchedulerRunRetentionBatch = 5000 + defaultSchedulerRunRetentionBatches = 100 +) + +type SchedulerRunRetentionWorker struct { + pool *pgxpool.Pool + logger *zap.Logger +} + +type schedulerRunRetentionTarget struct { + Name string + DeleteSQL string + ProbeSQL string +} + +func NewSchedulerRunRetentionWorker(pool *pgxpool.Pool, logger *zap.Logger) *SchedulerRunRetentionWorker { + return &SchedulerRunRetentionWorker{pool: pool, logger: logger} +} + +func (w *SchedulerRunRetentionWorker) RunOnce(ctx context.Context, run JobRunContext) (map[string]any, error) { + if w == nil || w.pool == nil { + return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil + } + + startedAt := time.Now() + requestedRetentionDays := AsInt(run.Config, "retention_days", defaultSchedulerRunRetentionDays) + retentionDays := requestedRetentionDays + retentionDaysClamped := false + if requestedRetentionDays < defaultSchedulerRunRetentionDays { + retentionDays = defaultSchedulerRunRetentionDays + retentionDaysClamped = true + if w.logger != nil { + w.logger.Warn("scheduler run retention_days below hard floor", + zap.Int("requested_retention_days", requestedRetentionDays), + zap.Int("effective_retention_days", retentionDays), + ) + } + } + batchSize := AsInt(run.Config, "batch_size", defaultSchedulerRunRetentionBatch) + if run.Job != nil && run.Job.BatchSize != nil { + batchSize = *run.Job.BatchSize + } + if batchSize <= 0 { + batchSize = defaultSchedulerRunRetentionBatch + } + maxBatches := AsInt(run.Config, "max_batches_per_run", defaultSchedulerRunRetentionBatches) + if maxBatches <= 0 { + maxBatches = defaultSchedulerRunRetentionBatches + } + statementTimeout := AsDuration(run.Config, "statement_timeout", 5*time.Second) + cutoff := schedulerRunRetentionCutoff(time.Now(), retentionDays) + dryRun := run.DryRun + + stats := map[string]any{ + "retention_days": retentionDays, + "requested_retention_days": requestedRetentionDays, + "retention_days_clamped": retentionDaysClamped, + "cutoff_at": cutoff, + "batch_size": batchSize, + "max_batches_per_run": maxBatches, + "dry_run": dryRun, + } + + conn, err := w.pool.Acquire(ctx) + if err != nil { + return stats, fmt.Errorf("acquire ops connection: %w", err) + } + defer conn.Release() + + if statementTimeout > 0 { + if _, err := conn.Exec(ctx, `SELECT set_config('statement_timeout', $1, false)`, fmt.Sprintf("%dms", statementTimeout.Milliseconds())); err != nil { + return stats, fmt.Errorf("set statement timeout: %w", err) + } + defer conn.Exec(context.Background(), `RESET statement_timeout`) + } + + totalDeleted := int64(0) + totalCandidateProbe := int64(0) + perTable := map[string]any{} + for _, target := range schedulerRunRetentionTargets() { + tableStats := map[string]any{} + if dryRun { + candidateProbe, err := w.probeTarget(ctx, conn, target, cutoff, batchSize) + if err != nil { + return stats, fmt.Errorf("probe %s: %w", target.Name, err) + } + tableStats["candidate_probe_count"] = candidateProbe + tableStats["candidate_probe_limited"] = candidateProbe >= int64(batchSize) + totalCandidateProbe += candidateProbe + } else { + deleted, batches, err := w.deleteTarget(ctx, conn, target, cutoff, batchSize, maxBatches) + if err != nil { + return stats, fmt.Errorf("delete %s: %w", target.Name, err) + } + tableStats["deleted_count"] = deleted + tableStats["batch_count"] = batches + totalDeleted += deleted + } + perTable[target.Name] = tableStats + } + + if dryRun { + stats["candidate_probe_count"] = totalCandidateProbe + } + stats["deleted_count"] = totalDeleted + stats["tables"] = perTable + stats["duration_ms"] = time.Since(startedAt).Milliseconds() + return stats, nil +} + +func schedulerRunRetentionCutoff(now time.Time, retentionDays int) time.Time { + if retentionDays < defaultSchedulerRunRetentionDays { + retentionDays = defaultSchedulerRunRetentionDays + } + loc, err := time.LoadLocation("Asia/Shanghai") + if err != nil { + loc = time.FixedZone("UTC+8", 8*60*60) + } + localNow := now.In(loc) + today := time.Date(localNow.Year(), localNow.Month(), localNow.Day(), 0, 0, 0, 0, loc) + return today.AddDate(0, 0, -(retentionDays - 1)).UTC() +} + +func (w *SchedulerRunRetentionWorker) probeTarget(ctx context.Context, conn *pgxpool.Conn, target schedulerRunRetentionTarget, cutoff time.Time, limit int) (int64, error) { + var count int64 + if limit <= 0 { + limit = defaultSchedulerRunRetentionBatch + } + if err := conn.QueryRow(ctx, target.ProbeSQL, cutoff.UTC(), limit).Scan(&count); err != nil { + return 0, err + } + return count, nil +} + +func (w *SchedulerRunRetentionWorker) deleteTarget(ctx context.Context, conn *pgxpool.Conn, target schedulerRunRetentionTarget, cutoff time.Time, batchSize, maxBatches int) (int64, int, error) { + total := int64(0) + batches := 0 + for batches < maxBatches { + tag, err := conn.Exec(ctx, target.DeleteSQL, cutoff.UTC(), batchSize) + if err != nil { + return total, batches, err + } + affected := tag.RowsAffected() + if affected == 0 { + break + } + total += affected + batches++ + if affected < int64(batchSize) { + break + } + } + return total, batches, nil +} + +func schedulerRunRetentionTargets() []schedulerRunRetentionTarget { + return []schedulerRunRetentionTarget{ + { + Name: "scheduler_job_triggers", + ProbeSQL: schedulerRunRetentionTriggerProbeSQL(), + DeleteSQL: schedulerRunRetentionTriggerDeleteSQL(), + }, + { + Name: "scheduler_job_runs", + ProbeSQL: schedulerRunRetentionRunProbeSQL(), + DeleteSQL: schedulerRunRetentionRunDeleteSQL(), + }, + } +} + +func schedulerRunRetentionTriggerProbeSQL() string { + return schedulerRunRetentionProbeSQL( + "ops.scheduler_job_triggers", + "id", + "status IN ('success', 'failed', 'skipped') AND requested_at < $1", + "requested_at ASC, id ASC", + ) +} + +func schedulerRunRetentionTriggerDeleteSQL() string { + return schedulerRunRetentionDeleteSQL( + "ops.scheduler_job_triggers", + "id", + "status IN ('success', 'failed', 'skipped') AND requested_at < $1", + "requested_at ASC, id ASC", + ) +} + +func schedulerRunRetentionRunProbeSQL() string { + return schedulerRunRetentionProbeSQL( + "ops.scheduler_job_runs r", + "r.id", + "r.status IN ('success', 'failed', 'skipped') AND r.started_at < $1 AND NOT EXISTS (SELECT 1 FROM ops.scheduler_job_triggers t WHERE t.run_id = r.id)", + "r.started_at ASC, r.id ASC", + ) +} + +func schedulerRunRetentionRunDeleteSQL() string { + return schedulerRunRetentionDeleteSQL( + "ops.scheduler_job_runs r", + "r.id", + "r.status IN ('success', 'failed', 'skipped') AND r.started_at < $1 AND NOT EXISTS (SELECT 1 FROM ops.scheduler_job_triggers t WHERE t.run_id = r.id)", + "r.started_at ASC, r.id ASC", + ) +} + +func schedulerRunRetentionProbeSQL(table, idColumn, predicate, orderBy string) string { + return strings.Join([]string{ + "SELECT COUNT(*)::bigint", + "FROM (", + "SELECT " + idColumn, + "FROM " + table, + "WHERE " + predicate, + "ORDER BY " + orderBy, + "LIMIT $2", + ") candidates", + }, "\n") +} + +func schedulerRunRetentionDeleteSQL(table, idColumn, predicate, orderBy string) string { + return strings.Join([]string{ + "WITH doomed AS (", + "SELECT " + idColumn, + "FROM " + table, + "WHERE " + predicate, + "ORDER BY " + orderBy, + "LIMIT $2", + "FOR UPDATE SKIP LOCKED", + ")", + "DELETE FROM " + strings.Fields(table)[0] + " target", + "USING doomed", + "WHERE target.id = doomed.id", + }, "\n") +} diff --git a/server/internal/scheduler/scheduler_run_retention_worker_test.go b/server/internal/scheduler/scheduler_run_retention_worker_test.go new file mode 100644 index 0000000..09570cf --- /dev/null +++ b/server/internal/scheduler/scheduler_run_retention_worker_test.go @@ -0,0 +1,53 @@ +package scheduler + +import ( + "strings" + "testing" + "time" +) + +func TestSchedulerRunRetentionCutoffKeepsFifteenDays(t *testing.T) { + now := time.Date(2026, 5, 20, 22, 0, 0, 0, time.FixedZone("UTC+8", 8*60*60)) + + got := schedulerRunRetentionCutoff(now, defaultSchedulerRunRetentionDays) + want := time.Date(2026, 5, 6, 0, 0, 0, 0, time.FixedZone("UTC+8", 8*60*60)).UTC() + if !got.Equal(want) { + t.Fatalf("cutoff = %s, want %s", got.Format(time.RFC3339), want.Format(time.RFC3339)) + } +} + +func TestSchedulerRunRetentionCutoffHasHardFloor(t *testing.T) { + now := time.Date(2026, 5, 20, 22, 0, 0, 0, time.FixedZone("UTC+8", 8*60*60)) + + got := schedulerRunRetentionCutoff(now, 7) + want := schedulerRunRetentionCutoff(now, defaultSchedulerRunRetentionDays) + if !got.Equal(want) { + t.Fatalf("cutoff = %s, want hard floor cutoff %s", got.Format(time.RFC3339), want.Format(time.RFC3339)) + } +} + +func TestSchedulerRunRetentionSQLIsBoundedAndFKSafe(t *testing.T) { + triggerDeleteSQL := schedulerRunRetentionTriggerDeleteSQL() + for _, want := range []string{ + "status IN ('success', 'failed', 'skipped')", + "LIMIT $2", + "FOR UPDATE SKIP LOCKED", + "DELETE FROM ops.scheduler_job_triggers target", + } { + if !strings.Contains(triggerDeleteSQL, want) { + t.Fatalf("trigger retention SQL missing %q in SQL:\n%s", want, triggerDeleteSQL) + } + } + + runDeleteSQL := schedulerRunRetentionRunDeleteSQL() + for _, want := range []string{ + "NOT EXISTS (SELECT 1 FROM ops.scheduler_job_triggers t WHERE t.run_id = r.id)", + "LIMIT $2", + "FOR UPDATE SKIP LOCKED", + "DELETE FROM ops.scheduler_job_runs target", + } { + if !strings.Contains(runDeleteSQL, want) { + t.Fatalf("run retention SQL missing %q in SQL:\n%s", want, runDeleteSQL) + } + } +} diff --git a/server/migrations_ops/20260520151000_add_scheduler_run_retention_job.down.sql b/server/migrations_ops/20260520151000_add_scheduler_run_retention_job.down.sql new file mode 100644 index 0000000..e02db4a --- /dev/null +++ b/server/migrations_ops/20260520151000_add_scheduler_run_retention_job.down.sql @@ -0,0 +1,11 @@ +DELETE FROM ops.scheduler_job_triggers +WHERE job_key = 'scheduler_run_retention'; + +DELETE FROM ops.scheduler_job_runs +WHERE job_key = 'scheduler_run_retention'; + +DELETE FROM ops.scheduler_jobs +WHERE job_key = 'scheduler_run_retention'; + +DROP INDEX IF EXISTS ops.idx_scheduler_job_triggers_finished_requested; +DROP INDEX IF EXISTS ops.idx_scheduler_job_triggers_run_id; diff --git a/server/migrations_ops/20260520151000_add_scheduler_run_retention_job.up.sql b/server/migrations_ops/20260520151000_add_scheduler_run_retention_job.up.sql new file mode 100644 index 0000000..8623fa9 --- /dev/null +++ b/server/migrations_ops/20260520151000_add_scheduler_run_retention_job.up.sql @@ -0,0 +1,19 @@ +CREATE INDEX IF NOT EXISTS idx_scheduler_job_triggers_run_id + ON ops.scheduler_job_triggers (run_id) + WHERE run_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_scheduler_job_triggers_finished_requested + ON ops.scheduler_job_triggers (requested_at ASC, id ASC) + WHERE status IN ('success', 'failed', 'skipped'); + +INSERT INTO ops.scheduler_jobs + (job_key, display_name, category, description, enabled, schedule_type, interval_seconds, timezone, timeout_seconds, batch_size, config) +VALUES + ('scheduler_run_retention', '调度运行记录留存清理', 'ops', '限批删除 15 天以前已结束的调度触发与运行记录，控制高频 scheduler 审计表体积。', true, 'interval', 86400, 'Asia/Shanghai', 300, 5000, + '{"retention_days":15,"batch_size":5000,"max_batches_per_run":100,"dry_run":false,"statement_timeout":"5s","run_at_local":"03:20"}'::jsonb) +ON CONFLICT (job_key) DO UPDATE +SET display_name = EXCLUDED.display_name, + category = EXCLUDED.category, + description = EXCLUDED.description, + config = ops.scheduler_jobs.config || EXCLUDED.config, + updated_at = NOW(); diff --git a/server/migrations_ops/20260521100000_tune_scheduler_polling_noise.down.sql b/server/migrations_ops/20260521100000_tune_scheduler_polling_noise.down.sql new file mode 100644 index 0000000..c30c7d8 --- /dev/null +++ b/server/migrations_ops/20260521100000_tune_scheduler_polling_noise.down.sql @@ -0,0 +1,10 @@ +UPDATE ops.scheduler_jobs +SET interval_seconds = 60, + config = config - 'claim_ttl' - 'suppress_empty_success_runs' - 'empty_success_metric_keys', + updated_at = NOW() +WHERE job_key = 'monitoring_result_recovery'; + +UPDATE ops.scheduler_jobs +SET config = config - 'suppress_empty_success_runs' - 'empty_success_metric_keys', + updated_at = NOW() +WHERE job_key = 'schedule_dispatch'; diff --git a/server/migrations_ops/20260521100000_tune_scheduler_polling_noise.up.sql b/server/migrations_ops/20260521100000_tune_scheduler_polling_noise.up.sql new file mode 100644 index 0000000..171fbd4 --- /dev/null +++ b/server/migrations_ops/20260521100000_tune_scheduler_polling_noise.up.sql @@ -0,0 +1,15 @@ +UPDATE ops.scheduler_jobs +SET interval_seconds = 120, + config = config || '{"claim_ttl":"5m"}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_result_recovery'; + +UPDATE ops.scheduler_jobs +SET config = config || '{"suppress_empty_success_runs":true,"empty_success_metric_keys":["hydrated_count","claimed_count"]}'::jsonb, + updated_at = NOW() +WHERE job_key = 'schedule_dispatch'; + +UPDATE ops.scheduler_jobs +SET config = config || '{"suppress_empty_success_runs":true,"empty_success_metric_keys":["recoverable_count","republished_count"]}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_result_recovery';