From 082f91a6a900c7d04ed4cb824fa6dd139d7f8191 Mon Sep 17 00:00:00 2001 From: liangxu Date: Mon, 22 Jun 2026 21:47:26 +0800 Subject: [PATCH] Fix monitor collect-now queue fairness --- apps/desktop-client/package.json | 2 +- .../src/main/monitor-scheduler.test.ts | 135 ++++++++++++++++++ .../src/main/monitor-scheduler.ts | 14 +- .../src/main/runtime-controller.ts | 54 ++++--- .../tenant/app/desktop_task_service_test.go | 1 + .../app/monitoring_collect_outbox_worker.go | 23 --- .../app/monitoring_daily_task_worker_test.go | 4 + .../app/monitoring_phase2_desktop_tasks.go | 117 --------------- .../monitoring_phase2_desktop_tasks_test.go | 14 -- .../internal/tenant/app/monitoring_service.go | 92 ++---------- .../app/monitoring_service_date_test.go | 24 ++++ 11 files changed, 222 insertions(+), 258 deletions(-) create mode 100644 apps/desktop-client/src/main/monitor-scheduler.test.ts diff --git a/apps/desktop-client/package.json b/apps/desktop-client/package.json index f13d0ea..b6d3b0f 100644 --- a/apps/desktop-client/package.json +++ b/apps/desktop-client/package.json @@ -1,6 +1,6 @@ { "name": "@geo/desktop-client", - "version": "0.1.2", + "version": "0.1.3", "private": true, "description": "省心推客户端 — 连接和管理您的媒体账号，轻松发布和监控内容表现。", "author": { diff --git a/apps/desktop-client/src/main/monitor-scheduler.test.ts b/apps/desktop-client/src/main/monitor-scheduler.test.ts new file mode 100644 index 0000000..0289317 --- /dev/null +++ b/apps/desktop-client/src/main/monitor-scheduler.test.ts @@ -0,0 +1,135 @@ +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +vi.mock('electron/main', () => ({ + app: { + getPath: () => join(tmpdir(), 'geo-rankly-monitor-scheduler-test'), + }, +})) + +import { + enqueueMonitorLeaseTask, + initMonitorScheduler, + noteMonitorTaskActivated, + noteMonitorTaskCompleted, + selectNextMonitorTask, +} from './monitor-scheduler' + +describe('monitor scheduler', () => { + beforeEach(() => { + vi.useFakeTimers() + vi.setSystemTime(new Date('2026-06-22T00:00:00.000Z')) + initMonitorScheduler() + }) + + afterEach(() => { + vi.useRealTimers() + initMonitorScheduler() + }) + + it('does not let high-priority monitor tasks bypass same-platform cooldown', () => { + enqueueMonitorLeaseTask({ + taskId: 'first', + clientId: 'client-1', + platform: 'doubao', + routing: 'rabbitmq-primary', + questionKey: 'q:first', + }) + expect( + selectNextMonitorTask({ + maxConcurrency: 2, + activePlatforms: new Set(), + activeQuestionKeys: new Set(), + activeCount: 0, + })?.taskId, + ).toBe('first') + noteMonitorTaskActivated('first') + noteMonitorTaskCompleted('first') + + enqueueMonitorLeaseTask({ + taskId: 'urgent', + clientId: 'client-1', + platform: 'doubao', + routing: 'rabbitmq-primary', + priority: 5000, + lane: 'high', + questionKey: 'q:urgent', + }) + + expect( + selectNextMonitorTask({ + maxConcurrency: 2, + activePlatforms: new Set(), + activeQuestionKeys: new Set(), + activeCount: 0, + }), + ).toBeNull() + + vi.advanceTimersByTime(5_001) + + expect( + selectNextMonitorTask({ + maxConcurrency: 2, + activePlatforms: new Set(), + activeQuestionKeys: new Set(), + activeCount: 0, + })?.taskId, + ).toBe('urgent') + }) + + it('does not let high-priority monitor tasks bypass an active same-platform task', () => { + enqueueMonitorLeaseTask({ + taskId: 'urgent-doubao', + clientId: 'client-1', + platform: 'doubao', + routing: 'rabbitmq-primary', + priority: 5000, + lane: 'high', + questionKey: 'q:urgent', + }) + + expect( + selectNextMonitorTask({ + maxConcurrency: 2, + activePlatforms: new Set(['doubao']), + activeQuestionKeys: new Set(), + activeCount: 1, + }), + ).toBeNull() + }) + + it('prioritizes collect-now high lane before older normal tasks when platform is available', () => { + enqueueMonitorLeaseTask({ + taskId: 'old-normal', + clientId: 'client-1', + platform: 'qwen', + routing: 'rabbitmq-primary', + priority: 100, + lane: 'normal', + questionKey: 'q:normal', + }) + + vi.advanceTimersByTime(1_000) + + enqueueMonitorLeaseTask({ + taskId: 'collect-now', + clientId: 'client-1', + platform: 'doubao', + routing: 'rabbitmq-primary', + priority: 5000, + lane: 'high', + questionKey: 'q:collect-now', + }) + + expect( + selectNextMonitorTask({ + maxConcurrency: 2, + activePlatforms: new Set(), + activeQuestionKeys: new Set(), + activeCount: 0, + })?.taskId, + ).toBe('collect-now') + }) +}) diff --git a/apps/desktop-client/src/main/monitor-scheduler.ts b/apps/desktop-client/src/main/monitor-scheduler.ts index fd38b59..408e96a 100644 --- a/apps/desktop-client/src/main/monitor-scheduler.ts +++ b/apps/desktop-client/src/main/monitor-scheduler.ts @@ -74,7 +74,7 @@ interface MonitorTaskMetadata { questionText: string | null } -const schedulerStateVersion = 5 +const schedulerStateVersion = 6 const schedulerQuestionCooldownMs = 45_000 const schedulerPlatformCooldownMs = 5_000 const schedulerRestartRecoveryDelayMs = 90_000 @@ -492,10 +492,10 @@ export function selectNextMonitorTask( continue } - const bypassCooldowns = candidate.lane === 'high' + const bypassQuestionCooldown = candidate.lane === 'high' const platformCooldown = draft.platformCooldowns[candidate.platform] ?? 0 - if (!bypassCooldowns && platformCooldown > now) { + if (platformCooldown > now) { continue } @@ -508,7 +508,7 @@ export function selectNextMonitorTask( continue } const questionCooldown = draft.questionCooldowns[candidate.questionKey] ?? 0 - if (!bypassCooldowns && questionCooldown > now) { + if (!bypassQuestionCooldown && questionCooldown > now) { continue } } @@ -553,12 +553,12 @@ export function getMonitorSchedulerSnapshot(): MonitorSchedulerSnapshot { function leaseMissBackoffMs(leaseMisses: number): number { if (leaseMisses <= 1) { - return 20_000 + return 15_000 } if (leaseMisses === 2) { - return 60_000 + return 30_000 } - return 3 * 60_000 + return 60_000 } function metadataFromEvent(event: DesktopTaskEventMessage): MonitorTaskMetadata { diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index 459e0dc..afa907f 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -63,13 +63,11 @@ import { } from './lease-manager' import { enqueueMonitorLeaseTask, - enqueueMonitorTaskFromEvent, getMonitorSchedulerSnapshot, initMonitorScheduler, noteMonitorTaskActivated, noteMonitorTaskCanceled, noteMonitorTaskCompleted, - noteMonitorTaskLeaseMiss, noteMonitorTaskLeased, selectNextMonitorTask, } from './monitor-scheduler' @@ -149,6 +147,7 @@ type MonitorExecutionPhase = const heartbeatIntervalMs = 15_000 const pullIntervalMs = 60_000 const publishPullIntervalMs = 30_000 +const monitorPullFallbackIntervalMs = 10_000 const accountSyncIntervalMs = 30_000 const pumpWatchdogIntervalMs = 30_000 const leaseExtendIntervalMs = 60_000 @@ -256,6 +255,7 @@ interface RuntimeState { activeExecutions: Map leaseInFlightKinds: Set<'publish' | 'monitor'> publishFallbackBackoffUntil: number + monitorFallbackBackoffUntil: number heartbeatTimer: ReturnType | null pullTimer: ReturnType | null accountSyncTimer: ReturnType | null @@ -289,6 +289,7 @@ const state: RuntimeState = { activeExecutions: new Map(), leaseInFlightKinds: new Set<'publish' | 'monitor'>(), publishFallbackBackoffUntil: 0, + monitorFallbackBackoffUntil: 0, heartbeatTimer: null, pullTimer: null, accountSyncTimer: null, @@ -806,6 +807,7 @@ function clearRuntimeState(): void { state.onlineClientCount = null state.lastError = null state.publishFallbackBackoffUntil = 0 + state.monitorFallbackBackoffUntil = 0 setSchedulerQueueDepth(localQueueDepth()) } @@ -826,11 +828,11 @@ function schedulePullLoop(): void { clearInterval(state.pullTimer) } - setSchedulerNextPull(Date.now() + publishPullIntervalMs) + setSchedulerNextPull(Date.now() + monitorPullFallbackIntervalMs) state.pullTimer = setInterval(() => { - setSchedulerNextPull(Date.now() + publishPullIntervalMs) + setSchedulerNextPull(Date.now() + monitorPullFallbackIntervalMs) pumpExecutionLoop() - }, publishPullIntervalMs) + }, monitorPullFallbackIntervalMs) } function scheduleAccountSyncLoop(): void { @@ -986,7 +988,7 @@ function handleTaskEvent(event: DesktopTaskEventMessage): void { if (event.kind === 'monitor') { if (event.type === 'task_available') { - enqueueMonitorTaskFromEvent(event, 'rabbitmq-primary') + state.monitorFallbackBackoffUntil = 0 } if ( event.type === 'task_completed' || @@ -1027,13 +1029,8 @@ async function handleMonitoringDispatchSignal(event: DesktopTaskEventMessage): P ) } - if (event.task_id.trim()) { - enqueueMonitorTaskFromEvent(event, 'rabbitmq-primary') - pumpExecutionLoop() - return - } - - await pullNextTask('monitor') + state.monitorFallbackBackoffUntil = 0 + pumpExecutionLoop() } async function handleMonitoringTaskControl(event: DesktopTaskEventMessage): Promise { @@ -1889,11 +1886,21 @@ function pumpExecutionLoop(): void { activeCount: activeMonitorCount(), }) if (selected) { - void leaseSpecificTask(selected, 'monitor') + if (selected.source === 'monitoring_lease') { + void leaseSpecificTask(selected, 'monitor') + } else { + noteMonitorTaskCanceled(selected.taskId) + void pullNextTask('monitor') + } return } } + if (shouldPullMonitorFallback()) { + void pullNextTask('monitor') + return + } + return } @@ -1917,7 +1924,7 @@ async function leaseSpecificTask( const leased = await leaseDesktopTask({ task_id: request.taskId }) if (!leased.task) { if (expectedKind === 'monitor') { - noteMonitorTaskLeaseMiss(request.taskId) + noteMonitorTaskCanceled(request.taskId) } else { notePublishTaskLeaseMiss(request.taskId) } @@ -1938,7 +1945,7 @@ async function leaseSpecificTask( state.tasks.set(request.taskId, existing) } } else { - noteMonitorTaskLeaseMiss(request.taskId) + noteMonitorTaskCanceled(request.taskId) } } else { notePublishTaskLeaseMiss(request.taskId) @@ -1973,14 +1980,17 @@ async function pullNextTask(kind: 'publish' | 'monitor'): Promise { noteSchedulerPull() if (leased.task) { + state.monitorFallbackBackoffUntil = 0 state.leaseInFlightKinds.delete('monitor') await executeLeasedTask(leased, 'rabbitmq-primary') return } + state.monitorFallbackBackoffUntil = Date.now() + monitorPullFallbackIntervalMs } catch (error) { state.lastPullAt = Date.now() state.lastPullStatus = 'failed' state.lastError = errorMessage(error) + state.monitorFallbackBackoffUntil = Date.now() + monitorPullFallbackIntervalMs noteTransportPull(false) setSchedulerError(state.lastError) } finally { @@ -2052,10 +2062,12 @@ async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise { noteSchedulerPull() enqueueLeasedMonitoringTasks(leased.tasks, routing) + state.monitorFallbackBackoffUntil = leased.tasks.length > 0 ? 0 : Date.now() + monitorPullFallbackIntervalMs } catch (error) { state.lastPullAt = Date.now() state.lastPullStatus = 'failed' state.lastError = errorMessage(error) + state.monitorFallbackBackoffUntil = Date.now() + monitorPullFallbackIntervalMs noteTransportPull(false) setSchedulerError(state.lastError) @@ -2176,7 +2188,7 @@ async function executeLeasedMonitoringTask( ): Promise { const taskRecord = state.tasks.get(taskId) if (!taskRecord || taskRecord.kind !== 'monitor' || !taskRecord.leaseToken) { - noteMonitorTaskLeaseMiss(taskId) + noteMonitorTaskCanceled(taskId) return } @@ -3609,6 +3621,14 @@ function shouldPullPublishFallback(): boolean { ) } +function shouldPullMonitorFallback(): boolean { + return ( + Date.now() >= state.monitorFallbackBackoffUntil && + !hasPublishBacklog() && + canStartAnotherMonitorExecution() + ) +} + function canStartAnotherMonitorExecution(): boolean { if (hasPublishBacklog() || isLeaseInFlight('publish') || isLeaseInFlight('monitor')) { return false diff --git a/server/internal/tenant/app/desktop_task_service_test.go b/server/internal/tenant/app/desktop_task_service_test.go index acb50f1..2c5bb45 100644 --- a/server/internal/tenant/app/desktop_task_service_test.go +++ b/server/internal/tenant/app/desktop_task_service_test.go @@ -256,6 +256,7 @@ func TestLeaseNextQueuedMonitorTaskSQLSerializesPerClientPlatform(t *testing.T) "recent.updated_at >= now() - interval '2 seconds'", "recent.status IN ('failed', 'unknown', 'aborted')", "recent.updated_at >= now() - interval '30 seconds'", + "ORDER BY dt.lane_weight DESC, dt.priority DESC, COALESCE(dt.enqueued_at, dt.created_at) ASC", } { if !strings.Contains(normalized, fragment) { t.Fatalf("monitor lease query missing %q: %s", fragment, normalized) diff --git a/server/internal/tenant/app/monitoring_collect_outbox_worker.go b/server/internal/tenant/app/monitoring_collect_outbox_worker.go index 452fd11..9c816c1 100644 --- a/server/internal/tenant/app/monitoring_collect_outbox_worker.go +++ b/server/internal/tenant/app/monitoring_collect_outbox_worker.go @@ -871,7 +871,6 @@ func (s *MonitoringService) enqueueCollectNowOutbox( targetClientIDs []uuid.UUID, affectedTaskCount int64, interruptGeneration int, - interruptTaskIDs []int64, ) error { if tx == nil { return nil @@ -896,28 +895,6 @@ func (s *MonitoringService) enqueueCollectNowOutbox( } } - seenTaskIDs := make(map[int64]struct{}, len(interruptTaskIDs)) - for _, taskID := range interruptTaskIDs { - if taskID <= 0 { - continue - } - if _, exists := seenTaskIDs[taskID]; exists { - continue - } - seenTaskIDs[taskID] = struct{}{} - for _, targetClientID := range targetClientIDs { - if _, err := tx.Exec(ctx, ` - INSERT INTO monitoring_collect_dispatch_outbox ( - request_id, workspace_id, target_client_id, event_kind, task_id, - lane, priority, interrupt_generation, reason - ) - VALUES ($1, $2, $3, 'interrupt_requested', $4, 'high', 5000, $5, 'collect_now_preempt') - `, requestID, workspaceID, targetClientID, taskID, interruptGeneration); err != nil { - return response.ErrInternal(50041, "outbox_insert_failed", "failed to enqueue collect-now interrupt signal") - } - } - } - return nil } 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 0384e19..22dc6cc 100644 --- a/server/internal/tenant/app/monitoring_daily_task_worker_test.go +++ b/server/internal/tenant/app/monitoring_daily_task_worker_test.go @@ -504,6 +504,8 @@ type captureDailyMonitorTaskTx struct { execCalled bool execSQL string execArgs []any + execSQLs []string + execArgses [][]any commandTag pgconn.CommandTag } @@ -524,6 +526,8 @@ func (tx *captureDailyMonitorTaskTx) Exec(_ context.Context, sql string, argumen tx.execCalled = true tx.execSQL = sql tx.execArgs = arguments + tx.execSQLs = append(tx.execSQLs, sql) + tx.execArgses = append(tx.execArgses, arguments) return tx.commandTag, nil } func (tx *captureDailyMonitorTaskTx) Query(context.Context, string, ...any) (pgx.Rows, error) { diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go index 9f9b9b7..7e4e0b4 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go @@ -12,7 +12,6 @@ import ( "github.com/jackc/pgx/v5/pgtype" "go.uber.org/zap" - "github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq" "github.com/geo-platform/tenant-api/internal/shared/response" "github.com/geo-platform/tenant-api/internal/shared/stream" "github.com/geo-platform/tenant-api/internal/tenant/repository" @@ -66,14 +65,6 @@ type monitorDesktopTaskTargetCandidate struct { ClientOrder int } -type monitorDesktopInterruptTarget struct { - TaskID string - MonitorTaskID int64 - TargetClientID string - WorkspaceID int64 - InterruptGeneration int -} - func monitorDesktopTaskSpecIDs(specs []monitorDesktopTaskSpec) []int64 { if len(specs) == 0 { return nil @@ -747,84 +738,6 @@ func (s *MonitoringService) fallbackMonitorDesktopTasksToLegacy(ctx context.Cont return nil } -func (s *MonitoringService) requestPhase2MonitorInterrupts( - ctx context.Context, - targetClientID uuid.UUID, - excludedMonitorTaskIDs []int64, - excludedPlatformIDs []string, - interruptGeneration int, -) ([]monitorDesktopInterruptTarget, error) { - if s == nil || s.businessPool == nil { - return nil, nil - } - - queryArgs := []any{targetClientID, interruptGeneration} - nextParam := 3 - query := ` - UPDATE desktop_tasks - SET control_flags = ( - CASE - WHEN control_flags IS NULL OR jsonb_typeof(control_flags) <> 'object' - THEN '{}'::jsonb - ELSE control_flags - END - ) || jsonb_build_object( - 'interrupt_requested', true, - 'interrupt_generation', $2::integer, - 'interrupt_reason', 'collect_now_preempt' - ), - interrupt_generation = GREATEST(interrupt_generation, $2::integer), - interrupted_at = COALESCE(interrupted_at, NOW()), - interrupt_reason = 'collect_now_preempt', - updated_at = NOW() - WHERE kind = 'monitor' - AND target_client_id = $1 - AND status = 'in_progress' - AND lane IN ('normal', 'normal_boosted', 'retry') - AND monitor_task_id IS NOT NULL - ` - if len(excludedMonitorTaskIDs) > 0 { - query += fmt.Sprintf(` AND NOT (monitor_task_id = ANY($%d::bigint[]))`, nextParam) - queryArgs = append(queryArgs, excludedMonitorTaskIDs) - nextParam++ - } - excludedPlatformIDs = reconcileEnabledMonitoringPlatforms(excludedPlatformIDs) - if len(excludedPlatformIDs) > 0 { - query += fmt.Sprintf(` AND NOT (platform_id = ANY($%d::text[]))`, nextParam) - queryArgs = append(queryArgs, excludedPlatformIDs) - } - query += ` - RETURNING desktop_id::text, monitor_task_id, target_client_id::text, workspace_id, interrupt_generation - ` - - rows, err := s.businessPool.Query(ctx, query, queryArgs...) - if err != nil { - if s.logger != nil { - s.logger.Error("phase2 monitor interrupt query failed", - zap.Error(err), - zap.String("target_client_id", targetClientID.String()), - zap.Int("interrupt_generation", interruptGeneration), - zap.Int("excluded_monitor_task_count", len(excludedMonitorTaskIDs)), - ) - } - return nil, response.ErrInternal(50121, "desktop_task_interrupt_failed", "failed to mark active phase2 monitor desktop tasks for interrupt") - } - defer rows.Close() - - result := make([]monitorDesktopInterruptTarget, 0) - for rows.Next() { - var item monitorDesktopInterruptTarget - if scanErr := rows.Scan(&item.TaskID, &item.MonitorTaskID, &item.TargetClientID, &item.WorkspaceID, &item.InterruptGeneration); scanErr != nil { - return nil, response.ErrInternal(50121, "desktop_task_interrupt_failed", "failed to parse active phase2 monitor interrupt targets") - } - result = append(result, item) - } - if err := rows.Err(); err != nil { - return nil, response.ErrInternal(50121, "desktop_task_interrupt_failed", "failed to iterate active phase2 monitor interrupt targets") - } - return result, nil -} - func (s *MonitoringService) deferQueuedPhase2MonitorTasks( ctx context.Context, targetClientID uuid.UUID, @@ -953,36 +866,6 @@ func publishMonitorDesktopTaskAvailable( return nil } -func publishPhase2MonitorControlEvent( - ctx context.Context, - client *rabbitmq.Client, - logger *zap.Logger, - target monitorDesktopInterruptTarget, -) { - if client == nil || strings.TrimSpace(target.TargetClientID) == "" || strings.TrimSpace(target.TaskID) == "" { - return - } - if err := publishDesktopDispatchEvent(ctx, client, logger, stream.DesktopDispatchEvent{ - Type: "task_control", - TaskID: target.TaskID, - WorkspaceID: target.WorkspaceID, - TargetClientID: target.TargetClientID, - Status: "in_progress", - Kind: "monitor", - Lane: "high", - InterruptGeneration: target.InterruptGeneration, - Control: "interrupt_requested", - Reason: "collect_now_preempt", - UpdatedAt: time.Now().UTC(), - }); err != nil && logger != nil { - logger.Warn("phase2 monitor control event publish failed", - zap.String("task_id", target.TaskID), - zap.Int64("monitor_task_id", target.MonitorTaskID), - zap.Error(err), - ) - } -} - func monitoringLaneWeight(lane string) int { switch strings.TrimSpace(lane) { case "high": 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 dd6ffc5..2578ddc 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go @@ -206,17 +206,3 @@ func TestMonitoringSkipOutcomeKeepsOrdinarySkip(t *testing.T) { assert.Equal(t, "skipped", outcome.TaskStatus) assert.Equal(t, "stale_business_date", outcome.StoredSkipReason) } - -func TestRequestPhase2MonitorInterruptsAcceptsExcludedPlatforms(t *testing.T) { - clientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000701") - targets, err := (&MonitoringService{}).requestPhase2MonitorInterrupts( - t.Context(), - clientID, - []int64{101, 102}, - []string{"doubao", "qwen"}, - 7, - ) - - require.NoError(t, err) - assert.Empty(t, targets) -} diff --git a/server/internal/tenant/app/monitoring_service.go b/server/internal/tenant/app/monitoring_service.go index 2dd7f1d..44734b4 100644 --- a/server/internal/tenant/app/monitoring_service.go +++ b/server/internal/tenant/app/monitoring_service.go @@ -882,7 +882,7 @@ func (s *MonitoringService) CollectNow( return nil, err } - refreshedCount, createdCount, leasedCount, completedCount, interruptTaskIDs, err := s.ensureCollectNowTasks( + refreshedCount, createdCount, leasedCount, completedCount, err := s.ensureCollectNowTasks( ctx, tx, actor.TenantID, @@ -969,7 +969,7 @@ func (s *MonitoringService) CollectNow( $12, $13, $14, $15, $16 ) - `, requestID, actor.TenantID, workspaceID, brand.ID, nullableInt64(keywordID), actor.UserID, *clientID, scopeHash, requestScopeJSON, affectedTaskCount, affectedTaskCount, abortedQueuedCount, int64(len(interruptTaskIDs)), interruptGeneration, message, ttlExpiresAt); err != nil { + `, requestID, actor.TenantID, workspaceID, brand.ID, nullableInt64(keywordID), actor.UserID, *clientID, scopeHash, requestScopeJSON, affectedTaskCount, affectedTaskCount, abortedQueuedCount, int64(0), interruptGeneration, message, ttlExpiresAt); err != nil { return nil, response.ErrInternal(50041, "request_create_failed", "failed to persist monitoring collect-now request") } @@ -981,12 +981,6 @@ func (s *MonitoringService) CollectNow( dispatchTargetClientIDs, affectedTaskCount, interruptGeneration, - func() []int64 { - if !options.Preempt { - return nil - } - return collectNowInterruptTaskIDs(interruptTaskIDs) - }(), ); err != nil { return nil, err } @@ -1022,14 +1016,11 @@ func (s *MonitoringService) CollectNow( } } - phase2InterruptTargets := make([]monitorDesktopInterruptTarget, 0) 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 { @@ -1041,28 +1032,9 @@ func (s *MonitoringService) CollectNow( return nil, deferErr } phase2DeferredTasks = append(phase2DeferredTasks, deferredTasks...) - interruptTargets, interruptErr := s.requestPhase2MonitorInterrupts(ctx, phase2TargetClientID, excludedMonitorTaskIDs, excludedPlatformIDs, interruptGeneration) - if interruptErr != nil { - return nil, interruptErr - } - phase2InterruptTargets = append(phase2InterruptTargets, interruptTargets...) } if len(phase2DeferredTasks) > 0 { abortedQueuedCount += int64(len(phase2DeferredTasks)) - } - if len(phase2InterruptTargets) > 0 { - // Keep collect-now request counters in sync after per-platform dispatch - // fan-out chooses more than one desktop client. - if _, updateErr := s.monitoringPool.Exec(ctx, ` - UPDATE monitoring_collect_requests - SET aborted_queued_count = aborted_queued_count + $2, - interrupt_requested_count = interrupt_requested_count + $3, - updated_at = NOW() - WHERE request_id = $1 - `, requestID, len(phase2DeferredTasks), len(phase2InterruptTargets)); updateErr != nil { - return nil, response.ErrInternal(50041, "request_update_failed", "failed to persist phase2 preempt counts") - } - } else if len(phase2DeferredTasks) > 0 { if _, updateErr := s.monitoringPool.Exec(ctx, ` UPDATE monitoring_collect_requests SET aborted_queued_count = aborted_queued_count + $2, @@ -1107,12 +1079,8 @@ func (s *MonitoringService) CollectNow( ) } } - for _, target := range phase2InterruptTargets { - publishPhase2MonitorControlEvent(context.Background(), s.rabbitMQ, s.logger, target) - } hasEffectiveSnapshot := leasedCount > 0 || completedCount > 0 - totalInterruptRequestedCount := int64(len(interruptTaskIDs) + len(phase2InterruptTargets)) responseData := &MonitoringCollectNowResponse{ CollectionMode: quota.CollectionMode, RefreshedTaskCount: refreshedCount, @@ -1125,7 +1093,7 @@ func (s *MonitoringService) CollectNow( AffectedTaskCount: affectedTaskCount, PromotedTaskCount: affectedTaskCount, AbortedQueuedCount: abortedQueuedCount, - InterruptRequestedCount: totalInterruptRequestedCount, + InterruptRequestedCount: 0, InterruptGeneration: interruptGeneration, TTLExpiresAt: ttlExpiresAt.Format(time.RFC3339), Message: message, @@ -1698,7 +1666,7 @@ func (s *MonitoringService) ensureCollectNowTasks( targetClientID uuid.UUID, interruptGeneration int, executionOwner string, -) (int64, int64, int64, int64, []int64, error) { +) (int64, int64, int64, int64, error) { const runMode = "plugin_standard" if strings.TrimSpace(executionOwner) == "" { executionOwner = "legacy" @@ -1707,12 +1675,11 @@ func (s *MonitoringService) ensureCollectNowTasks( dateText := businessDate.Format("2006-01-02") var refreshedCount int64 var createdCount int64 - interruptTaskIDs := make([]int64, 0) for _, question := range questions { questionText, textErr := monitoringQuestionTextSnapshot(question) if textErr != nil { - return 0, 0, 0, 0, nil, textErr + return 0, 0, 0, 0, textErr } for _, platform := range platforms { platformID := normalizeMonitoringPlatformID(platform.ID) @@ -1762,11 +1729,11 @@ func (s *MonitoringService) ensureCollectNowTasks( AND status IN ('pending', 'expired', 'failed', 'completed', 'skipped') `, 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") + return 0, 0, 0, 0, response.ErrInternal(50041, "update_failed", "failed to refresh monitoring tasks") } refreshedCount += tag.RowsAffected() - rows, err := tx.Query(ctx, ` + tag, err = tx.Exec(ctx, ` UPDATE monitoring_collect_tasks SET question_hash = $7, question_text_snapshot = $11, @@ -1793,24 +1760,10 @@ func (s *MonitoringService) ensureCollectNowTasks( AND business_date = $8::date AND status = 'leased' AND COALESCE(execution_owner, 'legacy') = 'legacy' - RETURNING id `, 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") + return 0, 0, 0, 0, response.ErrInternal(50041, "update_failed", "failed to promote leased monitoring tasks") } - for rows.Next() { - var taskID int64 - if scanErr := rows.Scan(&taskID); scanErr != nil { - rows.Close() - return 0, 0, 0, 0, nil, response.ErrInternal(50041, "scan_failed", "failed to parse promoted leased monitoring task") - } - interruptTaskIDs = append(interruptTaskIDs, taskID) - } - if err := rows.Err(); err != nil { - rows.Close() - return 0, 0, 0, 0, nil, response.ErrInternal(50041, "scan_failed", "failed to iterate promoted leased monitoring tasks") - } - rows.Close() tag, err = tx.Exec(ctx, ` INSERT INTO monitoring_collect_tasks ( @@ -1831,7 +1784,7 @@ func (s *MonitoringService) ensureCollectNowTasks( DO NOTHING `, 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") + return 0, 0, 0, 0, response.ErrInternal(50041, "insert_failed", "failed to create monitoring tasks") } createdCount += tag.RowsAffected() } @@ -1847,13 +1800,13 @@ func (s *MonitoringService) ensureCollectNowTasks( WHERE tenant_id = $1 AND brand_id = $2 AND collector_type = $3 - AND business_date = $4::date - AND question_id = ANY($5) + AND business_date = $4::date + AND question_id = ANY($5) `, tenantID, brandID, monitoringCollectorType, dateText, configuredQuestionIDs(questions)).Scan(&leasedCount, &completedCount); err != nil { - return 0, 0, 0, 0, nil, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring tasks") + return 0, 0, 0, 0, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring tasks") } - return refreshedCount, createdCount, leasedCount, completedCount, interruptTaskIDs, nil + return refreshedCount, createdCount, leasedCount, completedCount, nil } func (s *MonitoringService) deferQueuedNormalTasks( @@ -2255,25 +2208,6 @@ func uuidStrings(values []uuid.UUID) []string { return result } -func collectNowInterruptTaskIDs(taskIDs []int64) []int64 { - if len(taskIDs) == 0 { - return nil - } - seen := make(map[int64]struct{}, len(taskIDs)) - result := make([]int64, 0, len(taskIDs)) - for _, taskID := range taskIDs { - if taskID <= 0 { - continue - } - if _, ok := seen[taskID]; ok { - continue - } - seen[taskID] = struct{}{} - result = append(result, taskID) - } - return result -} - func desktopTaskClientIDs(tasks []*repository.DesktopTask) []uuid.UUID { if len(tasks) == 0 { return nil diff --git a/server/internal/tenant/app/monitoring_service_date_test.go b/server/internal/tenant/app/monitoring_service_date_test.go index ecb9349..0c08465 100644 --- a/server/internal/tenant/app/monitoring_service_date_test.go +++ b/server/internal/tenant/app/monitoring_service_date_test.go @@ -5,6 +5,7 @@ import ( "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgconn" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -153,6 +154,29 @@ func TestCollectNowDispatchTargetClientIDsDeduplicatesByPlatformOrder(t *testing assert.Equal(t, []uuid.UUID{sharedClientID, qwenClientID}, targets) } +func TestEnqueueCollectNowOutboxDoesNotInterruptActiveTasks(t *testing.T) { + ctx := t.Context() + targetClientID := mustParseDailyMonitoringUUID(t, "00000000-0000-0000-0000-000000000503") + tx := &captureDailyMonitorTaskTx{commandTag: pgconn.NewCommandTag("INSERT 0 1")} + + err := (&MonitoringService{}).enqueueCollectNowOutbox( + ctx, + tx, + uuid.New(), + 11, + []uuid.UUID{targetClientID}, + 3, + 9, + ) + + require.NoError(t, err) + require.Len(t, tx.execSQLs, 1) + normalized := normalizeSQLWhitespace(tx.execSQLs[0]) + assert.Contains(t, normalized, "'dispatch_high'") + assert.NotContains(t, normalized, "'interrupt_requested'") + assert.NotContains(t, normalized, "task_control") +} + func TestMonitoringPlatformAuthorizationStatus(t *testing.T) { clientID := uuid.New()