From b9b1db6e5df712dfa6beef512b55a6b7052efe61 Mon Sep 17 00:00:00 2001 From: liangxu Date: Sun, 28 Jun 2026 12:26:27 +0800 Subject: [PATCH] feat: implement batch processing for daily monitoring tasks across platforms --- .../app/monitoring_daily_task_worker.go | 153 +++++++++++++++++- .../app/monitoring_daily_task_worker_test.go | 137 +++++++++++++++- .../app/monitoring_phase2_desktop_tasks.go | 2 +- .../monitoring_phase2_desktop_tasks_test.go | 34 ++++ 4 files changed, 317 insertions(+), 9 deletions(-) diff --git a/server/internal/tenant/app/monitoring_daily_task_worker.go b/server/internal/tenant/app/monitoring_daily_task_worker.go index 06afe5f..cd1cb83 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker.go @@ -26,13 +26,14 @@ const ( defaultMonitoringDailyLookahead = 10 * time.Minute defaultMonitoringDailyDesktopRetryCooldown = 15 * time.Minute defaultMonitoringDailyPublishTimeout = 5 * time.Second + defaultMonitoringDailyPlatformBatchSize = 2 defaultMonitoringDailyMaxMaterializePerRun = 0 defaultMonitoringDailyMaxMaterializePerPlan = 0 defaultMonitoringDailyMaxMaterializePerBrand = 0 defaultMonitoringDailyMaxDesktopDispatch = 0 defaultMonitoringDailyMaxDesktopClientBacklog = 12 monitoringDailyPlatformActiveWindow = 24 * time.Hour - monitoringDailyMaterializeStartOffset = 5 * time.Minute + monitoringDailyMaterializeStartOffset = 0 ) type MonitoringDailyTaskWorker struct { @@ -953,7 +954,7 @@ func (s *MonitoringService) createDispatchableDailyMonitoringTasksForBrand( if err != nil { return 0, 0, nil, err } - dueCandidates := selectDueMonitoringDailyCandidates(candidates, existingKeys, now, defaultMonitoringDailyLookahead, limit) + dueCandidates := selectDueMonitoringDailyBatchCandidates(candidates, existingKeys, now, defaultMonitoringDailyLookahead, limit) dueCount := int64(len(dueCandidates)) if len(dueCandidates) == 0 { if err := tx.Commit(ctx); err != nil { @@ -1457,27 +1458,165 @@ func selectDueMonitoringDailyCandidates( now time.Time, lookahead time.Duration, limit int, +) []monitoringDailyTaskCandidate { + return selectDueMonitoringDailyCandidatesWithPlatformBatch(candidates, existing, now, lookahead, limit, 0) +} + +func selectDueMonitoringDailyBatchCandidates( + candidates []monitoringDailyTaskCandidate, + existing map[monitoringDailyTaskKey]struct{}, + now time.Time, + lookahead time.Duration, + limit int, +) []monitoringDailyTaskCandidate { + return selectDueMonitoringDailyCandidatesWithPlatformBatch(candidates, existing, now, lookahead, limit, defaultMonitoringDailyPlatformBatchSize) +} + +func selectDueMonitoringDailyCandidatesWithPlatformBatch( + candidates []monitoringDailyTaskCandidate, + existing map[monitoringDailyTaskKey]struct{}, + now time.Time, + lookahead time.Duration, + limit int, + platformBatchSize int, ) []monitoringDailyTaskCandidate { if len(candidates) == 0 || limit <= 0 { return nil } cutoff := now.Add(lookahead) - result := make([]monitoringDailyTaskCandidate, 0, limit) + due := make([]monitoringDailyTaskCandidate, 0, minMonitoringDailyInt(limit, len(candidates))) + dueByPlatform := make(map[string][]monitoringDailyTaskCandidate) for _, candidate := range candidates { if candidate.MaterializeAt.IsZero() || candidate.MaterializeAt.After(cutoff) { continue } + platformID := normalizeMonitoringPlatformID(candidate.Platform.ID) key := monitoringDailyTaskKey{ QuestionID: candidate.Question.ID, - PlatformID: normalizeMonitoringPlatformID(candidate.Platform.ID), + PlatformID: platformID, } if _, ok := existing[key]; ok { continue } - result = append(result, candidate) - if len(result) >= limit { - return result + due = append(due, candidate) + if platformID != "" { + dueByPlatform[platformID] = append(dueByPlatform[platformID], candidate) + } + } + if platformBatchSize <= 0 || len(dueByPlatform) <= platformBatchSize { + return limitMonitoringDailyCandidates(due, limit) + } + + platformIDs := make([]string, 0, len(dueByPlatform)) + for platformID := range dueByPlatform { + platformIDs = append(platformIDs, platformID) + } + selectedPlatforms := selectMonitoringDailyBatchPlatformIDs(platformIDs, now, platformBatchSize) + return selectMonitoringDailyCandidatesByPlatformQuota(dueByPlatform, selectedPlatforms, limit) +} + +func limitMonitoringDailyCandidates(candidates []monitoringDailyTaskCandidate, limit int) []monitoringDailyTaskCandidate { + if limit <= 0 || len(candidates) == 0 { + return nil + } + if len(candidates) <= limit { + return candidates + } + return candidates[:limit] +} + +func selectMonitoringDailyBatchPlatformIDs(platformIDs []string, now time.Time, limit int) []string { + if len(platformIDs) == 0 || limit <= 0 { + return nil + } + seen := make(map[string]struct{}, len(platformIDs)) + ordered := make([]string, 0, len(platformIDs)) + for _, platformID := range platformIDs { + platformID = normalizeMonitoringPlatformID(platformID) + if platformID == "" { + continue + } + if _, ok := seen[platformID]; ok { + continue + } + seen[platformID] = struct{}{} + ordered = append(ordered, platformID) + } + if len(ordered) == 0 { + return nil + } + + bucket := int64(0) + if defaultMonitoringDailyTaskInterval > 0 { + bucket = now.Unix() / int64(defaultMonitoringDailyTaskInterval/time.Second) + } + businessDate := monitoringBusinessDateText(now) + sort.SliceStable(ordered, func(i, j int) bool { + left := monitoringDailyStableHash("platform_batch_order", businessDate, int64Text(bucket), ordered[i]) + right := monitoringDailyStableHash("platform_batch_order", businessDate, int64Text(bucket), ordered[j]) + if left == right { + leftIndex := monitoringPlatformSortIndex(ordered[i]) + rightIndex := monitoringPlatformSortIndex(ordered[j]) + if leftIndex == rightIndex { + return ordered[i] < ordered[j] + } + return leftIndex < rightIndex + } + return left < right + }) + if len(ordered) > limit { + ordered = ordered[:limit] + } + return ordered +} + +func selectMonitoringDailyCandidatesByPlatformQuota( + dueByPlatform map[string][]monitoringDailyTaskCandidate, + platformIDs []string, + limit int, +) []monitoringDailyTaskCandidate { + if len(dueByPlatform) == 0 || len(platformIDs) == 0 || limit <= 0 { + return nil + } + result := make([]monitoringDailyTaskCandidate, 0, limit) + usedByPlatform := make(map[string]int, len(platformIDs)) + baseQuota := limit / len(platformIDs) + extraQuota := limit % len(platformIDs) + + for index, platformID := range platformIDs { + platformID = normalizeMonitoringPlatformID(platformID) + items := dueByPlatform[platformID] + if len(items) == 0 { + continue + } + quota := baseQuota + if index < extraQuota { + quota++ + } + take := minMonitoringDailyInt(quota, len(items)) + result = append(result, items[:take]...) + usedByPlatform[platformID] = take + } + + for len(result) < limit { + added := false + for _, platformID := range platformIDs { + platformID = normalizeMonitoringPlatformID(platformID) + items := dueByPlatform[platformID] + used := usedByPlatform[platformID] + if used >= len(items) { + continue + } + result = append(result, items[used]) + usedByPlatform[platformID] = used + 1 + added = true + if len(result) >= limit { + return result + } + } + if !added { + break } } return result 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 e8ef704..0d9af31 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker_test.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker_test.go @@ -31,7 +31,7 @@ func TestBuildMonitoringDailyTaskCandidatesMaterializesWholeDayImmediately(t *te candidates := buildMonitoringDailyTaskCandidates(7, 11, businessDay, questions, platforms, 4) require.Len(t, candidates, 4) - dueAt := monitoringCanonicalBusinessDay(businessDay).Add(monitoringDailyMaterializeStartOffset).UTC() + dueAt := monitoringCanonicalBusinessDay(businessDay).UTC() for i, candidate := range candidates { assert.Equal(t, dueAt, candidate.MaterializeAt) assert.Equal(t, candidate.MaterializeAt, candidate.DispatchAfter) @@ -41,6 +41,141 @@ func TestBuildMonitoringDailyTaskCandidatesMaterializesWholeDayImmediately(t *te } } +func TestSelectDueMonitoringDailyCandidatesSplitsBatchAcrossTwoPlatforms(t *testing.T) { + businessDay := time.Date(2026, 6, 28, 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)), + }) + } + candidates := buildMonitoringDailyTaskCandidates( + 7, + 7, + businessDay, + questions, + []monitoringPlatformMetadata{ + {ID: "deepseek"}, + {ID: "qwen"}, + {ID: "doubao"}, + {ID: "kimi"}, + }, + 17*4, + ) + existing := make(map[monitoringDailyTaskKey]struct{}, 17) + for _, candidate := range candidates { + if candidate.Platform.ID == "deepseek" { + existing[monitoringDailyTaskKey{ + QuestionID: candidate.Question.ID, + PlatformID: candidate.Platform.ID, + }] = struct{}{} + } + } + now := monitoringCanonicalBusinessDay(businessDay).Add(10 * time.Hour) + + due := selectDueMonitoringDailyBatchCandidates(candidates, existing, now, defaultMonitoringDailyLookahead, 12) + + require.Len(t, due, 12) + counts := make(map[string]int) + for _, candidate := range due { + assert.NotEqual(t, "deepseek", candidate.Platform.ID) + counts[candidate.Platform.ID]++ + } + require.Len(t, counts, 2) + for _, count := range counts { + assert.Equal(t, 6, count) + } +} + +func TestSelectDueMonitoringDailyCandidatesLimitsSmallRemainderToTwoPlatforms(t *testing.T) { + businessDay := time.Date(2026, 6, 28, 0, 0, 0, 0, monitoringBusinessLocation()) + questions := make([]monitoringConfiguredQuestion, 0, 3) + for id := int64(1); id <= 3; id++ { + questions = append(questions, monitoringConfiguredQuestion{ + ID: id, + QuestionText: "用户问题 " + int64Text(id), + QuestionHash: []byte("q" + int64Text(id)), + }) + } + candidates := buildMonitoringDailyTaskCandidates( + 7, + 7, + businessDay, + questions, + []monitoringPlatformMetadata{ + {ID: "qwen"}, + {ID: "doubao"}, + {ID: "kimi"}, + }, + 9, + ) + now := monitoringCanonicalBusinessDay(businessDay).Add(10 * time.Hour) + + due := selectDueMonitoringDailyBatchCandidates(candidates, nil, now, defaultMonitoringDailyLookahead, 12) + + require.Len(t, due, 6) + counts := make(map[string]int) + for _, candidate := range due { + counts[candidate.Platform.ID]++ + } + require.Len(t, counts, 2) + for _, count := range counts { + assert.Equal(t, 3, count) + } +} + +func TestSelectDueMonitoringDailyCandidatesAllowsSinglePlatformRemainder(t *testing.T) { + businessDay := time.Date(2026, 6, 28, 0, 0, 0, 0, monitoringBusinessLocation()) + questions := make([]monitoringConfiguredQuestion, 0, 9) + for id := int64(1); id <= 9; id++ { + questions = append(questions, monitoringConfiguredQuestion{ + ID: id, + QuestionText: "用户问题 " + int64Text(id), + QuestionHash: []byte("q" + int64Text(id)), + }) + } + candidates := buildMonitoringDailyTaskCandidates( + 7, + 7, + businessDay, + questions, + []monitoringPlatformMetadata{{ID: "qwen"}}, + 9, + ) + now := monitoringCanonicalBusinessDay(businessDay).Add(10 * time.Hour) + + due := selectDueMonitoringDailyBatchCandidates(candidates, nil, now, defaultMonitoringDailyLookahead, 12) + + require.Len(t, due, 9) + for _, candidate := range due { + assert.Equal(t, "qwen", candidate.Platform.ID) + } +} + +func TestSelectDueMonitoringDailyCandidatesUsesMidnightStart(t *testing.T) { + businessDay := time.Date(2026, 6, 28, 0, 0, 0, 0, monitoringBusinessLocation()) + candidates := buildMonitoringDailyTaskCandidates( + 7, + 7, + businessDay, + []monitoringConfiguredQuestion{{ + ID: 1, + QuestionText: "用户问题", + QuestionHash: []byte("q1"), + }}, + []monitoringPlatformMetadata{{ID: "qwen"}}, + 1, + ) + now := monitoringCanonicalBusinessDay(businessDay) + + due := selectDueMonitoringDailyCandidates(candidates, nil, now, defaultMonitoringDailyLookahead, 1) + + require.Len(t, due, 1) + assert.Equal(t, "qwen", due[0].Platform.ID) +} + func TestNormalizeMonitoringDailyRunOptionsDefaultsDoNotThrottleByPlanOrBrand(t *testing.T) { got := normalizeMonitoringDailyRunOptions() diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go index b19ec6d..3ced8cd 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go @@ -682,7 +682,7 @@ func selectMonitorDesktopTaskTargetsWithOptions( continue } result[platformID] = candidate.Target - if options.MaxClientBacklog > 0 { + if options.MaxClientBacklog > 0 && !options.RequireIdleClient { clientBacklog[candidate.Target.ClientID]++ } break 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 007428b..dd395cb 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go @@ -267,6 +267,40 @@ func TestSelectMonitorDesktopTaskTargetsRequireIdleSkipsAnyActiveBacklog(t *test assert.Empty(t, targets) } +func TestSelectMonitorDesktopTaskTargetsRequireIdleKeepsMultiplePlatformsOnSameIdleClient(t *testing.T) { + clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000709") + qwenAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000809") + doubaoAccountID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000810") + + targets := selectMonitorDesktopTaskTargetsWithOptions( + []monitorDesktopAccountCandidate{ + { + PlatformID: "doubao", + AccountID: doubaoAccountID, + DBClientID: &clientID, + }, + { + PlatformID: "qwen", + AccountID: qwenAccountID, + DBClientID: &clientID, + }, + }, + nil, + map[uuid.UUID]bool{ + clientID: true, + }, + map[uuid.UUID]int{}, + monitorDesktopTaskTargetOptions{ + MaxClientBacklog: 12, + RequireIdleClient: true, + }, + ) + + require.Len(t, targets, 2) + assert.Equal(t, clientID, targets["doubao"].ClientID) + assert.Equal(t, clientID, targets["qwen"].ClientID) +} + func TestReserveMonitorDesktopClientBacklogAllowsExistingSameClientWithoutConsumingSlot(t *testing.T) { clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000704") backlog := map[uuid.UUID]int{