diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 0873b35..d121ce3 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -27,6 +27,8 @@ import ( const desktopPublishMaxAttempts = 3 +var errDesktopTaskLeaseDeferred = errors.New("desktop task lease deferred") + type DesktopTaskService struct { pool *pgxpool.Pool monitoringPool *pgxpool.Pool @@ -247,7 +249,11 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt if taskID == nil { return &LeaseDesktopTaskResponse{}, nil } - return nil, s.classifyLeaseError(ctx, client, taskID) + classifiedErr := s.classifyLeaseError(ctx, client, taskID) + if errors.Is(classifiedErr, errDesktopTaskLeaseDeferred) { + return &LeaseDesktopTaskResponse{}, nil + } + return nil, classifiedErr } fields := []zap.Field{ zap.String("client_id", client.ID.String()), @@ -966,6 +972,9 @@ func (s *DesktopTaskService) Complete(ctx context.Context, client *repository.De taskRepo := repository.NewDesktopTaskRepository(tx) previousTask, _ := taskRepo.GetByDesktopID(ctx, desktopID, client.WorkspaceID) previousAttemptID := activeAttemptID(previousTask) + if previousTask != nil { + finalStatus = normalizeDesktopTaskCompletionStatusForKind(previousTask.Kind, finalStatus) + } task, err := taskRepo.Complete(ctx, desktopID, HashDesktopClientToken(strings.TrimSpace(req.LeaseToken)), finalStatus, resultJSON, errorJSON) if err != nil { @@ -1558,10 +1567,19 @@ func (s *DesktopTaskService) classifyLeaseError(ctx context.Context, client *rep return response.ErrForbidden(40381, "desktop_task_not_target_client", "desktop task is assigned to another desktop client") } + return classifyDesktopTaskLeaseState(task) +} + +func classifyDesktopTaskLeaseState(task *repository.DesktopTask) error { + if task == nil { + return response.ErrNotFound(40482, "desktop_task_not_found", "desktop task not found") + } + if task.Kind == "monitor" && task.Status == "queued" { + return errDesktopTaskLeaseDeferred + } if task.Status == "in_progress" { return response.ErrConflict(40989, "desktop_task_already_in_progress", "desktop task is already in progress") } - return response.ErrConflict(40991, "desktop_task_not_queued", "desktop task is not queued") } @@ -1646,13 +1664,22 @@ func normalizeDesktopTaskCompletionStatus(status string) string { } } -func normalizeDesktopTaskViewStatus(kind string, status string) string { - if strings.TrimSpace(kind) == "publish" { - return normalizeDesktopTaskTerminalStatus(status) +func normalizeDesktopTaskCompletionStatusForKind(kind string, status string) string { + if strings.TrimSpace(kind) == "monitor" && strings.TrimSpace(status) == "unknown" { + return "failed" } return strings.TrimSpace(status) } +func normalizeDesktopTaskViewStatus(kind string, status string) string { + switch strings.TrimSpace(kind) { + case "monitor", "publish": + return normalizeDesktopTaskTerminalStatus(status) + default: + return strings.TrimSpace(status) + } +} + type desktopTaskRecoveryMode string const ( diff --git a/server/internal/tenant/app/desktop_task_service_test.go b/server/internal/tenant/app/desktop_task_service_test.go index 5e45069..20f08bd 100644 --- a/server/internal/tenant/app/desktop_task_service_test.go +++ b/server/internal/tenant/app/desktop_task_service_test.go @@ -1,11 +1,13 @@ package app import ( + "errors" "regexp" "strings" "testing" "github.com/geo-platform/tenant-api/internal/shared/response" + "github.com/geo-platform/tenant-api/internal/tenant/repository" ) func TestDesktopTaskRecoverySelectQueryMatchesRecoveredLeaseScanOrder(t *testing.T) { @@ -138,19 +140,34 @@ func TestResolvePublishRecoveryOutcome(t *testing.T) { } } -func TestNormalizeDesktopTaskCompletionStatusPreservesUnknown(t *testing.T) { +func TestNormalizeDesktopTaskCompletionStatusForKindFailsMonitorUnknown(t *testing.T) { t.Parallel() - if got := normalizeDesktopTaskCompletionStatus("unknown"); got != "unknown" { - t.Fatalf("normalizeDesktopTaskCompletionStatus(unknown) = %q, want unknown", got) + if got := normalizeDesktopTaskCompletionStatusForKind("monitor", "unknown"); got != "failed" { + t.Fatalf("monitor completion status = %q, want failed", got) + } + if got := normalizeDesktopTaskCompletionStatusForKind("publish", "unknown"); got != "unknown" { + t.Fatalf("publish completion status = %q, want unknown", got) } } -func TestNormalizeDesktopTaskViewStatusOnlyMapsPublishUnknownToFailed(t *testing.T) { +func TestClassifyLeaseErrorDefersQueuedMonitorTask(t *testing.T) { t.Parallel() - if got := normalizeDesktopTaskViewStatus("monitor", "unknown"); got != "unknown" { - t.Fatalf("monitor unknown view status = %q, want unknown", got) + task := &repository.DesktopTask{ + Kind: "monitor", + Status: "queued", + } + if got := classifyDesktopTaskLeaseState(task); !errors.Is(got, errDesktopTaskLeaseDeferred) { + t.Fatalf("classifyDesktopTaskLeaseState() = %v, want deferred", got) + } +} + +func TestNormalizeDesktopTaskViewStatusMapsMonitorAndPublishUnknownToFailed(t *testing.T) { + t.Parallel() + + if got := normalizeDesktopTaskViewStatus("monitor", "unknown"); got != "failed" { + t.Fatalf("monitor unknown view status = %q, want failed", got) } if got := normalizeDesktopTaskViewStatus("publish", "unknown"); got != "failed" { t.Fatalf("publish unknown view status = %q, want failed", got) diff --git a/server/internal/tenant/app/monitoring_callback_service.go b/server/internal/tenant/app/monitoring_callback_service.go index 4b9edc3..7ea2262 100644 --- a/server/internal/tenant/app/monitoring_callback_service.go +++ b/server/internal/tenant/app/monitoring_callback_service.go @@ -945,6 +945,7 @@ func (s *MonitoringCallbackService) skipTaskForInstallation( } skipReason := normalizeSkipReason(req.SkipReason) + skipOutcome := monitoringSkipOutcomeForReason(skipReason) tx, err := s.monitoringPool.Begin(ctx) if err != nil { @@ -954,14 +955,14 @@ func (s *MonitoringCallbackService) skipTaskForInstallation( if _, err := tx.Exec(ctx, ` UPDATE monitoring_collect_tasks - SET status = 'skipped', + SET status = $2, callback_received_at = NOW(), - completed_at = $2, - skip_reason = $3, - error_message = $4, + completed_at = $3, + skip_reason = $4, + error_message = $5, updated_at = NOW() WHERE id = $1 - `, task.ID, completedAt, skipReason, nullableString(req.ErrorMessage)); err != nil { + `, task.ID, skipOutcome.TaskStatus, completedAt, nullableStringPtr(skipOutcome.StoredSkipReason), nullableString(req.ErrorMessage)); err != nil { return nil, response.ErrInternal(50041, "task_update_failed", "failed to update monitoring task skip status") } @@ -978,12 +979,34 @@ func (s *MonitoringCallbackService) skipTaskForInstallation( completedAtText := completedAt.Format(time.RFC3339) return &MonitoringSkipTaskResponse{ TaskID: task.ID, - TaskStatus: "skipped", + TaskStatus: skipOutcome.TaskStatus, SkipReason: skipReason, CompletedAt: &completedAtText, }, nil } +type monitoringSkipOutcome struct { + TaskStatus string + StoredSkipReason string +} + +func monitoringSkipOutcomeForReason(skipReason string) monitoringSkipOutcome { + switch strings.TrimSpace(skipReason) { + case "runtime_unknown": + return monitoringSkipOutcome{TaskStatus: "failed"} + default: + return monitoringSkipOutcome{TaskStatus: "skipped", StoredSkipReason: strings.TrimSpace(skipReason)} + } +} + +func nullableStringPtr(value string) *string { + trimmed := strings.TrimSpace(value) + if trimmed == "" { + return nil + } + return &trimmed +} + func (s *MonitoringCallbackService) CancelTaskForDesktopClient( ctx context.Context, client *repository.DesktopClient, @@ -1504,7 +1527,7 @@ func (s *MonitoringCallbackService) loadActiveDesktopMonitorExecutionLease( FROM desktop_tasks WHERE kind = 'monitor' AND monitor_task_id = $1 - AND status IN ('queued', 'in_progress', 'unknown') + AND status IN ('queued', 'in_progress') ORDER BY created_at DESC, id DESC LIMIT 1 `, monitorTaskID).Scan( @@ -2349,16 +2372,7 @@ func (s *MonitoringCallbackService) completeLinkedDesktopMonitorTask( workspaceID int64 activeAttemptID pgtype.UUID ) - if err := tx.QueryRow(ctx, ` - SELECT desktop_id, workspace_id, active_attempt_id - FROM desktop_tasks - WHERE kind = 'monitor' - AND monitor_task_id = $1 - AND status IN ('in_progress', 'unknown') - ORDER BY created_at DESC, id DESC - LIMIT 1 - FOR UPDATE - `, task.ID).Scan(&desktopID, &workspaceID, &activeAttemptID); err != nil { + if err := tx.QueryRow(ctx, completeLinkedDesktopMonitorTaskLookupSQL(), task.ID).Scan(&desktopID, &workspaceID, &activeAttemptID); err != nil { if err == pgx.ErrNoRows { return nil } @@ -2434,6 +2448,19 @@ func (s *MonitoringCallbackService) completeLinkedDesktopMonitorTask( return nil } +func completeLinkedDesktopMonitorTaskLookupSQL() string { + return ` + SELECT desktop_id, workspace_id, active_attempt_id + FROM desktop_tasks + WHERE kind = 'monitor' + AND monitor_task_id = $1 + AND status = 'in_progress' + ORDER BY created_at DESC, id DESC + LIMIT 1 + FOR UPDATE + ` +} + func (s *MonitoringCallbackService) replaceCitationFacts(ctx context.Context, tx pgx.Tx, tenantID, runID int64) error { if _, err := tx.Exec(ctx, ` DELETE FROM monitoring_citation_facts diff --git a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go index 8231731..c257568 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks.go @@ -612,16 +612,7 @@ func (s *MonitoringService) upsertMonitorDesktopTask( existingLeaseExpiresAt pgtype.Timestamptz existingActiveAttempt pgtype.UUID ) - err := tx.QueryRow(ctx, ` - SELECT desktop_id, status, lease_expires_at, active_attempt_id - FROM desktop_tasks - WHERE kind = 'monitor' - AND monitor_task_id = $1 - AND status IN ('queued', 'in_progress', 'unknown') - ORDER BY created_at DESC, id DESC - LIMIT 1 - FOR UPDATE - `, spec.MonitorTaskID).Scan(&existingDesktopID, &existingStatus, &existingLeaseExpiresAt, &existingActiveAttempt) + err := tx.QueryRow(ctx, upsertMonitorDesktopTaskActiveLookupSQL(), spec.MonitorTaskID).Scan(&existingDesktopID, &existingStatus, &existingLeaseExpiresAt, &existingActiveAttempt) switch err { case nil: task, getErr := repo.GetByDesktopID(ctx, existingDesktopID, spec.WorkspaceID) @@ -771,6 +762,19 @@ func (s *MonitoringService) upsertMonitorDesktopTask( return task, true, false, nil } +func upsertMonitorDesktopTaskActiveLookupSQL() string { + return ` + SELECT desktop_id, status, lease_expires_at, active_attempt_id + FROM desktop_tasks + WHERE kind = 'monitor' + AND monitor_task_id = $1 + AND status IN ('queued', 'in_progress') + ORDER BY created_at DESC, id DESC + LIMIT 1 + FOR UPDATE + ` +} + func (s *MonitoringService) fallbackMonitorDesktopTasksToLegacy(ctx context.Context, taskIDs []int64) error { if s == nil || s.monitoringPool == nil || len(taskIDs) == 0 { return nil 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 34ab2f2..707aa4b 100644 --- a/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go +++ b/server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go @@ -179,6 +179,34 @@ func TestMarshalMonitorDesktopTaskPayloadUsesQuestionTextNotFallbackTitle(t *tes assert.Equal(t, "通义千问 · 幽灵门轨道长度定制", decoded["title"]) } +func TestUpsertMonitorDesktopTaskIgnoresUnknownAsActive(t *testing.T) { + query := strings.Join(strings.Fields(upsertMonitorDesktopTaskActiveLookupSQL()), " ") + + assert.Contains(t, query, "status IN ('queued', 'in_progress')") + assert.NotContains(t, query, "'unknown'") +} + +func TestCompleteLinkedDesktopMonitorTaskOnlyFinalizesInProgress(t *testing.T) { + query := strings.Join(strings.Fields(completeLinkedDesktopMonitorTaskLookupSQL()), " ") + + assert.Contains(t, query, "status = 'in_progress'") + assert.NotContains(t, query, "'unknown'") +} + +func TestMonitoringSkipOutcomeForRuntimeUnknownFailsTask(t *testing.T) { + outcome := monitoringSkipOutcomeForReason("runtime_unknown") + + assert.Equal(t, "failed", outcome.TaskStatus) + assert.Empty(t, outcome.StoredSkipReason) +} + +func TestMonitoringSkipOutcomeKeepsOrdinarySkip(t *testing.T) { + outcome := monitoringSkipOutcomeForReason("stale_business_date") + + assert.Equal(t, "skipped", outcome.TaskStatus) + assert.Equal(t, "stale_business_date", outcome.StoredSkipReason) +} + func TestMonitorDesktopTaskPlatformSlotBusySQLThrottlesPerClientPlatformInProgress(t *testing.T) { query := strings.Join(strings.Fields(monitorDesktopTaskPlatformSlotBusySQL()), " ") diff --git a/server/migrations/20260422120000_add_monitor_phase2_desktop_task_fields.up.sql b/server/migrations/20260422120000_add_monitor_phase2_desktop_task_fields.up.sql index bc03e0a..dc669ca 100644 --- a/server/migrations/20260422120000_add_monitor_phase2_desktop_task_fields.up.sql +++ b/server/migrations/20260422120000_add_monitor_phase2_desktop_task_fields.up.sql @@ -54,4 +54,4 @@ CREATE UNIQUE INDEX IF NOT EXISTS uq_desktop_tasks_monitor_active ON desktop_tasks (monitor_task_id) WHERE kind = 'monitor' AND monitor_task_id IS NOT NULL - AND status IN ('queued', 'in_progress', 'unknown'); + AND status IN ('queued', 'in_progress'); diff --git a/server/migrations/20260622152000_fail_monitor_unknown_tasks.down.sql b/server/migrations/20260622152000_fail_monitor_unknown_tasks.down.sql new file mode 100644 index 0000000..ed79068 --- /dev/null +++ b/server/migrations/20260622152000_fail_monitor_unknown_tasks.down.sql @@ -0,0 +1,7 @@ +DROP INDEX IF EXISTS uq_desktop_tasks_monitor_active; + +CREATE UNIQUE INDEX IF NOT EXISTS uq_desktop_tasks_monitor_active + ON desktop_tasks (monitor_task_id) + WHERE kind = 'monitor' + AND monitor_task_id IS NOT NULL + AND status IN ('queued', 'in_progress', 'unknown'); diff --git a/server/migrations/20260622152000_fail_monitor_unknown_tasks.up.sql b/server/migrations/20260622152000_fail_monitor_unknown_tasks.up.sql new file mode 100644 index 0000000..6e5d011 --- /dev/null +++ b/server/migrations/20260622152000_fail_monitor_unknown_tasks.up.sql @@ -0,0 +1,13 @@ +UPDATE desktop_tasks +SET status = 'failed', + updated_at = NOW() +WHERE kind = 'monitor' + AND status = 'unknown'; + +DROP INDEX IF EXISTS uq_desktop_tasks_monitor_active; + +CREATE UNIQUE INDEX IF NOT EXISTS uq_desktop_tasks_monitor_active + ON desktop_tasks (monitor_task_id) + WHERE kind = 'monitor' + AND monitor_task_id IS NOT NULL + AND status IN ('queued', 'in_progress');