From 5fb9d0b0dd762ffafd50a0c7a27a99aeb3099f21 Mon Sep 17 00:00:00 2001 From: liangxu Date: Wed, 20 May 2026 12:19:27 +0800 Subject: [PATCH] perf(scheduler): harden jobs for scalable execution --- server/cmd/scheduler/main.go | 73 +++++++++++--- server/cmd/scheduler/main_test.go | 45 +++++++++ server/internal/ops/repository/scheduler.go | 20 ++++ server/internal/scheduler/control_plane.go | 96 ++++++++++++++----- .../internal/scheduler/control_plane_test.go | 47 +++++++++ .../monitoring_lease_recovery_worker.go | 23 ++++- .../monitoring_received_inspection_worker.go | 24 ++++- .../monitoring_result_recovery_worker.go | 96 ++++++++++++++----- .../monitoring_result_recovery_worker_test.go | 75 +++++++++++++++ .../scheduler/monitoring_retention_worker.go | 86 ++++++++++------- .../monitoring_retention_worker_test.go | 36 +++++++ .../scheduler/schedule_dispatch_worker.go | 74 ++++++++++---- .../app/monitoring_daily_task_worker.go | 49 +++++++++- .../tenant/app/monitoring_lease_recovery.go | 46 ++++++--- .../app/monitoring_received_inspection.go | 2 +- .../article_generation_state_check_worker.go | 27 +++++- ...141000_add_scheduler_safe_indexes.down.sql | 6 ++ ...20141000_add_scheduler_safe_indexes.up.sql | 30 ++++++ ...140000_add_scheduler_safe_indexes.down.sql | 13 +++ ...20140000_add_scheduler_safe_indexes.up.sql | 87 +++++++++++++++++ ...000_create_scheduler_control_tables.up.sql | 19 ++-- ...000_harden_scheduler_jobs_for_k3s.down.sql | 1 + ...43000_harden_scheduler_jobs_for_k3s.up.sql | 59 ++++++++++++ 23 files changed, 887 insertions(+), 147 deletions(-) create mode 100644 server/internal/scheduler/control_plane_test.go create mode 100644 server/internal/scheduler/monitoring_result_recovery_worker_test.go create mode 100644 server/internal/scheduler/monitoring_retention_worker_test.go create mode 100644 server/migrations/20260520141000_add_scheduler_safe_indexes.down.sql create mode 100644 server/migrations/20260520141000_add_scheduler_safe_indexes.up.sql create mode 100644 server/migrations_monitoring/20260520140000_add_scheduler_safe_indexes.down.sql create mode 100644 server/migrations_monitoring/20260520140000_add_scheduler_safe_indexes.up.sql create mode 100644 server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.down.sql create mode 100644 server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.up.sql diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go index b62ac31..d5ab43d 100644 --- a/server/cmd/scheduler/main.go +++ b/server/cmd/scheduler/main.go @@ -99,38 +99,47 @@ func main() { controlPlane := internalscheduler.NewControlPlane(app.DB, app.Logger, []internalscheduler.ControlledJob{ { Key: "schedule_dispatch", - Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { - return scheduleDispatchWorker.RunOnce(runCtx) + Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { + return scheduleDispatchWorker.RunOnce(runCtx, jobCtx) }, }, { Key: "monitoring_daily_task", - Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { - return monitoringDailyTaskWorker.RunOnce(runCtx) + Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { + timeout := 45 * time.Second + if jobCtx.Job != nil && jobCtx.Job.TimeoutSeconds > 0 { + timeout = time.Duration(jobCtx.Job.TimeoutSeconds) * time.Second + } + if timeout <= 0 { + timeout = 45 * time.Second + } + taskCtx, cancel := context.WithTimeout(runCtx, timeout) + defer cancel() + return monitoringDailyTaskWorker.RunOnce(taskCtx, monitoringDailyRunOptions(jobCtx)) }, }, { Key: "monitoring_result_recovery", - Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { - return monitoringResultRecoveryWorker.RunOnce(runCtx) + Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { + return monitoringResultRecoveryWorker.RunOnce(runCtx, jobCtx) }, }, { Key: "monitoring_lease_recovery", - Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { - return monitoringLeaseRecoveryWorker.RunOnce(runCtx) + Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { + return monitoringLeaseRecoveryWorker.RunOnce(runCtx, jobCtx) }, }, { Key: "monitoring_received_inspection", - Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { - return monitoringReceivedInspectionWorker.RunOnce(runCtx) + Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { + return monitoringReceivedInspectionWorker.RunOnce(runCtx, jobCtx) }, }, { Key: "generation_state_check", - Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { - return generationStateCheckWorker.RunOnce(runCtx) + Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) { + return generationStateCheckWorker.RunOnce(runCtx, generationStateCheckRunOptions(jobCtx)) }, }, { @@ -260,6 +269,46 @@ func shutdownSchedulerMetricsServer(server *http.Server, logger interface { logger.Infof("scheduler metrics http server stopped") } +func monitoringDailyRunOptions(jobCtx internalscheduler.JobRunContext) tenantapp.MonitoringDailyTaskRunOptions { + batchSize := 0 + if jobCtx.Job != nil && jobCtx.Job.BatchSize != nil { + batchSize = *jobCtx.Job.BatchSize + } + if batchSize <= 0 { + batchSize = internalscheduler.AsInt(jobCtx.Config, "max_materialize_per_run", 0) + } + if batchSize <= 0 { + batchSize = internalscheduler.AsInt(jobCtx.Config, "batch_size", 0) + } + return tenantapp.MonitoringDailyTaskRunOptions{ + MaxMaterializePerRun: batchSize, + MaxMaterializePerPlan: internalscheduler.AsInt(jobCtx.Config, "max_materialize_per_plan", 0), + MaxMaterializePerBrand: internalscheduler.AsInt(jobCtx.Config, "max_materialize_per_brand", 0), + } +} + +func generationStateCheckRunOptions(jobCtx internalscheduler.JobRunContext) generateworker.GenerationTaskStateCheckRunOptions { + batchSize := 0 + if jobCtx.Job != nil && jobCtx.Job.BatchSize != nil { + batchSize = *jobCtx.Job.BatchSize + } + if batchSize <= 0 { + batchSize = internalscheduler.AsInt(jobCtx.Config, "batch_size", 0) + } + timeout := time.Duration(0) + if jobCtx.Job != nil && jobCtx.Job.TimeoutSeconds > 0 { + timeout = time.Duration(jobCtx.Job.TimeoutSeconds) * time.Second + } + if timeout <= 0 { + timeout = internalscheduler.AsDuration(jobCtx.Config, "timeout", 0) + } + return generateworker.GenerationTaskStateCheckRunOptions{ + Timeout: timeout, + BatchSize: batchSize, + Lookback: internalscheduler.AsDuration(jobCtx.Config, "lookback", 0), + } +} + func schedulerHTTPAddr(host string, port int) string { host = strings.TrimSpace(host) if host == "" { diff --git a/server/cmd/scheduler/main_test.go b/server/cmd/scheduler/main_test.go index 9133a9f..1b7974c 100644 --- a/server/cmd/scheduler/main_test.go +++ b/server/cmd/scheduler/main_test.go @@ -9,6 +9,9 @@ import ( "time" "github.com/gin-gonic/gin" + + opsdomain "github.com/geo-platform/tenant-api/internal/ops/domain" + internalscheduler "github.com/geo-platform/tenant-api/internal/scheduler" ) func TestSchedulerHTTPAddrDefaultsToLoopback(t *testing.T) { @@ -56,6 +59,48 @@ func TestSchedulerMetricsAuthMiddleware(t *testing.T) { } } +func TestMonitoringDailyRunOptionsUsesJobBatchSize(t *testing.T) { + batchSize := 88 + got := monitoringDailyRunOptions(internalscheduler.JobRunContext{ + Job: &opsdomain.SchedulerJob{BatchSize: &batchSize}, + Config: map[string]any{ + "max_materialize_per_run": 999, + "max_materialize_per_plan": 12, + "max_materialize_per_brand": 4, + }, + }) + + if got.MaxMaterializePerRun != 88 { + t.Fatalf("MaxMaterializePerRun = %d, want 88", got.MaxMaterializePerRun) + } + if got.MaxMaterializePerPlan != 12 || got.MaxMaterializePerBrand != 4 { + t.Fatalf("unexpected plan/brand options: %#v", got) + } +} + +func TestGenerationStateCheckRunOptionsUsesOpsConfig(t *testing.T) { + batchSize := 50 + got := generationStateCheckRunOptions(internalscheduler.JobRunContext{ + Job: &opsdomain.SchedulerJob{ + BatchSize: &batchSize, + TimeoutSeconds: 7, + }, + Config: map[string]any{ + "lookback": "2h", + }, + }) + + if got.BatchSize != 50 { + t.Fatalf("BatchSize = %d, want 50", got.BatchSize) + } + if got.Timeout != 7*time.Second { + t.Fatalf("Timeout = %s, want 7s", got.Timeout) + } + if got.Lookback != 2*time.Hour { + t.Fatalf("Lookback = %s, want 2h", got.Lookback) + } +} + func TestStartSchedulerWorkerWaitsForRunToReturn(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/server/internal/ops/repository/scheduler.go b/server/internal/ops/repository/scheduler.go index 00c12dc..26384cb 100644 --- a/server/internal/ops/repository/scheduler.go +++ b/server/internal/ops/repository/scheduler.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" @@ -188,6 +189,25 @@ func (r *SchedulerRepository) GetJobForRuntime(ctx context.Context, key string) WHERE j.job_key = $1`, key)) } +func (r *SchedulerRepository) LastAutoRunStartedAt(ctx context.Context, key string) (time.Time, bool, error) { + var startedAt time.Time + err := r.pool.QueryRow(ctx, ` + SELECT started_at + FROM ops.scheduler_job_runs + WHERE job_key = $1 + AND trigger_type = 'auto' + ORDER BY started_at DESC, id DESC + LIMIT 1 + `, key).Scan(&startedAt) + if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return time.Time{}, false, nil + } + return time.Time{}, false, err + } + return startedAt, true, nil +} + func (r *SchedulerRepository) UpdateJob(ctx context.Context, key string, in SchedulerJobUpdate) (*domain.SchedulerJob, error) { configJSON, err := marshalSchedulerJSONMap(in.Config) if err != nil { diff --git a/server/internal/scheduler/control_plane.go b/server/internal/scheduler/control_plane.go index 8406015..1b217a8 100644 --- a/server/internal/scheduler/control_plane.go +++ b/server/internal/scheduler/control_plane.go @@ -33,6 +33,13 @@ type JobRunContext struct { Config map[string]any } +const ( + schedulerManualPollInterval = 15 * time.Second + schedulerConfigRetryInterval = 30 * time.Second + schedulerFinishTimeout = 10 * time.Second + schedulerMaxManualTriggersDrain = 100 +) + type ControlPlane struct { pool *pgxpool.Pool repo *opsrepo.SchedulerRepository @@ -137,36 +144,35 @@ func (p *ControlPlane) heartbeat(ctx context.Context) { } func (p *ControlPlane) runJobLoop(ctx context.Context, job ControlledJob) { - nextAutoAt := p.nextAutoAt(context.Background(), job.Key, time.Now(), true) + nextAutoAt := p.nextAutoAt(ctx, job.Key, time.Now(), true) for { - p.drainManualTriggers(context.Background(), job) + if ctx.Err() != nil { + return + } + p.drainManualTriggers(ctx, job) now := time.Now() if !now.Before(nextAutoAt) { - p.runJobOnce(context.Background(), job, opsdomain.SchedulerTriggerAuto, nil, nil) - nextAutoAt = p.nextAutoAt(context.Background(), job.Key, time.Now(), false) - } - sleepFor := time.Until(nextAutoAt) - if sleepFor > 15*time.Second { - sleepFor = 15 * time.Second - } - if sleepFor <= 0 { - sleepFor = time.Second + p.runJobOnce(ctx, job, opsdomain.SchedulerTriggerAuto, nil, nil) + nextAutoAt = p.nextAutoAt(ctx, job.Key, time.Now(), false) + continue } + sleepFor := schedulerLoopSleep(time.Now(), nextAutoAt) timer := time.NewTimer(sleepFor) select { case <-ctx.Done(): timer.Stop() return case <-timer.C: - p.drainManualTriggers(context.Background(), job) - p.runJobOnce(context.Background(), job, opsdomain.SchedulerTriggerAuto, nil, nil) } } } func (p *ControlPlane) drainManualTriggers(parent context.Context, job ControlledJob) { - for { + for claimed := 0; claimed < schedulerMaxManualTriggersDrain; claimed++ { + if parent.Err() != nil { + return + } trigger, ok, err := p.repo.ClaimPendingTrigger(parent, job.Key, p.instanceID) if err != nil { if p.logger != nil { @@ -180,15 +186,21 @@ func (p *ControlPlane) drainManualTriggers(parent context.Context, job Controlle triggerID := trigger.ID p.runJobOnce(parent, job, trigger.TriggerType, &triggerID, trigger.ConfigOverride) } + if p.logger != nil { + p.logger.Warn("scheduler manual trigger drain reached batch limit", + zap.String("job_key", job.Key), + zap.Int("limit", schedulerMaxManualTriggersDrain), + ) + } } func (p *ControlPlane) nextAutoAt(ctx context.Context, key string, now time.Time, initial bool) time.Time { job, err := p.repo.GetJobForRuntime(ctx, key) if err != nil || job == nil { - return now.Add(30 * time.Second) + return now.Add(schedulerConfigRetryInterval) } if job.ScheduleType == "manual" { - return now.Add(15 * time.Second) + return now.Add(schedulerManualPollInterval) } if job.ScheduleType == "cron" && job.CronExpr != nil { loc, locErr := time.LoadLocation(job.Timezone) @@ -204,7 +216,7 @@ func (p *ControlPlane) nextAutoAt(ctx context.Context, key string, now time.Time zap.Error(parseErr), ) } - return now.Add(30 * time.Second) + return now.Add(schedulerConfigRetryInterval) } base := now.In(loc) return schedule.Next(base) @@ -226,12 +238,23 @@ func (p *ControlPlane) nextAutoAt(ctx context.Context, key string, now time.Time } interval := time.Duration(job.IntervalSeconds) * time.Second if interval <= 0 { - interval = 30 * time.Second + interval = schedulerConfigRetryInterval } + var lastStartedAt *time.Time if initial { - return now + 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 now.Add(interval) + return nextIntervalAutoAt(now, interval, initial, lastStartedAt) } func (p *ControlPlane) runJobOnce(parent context.Context, job ControlledJob, triggerType string, triggerID *int64, configOverride map[string]any) { @@ -313,7 +336,9 @@ func (p *ControlPlane) runJobOnce(parent context.Context, job ControlledJob, tri } stats["dry_run"] = dryRun - finished, finishErr := p.repo.FinishRun(parent, opsrepo.SchedulerRunFinish{ + finishCtx, finishCancel := context.WithTimeout(context.Background(), schedulerFinishTimeout) + defer finishCancel() + finished, finishErr := p.repo.FinishRun(finishCtx, opsrepo.SchedulerRunFinish{ RunID: run.ID, Status: status, Stats: stats, @@ -328,7 +353,7 @@ func (p *ControlPlane) runJobOnce(parent context.Context, job ControlledJob, tri triggerStatus = opsdomain.SchedulerRunFailed } runID := run.ID - _ = p.repo.FinishTrigger(parent, *triggerID, &runID, triggerStatus, errMsg) + _ = p.repo.FinishTrigger(finishCtx, *triggerID, &runID, triggerStatus, errMsg) } if runErr != nil { if p.logger != nil { @@ -382,6 +407,33 @@ func advisoryLockKey(value string) int64 { return int64(hash.Sum64() & 0x7fffffffffffffff) } +func schedulerLoopSleep(now, nextAutoAt time.Time) time.Duration { + sleepFor := nextAutoAt.Sub(now) + if sleepFor > schedulerManualPollInterval { + sleepFor = schedulerManualPollInterval + } + if sleepFor <= 0 { + sleepFor = time.Second + } + return sleepFor +} + +func nextIntervalAutoAt(now time.Time, interval time.Duration, initial bool, lastStartedAt *time.Time) time.Time { + if interval <= 0 { + interval = schedulerConfigRetryInterval + } + if initial { + if lastStartedAt != nil && !lastStartedAt.IsZero() { + dueAt := lastStartedAt.Add(interval) + if dueAt.After(now) { + return dueAt + } + } + return now + } + return now.Add(interval) +} + func schedulerConfigSnapshot(job *opsdomain.SchedulerJob, config map[string]any) map[string]any { out := map[string]any{ "enabled": job.Enabled, diff --git a/server/internal/scheduler/control_plane_test.go b/server/internal/scheduler/control_plane_test.go new file mode 100644 index 0000000..dc284e8 --- /dev/null +++ b/server/internal/scheduler/control_plane_test.go @@ -0,0 +1,47 @@ +package scheduler + +import ( + "testing" + "time" +) + +func TestSchedulerLoopSleepCapsManualPollWithoutForcingAutoRun(t *testing.T) { + now := time.Date(2026, 5, 20, 10, 0, 0, 0, time.UTC) + nextAutoAt := now.Add(time.Minute) + + got := schedulerLoopSleep(now, nextAutoAt) + if got != schedulerManualPollInterval { + t.Fatalf("schedulerLoopSleep() = %s, want %s", got, schedulerManualPollInterval) + } +} + +func TestNextIntervalAutoAtUsesLastRunOnInitialStart(t *testing.T) { + now := time.Date(2026, 5, 20, 10, 0, 0, 0, time.UTC) + last := now.Add(-time.Minute) + + got := nextIntervalAutoAt(now, 5*time.Minute, true, &last) + want := last.Add(5 * time.Minute) + if !got.Equal(want) { + t.Fatalf("nextIntervalAutoAt() = %s, want %s", got, want) + } +} + +func TestNextIntervalAutoAtRunsImmediatelyWhenOverdue(t *testing.T) { + now := time.Date(2026, 5, 20, 10, 0, 0, 0, time.UTC) + last := now.Add(-10 * time.Minute) + + got := nextIntervalAutoAt(now, 5*time.Minute, true, &last) + if !got.Equal(now) { + t.Fatalf("nextIntervalAutoAt() = %s, want %s", got, now) + } +} + +func TestNextIntervalAutoAtSchedulesAfterCurrentCompletion(t *testing.T) { + now := time.Date(2026, 5, 20, 10, 0, 0, 0, time.UTC) + + got := nextIntervalAutoAt(now, 5*time.Minute, false, nil) + want := now.Add(5 * time.Minute) + if !got.Equal(want) { + t.Fatalf("nextIntervalAutoAt() = %s, want %s", got, want) + } +} diff --git a/server/internal/scheduler/monitoring_lease_recovery_worker.go b/server/internal/scheduler/monitoring_lease_recovery_worker.go index eb9636e..1103b13 100644 --- a/server/internal/scheduler/monitoring_lease_recovery_worker.go +++ b/server/internal/scheduler/monitoring_lease_recovery_worker.go @@ -13,6 +13,7 @@ import ( const ( defaultMonitoringLeaseRecoveryInterval = 30 * time.Minute defaultMonitoringLeaseRecoveryTimeout = 30 * time.Second + defaultMonitoringLeaseRecoveryLimit = 1000 ) type MonitoringLeaseRecoveryWorker struct { @@ -20,6 +21,7 @@ type MonitoringLeaseRecoveryWorker struct { logger *zap.Logger interval time.Duration timeout time.Duration + limit int } func NewMonitoringLeaseRecoveryWorker(monitoringPool *pgxpool.Pool, logger *zap.Logger) *MonitoringLeaseRecoveryWorker { @@ -28,6 +30,7 @@ func NewMonitoringLeaseRecoveryWorker(monitoringPool *pgxpool.Pool, logger *zap. logger: logger, interval: defaultMonitoringLeaseRecoveryInterval, timeout: defaultMonitoringLeaseRecoveryTimeout, + limit: defaultMonitoringLeaseRecoveryLimit, } } @@ -71,7 +74,7 @@ func (w *MonitoringLeaseRecoveryWorker) runOnce(parent context.Context) { } defer tx.Rollback(ctx) - expiredCount, err := tenantapp.ExpireMonitoringLeasedTasks(ctx, tx, nil, time.Now().UTC()) + expiredCount, err := tenantapp.ExpireMonitoringLeasedTasks(ctx, tx, nil, time.Now().UTC(), w.limit) if err != nil { if w.logger != nil { w.logger.Warn("monitoring lease recovery failed", tenantapp.MonitoringWorkerErrorFields(err)...) @@ -91,12 +94,23 @@ func (w *MonitoringLeaseRecoveryWorker) runOnce(parent context.Context) { } } -func (w *MonitoringLeaseRecoveryWorker) RunOnce(parent context.Context) (map[string]any, error) { +func (w *MonitoringLeaseRecoveryWorker) RunOnce(parent context.Context, run JobRunContext) (map[string]any, error) { if w == nil || w.monitoringPool == nil { return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil } startedAt := time.Now() - ctx, cancel := context.WithTimeout(parent, w.timeout) + timeout := w.timeout + if run.Job != nil && run.Job.TimeoutSeconds > 0 { + timeout = time.Duration(run.Job.TimeoutSeconds) * time.Second + } + limit := w.limit + if run.Job != nil && run.Job.BatchSize != nil && *run.Job.BatchSize > 0 { + limit = *run.Job.BatchSize + } + if value := AsInt(run.Config, "batch_size", 0); value > 0 && (run.Job == nil || run.Job.BatchSize == nil) { + limit = value + } + ctx, cancel := context.WithTimeout(parent, timeout) defer cancel() tx, err := w.monitoringPool.Begin(ctx) @@ -105,7 +119,7 @@ func (w *MonitoringLeaseRecoveryWorker) RunOnce(parent context.Context) (map[str } defer tx.Rollback(ctx) - expiredCount, err := tenantapp.ExpireMonitoringLeasedTasks(ctx, tx, nil, time.Now().UTC()) + expiredCount, err := tenantapp.ExpireMonitoringLeasedTasks(ctx, tx, nil, time.Now().UTC(), limit) if err != nil { return map[string]any{"stage": "expire"}, err } @@ -115,5 +129,6 @@ func (w *MonitoringLeaseRecoveryWorker) RunOnce(parent context.Context) (map[str return map[string]any{ "expired_task_count": expiredCount, "duration_ms": time.Since(startedAt).Milliseconds(), + "batch_size": limit, }, nil } diff --git a/server/internal/scheduler/monitoring_received_inspection_worker.go b/server/internal/scheduler/monitoring_received_inspection_worker.go index 4121473..7f54ecb 100644 --- a/server/internal/scheduler/monitoring_received_inspection_worker.go +++ b/server/internal/scheduler/monitoring_received_inspection_worker.go @@ -117,12 +117,27 @@ func (w *MonitoringReceivedInspectionWorker) runOnce(parent context.Context) { } } -func (w *MonitoringReceivedInspectionWorker) RunOnce(parent context.Context) (map[string]any, error) { +func (w *MonitoringReceivedInspectionWorker) RunOnce(parent context.Context, run JobRunContext) (map[string]any, error) { if w == nil || w.monitoringPool == nil { return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil } startedAt := time.Now() - ctx, cancel := context.WithTimeout(parent, w.timeout) + timeout := w.timeout + if run.Job != nil && run.Job.TimeoutSeconds > 0 { + timeout = time.Duration(run.Job.TimeoutSeconds) * time.Second + } + limit := w.limit + if run.Job != nil && run.Job.BatchSize != nil && *run.Job.BatchSize > 0 { + limit = *run.Job.BatchSize + } + if value := AsInt(run.Config, "batch_size", 0); value > 0 && (run.Job == nil || run.Job.BatchSize == nil) { + limit = value + } + threshold := w.threshold + if value := AsDuration(run.Config, "threshold", 0); value > 0 { + threshold = value + } + ctx, cancel := context.WithTimeout(parent, timeout) defer cancel() tx, err := w.monitoringPool.Begin(ctx) @@ -132,7 +147,7 @@ func (w *MonitoringReceivedInspectionWorker) RunOnce(parent context.Context) (ma defer tx.Rollback(ctx) now := time.Now().UTC() - items, err := tenantapp.MarkMonitoringStaleReceivedTasks(ctx, tx, now, w.threshold, w.limit) + items, err := tenantapp.MarkMonitoringStaleReceivedTasks(ctx, tx, now, threshold, limit) if err != nil { return map[string]any{"stage": "inspect"}, err } @@ -141,7 +156,8 @@ func (w *MonitoringReceivedInspectionWorker) RunOnce(parent context.Context) (ma } return map[string]any{ "overdue_task_count": len(items), - "threshold_seconds": int64(w.threshold.Seconds()), + "threshold_seconds": int64(threshold.Seconds()), "duration_ms": time.Since(startedAt).Milliseconds(), + "batch_size": limit, }, nil } diff --git a/server/internal/scheduler/monitoring_result_recovery_worker.go b/server/internal/scheduler/monitoring_result_recovery_worker.go index 4e0b42b..82a7328 100644 --- a/server/internal/scheduler/monitoring_result_recovery_worker.go +++ b/server/internal/scheduler/monitoring_result_recovery_worker.go @@ -2,6 +2,7 @@ package scheduler import ( "context" + "fmt" "time" "github.com/jackc/pgx/v5" @@ -15,6 +16,7 @@ const ( defaultMonitoringResultRecoveryInterval = 1 * time.Minute defaultMonitoringResultRecoveryTimeout = 30 * time.Second defaultMonitoringResultRecoveryLimit = 100 + defaultMonitoringResultRecoveryClaimTTL = 5 * time.Minute ) type MonitoringResultRecoveryWorker struct { @@ -87,7 +89,7 @@ func (w *MonitoringResultRecoveryWorker) runOnce(parent context.Context) { } defer tx.Rollback(ctx) - items, err := loadMonitoringRecoverableTaskResults(ctx, tx, w.limit) + items, err := loadMonitoringRecoverableTaskResults(ctx, tx, w.limit, defaultMonitoringResultRecoveryClaimTTL) if err != nil { if w.logger != nil { w.logger.Warn("monitoring result recovery failed", tenantapp.MonitoringWorkerErrorFields(err)...) @@ -118,12 +120,23 @@ func (w *MonitoringResultRecoveryWorker) runOnce(parent context.Context) { } } -func (w *MonitoringResultRecoveryWorker) RunOnce(parent context.Context) (map[string]any, error) { +func (w *MonitoringResultRecoveryWorker) RunOnce(parent context.Context, run JobRunContext) (map[string]any, error) { if w == nil || w.monitoringPool == nil || w.service == nil { return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil } startedAt := time.Now() - ctx, cancel := context.WithTimeout(parent, w.timeout) + timeout := w.timeout + if run.Job != nil && run.Job.TimeoutSeconds > 0 { + timeout = time.Duration(run.Job.TimeoutSeconds) * time.Second + } + limit := w.limit + if run.Job != nil && run.Job.BatchSize != nil && *run.Job.BatchSize > 0 { + limit = *run.Job.BatchSize + } + if value := AsInt(run.Config, "batch_size", 0); value > 0 && (run.Job == nil || run.Job.BatchSize == nil) { + limit = value + } + ctx, cancel := context.WithTimeout(parent, timeout) defer cancel() tx, err := w.monitoringPool.Begin(ctx) @@ -132,7 +145,8 @@ func (w *MonitoringResultRecoveryWorker) RunOnce(parent context.Context) (map[st } defer tx.Rollback(ctx) - items, err := loadMonitoringRecoverableTaskResults(ctx, tx, w.limit) + claimTTL := AsDuration(run.Config, "claim_ttl", defaultMonitoringResultRecoveryClaimTTL) + items, err := loadMonitoringRecoverableTaskResults(ctx, tx, limit, claimTTL) if err != nil { return map[string]any{"stage": "load"}, err } @@ -150,6 +164,8 @@ func (w *MonitoringResultRecoveryWorker) RunOnce(parent context.Context) (map[st "recoverable_count": len(items), "republished_count": republishedCount, "duration_ms": time.Since(startedAt).Milliseconds(), + "batch_size": limit, + "claim_ttl_seconds": int64(claimTTL.Seconds()), }, nil } @@ -194,34 +210,70 @@ func (w *MonitoringResultRecoveryWorker) tryRepublish(parent context.Context, it return true } -func loadMonitoringRecoverableTaskResults(ctx context.Context, tx pgx.Tx, limit int) ([]monitoringRecoverableTaskResult, error) { +func loadMonitoringRecoverableTaskResults(ctx context.Context, tx pgx.Tx, limit int, claimTTL time.Duration) ([]monitoringRecoverableTaskResult, error) { if limit <= 0 { limit = defaultMonitoringResultRecoveryLimit } + if claimTTL <= 0 { + claimTTL = defaultMonitoringResultRecoveryClaimTTL + } rows, err := tx.Query(ctx, ` + WITH candidates AS ( + SELECT + id, + callback_received_at, + request_payload_json + FROM monitoring_collect_tasks + WHERE collector_type = $1 + AND ( + status = 'received' + OR ( + COALESCE(execution_owner, 'legacy') = 'desktop_tasks' + AND status = 'pending' + AND callback_received_at IS NOT NULL + ) + ) + AND callback_received_at IS NOT NULL + AND request_payload_json IS NOT NULL + AND jsonb_typeof(request_payload_json) = 'object' + AND ( + COALESCE(NULLIF(request_payload_json ->> 'queue_status', ''), 'pending') IN ('pending', 'publish_failed') + OR ( + COALESCE(NULLIF(request_payload_json ->> 'queue_status', ''), 'pending') = 'claiming' + AND COALESCE(NULLIF(request_payload_json ->> 'queue_claimed_at', ''), '') <> '' + AND NULLIF(request_payload_json ->> 'queue_claimed_at', '')::timestamptz < NOW() - $3::interval + ) + ) + ORDER BY callback_received_at ASC, id ASC + FOR UPDATE SKIP LOCKED + LIMIT $2 + ), + claimed AS ( + UPDATE monitoring_collect_tasks t + SET request_payload_json = ( + CASE + WHEN t.request_payload_json IS NULL OR jsonb_typeof(t.request_payload_json) <> 'object' + THEN '{}'::jsonb + ELSE t.request_payload_json + END + ) || jsonb_build_object( + 'queue_status', 'claiming', + 'queue_claimed_at', NOW()::text + ), + updated_at = NOW() + FROM candidates c + WHERE t.id = c.id + AND t.callback_received_at = c.callback_received_at + RETURNING c.id, c.callback_received_at, c.request_payload_json + ) SELECT id, callback_received_at, request_payload_json - FROM monitoring_collect_tasks - WHERE collector_type = $1 - AND ( - status = 'received' - OR ( - COALESCE(execution_owner, 'legacy') = 'desktop_tasks' - AND status = 'pending' - AND callback_received_at IS NOT NULL - ) - ) - AND callback_received_at IS NOT NULL - AND request_payload_json IS NOT NULL - AND jsonb_typeof(request_payload_json) = 'object' - AND COALESCE(NULLIF(request_payload_json ->> 'queue_status', ''), 'pending') IN ('pending', 'publish_failed') + FROM claimed ORDER BY callback_received_at ASC, id ASC - FOR UPDATE SKIP LOCKED - LIMIT $2 - `, "plugin", limit) + `, "desktop", limit, fmt.Sprintf("%d milliseconds", claimTTL.Milliseconds())) if err != nil { return nil, tenantapp.InternalMonitoringError(50041, "result_recovery_query_failed", "failed to load recoverable monitoring results", err) } diff --git a/server/internal/scheduler/monitoring_result_recovery_worker_test.go b/server/internal/scheduler/monitoring_result_recovery_worker_test.go new file mode 100644 index 0000000..4bdf71b --- /dev/null +++ b/server/internal/scheduler/monitoring_result_recovery_worker_test.go @@ -0,0 +1,75 @@ +package scheduler + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +type captureRecoveryTx struct { + sql string + args []any + rows pgx.Rows +} + +func (tx *captureRecoveryTx) Begin(context.Context) (pgx.Tx, error) { return nil, nil } +func (tx *captureRecoveryTx) Commit(context.Context) error { return nil } +func (tx *captureRecoveryTx) Rollback(context.Context) error { return nil } +func (tx *captureRecoveryTx) CopyFrom(context.Context, pgx.Identifier, []string, pgx.CopyFromSource) (int64, error) { + return 0, nil +} +func (tx *captureRecoveryTx) SendBatch(context.Context, *pgx.Batch) pgx.BatchResults { return nil } +func (tx *captureRecoveryTx) LargeObjects() pgx.LargeObjects { return pgx.LargeObjects{} } +func (tx *captureRecoveryTx) Prepare(context.Context, string, string) (*pgconn.StatementDescription, error) { + return nil, nil +} +func (tx *captureRecoveryTx) Exec(context.Context, string, ...any) (pgconn.CommandTag, error) { + return pgconn.CommandTag{}, nil +} +func (tx *captureRecoveryTx) Query(_ context.Context, sql string, args ...any) (pgx.Rows, error) { + tx.sql = sql + tx.args = args + return tx.rows, nil +} +func (tx *captureRecoveryTx) QueryRow(context.Context, string, ...any) pgx.Row { return nil } +func (tx *captureRecoveryTx) Conn() *pgx.Conn { return nil } + +type emptyRecoveryRows struct{} + +func (r emptyRecoveryRows) Close() {} +func (r emptyRecoveryRows) Err() error { return nil } +func (r emptyRecoveryRows) CommandTag() pgconn.CommandTag { return pgconn.CommandTag{} } +func (r emptyRecoveryRows) FieldDescriptions() []pgconn.FieldDescription { return nil } +func (r emptyRecoveryRows) Next() bool { return false } +func (r emptyRecoveryRows) Scan(...any) error { return nil } +func (r emptyRecoveryRows) Values() ([]any, error) { return nil, nil } +func (r emptyRecoveryRows) RawValues() [][]byte { return nil } +func (r emptyRecoveryRows) Conn() *pgx.Conn { return nil } + +func TestLoadMonitoringRecoverableTaskResultsClaimsRows(t *testing.T) { + tx := &captureRecoveryTx{rows: emptyRecoveryRows{}} + + _, err := loadMonitoringRecoverableTaskResults(context.Background(), tx, 42, 7*time.Minute) + if err != nil { + t.Fatalf("loadMonitoringRecoverableTaskResults() error = %v", err) + } + if !strings.Contains(tx.sql, "queue_status', 'claiming'") { + t.Fatalf("query should claim rows before publish, got SQL:\n%s", tx.sql) + } + if !strings.Contains(tx.sql, "FOR UPDATE SKIP LOCKED") { + t.Fatalf("query should use SKIP LOCKED, got SQL:\n%s", tx.sql) + } + if got := tx.args[0]; got != "desktop" { + t.Fatalf("collector arg = %v, want desktop", got) + } + if got := tx.args[1]; got != 42 { + t.Fatalf("limit arg = %v, want 42", got) + } + if got := tx.args[2]; got != "420000 milliseconds" { + t.Fatalf("claim ttl arg = %v, want 420000 milliseconds", got) + } +} diff --git a/server/internal/scheduler/monitoring_retention_worker.go b/server/internal/scheduler/monitoring_retention_worker.go index 650a3a7..8cd259c 100644 --- a/server/internal/scheduler/monitoring_retention_worker.go +++ b/server/internal/scheduler/monitoring_retention_worker.go @@ -24,7 +24,7 @@ type MonitoringRetentionWorker struct { type retentionTarget struct { Name string DeleteSQL string - CountSQL string + ProbeSQL string } func NewMonitoringRetentionWorker(monitoringPool *pgxpool.Pool, logger *zap.Logger) *MonitoringRetentionWorker { @@ -77,18 +77,19 @@ func (w *MonitoringRetentionWorker) RunOnce(ctx context.Context, run JobRunConte } totalDeleted := int64(0) - totalCandidates := int64(0) + totalCandidateProbe := int64(0) perTable := map[string]any{} for _, target := range retentionTargets() { - candidateCount, err := w.countTarget(ctx, conn, target, cutoffDate) - if err != nil { - return stats, fmt.Errorf("count %s: %w", target.Name, err) - } - totalCandidates += candidateCount - tableStats := map[string]any{ - "candidate_count": candidateCount, - } - if !dryRun && candidateCount > 0 { + tableStats := map[string]any{} + if dryRun { + candidateProbe, err := w.probeTarget(ctx, conn, target, cutoffDate, 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, cutoffDate, batchSize, maxBatches) if err != nil { return stats, fmt.Errorf("delete %s: %w", target.Name, err) @@ -100,7 +101,9 @@ func (w *MonitoringRetentionWorker) RunOnce(ctx context.Context, run JobRunConte perTable[target.Name] = tableStats } - stats["candidate_count"] = totalCandidates + if dryRun { + stats["candidate_probe_count"] = totalCandidateProbe + } stats["deleted_count"] = totalDeleted stats["tables"] = perTable stats["duration_ms"] = time.Since(startedAt).Milliseconds() @@ -118,9 +121,12 @@ func monitoringRetentionCutoffDate(now time.Time, retentionDays int) string { return cutoff.Format("2006-01-02") } -func (w *MonitoringRetentionWorker) countTarget(ctx context.Context, conn *pgxpool.Conn, target retentionTarget, cutoffDate string) (int64, error) { +func (w *MonitoringRetentionWorker) probeTarget(ctx context.Context, conn *pgxpool.Conn, target retentionTarget, cutoffDate string, limit int) (int64, error) { var count int64 - if err := conn.QueryRow(ctx, target.CountSQL, cutoffDate).Scan(&count); err != nil { + if limit <= 0 { + limit = defaultMonitoringRetentionBatch + } + if err := conn.QueryRow(ctx, target.ProbeSQL, cutoffDate, limit).Scan(&count); err != nil { return 0, err } return count, nil @@ -151,53 +157,69 @@ func retentionTargets() []retentionTarget { return []retentionTarget{ { Name: "monitoring_collect_dispatch_outbox", - CountSQL: `SELECT COUNT(*) FROM monitoring_collect_dispatch_outbox WHERE created_at < $1::date`, - DeleteSQL: batchDeleteSQL("monitoring_collect_dispatch_outbox", "id", "created_at < $1::date"), + ProbeSQL: batchProbeSQL("monitoring_collect_dispatch_outbox", "id", "created_at < $1::date", "created_at ASC, id ASC"), + DeleteSQL: batchDeleteSQL("monitoring_collect_dispatch_outbox", "id", "created_at < $1::date", "created_at ASC, id ASC"), }, { Name: "monitoring_collect_requests", - CountSQL: `SELECT COUNT(*) FROM monitoring_collect_requests WHERE created_at < $1::date`, - DeleteSQL: batchDeleteSQL("monitoring_collect_requests", "id", "created_at < $1::date"), + ProbeSQL: batchProbeSQL("monitoring_collect_requests", "id", "created_at < $1::date", "created_at ASC, id ASC"), + DeleteSQL: batchDeleteSQL("monitoring_collect_requests", "id", "created_at < $1::date", "created_at ASC, id ASC"), }, { Name: "monitoring_brand_platform_daily", - CountSQL: `SELECT COUNT(*) FROM monitoring_brand_platform_daily WHERE business_date < $1::date`, - DeleteSQL: batchDeleteSQL("monitoring_brand_platform_daily", "id", "business_date < $1::date"), + ProbeSQL: batchProbeSQL("monitoring_brand_platform_daily", "id", "business_date < $1::date", "business_date ASC, id ASC"), + DeleteSQL: batchDeleteSQL("monitoring_brand_platform_daily", "id", "business_date < $1::date", "business_date ASC, id ASC"), }, { Name: "monitoring_brand_daily", - CountSQL: `SELECT COUNT(*) FROM monitoring_brand_daily WHERE business_date < $1::date`, - DeleteSQL: batchDeleteSQL("monitoring_brand_daily", "id", "business_date < $1::date"), + ProbeSQL: batchProbeSQL("monitoring_brand_daily", "id", "business_date < $1::date", "business_date ASC, id ASC"), + DeleteSQL: batchDeleteSQL("monitoring_brand_daily", "id", "business_date < $1::date", "business_date ASC, id ASC"), }, { Name: "monitoring_platform_access_snapshots", - CountSQL: `SELECT COUNT(*) FROM monitoring_platform_access_snapshots WHERE business_date < $1::date`, - DeleteSQL: batchDeleteSQL("monitoring_platform_access_snapshots", "id", "business_date < $1::date"), + ProbeSQL: batchProbeSQL("monitoring_platform_access_snapshots", "id", "business_date < $1::date", "business_date ASC, id ASC"), + DeleteSQL: batchDeleteSQL("monitoring_platform_access_snapshots", "id", "business_date < $1::date", "business_date ASC, id ASC"), }, { Name: "monitoring_collect_tasks", - CountSQL: `SELECT COUNT(*) FROM monitoring_collect_tasks WHERE business_date < $1::date`, - DeleteSQL: batchDeleteSQL("monitoring_collect_tasks", "id", "business_date < $1::date"), + ProbeSQL: batchProbeSQL("monitoring_collect_tasks", "id", "business_date < $1::date", "business_date ASC, id ASC"), + DeleteSQL: batchDeleteSQL("monitoring_collect_tasks", "id", "business_date < $1::date", "business_date ASC, id ASC"), }, { Name: "question_monitor_runs", - CountSQL: `SELECT COUNT(*) FROM question_monitor_runs WHERE business_date < $1::date`, - DeleteSQL: batchDeleteSQL("question_monitor_runs", "id", "business_date < $1::date"), + ProbeSQL: batchProbeSQL("question_monitor_runs", "id", "business_date < $1::date", "business_date ASC, id ASC"), + DeleteSQL: batchDeleteSQL("question_monitor_runs", "id", "business_date < $1::date", "business_date ASC, id ASC"), }, } } -func batchDeleteSQL(table, idColumn, predicate string) string { +func batchProbeSQL(table, idColumn, predicate, orderBy string) string { + parts := []string{ + "SELECT COUNT(*)::bigint", + "FROM (", + "SELECT " + idColumn, + "FROM " + table, + "WHERE " + predicate, + "ORDER BY " + orderBy, + "LIMIT $2", + ") candidates", + } + return strings.Join(parts, "\n") +} + +func batchDeleteSQL(table, idColumn, predicate, orderBy string) string { parts := []string{ "WITH doomed AS (", "SELECT " + idColumn, "FROM " + table, "WHERE " + predicate, - "ORDER BY " + idColumn + " ASC", + "ORDER BY " + orderBy, "LIMIT $2", + "FOR UPDATE SKIP LOCKED", ")", - "DELETE FROM " + table, - "WHERE " + idColumn + " IN (SELECT " + idColumn + " FROM doomed)", + "DELETE FROM " + table + " t", + "USING doomed", + "WHERE t." + idColumn + " = doomed." + idColumn, } return strings.Join(parts, "\n") } diff --git a/server/internal/scheduler/monitoring_retention_worker_test.go b/server/internal/scheduler/monitoring_retention_worker_test.go new file mode 100644 index 0000000..4891d4b --- /dev/null +++ b/server/internal/scheduler/monitoring_retention_worker_test.go @@ -0,0 +1,36 @@ +package scheduler + +import ( + "strings" + "testing" +) + +func TestBatchDeleteSQLUsesBoundedSkipLockedDelete(t *testing.T) { + sql := batchDeleteSQL("monitoring_collect_tasks", "id", "business_date < $1::date", "business_date ASC, id ASC") + + for _, want := range []string{ + "LIMIT $2", + "FOR UPDATE SKIP LOCKED", + "DELETE FROM monitoring_collect_tasks t", + "USING doomed", + "WHERE t.id = doomed.id", + } { + if !strings.Contains(sql, want) { + t.Fatalf("batchDeleteSQL missing %q in SQL:\n%s", want, sql) + } + } + if strings.Contains(sql, "COUNT(*) FROM monitoring_collect_tasks") { + t.Fatalf("batchDeleteSQL must not count full candidate set:\n%s", sql) + } +} + +func TestBatchProbeSQLCapsDryRunProbe(t *testing.T) { + sql := batchProbeSQL("monitoring_collect_tasks", "id", "business_date < $1::date", "business_date ASC, id ASC") + + if !strings.Contains(sql, "LIMIT $2") { + t.Fatalf("batchProbeSQL should cap dry-run probe:\n%s", sql) + } + if !strings.Contains(sql, "SELECT COUNT(*)::bigint") { + t.Fatalf("batchProbeSQL should count only the limited candidate subquery:\n%s", sql) + } +} diff --git a/server/internal/scheduler/schedule_dispatch_worker.go b/server/internal/scheduler/schedule_dispatch_worker.go index a790f1a..4597889 100644 --- a/server/internal/scheduler/schedule_dispatch_worker.go +++ b/server/internal/scheduler/schedule_dispatch_worker.go @@ -130,8 +130,9 @@ func (w *ScheduleDispatchWorker) run(ctx context.Context) { } func (w *ScheduleDispatchWorker) runOnce(parent context.Context) { - ctx, cancel := w.stageContext(parent) - count, err := w.hydrateMissingNextRuns(ctx) + runtimeCfg := w.runtimeConfig() + ctx, cancel := w.stageContext(parent, runtimeCfg) + count, err := w.hydrateMissingNextRuns(ctx, runtimeCfg) cancel() if err != nil { if w.logger != nil { @@ -141,8 +142,8 @@ func (w *ScheduleDispatchWorker) runOnce(parent context.Context) { w.logger.Info("schedule dispatch hydrated next_run_at", zap.Int("task_count", count)) } - ctx, cancel = w.stageContext(parent) - paused, stats, err := w.shouldPauseDispatch(ctx) + ctx, cancel = w.stageContext(parent, runtimeCfg) + paused, stats, err := w.shouldPauseDispatch(ctx, runtimeCfg) cancel() if err != nil { if w.logger != nil { @@ -160,8 +161,8 @@ func (w *ScheduleDispatchWorker) runOnce(parent context.Context) { return } - ctx, cancel = w.stageContext(parent) - tasks, err := w.claimDueSchedules(ctx) + ctx, cancel = w.stageContext(parent, runtimeCfg) + tasks, err := w.claimDueSchedules(ctx, runtimeCfg) cancel() if err != nil { if w.logger != nil { @@ -173,14 +174,16 @@ func (w *ScheduleDispatchWorker) runOnce(parent context.Context) { w.dispatchBatch(parent, tasks) } -func (w *ScheduleDispatchWorker) RunOnce(ctx context.Context) (map[string]any, error) { +func (w *ScheduleDispatchWorker) RunOnce(ctx context.Context, run JobRunContext) (map[string]any, error) { if w == nil || w.pool == nil || w.promptRuleService == nil { return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil } startedAt := time.Now() + runtimeCfg := w.runtimeConfig() + runtimeCfg.ApplyJobRun(run) - stageCtx, cancel := w.stageContext(ctx) - hydratedCount, err := w.hydrateMissingNextRuns(stageCtx) + stageCtx, cancel := w.stageContext(ctx, runtimeCfg) + hydratedCount, err := w.hydrateMissingNextRuns(stageCtx, runtimeCfg) cancel() if err != nil { return map[string]any{ @@ -188,8 +191,8 @@ func (w *ScheduleDispatchWorker) RunOnce(ctx context.Context) (map[string]any, e }, err } - stageCtx, cancel = w.stageContext(ctx) - paused, stats, err := w.shouldPauseDispatch(stageCtx) + stageCtx, cancel = w.stageContext(ctx, runtimeCfg) + paused, stats, err := w.shouldPauseDispatch(stageCtx, runtimeCfg) cancel() if err != nil { return map[string]any{ @@ -204,11 +207,12 @@ func (w *ScheduleDispatchWorker) RunOnce(ctx context.Context) (map[string]any, e "queue_depth": stats.Messages, "queue_consumers": stats.Consumers, "duration_ms": time.Since(startedAt).Milliseconds(), + "batch_size": runtimeCfg.BatchSize, }, nil } - stageCtx, cancel = w.stageContext(ctx) - tasks, err := w.claimDueSchedules(stageCtx) + stageCtx, cancel = w.stageContext(ctx, runtimeCfg) + tasks, err := w.claimDueSchedules(stageCtx, runtimeCfg) cancel() if err != nil { return map[string]any{ @@ -221,11 +225,11 @@ func (w *ScheduleDispatchWorker) RunOnce(ctx context.Context) (map[string]any, e "hydrated_count": hydratedCount, "claimed_count": len(tasks), "duration_ms": time.Since(startedAt).Milliseconds(), + "batch_size": runtimeCfg.BatchSize, }, nil } -func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context) (int, error) { - runtimeCfg := w.runtimeConfig() +func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context, runtimeCfg scheduleDispatchRuntimeConfig) (int, error) { tx, err := w.pool.Begin(ctx) if err != nil { return 0, err @@ -318,8 +322,7 @@ func (w *ScheduleDispatchWorker) hydrateMissingNextRuns(ctx context.Context) (in return count, nil } -func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context) ([]dueScheduleTask, error) { - runtimeCfg := w.runtimeConfig() +func (w *ScheduleDispatchWorker) claimDueSchedules(ctx context.Context, runtimeCfg scheduleDispatchRuntimeConfig) ([]dueScheduleTask, error) { tx, err := w.pool.Begin(ctx) if err != nil { return nil, err @@ -539,11 +542,10 @@ func (w *ScheduleDispatchWorker) dispatchBatch(parent context.Context, tasks []d wg.Wait() } -func (w *ScheduleDispatchWorker) shouldPauseDispatch(ctx context.Context) (bool, rabbitmq.QueueStats, error) { +func (w *ScheduleDispatchWorker) shouldPauseDispatch(ctx context.Context, runtimeCfg scheduleDispatchRuntimeConfig) (bool, rabbitmq.QueueStats, error) { if w == nil || w.rabbitMQ == nil { return false, rabbitmq.QueueStats{}, nil } - runtimeCfg := w.runtimeConfig() if runtimeCfg.QueueBackpressure <= 0 { return false, rabbitmq.QueueStats{}, nil } @@ -565,6 +567,32 @@ func scheduleDispatchRuntimeFromConfig(cfg config.SchedulerConfig) scheduleDispa } } +func (cfg *scheduleDispatchRuntimeConfig) ApplyJobRun(run JobRunContext) { + if cfg == nil { + return + } + if run.Job != nil { + if run.Job.BatchSize != nil && *run.Job.BatchSize > 0 { + cfg.BatchSize = *run.Job.BatchSize + } + if run.Job.TimeoutSeconds > 0 { + cfg.Timeout = time.Duration(run.Job.TimeoutSeconds) * time.Second + } + } + if value := AsInt(run.Config, "dispatch_workers", 0); value > 0 { + cfg.DispatchWorkers = value + } + if value := AsInt(run.Config, "queue_backpressure", 0); value > 0 { + cfg.QueueBackpressure = value + } + if value := AsInt(run.Config, "batch_size", 0); value > 0 && (run.Job == nil || run.Job.BatchSize == nil) { + cfg.BatchSize = value + } + if timeout := AsDuration(run.Config, "timeout", 0); timeout > 0 && (run.Job == nil || run.Job.TimeoutSeconds <= 0) { + cfg.Timeout = timeout + } +} + func schedulerDispatchInterval(cfg config.SchedulerConfig) time.Duration { if cfg.DispatchInterval > 0 { return cfg.DispatchInterval @@ -593,11 +621,15 @@ func schedulerDispatchConcurrency(cfg config.SchedulerConfig) int { return 1 } -func (w *ScheduleDispatchWorker) stageContext(parent context.Context) (context.Context, context.CancelFunc) { +func (w *ScheduleDispatchWorker) stageContext(parent context.Context, runtimeCfg ...scheduleDispatchRuntimeConfig) (context.Context, context.CancelFunc) { if parent == nil { parent = context.Background() } - return context.WithTimeout(parent, w.runtimeConfig().Timeout) + cfg := w.runtimeConfig() + if len(runtimeCfg) > 0 { + cfg = runtimeCfg[0] + } + return context.WithTimeout(parent, cfg.Timeout) } func (w *ScheduleDispatchWorker) invalidateScheduleTaskCaches(updates []scheduleTaskCacheUpdate) { diff --git a/server/internal/tenant/app/monitoring_daily_task_worker.go b/server/internal/tenant/app/monitoring_daily_task_worker.go index 4882142..01a04ca 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker.go @@ -56,6 +56,39 @@ type MonitoringDailyTaskGenerationSummary struct { SkippedPlanCount int } +type MonitoringDailyTaskRunOptions struct { + MaxMaterializePerRun int + MaxMaterializePerPlan int + MaxMaterializePerBrand int +} + +func normalizeMonitoringDailyRunOptions(options ...MonitoringDailyTaskRunOptions) MonitoringDailyTaskRunOptions { + out := MonitoringDailyTaskRunOptions{ + MaxMaterializePerRun: defaultMonitoringDailyMaxMaterializePerRun, + MaxMaterializePerPlan: defaultMonitoringDailyMaxMaterializePerPlan, + MaxMaterializePerBrand: defaultMonitoringDailyMaxMaterializePerBrand, + } + if len(options) > 0 { + in := options[0] + if in.MaxMaterializePerRun > 0 { + out.MaxMaterializePerRun = in.MaxMaterializePerRun + } + if in.MaxMaterializePerPlan > 0 { + out.MaxMaterializePerPlan = in.MaxMaterializePerPlan + } + if in.MaxMaterializePerBrand > 0 { + out.MaxMaterializePerBrand = in.MaxMaterializePerBrand + } + } + if out.MaxMaterializePerPlan > out.MaxMaterializePerRun { + out.MaxMaterializePerPlan = out.MaxMaterializePerRun + } + if out.MaxMaterializePerBrand > out.MaxMaterializePerPlan { + out.MaxMaterializePerBrand = out.MaxMaterializePerPlan + } + return out +} + type monitoringDailyPlan struct { TenantID int64 WorkspaceID int64 @@ -174,15 +207,16 @@ func (w *MonitoringDailyTaskWorker) runOnce(parent context.Context) { ) } -func (w *MonitoringDailyTaskWorker) RunOnce(parent context.Context) (map[string]any, error) { +func (w *MonitoringDailyTaskWorker) RunOnce(parent context.Context, options ...MonitoringDailyTaskRunOptions) (map[string]any, error) { if w == nil || w.service == nil || w.service.businessPool == nil || w.service.monitoringPool == nil { return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil } + runOptions := normalizeMonitoringDailyRunOptions(options...) ctx, cancel := context.WithTimeout(parent, w.timeout) defer cancel() startedAt := time.Now() - summary, err := w.service.GenerateDailyMonitoringTasks(ctx, w.projectionService, time.Now().UTC()) + summary, err := w.service.GenerateDailyMonitoringTasks(ctx, w.projectionService, time.Now().UTC(), runOptions) duration := time.Since(startedAt) recordMonitoringDailyTaskMetrics(summary, duration, err) stats := map[string]any{ @@ -201,6 +235,9 @@ func (w *MonitoringDailyTaskWorker) RunOnce(parent context.Context) (map[string] stats["deferred_task_count"] = summary.DeferredTaskCount stats["skipped_plan_count"] = summary.SkippedPlanCount } + stats["max_materialize_per_run"] = runOptions.MaxMaterializePerRun + stats["max_materialize_per_plan"] = runOptions.MaxMaterializePerPlan + stats["max_materialize_per_brand"] = runOptions.MaxMaterializePerBrand if err != nil { if w.logger != nil { w.logger.Warn("monitoring daily task generation failed", MonitoringWorkerErrorFields(err)...) @@ -214,6 +251,7 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( ctx context.Context, projectionService *MonitoringCallbackService, now time.Time, + runOptions ...MonitoringDailyTaskRunOptions, ) (*MonitoringDailyTaskGenerationSummary, error) { if s == nil || s.businessPool == nil || s.monitoringPool == nil { return nil, nil @@ -230,7 +268,8 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( BusinessDate: businessDate, TenantCount: len(plans), } - globalRemaining := defaultMonitoringDailyMaxMaterializePerRun + options := normalizeMonitoringDailyRunOptions(runOptions...) + globalRemaining := options.MaxMaterializePerRun for _, plan := range plans { if globalRemaining <= 0 { @@ -271,7 +310,7 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( continue } - planRunRemaining := minMonitoringDailyInt(defaultMonitoringDailyMaxMaterializePerPlan, globalRemaining) + planRunRemaining := minMonitoringDailyInt(options.MaxMaterializePerPlan, globalRemaining) for _, brand := range brands { if planRunRemaining <= 0 || globalRemaining <= 0 { @@ -308,7 +347,7 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( continue } - dispatchLimit := minMonitoringDailyInt(defaultMonitoringDailyMaxMaterializePerBrand, planRunRemaining, globalRemaining) + dispatchLimit := minMonitoringDailyInt(options.MaxMaterializePerBrand, planRunRemaining, globalRemaining) materializeLimit := minMonitoringDailyInt(dispatchLimit, dailyRemaining) createdCount, dueCount, desktopCount, deferredCount, err := s.generateDailyMonitoringTasksForBrand( ctx, diff --git a/server/internal/tenant/app/monitoring_lease_recovery.go b/server/internal/tenant/app/monitoring_lease_recovery.go index ce01142..62bbcc0 100644 --- a/server/internal/tenant/app/monitoring_lease_recovery.go +++ b/server/internal/tenant/app/monitoring_lease_recovery.go @@ -14,16 +14,35 @@ const ( monitoringStaleTaskDropError = "monitoring task dropped after business day rollover" ) -func expireMonitoringLeasedTasks(ctx context.Context, tx pgx.Tx, tenantID *int64, now time.Time) (int64, error) { +func expireMonitoringLeasedTasks(ctx context.Context, tx pgx.Tx, tenantID *int64, now time.Time, limits ...int) (int64, error) { + limit := 0 + if len(limits) > 0 { + limit = limits[0] + } + if limit <= 0 { + limit = 1000 + } tag, err := tx.Exec(ctx, ` + WITH candidates AS ( + SELECT id + FROM monitoring_collect_tasks + WHERE collector_type = $2 + AND status = 'leased' + AND lease_expires_at IS NOT NULL + AND lease_expires_at < $3 + AND ($4::bigint IS NULL OR tenant_id = $4) + ORDER BY lease_expires_at ASC, id ASC + FOR UPDATE SKIP LOCKED + LIMIT $5 + ) UPDATE monitoring_collect_tasks SET status = 'expired', - lease_token_hash = NULL, - leased_to_executor = NULL, - leased_at = NULL, - lease_expires_at = NULL, - error_message = COALESCE(NULLIF(error_message, ''), $1), - request_payload_json = ( + lease_token_hash = NULL, + leased_to_executor = NULL, + leased_at = NULL, + lease_expires_at = NULL, + error_message = COALESCE(NULLIF(error_message, ''), $1), + request_payload_json = ( CASE WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object' THEN '{}'::jsonb @@ -40,12 +59,9 @@ func expireMonitoringLeasedTasks(ctx context.Context, tx pgx.Tx, tenantID *int64 ) + 1 ), updated_at = NOW() - WHERE collector_type = $2 - AND status = 'leased' - AND lease_expires_at IS NOT NULL - AND lease_expires_at < $3 - AND ($4::bigint IS NULL OR tenant_id = $4) - `, monitoringLeaseExpiredErrPrefix, monitoringCollectorType, now.UTC(), nullableInt64(tenantID)) + FROM candidates c + WHERE monitoring_collect_tasks.id = c.id + `, monitoringLeaseExpiredErrPrefix, monitoringCollectorType, now.UTC(), nullableInt64(tenantID), limit) if err != nil { return 0, monitoringInternalError(50041, "lease_expire_failed", "failed to expire overdue monitoring lease tasks", err) } @@ -118,6 +134,6 @@ func skipMonitoringTasksBeforeBusinessDate( return tag.RowsAffected(), nil } -func ExpireMonitoringLeasedTasks(ctx context.Context, tx pgx.Tx, tenantID *int64, now time.Time) (int64, error) { - return expireMonitoringLeasedTasks(ctx, tx, tenantID, now) +func ExpireMonitoringLeasedTasks(ctx context.Context, tx pgx.Tx, tenantID *int64, now time.Time, limit int) (int64, error) { + return expireMonitoringLeasedTasks(ctx, tx, tenantID, now, limit) } diff --git a/server/internal/tenant/app/monitoring_received_inspection.go b/server/internal/tenant/app/monitoring_received_inspection.go index 3ea6bbf..8b657db 100644 --- a/server/internal/tenant/app/monitoring_received_inspection.go +++ b/server/internal/tenant/app/monitoring_received_inspection.go @@ -56,7 +56,7 @@ func markMonitoringStaleReceivedTasks(ctx context.Context, tx pgx.Tx, now time.T AND ( request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object' - OR COALESCE(NULLIF(request_payload_json ->> 'received_alerted_at', ''), '') = '' + OR NOT (request_payload_json ? 'received_alerted_at') ) ORDER BY callback_received_at ASC, id ASC FOR UPDATE SKIP LOCKED diff --git a/server/internal/worker/generate/article_generation_state_check_worker.go b/server/internal/worker/generate/article_generation_state_check_worker.go index 282ae23..02a8d94 100644 --- a/server/internal/worker/generate/article_generation_state_check_worker.go +++ b/server/internal/worker/generate/article_generation_state_check_worker.go @@ -20,6 +20,12 @@ type GenerationTaskStateCheckWorker struct { cache sharedcache.Cache } +type GenerationTaskStateCheckRunOptions struct { + Timeout time.Duration + BatchSize int + Lookback time.Duration +} + func NewGenerationTaskStateCheckWorker(pool *pgxpool.Pool, logger *zap.Logger, cfg config.GenerationConfig) *GenerationTaskStateCheckWorker { return &GenerationTaskStateCheckWorker{ pool: pool, @@ -92,13 +98,30 @@ func (w *GenerationTaskStateCheckWorker) runOnce(parent context.Context) { }, zap.Int("anomaly_count", count)) } -func (w *GenerationTaskStateCheckWorker) RunOnce(parent context.Context) (map[string]any, error) { +func (w *GenerationTaskStateCheckWorker) RunOnce(parent context.Context, options ...GenerationTaskStateCheckRunOptions) (map[string]any, error) { if w == nil || w.pool == nil { return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil } cfg := w.runtimeConfig() + if len(options) > 0 { + if options[0].BatchSize > 0 { + cfg.BatchSize = options[0].BatchSize + } + if options[0].Timeout > 0 { + cfg.Timeout = options[0].Timeout + } + if options[0].Lookback > 0 { + cfg.Lookback = options[0].Lookback + } + } startedAt := time.Now() - count, err := CheckGenerationTaskStateConsistency(parent, w.pool, w.logger, w.cache, cfg) + checkCtx := parent + var cancel context.CancelFunc = func() {} + if cfg.Timeout > 0 { + checkCtx, cancel = context.WithTimeout(parent, cfg.Timeout) + } + defer cancel() + count, err := CheckGenerationTaskStateConsistency(checkCtx, w.pool, w.logger, w.cache, cfg) duration := time.Since(startedAt) stats := map[string]any{ "anomaly_count": count, diff --git a/server/migrations/20260520141000_add_scheduler_safe_indexes.down.sql b/server/migrations/20260520141000_add_scheduler_safe_indexes.down.sql new file mode 100644 index 0000000..64b3f7b --- /dev/null +++ b/server/migrations/20260520141000_add_scheduler_safe_indexes.down.sql @@ -0,0 +1,6 @@ +DROP INDEX IF EXISTS idx_tenant_plan_active_window; +DROP INDEX IF EXISTS idx_brands_monitoring_daily_candidate; +DROP INDEX IF EXISTS idx_platform_accounts_monitoring_client_platform; +DROP INDEX IF EXISTS idx_desktop_clients_active_plan_candidate; +DROP INDEX IF EXISTS idx_schedule_task_due_dispatch; +DROP INDEX IF EXISTS idx_schedule_task_missing_next_run; diff --git a/server/migrations/20260520141000_add_scheduler_safe_indexes.up.sql b/server/migrations/20260520141000_add_scheduler_safe_indexes.up.sql new file mode 100644 index 0000000..6680785 --- /dev/null +++ b/server/migrations/20260520141000_add_scheduler_safe_indexes.up.sql @@ -0,0 +1,30 @@ +CREATE INDEX IF NOT EXISTS idx_schedule_task_missing_next_run + ON schedule_tasks (created_at, id) + WHERE deleted_at IS NULL + AND status = 'enabled' + AND next_run_at IS NULL; + +CREATE INDEX IF NOT EXISTS idx_schedule_task_due_dispatch + ON schedule_tasks (next_run_at, id) + WHERE deleted_at IS NULL + AND status = 'enabled' + AND next_run_at IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_desktop_clients_active_plan_candidate + ON desktop_clients (tenant_id, workspace_id, created_at, id) + INCLUDE (last_seen_at) + WHERE revoked_at IS NULL; + +CREATE INDEX IF NOT EXISTS idx_platform_accounts_monitoring_client_platform + ON platform_accounts (tenant_id, workspace_id, client_id, platform_id) + WHERE deleted_at IS NULL + AND client_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_brands_monitoring_daily_candidate + ON brands (tenant_id, created_at, id) + WHERE deleted_at IS NULL; + +CREATE INDEX IF NOT EXISTS idx_tenant_plan_active_window + ON tenant_plan_subscriptions (tenant_id, start_at, end_at, plan_id) + WHERE status = 'active' + AND deleted_at IS NULL; diff --git a/server/migrations_monitoring/20260520140000_add_scheduler_safe_indexes.down.sql b/server/migrations_monitoring/20260520140000_add_scheduler_safe_indexes.down.sql new file mode 100644 index 0000000..adb4450 --- /dev/null +++ b/server/migrations_monitoring/20260520140000_add_scheduler_safe_indexes.down.sql @@ -0,0 +1,13 @@ +DROP INDEX IF EXISTS idx_question_monitor_runs_retention; +DROP INDEX IF EXISTS idx_collect_tasks_retention; +DROP INDEX IF EXISTS idx_platform_access_snapshots_retention; +DROP INDEX IF EXISTS idx_monitoring_brand_daily_retention; +DROP INDEX IF EXISTS idx_monitoring_brand_platform_daily_retention; +DROP INDEX IF EXISTS idx_collect_requests_retention; +DROP INDEX IF EXISTS idx_collect_dispatch_outbox_retention; +DROP INDEX IF EXISTS idx_collect_tasks_daily_dispatch_due; +DROP INDEX IF EXISTS idx_collect_tasks_brand_day_materialized; +DROP INDEX IF EXISTS idx_collect_tasks_brand_day_existing; +DROP INDEX IF EXISTS idx_collect_tasks_received_watchdog; +DROP INDEX IF EXISTS idx_collect_tasks_result_recovery_queue; +DROP INDEX IF EXISTS idx_collect_tasks_global_lease_expired; diff --git a/server/migrations_monitoring/20260520140000_add_scheduler_safe_indexes.up.sql b/server/migrations_monitoring/20260520140000_add_scheduler_safe_indexes.up.sql new file mode 100644 index 0000000..c228a86 --- /dev/null +++ b/server/migrations_monitoring/20260520140000_add_scheduler_safe_indexes.up.sql @@ -0,0 +1,87 @@ +CREATE INDEX IF NOT EXISTS idx_collect_tasks_global_lease_expired + ON monitoring_collect_tasks (lease_expires_at, id) + INCLUDE (tenant_id) + WHERE collector_type = 'desktop' + AND status = 'leased' + AND lease_expires_at IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_collect_tasks_result_recovery_queue + ON monitoring_collect_tasks (callback_received_at, id) + WHERE collector_type = 'desktop' + AND callback_received_at IS NOT NULL + AND request_payload_json IS NOT NULL + AND jsonb_typeof(request_payload_json) = 'object' + AND ( + status = 'received' + OR ( + COALESCE(execution_owner, 'legacy') = 'desktop_tasks' + AND status = 'pending' + ) + ); + +CREATE INDEX IF NOT EXISTS idx_collect_tasks_received_watchdog + ON monitoring_collect_tasks (callback_received_at, id) + WHERE collector_type = 'desktop' + AND callback_received_at IS NOT NULL + AND ( + status = 'received' + OR ( + COALESCE(execution_owner, 'legacy') = 'desktop_tasks' + AND status = 'pending' + ) + ) + AND ( + request_payload_json IS NULL + OR jsonb_typeof(request_payload_json) <> 'object' + OR NOT (request_payload_json ? 'received_alerted_at') + ); + +CREATE INDEX IF NOT EXISTS idx_collect_tasks_brand_day_existing + ON monitoring_collect_tasks (tenant_id, brand_id, business_date, question_id, ai_platform_id) + WHERE collector_type = 'desktop' + AND run_mode = 'desktop_standard'; + +CREATE INDEX IF NOT EXISTS idx_collect_tasks_brand_day_materialized + ON monitoring_collect_tasks (tenant_id, brand_id, business_date) + WHERE collector_type = 'desktop' + AND trigger_source = 'automatic' + AND run_mode = 'desktop_standard' + AND COALESCE(execution_owner, 'legacy') = 'desktop_tasks'; + +CREATE INDEX IF NOT EXISTS idx_collect_tasks_daily_dispatch_due + ON monitoring_collect_tasks ( + tenant_id, + brand_id, + business_date, + dispatch_after, + last_dispatched_at, + dispatch_priority DESC, + id + ) + WHERE collector_type = 'desktop' + AND status = 'pending' + AND callback_received_at IS NULL + AND trigger_source = 'automatic' + AND run_mode = 'desktop_standard' + AND COALESCE(execution_owner, 'legacy') = 'desktop_tasks'; + +CREATE INDEX IF NOT EXISTS idx_collect_dispatch_outbox_retention + ON monitoring_collect_dispatch_outbox (created_at, id); + +CREATE INDEX IF NOT EXISTS idx_collect_requests_retention + ON monitoring_collect_requests (created_at, id); + +CREATE INDEX IF NOT EXISTS idx_monitoring_brand_platform_daily_retention + ON monitoring_brand_platform_daily (business_date, id); + +CREATE INDEX IF NOT EXISTS idx_monitoring_brand_daily_retention + ON monitoring_brand_daily (business_date, id); + +CREATE INDEX IF NOT EXISTS idx_platform_access_snapshots_retention + ON monitoring_platform_access_snapshots (business_date, id); + +CREATE INDEX IF NOT EXISTS idx_collect_tasks_retention + ON monitoring_collect_tasks (business_date, id); + +CREATE INDEX IF NOT EXISTS idx_question_monitor_runs_retention + ON question_monitor_runs (business_date, id); diff --git a/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql b/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql index b282d44..ac42410 100644 --- a/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql +++ b/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql @@ -63,6 +63,10 @@ CREATE INDEX IF NOT EXISTS idx_scheduler_job_runs_job_started CREATE INDEX IF NOT EXISTS idx_scheduler_job_runs_status_started ON ops.scheduler_job_runs (status, started_at DESC); +CREATE INDEX IF NOT EXISTS idx_scheduler_job_runs_auto_started + ON ops.scheduler_job_runs (job_key, started_at DESC, id DESC) + WHERE trigger_type = 'auto'; + CREATE TABLE IF NOT EXISTS ops.scheduler_job_triggers ( id BIGSERIAL PRIMARY KEY, job_key VARCHAR(96) NOT NULL REFERENCES ops.scheduler_jobs(job_key), @@ -88,16 +92,17 @@ CREATE INDEX IF NOT EXISTS idx_scheduler_job_triggers_pending INSERT INTO ops.scheduler_jobs (job_key, display_name, category, description, enabled, schedule_type, interval_seconds, timezone, timeout_seconds, batch_size, config) VALUES - ('schedule_dispatch', '内容定时任务派发', 'content', '扫描内容定时任务并投递生成队列。', true, 'interval', 15, 'Asia/Shanghai', 30, 100, '{}'::jsonb), - ('monitoring_daily_task', 'AI 监控每日任务生成', 'monitoring', '为品牌问题按业务日生成 AI 监控采集任务。', true, 'interval', 300, 'Asia/Shanghai', 45, 64, '{}'::jsonb), - ('monitoring_result_recovery', '监控结果恢复', 'monitoring', '恢复已回调但未成功进入结果队列的监控采集结果。', true, 'interval', 60, 'Asia/Shanghai', 30, 100, '{}'::jsonb), - ('monitoring_lease_recovery', '监控租约回收', 'monitoring', '回收超过租约时间仍未回调的监控采集任务。', true, 'interval', 1800, 'Asia/Shanghai', 30, NULL, '{}'::jsonb), - ('monitoring_received_inspection', '监控 received 卡死检查', 'monitoring', '检查长时间停留在 received 状态的监控任务。', true, 'interval', 300, 'Asia/Shanghai', 30, 100, '{}'::jsonb), - ('generation_state_check', '生成任务状态巡检', 'generation', '低频巡检文章生成任务与文章状态一致性。', true, 'interval', 300, 'Asia/Shanghai', 10, 100, '{}'::jsonb), - ('monitoring_retention_cleanup', '监控历史保留清理', 'monitoring', '按业务日期清理最近 30 天窗口以前的监控历史数据。', true, 'interval', 86400, 'Asia/Shanghai', 900, 5000, + ('schedule_dispatch', '内容定时任务派发', 'content', '索引驱动的延迟队列派发，仅认领到期内容任务并投递生成队列。', true, 'interval', 15, 'Asia/Shanghai', 30, 100, '{}'::jsonb), + ('monitoring_daily_task', 'AI 监控每日任务生成', 'monitoring', '按业务日小批量物化监控采集任务并派发到桌面任务队列。', true, 'interval', 300, 'Asia/Shanghai', 45, 64, '{"max_materialize_per_plan":12,"max_materialize_per_brand":4}'::jsonb), + ('monitoring_result_recovery', '监控结果恢复', 'monitoring', '按回调结果队列索引恢复未成功入队的监控结果。', true, 'interval', 60, 'Asia/Shanghai', 30, 100, '{}'::jsonb), + ('monitoring_lease_recovery', '监控租约回收', 'monitoring', '按过期租约索引限批回收超过租约时间仍未回调的监控任务。', true, 'interval', 1800, 'Asia/Shanghai', 30, 1000, '{}'::jsonb), + ('monitoring_received_inspection', '监控 received 卡死检查', 'monitoring', '按 received watchdog 索引限批标记长时间未入队的监控任务。', true, 'interval', 300, 'Asia/Shanghai', 30, 100, '{"threshold":"15m"}'::jsonb), + ('generation_state_check', '生成任务状态巡检', 'generation', '低频、限批检查文章生成任务与文章状态一致性。', true, 'interval', 300, 'Asia/Shanghai', 10, 100, '{"lookback":"24h"}'::jsonb), + ('monitoring_retention_cleanup', '监控历史保留清理', 'monitoring', '按保留窗口和索引限批删除最近 30 天以前的监控历史数据。', true, 'interval', 86400, 'Asia/Shanghai', 900, 5000, '{"retention_days":30,"batch_size":5000,"max_batches_per_run":200,"dry_run":false,"statement_timeout":"5s","run_at_local":"02:30"}'::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/20260520143000_harden_scheduler_jobs_for_k3s.down.sql b/server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.down.sql new file mode 100644 index 0000000..6b083ed --- /dev/null +++ b/server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.down.sql @@ -0,0 +1 @@ +DROP INDEX IF EXISTS ops.idx_scheduler_job_runs_auto_started; diff --git a/server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.up.sql b/server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.up.sql new file mode 100644 index 0000000..e820b85 --- /dev/null +++ b/server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.up.sql @@ -0,0 +1,59 @@ +CREATE INDEX IF NOT EXISTS idx_scheduler_job_runs_auto_started + ON ops.scheduler_job_runs (job_key, started_at DESC, id DESC) + WHERE trigger_type = 'auto'; + +UPDATE ops.scheduler_jobs +SET description = '索引驱动的延迟队列派发，仅认领到期内容任务并投递生成队列。', + timeout_seconds = 30, + batch_size = 100, + config = config || '{}'::jsonb, + updated_at = NOW() +WHERE job_key = 'schedule_dispatch'; + +UPDATE ops.scheduler_jobs +SET description = '按业务日小批量物化监控采集任务并派发到桌面任务队列。', + timeout_seconds = 45, + batch_size = 64, + config = config || '{"max_materialize_per_plan":12,"max_materialize_per_brand":4}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_daily_task'; + +UPDATE ops.scheduler_jobs +SET description = '按回调结果队列索引恢复未成功入队的监控结果。', + timeout_seconds = 30, + batch_size = 100, + config = config || '{"claim_ttl":"5m"}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_result_recovery'; + +UPDATE ops.scheduler_jobs +SET description = '按过期租约索引限批回收超过租约时间仍未回调的监控任务。', + timeout_seconds = 30, + batch_size = 1000, + config = config || '{}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_lease_recovery'; + +UPDATE ops.scheduler_jobs +SET description = '按 received watchdog 索引限批标记长时间未入队的监控任务。', + timeout_seconds = 30, + batch_size = 100, + config = config || '{"threshold":"15m"}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_received_inspection'; + +UPDATE ops.scheduler_jobs +SET description = '低频、限批检查文章生成任务与文章状态一致性。', + timeout_seconds = 10, + batch_size = 100, + config = config || '{"lookback":"24h"}'::jsonb, + updated_at = NOW() +WHERE job_key = 'generation_state_check'; + +UPDATE ops.scheduler_jobs +SET description = '按保留窗口和索引限批删除最近 30 天以前的监控历史数据。', + timeout_seconds = 900, + batch_size = 5000, + config = config || '{"retention_days":30,"batch_size":5000,"max_batches_per_run":200,"dry_run":false,"statement_timeout":"5s","run_at_local":"02:30"}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_retention_cleanup';