diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go index 5ed6da0..151e646 100644 --- a/server/cmd/scheduler/main.go +++ b/server/cmd/scheduler/main.go @@ -344,20 +344,19 @@ func shutdownSchedulerMetricsServer(server *http.Server, logger interface { } func monitoringDailyRunOptions(jobCtx internalscheduler.JobRunContext) tenantapp.MonitoringDailyTaskRunOptions { - batchSize := 0 + maxDesktopDispatch := 0 if jobCtx.Job != nil && jobCtx.Job.BatchSize != nil { - batchSize = *jobCtx.Job.BatchSize + maxDesktopDispatch = *jobCtx.Job.BatchSize } - if batchSize <= 0 { - batchSize = internalscheduler.AsInt(jobCtx.Config, "max_materialize_per_run", 0) + if maxDesktopDispatch <= 0 { + maxDesktopDispatch = internalscheduler.AsInt(jobCtx.Config, "max_desktop_dispatch_per_run", 0) } - if batchSize <= 0 { - batchSize = internalscheduler.AsInt(jobCtx.Config, "batch_size", 0) + if maxDesktopDispatch <= 0 { + maxDesktopDispatch = 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), + MaxDesktopDispatchPerRun: maxDesktopDispatch, + MaxDesktopClientBacklog: internalscheduler.AsInt(jobCtx.Config, "max_desktop_client_backlog", 0), } } diff --git a/server/cmd/scheduler/main_test.go b/server/cmd/scheduler/main_test.go index 1b7974c..bf132b9 100644 --- a/server/cmd/scheduler/main_test.go +++ b/server/cmd/scheduler/main_test.go @@ -59,22 +59,29 @@ func TestSchedulerMetricsAuthMiddleware(t *testing.T) { } } -func TestMonitoringDailyRunOptionsUsesJobBatchSize(t *testing.T) { +func TestMonitoringDailyRunOptionsUsesJobBatchSizeForDesktopDispatchOnly(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, + "max_materialize_per_run": 999, + "max_materialize_per_plan": 12, + "max_materialize_per_brand": 4, + "max_desktop_client_backlog": 6, }, }) - if got.MaxMaterializePerRun != 88 { - t.Fatalf("MaxMaterializePerRun = %d, want 88", got.MaxMaterializePerRun) + if got.MaxDesktopDispatchPerRun != 88 { + t.Fatalf("MaxDesktopDispatchPerRun = %d, want 88", got.MaxDesktopDispatchPerRun) } - if got.MaxMaterializePerPlan != 12 || got.MaxMaterializePerBrand != 4 { - t.Fatalf("unexpected plan/brand options: %#v", got) + if got.MaxMaterializePerPlan != 0 || got.MaxMaterializePerBrand != 0 { + t.Fatalf("legacy plan/brand throttles should be ignored: %#v", got) + } + if got.MaxMaterializePerRun != 0 { + t.Fatalf("batch_size must not throttle daily materialization: %#v", got) + } + if got.MaxDesktopClientBacklog != 6 { + t.Fatalf("MaxDesktopClientBacklog = %d, want 6", got.MaxDesktopClientBacklog) } } diff --git a/server/internal/tenant/app/monitoring_daily_task_worker.go b/server/internal/tenant/app/monitoring_daily_task_worker.go index d511d4a..8d32ef0 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker.go @@ -19,19 +19,20 @@ import ( ) const ( - defaultMonitoringDailyTaskInterval = 5 * time.Minute - defaultMonitoringDailyTaskTimeout = 45 * time.Second - defaultMonitoringDailyTaskHardCap = 36 - defaultMonitoringDailyBrandLimit = 1 - defaultMonitoringDailyLookahead = 10 * time.Minute - defaultMonitoringDailyMaterializeWindow = 2 * time.Hour - defaultMonitoringDailyDesktopRetryCooldown = 15 * time.Minute - defaultMonitoringDailyPublishTimeout = 5 * time.Second - defaultMonitoringDailyMaxMaterializePerRun = 64 - defaultMonitoringDailyMaxMaterializePerPlan = 12 - defaultMonitoringDailyMaxMaterializePerBrand = 4 - monitoringDailyPlatformActiveWindow = 24 * time.Hour - monitoringDailyMaterializeStartOffset = 5 * time.Minute + defaultMonitoringDailyTaskInterval = 5 * time.Minute + defaultMonitoringDailyTaskTimeout = 45 * time.Second + defaultMonitoringDailyTaskHardCap = 36 + defaultMonitoringDailyBrandLimit = 1 + defaultMonitoringDailyLookahead = 10 * time.Minute + defaultMonitoringDailyDesktopRetryCooldown = 15 * time.Minute + defaultMonitoringDailyPublishTimeout = 5 * time.Second + defaultMonitoringDailyMaxMaterializePerRun = 0 + defaultMonitoringDailyMaxMaterializePerPlan = 0 + defaultMonitoringDailyMaxMaterializePerBrand = 0 + defaultMonitoringDailyMaxDesktopDispatch = 12 + defaultMonitoringDailyMaxDesktopClientBacklog = 12 + monitoringDailyPlatformActiveWindow = 24 * time.Hour + monitoringDailyMaterializeStartOffset = 5 * time.Minute ) type MonitoringDailyTaskWorker struct { @@ -57,16 +58,20 @@ type MonitoringDailyTaskGenerationSummary struct { } type MonitoringDailyTaskRunOptions struct { - MaxMaterializePerRun int - MaxMaterializePerPlan int - MaxMaterializePerBrand int + MaxMaterializePerRun int + MaxMaterializePerPlan int + MaxMaterializePerBrand int + MaxDesktopDispatchPerRun int + MaxDesktopClientBacklog int } func normalizeMonitoringDailyRunOptions(options ...MonitoringDailyTaskRunOptions) MonitoringDailyTaskRunOptions { out := MonitoringDailyTaskRunOptions{ - MaxMaterializePerRun: defaultMonitoringDailyMaxMaterializePerRun, - MaxMaterializePerPlan: defaultMonitoringDailyMaxMaterializePerPlan, - MaxMaterializePerBrand: defaultMonitoringDailyMaxMaterializePerBrand, + MaxMaterializePerRun: defaultMonitoringDailyMaxMaterializePerRun, + MaxMaterializePerPlan: defaultMonitoringDailyMaxMaterializePerPlan, + MaxMaterializePerBrand: defaultMonitoringDailyMaxMaterializePerBrand, + MaxDesktopDispatchPerRun: defaultMonitoringDailyMaxDesktopDispatch, + MaxDesktopClientBacklog: defaultMonitoringDailyMaxDesktopClientBacklog, } if len(options) > 0 { in := options[0] @@ -79,11 +84,17 @@ func normalizeMonitoringDailyRunOptions(options ...MonitoringDailyTaskRunOptions if in.MaxMaterializePerBrand > 0 { out.MaxMaterializePerBrand = in.MaxMaterializePerBrand } + if in.MaxDesktopDispatchPerRun > 0 { + out.MaxDesktopDispatchPerRun = in.MaxDesktopDispatchPerRun + } + if in.MaxDesktopClientBacklog > 0 { + out.MaxDesktopClientBacklog = in.MaxDesktopClientBacklog + } } - if out.MaxMaterializePerPlan > out.MaxMaterializePerRun { + if out.MaxMaterializePerRun > 0 && out.MaxMaterializePerPlan > out.MaxMaterializePerRun { out.MaxMaterializePerPlan = out.MaxMaterializePerRun } - if out.MaxMaterializePerBrand > out.MaxMaterializePerPlan { + if out.MaxMaterializePerPlan > 0 && out.MaxMaterializePerBrand > out.MaxMaterializePerPlan { out.MaxMaterializePerBrand = out.MaxMaterializePerPlan } return out @@ -236,6 +247,8 @@ func (w *MonitoringDailyTaskWorker) RunOnce(parent context.Context, options ...M stats["max_materialize_per_run"] = runOptions.MaxMaterializePerRun stats["max_materialize_per_plan"] = runOptions.MaxMaterializePerPlan stats["max_materialize_per_brand"] = runOptions.MaxMaterializePerBrand + stats["max_desktop_dispatch_per_run"] = runOptions.MaxDesktopDispatchPerRun + stats["max_desktop_client_backlog"] = runOptions.MaxDesktopClientBacklog if err != nil { if w.logger != nil { w.logger.Warn("monitoring daily task generation failed", MonitoringWorkerErrorFields(err)...) @@ -267,10 +280,11 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( TenantCount: len(plans), } options := normalizeMonitoringDailyRunOptions(runOptions...) - globalRemaining := options.MaxMaterializePerRun + globalMaterializeRemaining := options.MaxMaterializePerRun + desktopDispatchRemaining := options.MaxDesktopDispatchPerRun for _, plan := range plans { - if globalRemaining <= 0 { + if monitoringDailyLimitReached(globalMaterializeRemaining, options.MaxMaterializePerRun) { break } executionOwner, ok, skipReason := s.resolveDailyMonitoringExecutionOwner(plan) @@ -308,10 +322,14 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( continue } - planRunRemaining := minMonitoringDailyInt(options.MaxMaterializePerPlan, globalRemaining) + planMaterializeRemaining := options.MaxMaterializePerPlan + if options.MaxMaterializePerRun > 0 && (planMaterializeRemaining <= 0 || globalMaterializeRemaining < planMaterializeRemaining) { + planMaterializeRemaining = globalMaterializeRemaining + } for _, brand := range brands { - if planRunRemaining <= 0 || globalRemaining <= 0 { + if monitoringDailyLimitReached(planMaterializeRemaining, options.MaxMaterializePerPlan) || + monitoringDailyLimitReached(globalMaterializeRemaining, options.MaxMaterializePerRun) { break } @@ -334,8 +352,13 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( continue } - dispatchLimit := minMonitoringDailyInt(options.MaxMaterializePerBrand, planRunRemaining, globalRemaining) - materializeLimit := dispatchLimit + materializeLimit := monitoringDailyBoundedLimit( + len(candidates), + options.MaxMaterializePerBrand, + planMaterializeRemaining, + globalMaterializeRemaining, + ) + dispatchLimit := monitoringDailyDispatchLimit(len(candidates), desktopDispatchRemaining, options.MaxDesktopDispatchPerRun) createdCount, dueCount, desktopCount, deferredCount, err := s.generateDailyMonitoringTasksForBrand( ctx, projectionService, @@ -348,6 +371,7 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( materializeLimit, dispatchLimit, executionOwner, + options.MaxDesktopClientBacklog, ) if err != nil { if abortErr := monitoringDailyAbortErr(ctx); abortErr != nil { @@ -358,10 +382,14 @@ func (s *MonitoringService) GenerateDailyMonitoringTasks( continue } - processedCount := int(maxMonitoringDailyInt(dueCount, desktopCount+deferredCount)) - if processedCount > 0 { - planRunRemaining -= processedCount - globalRemaining -= processedCount + if options.MaxMaterializePerPlan > 0 { + planMaterializeRemaining -= int(dueCount) + } + if options.MaxMaterializePerRun > 0 { + globalMaterializeRemaining -= int(dueCount) + } + if options.MaxDesktopDispatchPerRun > 0 { + desktopDispatchRemaining -= int(desktopCount + deferredCount) } summary.BrandCount++ summary.PlannedTaskCount += int64(len(candidates)) @@ -431,6 +459,7 @@ func (s *MonitoringService) generateDailyMonitoringTasksForBrand( materializeLimit int, dispatchLimit int, executionOwner string, + maxDesktopClientBacklog int, ) (int64, int64, int64, int64, error) { if materializeLimit <= 0 && dispatchLimit <= 0 { return 0, 0, 0, 0, nil @@ -513,7 +542,7 @@ func (s *MonitoringService) generateDailyMonitoringTasksForBrand( return createdCount, dueCount, 0, 0, nil } - publishedTasks, deferredTaskIDs, err := s.dispatchMonitorDesktopTasks(ctx, specs) + publishedTasks, deferredTaskIDs, err := s.dispatchMonitorDesktopTasks(ctx, specs, maxDesktopClientBacklog) if err != nil { if s.logger != nil { s.logger.Warn("daily monitor desktop dispatch failed; pending tasks will be retried", @@ -1178,52 +1207,58 @@ func buildMonitoringDailyTaskCandidates( if len(candidates) > limit { candidates = candidates[:limit] } - assignMonitoringDailyScheduleTimes(tenantID, brandID, businessDay, candidates) + assignMonitoringDailyScheduleTimes(businessDay, candidates) return candidates } -func assignMonitoringDailyScheduleTimes(tenantID, brandID int64, businessDay time.Time, candidates []monitoringDailyTaskCandidate) { +func assignMonitoringDailyScheduleTimes(businessDay time.Time, candidates []monitoringDailyTaskCandidate) { if len(candidates) == 0 { return } - start := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset) - window := defaultMonitoringDailyMaterializeWindow - if window <= 0 { - window = time.Hour - } - slot := window / time.Duration(len(candidates)) - if slot <= 0 { - slot = time.Second - } - latest := start.Add(window) - businessDate := monitoringBusinessDateText(businessDay) - + dueAt := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset).UTC() for i := range candidates { - jitter := time.Duration(monitoringDailyStableHash( - "dispatch_jitter", - int64Text(tenantID), - int64Text(brandID), - businessDate, - int64Text(candidates[i].Question.ID), - candidates[i].Platform.ID, - ) % uint64(slot)) - materializeAt := start.Add(time.Duration(i)*slot + jitter) - if materializeAt.After(latest) { - materializeAt = latest - } - candidates[i].MaterializeAt = materializeAt.UTC() - candidates[i].DispatchAfter = materializeAt.UTC() + candidates[i].MaterializeAt = dueAt + candidates[i].DispatchAfter = dueAt } sort.SliceStable(candidates, func(i, j int) bool { - if candidates[i].MaterializeAt.Equal(candidates[j].MaterializeAt) { - return candidates[i].sortKey < candidates[j].sortKey + if candidates[i].sortKey == candidates[j].sortKey { + if candidates[i].Question.ID == candidates[j].Question.ID { + return candidates[i].Platform.ID < candidates[j].Platform.ID + } + return candidates[i].Question.ID < candidates[j].Question.ID } - return candidates[i].MaterializeAt.Before(candidates[j].MaterializeAt) + return candidates[i].sortKey < candidates[j].sortKey }) } +func monitoringDailyLimitReached(remaining, configuredLimit int) bool { + return configuredLimit > 0 && remaining <= 0 +} + +func monitoringDailyBoundedLimit(limit int, caps ...int) int { + if limit <= 0 { + return 0 + } + for _, cap := range caps { + if cap > 0 && cap < limit { + limit = cap + } + } + return limit +} + +func monitoringDailyDispatchLimit(candidateCount, remaining, configuredLimit int) int { + if configuredLimit <= 0 { + return candidateCount + } + if remaining <= 0 { + return 0 + } + return monitoringDailyBoundedLimit(candidateCount, remaining) +} + func selectDueMonitoringDailyCandidates( candidates []monitoringDailyTaskCandidate, existing map[monitoringDailyTaskKey]struct{}, diff --git a/server/internal/tenant/app/monitoring_daily_task_worker_test.go b/server/internal/tenant/app/monitoring_daily_task_worker_test.go index 22dc6cc..045438c 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker_test.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker_test.go @@ -16,7 +16,7 @@ import ( "github.com/stretchr/testify/require" ) -func TestBuildMonitoringDailyTaskCandidatesMaterializesAfterMidnight(t *testing.T) { +func TestBuildMonitoringDailyTaskCandidatesMaterializesWholeDayImmediately(t *testing.T) { businessDay := time.Date(2026, 4, 24, 0, 0, 0, 0, monitoringBusinessLocation()) questions := []monitoringConfiguredQuestion{ {ID: 101, QuestionHash: []byte("q101")}, @@ -31,18 +31,85 @@ func TestBuildMonitoringDailyTaskCandidatesMaterializesAfterMidnight(t *testing. candidates := buildMonitoringDailyTaskCandidates(7, 11, businessDay, questions, platforms, 4) require.Len(t, candidates, 4) - start := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset).UTC() - end := start.Add(defaultMonitoringDailyMaterializeWindow) + dueAt := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset).UTC() for i, candidate := range candidates { - assert.False(t, candidate.MaterializeAt.Before(start)) - assert.False(t, candidate.MaterializeAt.After(end)) + assert.Equal(t, dueAt, candidate.MaterializeAt) assert.Equal(t, candidate.MaterializeAt, candidate.DispatchAfter) if i > 0 { - assert.False(t, candidate.MaterializeAt.Before(candidates[i-1].MaterializeAt)) + assert.LessOrEqual(t, candidates[i-1].sortKey, candidate.sortKey) } } } +func TestNormalizeMonitoringDailyRunOptionsDefaultsDoNotThrottleByPlanOrBrand(t *testing.T) { + got := normalizeMonitoringDailyRunOptions() + + assert.Equal(t, 0, got.MaxMaterializePerRun) + assert.Equal(t, 0, got.MaxMaterializePerPlan) + assert.Equal(t, 0, got.MaxMaterializePerBrand) +} + +func TestMonitoringDailyBoundedLimitTreatsZeroCapsAsUnlimited(t *testing.T) { + assert.Equal(t, 102, monitoringDailyBoundedLimit(102, 0, 0, 0)) + assert.Equal(t, 64, monitoringDailyBoundedLimit(102, 0, 64, 0)) + assert.Equal(t, 12, monitoringDailyBoundedLimit(102, 0, 64, 12)) +} + +func TestMonitoringDailyDispatchLimitStopsWhenRunCapacityIsExhausted(t *testing.T) { + assert.Equal(t, 102, monitoringDailyDispatchLimit(102, 0, 0)) + assert.Equal(t, 12, monitoringDailyDispatchLimit(102, 12, 12)) + assert.Equal(t, 0, monitoringDailyDispatchLimit(102, 0, 12)) +} + +func TestAutomaticDailySelectionIncludesAllUserQuestionsAndPlatforms(t *testing.T) { + businessDay := time.Date(2026, 6, 27, 0, 0, 0, 0, monitoringBusinessLocation()) + questions := make([]monitoringConfiguredQuestion, 0, 17) + for id := int64(1); id <= 17; id++ { + questions = append(questions, monitoringConfiguredQuestion{ + ID: id, + QuestionText: "用户问题 " + int64Text(id), + QuestionHash: []byte("q" + int64Text(id)), + }) + } + platforms := []monitoringPlatformMetadata{ + {ID: "yuanbao"}, + {ID: "kimi"}, + {ID: "wenxin"}, + {ID: "doubao"}, + {ID: "deepseek"}, + {ID: "qwen"}, + } + + candidates := buildMonitoringDailyTaskCandidates( + 7, + 7, + businessDay, + questions, + platforms, + len(questions)*len(platforms), + ) + limit := monitoringDailyBoundedLimit(len(candidates), 0, 0, 0) + due := selectDueMonitoringDailyCandidates( + candidates, + nil, + monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset), + defaultMonitoringDailyLookahead, + limit, + ) + + require.Len(t, candidates, 17*6) + require.Len(t, due, 17*6) + seen := make(map[monitoringDailyTaskKey]struct{}, len(due)) + for _, candidate := range due { + seen[monitoringDailyTaskKey{ + QuestionID: candidate.Question.ID, + PlatformID: candidate.Platform.ID, + }] = struct{}{} + } + assert.Contains(t, seen, monitoringDailyTaskKey{QuestionID: 17, PlatformID: "qwen"}) + assert.Contains(t, seen, monitoringDailyTaskKey{QuestionID: 17, PlatformID: "deepseek"}) +} + func TestSelectDueMonitoringDailyCandidatesSkipsExistingAndHonorsLimit(t *testing.T) { businessDay := time.Date(2026, 4, 24, 0, 0, 0, 0, monitoringBusinessLocation()) candidates := buildMonitoringDailyTaskCandidates( @@ -68,7 +135,7 @@ func TestSelectDueMonitoringDailyCandidatesSkipsExistingAndHonorsLimit(t *testin PlatformID: candidates[0].Platform.ID, }: {}, } - now := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset + defaultMonitoringDailyMaterializeWindow) + now := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset) due := selectDueMonitoringDailyCandidates(candidates, existing, now, defaultMonitoringDailyLookahead, 2) @@ -110,7 +177,7 @@ func TestSelectDueMonitoringDailyCandidatesContinuesPastExistingTasks(t *testing }] = struct{}{} } - now := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset + defaultMonitoringDailyMaterializeWindow) + now := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset) due := selectDueMonitoringDailyCandidates(candidates, existing, now, defaultMonitoringDailyLookahead, 4) require.Len(t, due, 2) diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go index 0d616ea..2877a46 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "sort" "strings" "time" @@ -203,15 +204,34 @@ func (s *MonitoringService) loadMonitorDesktopTaskSpecs( func (s *MonitoringService) dispatchMonitorDesktopTasks( ctx context.Context, specs []monitorDesktopTaskSpec, + maxClientBacklog ...int, ) ([]*repository.DesktopTask, []int64, error) { if s == nil || s.businessPool == nil || len(specs) == 0 { return nil, nil, nil } + backlogLimit := 0 + if len(maxClientBacklog) > 0 && maxClientBacklog[0] > 0 { + backlogLimit = maxClientBacklog[0] + } var targets map[string]monitorDesktopTaskTarget if monitorDesktopTaskSpecsNeedTargetLookup(specs) { var err error - targets, err = s.loadMonitorDesktopTaskTargets(ctx, specs[0].TenantID, specs[0].WorkspaceID, specs[0].RequestedByUserID, specs) + targets, err = s.loadMonitorDesktopTaskTargets(ctx, specs[0].TenantID, specs[0].WorkspaceID, specs[0].RequestedByUserID, specs, backlogLimit) + if err != nil { + return nil, nil, err + } + } + clientBacklog := make(map[uuid.UUID]int) + if backlogLimit > 0 { + var err error + clientBacklog, err = s.loadMonitorDesktopClientBacklog( + ctx, + specs[0].TenantID, + specs[0].WorkspaceID, + monitorDesktopTaskDispatchClientIDs(specs, targets), + backlogLimit, + ) if err != nil { return nil, nil, err } @@ -237,11 +257,22 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks( spec.TargetClientID = target.ClientID } - task, shouldPublish, fallbackLegacy, upsertErr := s.upsertMonitorDesktopTask(ctx, tx, repo, spec) + task, shouldPublish, shouldDefer, upsertErr := s.upsertMonitorDesktopTask( + ctx, + tx, + repo, + spec, + func(oldClientID, newClientID uuid.UUID) bool { + return reserveMonitorDesktopClientBacklog(oldClientID, newClientID, clientBacklog, backlogLimit) + }, + func(oldClientID, newClientID uuid.UUID) (bool, error) { + return s.reserveMonitorDesktopClientBacklogTx(ctx, tx, spec.TenantID, spec.WorkspaceID, oldClientID, newClientID, backlogLimit) + }, + ) if upsertErr != nil { return nil, nil, upsertErr } - if fallbackLegacy { + if shouldDefer { deferredTaskIDs = append(deferredTaskIDs, spec.MonitorTaskID) continue } @@ -262,6 +293,26 @@ func (s *MonitoringService) dispatchMonitorDesktopTasks( return publishableTasks, deferredTaskIDs, nil } +func monitorDesktopTaskDispatchClientIDs(specs []monitorDesktopTaskSpec, targets map[string]monitorDesktopTaskTarget) []uuid.UUID { + if len(specs) == 0 { + return nil + } + result := make([]uuid.UUID, 0, len(specs)*2) + for _, spec := range specs { + if spec.TargetClientID != uuid.Nil { + result = append(result, spec.TargetClientID) + } + if targets == nil { + continue + } + target, ok := targets[normalizeMonitoringPlatformID(spec.PlatformID)] + if ok && target.ClientID != uuid.Nil { + result = append(result, target.ClientID) + } + } + return uniqueUUIDs(result) +} + func monitorDesktopTaskSpecsNeedTargetLookup(specs []monitorDesktopTaskSpec) bool { for _, spec := range specs { if spec.TargetAccountID == uuid.Nil || spec.TargetClientID == uuid.Nil { @@ -279,11 +330,16 @@ func (s *MonitoringService) loadMonitorDesktopTaskTargets( ctx context.Context, tenantID, workspaceID, userID int64, specs []monitorDesktopTaskSpec, + maxClientBacklog ...int, ) (map[string]monitorDesktopTaskTarget, error) { platformIDs := uniqueMonitorDesktopTaskPlatformIDs(specs) if len(platformIDs) == 0 { return nil, nil } + backlogLimit := 0 + if len(maxClientBacklog) > 0 && maxClientBacklog[0] > 0 { + backlogLimit = maxClientBacklog[0] + } rows, err := s.businessPool.Query(ctx, ` SELECT @@ -373,7 +429,11 @@ func (s *MonitoringService) loadMonitorDesktopTaskTargets( if err != nil { return nil, err } - return selectMonitorDesktopTaskTargets(candidates, accountPresence, onlineClients), nil + activeBacklog, err := s.loadMonitorDesktopClientBacklog(ctx, tenantID, workspaceID, clientIDs, backlogLimit) + if err != nil { + return nil, err + } + return selectMonitorDesktopTaskTargets(candidates, accountPresence, onlineClients, activeBacklog, backlogLimit), nil } func uniqueMonitorDesktopTaskPlatformIDs(specs []monitorDesktopTaskSpec) []string { @@ -463,13 +523,58 @@ func (s *MonitoringService) loadOnlineMonitorDesktopClients(ctx context.Context, return result, nil } +func (s *MonitoringService) loadMonitorDesktopClientBacklog( + ctx context.Context, + tenantID, workspaceID int64, + clientIDs []uuid.UUID, + limit int, +) (map[uuid.UUID]int, error) { + result := make(map[uuid.UUID]int) + clientIDs = uniqueUUIDs(clientIDs) + if s == nil || s.businessPool == nil || limit <= 0 || len(clientIDs) == 0 { + return result, nil + } + + rows, err := s.businessPool.Query(ctx, ` + SELECT target_client_id, COUNT(*)::int + FROM desktop_tasks + WHERE tenant_id = $1 + AND workspace_id = $2 + AND target_client_id = ANY($3::uuid[]) + AND kind = 'monitor' + AND status IN ('queued', 'in_progress') + GROUP BY target_client_id + `, tenantID, workspaceID, clientIDs) + if err != nil { + return nil, response.ErrInternal(50117, "desktop_client_backlog_lookup_failed", "failed to resolve monitor desktop client backlog") + } + defer rows.Close() + + for rows.Next() { + var ( + clientID uuid.UUID + count int + ) + if scanErr := rows.Scan(&clientID, &count); scanErr != nil { + return nil, response.ErrInternal(50117, "desktop_client_backlog_lookup_failed", "failed to parse monitor desktop client backlog") + } + result[clientID] = count + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50117, "desktop_client_backlog_lookup_failed", "failed to iterate monitor desktop client backlog") + } + return result, nil +} + func selectMonitorDesktopTaskTargets( candidates []monitorDesktopAccountCandidate, accountPresence map[uuid.UUID][]desktopAccountPresenceClient, onlineClients map[uuid.UUID]bool, + clientBacklog map[uuid.UUID]int, + maxClientBacklog int, ) map[string]monitorDesktopTaskTarget { result := make(map[string]monitorDesktopTaskTarget) - bestByPlatform := make(map[string]monitorDesktopTaskTargetCandidate) + candidatesByPlatform := make(map[string][]monitorDesktopTaskTargetCandidate) for _, candidate := range candidates { platformID := normalizeMonitoringPlatformID(candidate.PlatformID) if platformID == "" || candidate.AccountID == uuid.Nil { @@ -495,9 +600,7 @@ func selectMonitorDesktopTaskTargets( RecordOrder: candidate.RecordOrder, ClientOrder: index, } - if current, ok := bestByPlatform[platformID]; !ok || item.betterThan(current) { - bestByPlatform[platformID] = item - } + candidatesByPlatform[platformID] = append(candidatesByPlatform[platformID], item) } } if candidate.DBClientID == nil || *candidate.DBClientID == uuid.Nil || !onlineClients[*candidate.DBClientID] { @@ -516,16 +619,83 @@ func selectMonitorDesktopTaskTargets( SourceRank: 1, RecordOrder: candidate.RecordOrder, } - if current, ok := bestByPlatform[platformID]; !ok || item.betterThan(current) { - bestByPlatform[platformID] = item + candidatesByPlatform[platformID] = append(candidatesByPlatform[platformID], item) + } + + platformIDs := make([]string, 0, len(candidatesByPlatform)) + for platformID := range candidatesByPlatform { + platformIDs = append(platformIDs, platformID) + } + sort.Strings(platformIDs) + + for _, platformID := range platformIDs { + platformCandidates := candidatesByPlatform[platformID] + sort.SliceStable(platformCandidates, func(i, j int) bool { + return platformCandidates[i].betterThan(platformCandidates[j]) + }) + for _, candidate := range platformCandidates { + if monitorDesktopClientBacklogFull(candidate.Target.ClientID, clientBacklog, maxClientBacklog) { + continue + } + result[platformID] = candidate.Target + if maxClientBacklog > 0 { + clientBacklog[candidate.Target.ClientID]++ + } + break } } - for platformID, candidate := range bestByPlatform { - result[platformID] = candidate.Target - } return result } +func monitorDesktopClientBacklogFull(clientID uuid.UUID, clientBacklog map[uuid.UUID]int, maxClientBacklog int) bool { + return maxClientBacklog > 0 && clientID != uuid.Nil && clientBacklog[clientID] >= maxClientBacklog +} + +func reserveMonitorDesktopClientBacklog(oldClientID, newClientID uuid.UUID, clientBacklog map[uuid.UUID]int, maxClientBacklog int) bool { + if maxClientBacklog <= 0 || newClientID == uuid.Nil || oldClientID == newClientID { + return true + } + if clientBacklog == nil { + return true + } + if monitorDesktopClientBacklogFull(newClientID, clientBacklog, maxClientBacklog) { + return false + } + clientBacklog[newClientID]++ + if oldClientID != uuid.Nil && clientBacklog[oldClientID] > 0 { + clientBacklog[oldClientID]-- + } + return true +} + +func (s *MonitoringService) reserveMonitorDesktopClientBacklogTx( + ctx context.Context, + tx pgx.Tx, + tenantID, workspaceID int64, + oldClientID, newClientID uuid.UUID, + maxClientBacklog int, +) (bool, error) { + if tx == nil || maxClientBacklog <= 0 || newClientID == uuid.Nil || oldClientID == newClientID { + return true, nil + } + if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, monitorDesktopTaskClientSlotLockKey(newClientID)); err != nil { + return false, response.ErrInternal(50123, "desktop_client_backlog_lock_failed", "failed to lock monitor desktop client backlog") + } + var backlog int + if err := tx.QueryRow(ctx, ` + SELECT COUNT(*)::int + FROM desktop_tasks + WHERE tenant_id = $1 + AND workspace_id = $2 + AND target_client_id = $3 + AND kind = 'monitor' + AND status IN ('queued', 'in_progress') + `, tenantID, workspaceID, newClientID).Scan(&backlog); err != nil { + return false, response.ErrInternal(50124, "desktop_client_backlog_check_failed", "failed to check monitor desktop client backlog") + } + return backlog < maxClientBacklog, nil +} + func (candidate monitorDesktopTaskTargetCandidate) betterThan(current monitorDesktopTaskTargetCandidate) bool { if candidate.HealthRank != current.HealthRank { return candidate.HealthRank < current.HealthRank @@ -559,14 +729,17 @@ func (s *MonitoringService) upsertMonitorDesktopTask( tx pgx.Tx, repo repository.DesktopTaskRepository, spec monitorDesktopTaskSpec, + reserveClientBacklog func(oldClientID, newClientID uuid.UUID) bool, + reserveClientBacklogTx func(oldClientID, newClientID uuid.UUID) (bool, error), ) (*repository.DesktopTask, bool, bool, error) { var ( existingDesktopID uuid.UUID + existingTargetClientID uuid.UUID existingStatus string existingLeaseExpiresAt pgtype.Timestamptz existingActiveAttempt pgtype.UUID ) - err := tx.QueryRow(ctx, upsertMonitorDesktopTaskActiveLookupSQL(), spec.MonitorTaskID).Scan(&existingDesktopID, &existingStatus, &existingLeaseExpiresAt, &existingActiveAttempt) + err := tx.QueryRow(ctx, upsertMonitorDesktopTaskActiveLookupSQL(), spec.MonitorTaskID).Scan(&existingDesktopID, &existingTargetClientID, &existingStatus, &existingLeaseExpiresAt, &existingActiveAttempt) switch err { case nil: task, getErr := repo.GetByDesktopID(ctx, existingDesktopID, spec.WorkspaceID) @@ -583,6 +756,18 @@ func (s *MonitoringService) upsertMonitorDesktopTask( if targetAccountID == uuid.Nil { targetAccountID = task.TargetAccountID } + if reserveClientBacklog != nil && !reserveClientBacklog(existingTargetClientID, spec.TargetClientID) { + return nil, false, true, nil + } + if reserveClientBacklogTx != nil { + allowed, reserveErr := reserveClientBacklogTx(existingTargetClientID, spec.TargetClientID) + if reserveErr != nil { + return nil, false, false, reserveErr + } + if !allowed { + return nil, false, true, nil + } + } payloadJSON, payloadErr := marshalMonitorDesktopTaskPayload(spec) if payloadErr != nil { return nil, false, false, payloadErr @@ -658,6 +843,18 @@ func (s *MonitoringService) upsertMonitorDesktopTask( if targetAccountID == uuid.Nil || spec.TargetClientID == uuid.Nil { return nil, false, true, nil } + if reserveClientBacklog != nil && !reserveClientBacklog(uuid.Nil, spec.TargetClientID) { + return nil, false, true, nil + } + if reserveClientBacklogTx != nil { + allowed, reserveErr := reserveClientBacklogTx(uuid.Nil, spec.TargetClientID) + if reserveErr != nil { + return nil, false, false, reserveErr + } + if !allowed { + return nil, false, true, nil + } + } payloadJSON, payloadErr := marshalMonitorDesktopTaskPayload(spec) if payloadErr != nil { return nil, false, false, payloadErr @@ -718,7 +915,7 @@ func (s *MonitoringService) upsertMonitorDesktopTask( func upsertMonitorDesktopTaskActiveLookupSQL() string { return ` - SELECT desktop_id, status, lease_expires_at, active_attempt_id + SELECT desktop_id, target_client_id, status, lease_expires_at, active_attempt_id FROM desktop_tasks WHERE kind = 'monitor' AND monitor_task_id = $1 diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go index 987993c..a29398d 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go @@ -32,6 +32,8 @@ func TestSelectMonitorDesktopTaskTargetsPrefersPresentAccountClient(t *testing.T map[uuid.UUID]bool{ liveClientID: true, }, + nil, + 0, ) assert.Equal(t, monitorDesktopTaskTarget{ @@ -56,6 +58,8 @@ func TestSelectMonitorDesktopTaskTargetsFallsBackToOnlineDBClient(t *testing.T) map[uuid.UUID]bool{ clientID: true, }, + nil, + 0, ) assert.Equal(t, monitorDesktopTaskTarget{ @@ -78,6 +82,8 @@ func TestSelectMonitorDesktopTaskTargetsSkipsOfflineClients(t *testing.T) { }, nil, map[uuid.UUID]bool{}, + nil, + 0, ) assert.Empty(t, targets) @@ -111,6 +117,8 @@ func TestSelectMonitorDesktopTaskTargetsPrefersHealthierOnlineCandidate(t *testi riskClientID: true, liveClientID: true, }, + nil, + 0, ) assert.Equal(t, monitorDesktopTaskTarget{ @@ -144,6 +152,8 @@ func TestSelectMonitorDesktopTaskTargetsPrefersEarlierOnlineClientForSameAccount earlierClientID: true, laterClientID: true, }, + nil, + 0, ) assert.Equal(t, monitorDesktopTaskTarget{ @@ -152,6 +162,119 @@ func TestSelectMonitorDesktopTaskTargetsPrefersEarlierOnlineClientForSameAccount }, targets["doubao"]) } +func TestSelectMonitorDesktopTaskTargetsAppliesClientBackpressure(t *testing.T) { + fullClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000701") + availableClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000702") + fullAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000801") + availableAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000802") + + targets := selectMonitorDesktopTaskTargets( + []monitorDesktopAccountCandidate{ + { + PlatformID: "qwen", + AccountID: fullAccountID, + DBClientID: &fullClientID, + HealthRank: 0, + RecordOrder: 0, + }, + { + PlatformID: "qwen", + AccountID: availableAccountID, + DBClientID: &availableClientID, + HealthRank: 1, + RecordOrder: 1, + }, + }, + nil, + map[uuid.UUID]bool{ + fullClientID: true, + availableClientID: true, + }, + map[uuid.UUID]int{ + fullClientID: 12, + }, + 12, + ) + + assert.Equal(t, monitorDesktopTaskTarget{ + AccountID: availableAccountID, + ClientID: availableClientID, + }, targets["qwen"]) +} + +func TestSelectMonitorDesktopTaskTargetsConsumesBackpressureWithinRun(t *testing.T) { + clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000703") + qwenAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000803") + doubaoAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000804") + + targets := selectMonitorDesktopTaskTargets( + []monitorDesktopAccountCandidate{ + { + PlatformID: "doubao", + AccountID: doubaoAccountID, + DBClientID: &clientID, + HealthRank: 0, + RecordOrder: 0, + }, + { + PlatformID: "qwen", + AccountID: qwenAccountID, + DBClientID: &clientID, + HealthRank: 0, + RecordOrder: 1, + }, + }, + nil, + map[uuid.UUID]bool{ + clientID: true, + }, + map[uuid.UUID]int{ + clientID: 11, + }, + 12, + ) + + assert.Len(t, targets, 1) + assert.Contains(t, targets, "doubao") + assert.NotContains(t, targets, "qwen") +} + +func TestReserveMonitorDesktopClientBacklogAllowsExistingSameClientWithoutConsumingSlot(t *testing.T) { + clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000704") + backlog := map[uuid.UUID]int{ + clientID: 12, + } + + ok := reserveMonitorDesktopClientBacklog(clientID, clientID, backlog, 12) + + assert.True(t, ok) + assert.Equal(t, 12, backlog[clientID]) +} + +func TestReserveMonitorDesktopClientBacklogConsumesCapacityPerTask(t *testing.T) { + clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000705") + backlog := map[uuid.UUID]int{ + clientID: 11, + } + + assert.True(t, reserveMonitorDesktopClientBacklog(uuid.Nil, clientID, backlog, 12)) + assert.False(t, reserveMonitorDesktopClientBacklog(uuid.Nil, clientID, backlog, 12)) + assert.Equal(t, 12, backlog[clientID]) +} + +func TestReserveMonitorDesktopClientBacklogMovesCapacityWhenRetargeting(t *testing.T) { + oldClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000706") + newClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000707") + backlog := map[uuid.UUID]int{ + oldClientID: 4, + newClientID: 11, + } + + assert.True(t, reserveMonitorDesktopClientBacklog(oldClientID, newClientID, backlog, 12)) + assert.Equal(t, 3, backlog[oldClientID]) + assert.Equal(t, 12, backlog[newClientID]) +} + func TestMarshalMonitorDesktopTaskPayloadRequiresQuestionText(t *testing.T) { payload, err := marshalMonitorDesktopTaskPayload(monitorDesktopTaskSpec{ PlatformID: "qwen", 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 ac42410..ef51587 100644 --- a/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql +++ b/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql @@ -93,7 +93,7 @@ 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', '按业务日小批量物化监控采集任务并派发到桌面任务队列。', true, 'interval', 300, 'Asia/Shanghai', 45, 64, '{"max_materialize_per_plan":12,"max_materialize_per_brand":4}'::jsonb), + ('monitoring_daily_task', 'AI 监控每日任务生成', 'monitoring', '按业务日全量创建监控采集事实，按容量限速派发到桌面任务队列。', true, 'interval', 300, 'Asia/Shanghai', 45, 12, '{"max_desktop_client_backlog":12}'::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), 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 index e820b85..bfd9d4b 100644 --- a/server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.up.sql +++ b/server/migrations_ops/20260520143000_harden_scheduler_jobs_for_k3s.up.sql @@ -11,10 +11,11 @@ SET description = '索引驱动的延迟队列派发，仅认领到期内容任 WHERE job_key = 'schedule_dispatch'; UPDATE ops.scheduler_jobs -SET description = '按业务日小批量物化监控采集任务并派发到桌面任务队列。', +SET description = '按业务日全量创建监控采集事实，按容量限速派发到桌面任务队列。', timeout_seconds = 45, - batch_size = 64, - config = config || '{"max_materialize_per_plan":12,"max_materialize_per_brand":4}'::jsonb, + batch_size = 12, + config = (config - 'max_materialize_per_run' - 'max_materialize_per_plan' - 'max_materialize_per_brand') + || '{"max_desktop_client_backlog":12}'::jsonb, updated_at = NOW() WHERE job_key = 'monitoring_daily_task'; diff --git a/server/migrations_ops/20260627161000_rework_monitoring_daily_backpressure.down.sql b/server/migrations_ops/20260627161000_rework_monitoring_daily_backpressure.down.sql new file mode 100644 index 0000000..eeccaff --- /dev/null +++ b/server/migrations_ops/20260627161000_rework_monitoring_daily_backpressure.down.sql @@ -0,0 +1,7 @@ +UPDATE ops.scheduler_jobs +SET description = '按业务日小批量物化监控采集任务并派发到桌面任务队列。', + batch_size = 64, + config = (config - 'max_desktop_client_backlog') + || '{"max_materialize_per_plan":12,"max_materialize_per_brand":4}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_daily_task'; diff --git a/server/migrations_ops/20260627161000_rework_monitoring_daily_backpressure.up.sql b/server/migrations_ops/20260627161000_rework_monitoring_daily_backpressure.up.sql new file mode 100644 index 0000000..0689632 --- /dev/null +++ b/server/migrations_ops/20260627161000_rework_monitoring_daily_backpressure.up.sql @@ -0,0 +1,7 @@ +UPDATE ops.scheduler_jobs +SET description = '按业务日全量创建监控采集事实，按容量限速派发到桌面任务队列。', + batch_size = 12, + config = (config - 'max_materialize_per_run' - 'max_materialize_per_plan' - 'max_materialize_per_brand') + || '{"max_desktop_client_backlog":12}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_daily_task';