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

fix monitor desktop unknown recovery

Browse Source
This commit is contained in:
liangxu
2026-06-22 15:38:06 +08:00
parent 365a41597e
commit 12681105f2
8 changed files with 162 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/desktop_task_service.go
+32 -5
View File
@@ -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 (
server/internal/tenant/app/desktop_task_service_test.go
+23 -6
View File
@@ -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)
server/internal/tenant/app/monitoring_callback_service.go
+44 -17
View File
@@ -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
server/internal/tenant/app/monitoring_phase2_desktop_tasks.go
+14 -10
View File
@@ -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
server/internal/tenant/app/monitoring_phase2_desktop_tasks_test.go
+28
View File
@@ -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()), " ")
server/migrations/20260422120000_add_monitor_phase2_desktop_task_fields.up.sql
+1 -1
View File
@@ -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');
server/migrations/20260622152000_fail_monitor_unknown_tasks.down.sql
+7
View File
@@ -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');
server/migrations/20260622152000_fail_monitor_unknown_tasks.up.sql
+13
View File
@@ -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');