Fix monitor collect-now queue fairness

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)

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
 }
 

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) {

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":

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)
-}

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

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()