diff --git a/server/internal/tenant/app/monitoring_collect_outbox_worker.go b/server/internal/tenant/app/monitoring_collect_outbox_worker.go index 9c816c1..a3b2dd4 100644 --- a/server/internal/tenant/app/monitoring_collect_outbox_worker.go +++ b/server/internal/tenant/app/monitoring_collect_outbox_worker.go @@ -616,13 +616,7 @@ func (s *MonitoringService) loadCollectNowRequestTaskStats( return stats, nil } - if err := s.businessPool.QueryRow(ctx, ` - SELECT COUNT(*)::bigint - FROM desktop_tasks d - WHERE d.kind = 'monitor' - AND d.status = 'in_progress' - AND d.monitor_task_id = ANY($1::bigint[]) - `, monitorTaskIDs).Scan( + if err := s.businessPool.QueryRow(ctx, collectNowActiveDesktopTaskCountSQL(), monitorTaskIDs).Scan( &stats.ActiveDesktopTaskCount, ); err != nil { return stats, response.ErrInternal(50041, "query_failed", "failed to inspect collect-now desktop task progress") @@ -631,6 +625,16 @@ func (s *MonitoringService) loadCollectNowRequestTaskStats( return stats, nil } +func collectNowActiveDesktopTaskCountSQL() string { + return ` + SELECT COUNT(*)::bigint + FROM desktop_tasks d + WHERE d.kind = 'monitor' + AND d.status IN ('queued', 'in_progress') + AND d.monitor_task_id = ANY($1::bigint[]) + ` +} + func (s *MonitoringService) loadCollectNowMonitorTaskIDs( ctx context.Context, request monitoringCollectRequestReconcileRow, @@ -806,6 +810,9 @@ func deriveCollectNowRequestStatus( if now.Before(request.TTLExpiresAt) { return "", "", false } + if stats.Leased > 0 || stats.Received > 0 || stats.ActiveDesktopTaskCount > 0 { + return "", "", false + } if !request.FirstDispatchedAt.Valid { return "timed_out", "立即采集请求超时,5 分钟内未能派发首条任务。", request.Status != "timed_out" } diff --git a/server/internal/tenant/app/monitoring_collect_outbox_worker_test.go b/server/internal/tenant/app/monitoring_collect_outbox_worker_test.go new file mode 100644 index 0000000..36ada78 --- /dev/null +++ b/server/internal/tenant/app/monitoring_collect_outbox_worker_test.go @@ -0,0 +1,40 @@ +package app + +import ( + "database/sql" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestDeriveCollectNowRequestStatusKeepsRunningWhenDesktopTaskActiveAfterTTL(t *testing.T) { + now := time.Date(2026, 6, 27, 1, 37, 30, 0, time.UTC) + request := monitoringCollectRequestReconcileRow{ + Status: "running", + TTLExpiresAt: now.Add(-30 * time.Second), + FirstDispatchedAt: sql.NullTime{ + Time: now.Add(-5 * time.Minute), + Valid: true, + }, + } + stats := monitoringCollectRequestTaskStats{ + Total: 30, + Terminal: 6, + Completed: 6, + ActiveDesktopTaskCount: 1, + } + + nextStatus, nextMessage, shouldFinalize := deriveCollectNowRequestStatus(request, stats, now) + + assert.Empty(t, nextStatus) + assert.Empty(t, nextMessage) + assert.False(t, shouldFinalize) +} + +func TestLoadCollectNowRequestTaskStatsTreatsQueuedDesktopTasksAsActive(t *testing.T) { + query := strings.Join(strings.Fields(collectNowActiveDesktopTaskCountSQL()), " ") + + assert.Contains(t, query, "d.status IN ('queued', 'in_progress')") +}