root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: throttle desktop monitoring dispatch
Desktop Client Build / Resolve Build Metadata (push) Successful in 27s
Details
Backend CI / Backend (push) Successful in 15m57s
Details
Desktop Client Build / Build Desktop Client (push) Successful in 22m57s
Details
Desktop Client Build / Publish Client Artifacts to NAS (push) Successful in 27s
Details

Browse Source
This commit is contained in:
liangxu
2026-06-22 12:21:59 +08:00
parent 54c2f92e02
commit 8c6789dca6
23 changed files with 591 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/monitoring_service.go
+31 -7
View File
@@ -361,6 +361,14 @@ type monitoringConfiguredQuestion struct {
QuestionHash []byte
}
func monitoringQuestionTextSnapshot(question monitoringConfiguredQuestion) (string, error) {
questionText := strings.TrimSpace(question.QuestionText)
if questionText == "" {
return "", response.ErrInternal(50041, "missing_question_text_snapshot", "configured monitoring question text is empty")
}
return questionText, nil
}
var defaultMonitoringPlatforms = []monitoringPlatformMetadata{
{ID: "yuanbao", Name: "混元 / 元宝"},
{ID: "kimi", Name: "Kimi"},
@@ -1018,8 +1026,10 @@ func (s *MonitoringService) CollectNow(
phase2DeferredTasks := make([]*repository.DesktopTask, 0)
if options.Preempt && phase2DispatchReady {
excludedMonitorTaskIDs := make([]int64, 0, len(phase2TaskSpecs))
excludedPlatformIDs := make([]string, 0, len(phase2TaskSpecs))
for _, spec := range phase2TaskSpecs {
excludedMonitorTaskIDs = append(excludedMonitorTaskIDs, spec.MonitorTaskID)
excludedPlatformIDs = append(excludedPlatformIDs, spec.PlatformID)
}
phase2TargetClientIDs := desktopTaskClientIDs(phase2PublishedTasks)
if len(phase2TargetClientIDs) == 0 {
@@ -1031,7 +1041,7 @@ func (s *MonitoringService) CollectNow(
return nil, deferErr
}
phase2DeferredTasks = append(phase2DeferredTasks, deferredTasks...)
interruptTargets, interruptErr := s.requestPhase2MonitorInterrupts(ctx, phase2TargetClientID, excludedMonitorTaskIDs, interruptGeneration)
interruptTargets, interruptErr := s.requestPhase2MonitorInterrupts(ctx, phase2TargetClientID, excludedMonitorTaskIDs, excludedPlatformIDs, interruptGeneration)
if interruptErr != nil {
return nil, interruptErr
}
@@ -1534,6 +1544,10 @@ func (s *MonitoringService) loadConfiguredQuestions(ctx context.Context, tenantI
if scanErr := rows.Scan(&item.ID, &item.KeywordID, &item.QuestionText); scanErr != nil {
return nil, response.ErrInternal(50041, "scan_failed", "failed to parse configured monitoring questions")
}
item.QuestionText = strings.TrimSpace(item.QuestionText)
if item.QuestionText == "" {
return nil, response.ErrInternal(50041, "missing_question_text_snapshot", "configured monitoring question text is empty")
}
item.QuestionHash = seededQuestionHash(item.QuestionText)
items = append(items, item)
}
@@ -1631,6 +1645,10 @@ func (s *MonitoringService) syncMonitoringQuestionSnapshots(ctx context.Context,
}
for _, question := range questions {
questionText, textErr := monitoringQuestionTextSnapshot(question)
if textErr != nil {
return textErr
}
if _, err := tx.Exec(ctx, `
UPDATE monitoring_question_config_snapshots
SET superseded_at = COALESCE(superseded_at, NOW()),
@@ -1661,7 +1679,7 @@ func (s *MonitoringService) syncMonitoringQuestionSnapshots(ctx context.Context,
deleted_at = NULL,
projected_at = NOW(),
updated_at = NOW()
`, tenantID, brandID, question.ID, question.KeywordID, question.QuestionHash, question.QuestionText); err != nil {
`, tenantID, brandID, question.ID, question.KeywordID, question.QuestionHash, questionText); err != nil {
return response.ErrInternal(50041, "snapshot_upsert_failed", "failed to persist monitoring question snapshot")
}
}
@@ -1692,6 +1710,10 @@ func (s *MonitoringService) ensureCollectNowTasks(
interruptTaskIDs := make([]int64, 0)
for _, question := range questions {
questionText, textErr := monitoringQuestionTextSnapshot(question)
if textErr != nil {
return 0, 0, 0, 0, nil, textErr
}
for _, platform := range platforms {
platformID := normalizeMonitoringPlatformID(platform.ID)
if platformID == "" {
@@ -1704,6 +1726,7 @@ func (s *MonitoringService) ensureCollectNowTasks(
tag, err := tx.Exec(ctx, `
UPDATE monitoring_collect_tasks
SET question_hash = $7,
question_text_snapshot = $12,
status = 'pending',
planned_at = NOW(),
trigger_source = 'manual',
@@ -1737,7 +1760,7 @@ func (s *MonitoringService) ensureCollectNowTasks(
AND run_mode = $6
AND business_date = $8::date
AND status IN ('pending', 'expired', 'failed', 'completed', 'skipped')
`, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration, executionOwner)
`, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration, executionOwner, questionText)
if err != nil {
return 0, 0, 0, 0, nil, response.ErrInternal(50041, "update_failed", "failed to refresh monitoring tasks")
}
@@ -1746,6 +1769,7 @@ func (s *MonitoringService) ensureCollectNowTasks(
rows, err := tx.Query(ctx, `
UPDATE monitoring_collect_tasks
SET question_hash = $7,
question_text_snapshot = $11,
trigger_source = 'manual',
dispatch_priority = 5000,
dispatch_lane = 'high',
@@ -1770,7 +1794,7 @@ func (s *MonitoringService) ensureCollectNowTasks(
AND status = 'leased'
AND COALESCE(execution_owner, 'legacy') = 'legacy'
RETURNING id
`, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration)
`, tenantID, brandID, question.ID, platformID, monitoringCollectorType, runMode, question.QuestionHash, dateText, platformTargetClientID, interruptGeneration, questionText)
if err != nil {
return 0, 0, 0, 0, nil, response.ErrInternal(50041, "update_failed", "failed to promote leased monitoring tasks")
}
@@ -1790,13 +1814,13 @@ func (s *MonitoringService) ensureCollectNowTasks(
tag, err = tx.Exec(ctx, `
INSERT INTO monitoring_collect_tasks (
tenant_id, brand_id, question_id, question_hash, ai_platform_id,
tenant_id, brand_id, question_id, question_hash, question_text_snapshot, ai_platform_id,
collector_type, trigger_source, run_mode, business_date, planned_at, status,
dispatch_priority, dispatch_lane, target_client_id, dispatch_after,
interrupt_generation, ingest_shard_key, execution_owner
)
VALUES (
$1, $2, $3, $4, $5,
$1, $2, $3, $4, $13, $5,
$6, 'manual', $7, $8::date, NOW(), 'pending',
5000, 'high', $9, NOW(),
$10, $11, $12
@@ -1805,7 +1829,7 @@ func (s *MonitoringService) ensureCollectNowTasks(
tenant_id, brand_id, question_id, ai_platform_id, collector_type, run_mode, business_date
)
DO NOTHING
`, tenantID, brandID, question.ID, question.QuestionHash, platformID, monitoringCollectorType, runMode, dateText, platformTargetClientID, interruptGeneration, fmt.Sprintf("%d:%d:%s", tenantID, brandID, dateText), executionOwner)
`, tenantID, brandID, question.ID, question.QuestionHash, platformID, monitoringCollectorType, runMode, dateText, platformTargetClientID, interruptGeneration, fmt.Sprintf("%d:%d:%s", tenantID, brandID, dateText), executionOwner, questionText)
if err != nil {
return 0, 0, 0, 0, nil, response.ErrInternal(50041, "insert_failed", "failed to create monitoring tasks")
}