From d458dd74771b88e5d0cabcd5168dfb18fbb4578c Mon Sep 17 00:00:00 2001 From: liangxu Date: Tue, 23 Jun 2026 01:03:35 +0800 Subject: [PATCH] fix: fail expired monitor desktop leases --- server/cmd/scheduler/main.go | 4 +- .../monitoring_lease_recovery_worker.go | 40 +- .../tenant/app/desktop_task_service.go | 382 +++++++++++++++++- .../tenant/app/desktop_task_service_test.go | 91 +++++ ...monitor_task_lease_recovery_index.down.sql | 1 + ...d_monitor_task_lease_recovery_index.up.sql | 6 + ...00_tune_monitoring_lease_recovery.down.sql | 5 + ...3000_tune_monitoring_lease_recovery.up.sql | 6 + 8 files changed, 514 insertions(+), 21 deletions(-) create mode 100644 server/migrations/20260623103000_add_monitor_task_lease_recovery_index.down.sql create mode 100644 server/migrations/20260623103000_add_monitor_task_lease_recovery_index.up.sql create mode 100644 server/migrations_ops/20260623103000_tune_monitoring_lease_recovery.down.sql create mode 100644 server/migrations_ops/20260623103000_tune_monitoring_lease_recovery.up.sql diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go index b67b8d8..5ed6da0 100644 --- a/server/cmd/scheduler/main.go +++ b/server/cmd/scheduler/main.go @@ -114,10 +114,10 @@ func main() { desktopTaskService := tenantapp.NewDesktopTaskService(app.DB, app.MonitoringDB, app.RabbitMQ, app.Logger). WithCache(app.Cache). WithRedis(app.Redis) + publishLeaseRecoveryWorker := internalscheduler.NewPublishLeaseRecoveryWorker(desktopTaskService, app.Logger) monitoringDailyTaskWorker := tenantapp.NewMonitoringDailyTaskWorker(monitoringService, monitoringCallbackService, app.Logger) monitoringResultRecoveryWorker := internalscheduler.NewMonitoringResultRecoveryWorker(app.MonitoringDB, monitoringCallbackService, app.Logger) - monitoringLeaseRecoveryWorker := internalscheduler.NewMonitoringLeaseRecoveryWorker(app.MonitoringDB, app.Logger) - publishLeaseRecoveryWorker := internalscheduler.NewPublishLeaseRecoveryWorker(desktopTaskService, app.Logger) + monitoringLeaseRecoveryWorker := internalscheduler.NewMonitoringLeaseRecoveryWorker(app.MonitoringDB, desktopTaskService, app.Logger) monitoringReceivedInspectionWorker := internalscheduler.NewMonitoringReceivedInspectionWorker(app.MonitoringDB, app.Logger) generationStateCheckWorker := generateworker.NewGenerationTaskStateCheckWorker(app.DB, app.Logger, app.Config().Generation). WithConfigProvider(app.ConfigStore). diff --git a/server/internal/scheduler/monitoring_lease_recovery_worker.go b/server/internal/scheduler/monitoring_lease_recovery_worker.go index 1103b13..fd6e6bd 100644 --- a/server/internal/scheduler/monitoring_lease_recovery_worker.go +++ b/server/internal/scheduler/monitoring_lease_recovery_worker.go @@ -11,22 +11,24 @@ import ( ) const ( - defaultMonitoringLeaseRecoveryInterval = 30 * time.Minute + defaultMonitoringLeaseRecoveryInterval = 1 * time.Minute defaultMonitoringLeaseRecoveryTimeout = 30 * time.Second defaultMonitoringLeaseRecoveryLimit = 1000 ) type MonitoringLeaseRecoveryWorker struct { monitoringPool *pgxpool.Pool + desktopTasks *tenantapp.DesktopTaskService logger *zap.Logger interval time.Duration timeout time.Duration limit int } -func NewMonitoringLeaseRecoveryWorker(monitoringPool *pgxpool.Pool, logger *zap.Logger) *MonitoringLeaseRecoveryWorker { +func NewMonitoringLeaseRecoveryWorker(monitoringPool *pgxpool.Pool, desktopTasks *tenantapp.DesktopTaskService, logger *zap.Logger) *MonitoringLeaseRecoveryWorker { return &MonitoringLeaseRecoveryWorker{ monitoringPool: monitoringPool, + desktopTasks: desktopTasks, logger: logger, interval: defaultMonitoringLeaseRecoveryInterval, timeout: defaultMonitoringLeaseRecoveryTimeout, @@ -89,8 +91,23 @@ func (w *MonitoringLeaseRecoveryWorker) runOnce(parent context.Context) { return } - if expiredCount > 0 && w.logger != nil { - w.logger.Info("monitoring lease recovery completed", zap.Int64("expired_task_count", expiredCount)) + var monitorResult tenantapp.MonitorLeaseRecoveryResult + if w.desktopTasks != nil { + var recoverErr error + monitorResult, recoverErr = w.desktopTasks.RecoverExpiredMonitorTasks(ctx, w.limit) + if recoverErr != nil { + if w.logger != nil { + w.logger.Warn("monitor desktop lease recovery failed", zap.Error(recoverErr)) + } + return + } + } + + if (expiredCount > 0 || monitorResult.Failed > 0) && w.logger != nil { + w.logger.Info("monitoring lease recovery completed", + zap.Int64("expired_task_count", expiredCount), + zap.Int("failed_desktop_task_count", monitorResult.Failed), + ) } } @@ -126,9 +143,18 @@ func (w *MonitoringLeaseRecoveryWorker) RunOnce(parent context.Context, run JobR if err := tx.Commit(ctx); err != nil { return map[string]any{"stage": "commit"}, err } + var monitorResult tenantapp.MonitorLeaseRecoveryResult + if w.desktopTasks != nil { + var recoverErr error + monitorResult, recoverErr = w.desktopTasks.RecoverExpiredMonitorTasks(ctx, limit) + if recoverErr != nil { + return map[string]any{"stage": "recover_desktop_tasks"}, recoverErr + } + } return map[string]any{ - "expired_task_count": expiredCount, - "duration_ms": time.Since(startedAt).Milliseconds(), - "batch_size": limit, + "expired_task_count": expiredCount, + "failed_desktop_monitor_count": monitorResult.Failed, + "duration_ms": time.Since(startedAt).Milliseconds(), + "batch_size": limit, }, nil } diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index a88bbb0..9364993 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -1888,6 +1888,7 @@ type recoveredDesktopTaskLease struct { Kind string Status string ActiveAttempt pgtype.UUID + MonitorTaskID pgtype.Int8 Attempts int SubmitStarted bool ErrorJSON []byte @@ -1899,6 +1900,10 @@ type PublishLeaseRecoveryResult struct { Uncertain int `json:"uncertain"` } +type MonitorLeaseRecoveryResult struct { + Failed int `json:"failed"` +} + // resolvePublishRecoveryOutcome decides what to do with a publish task whose lease was lost. // // If the client had already entered the irreversible submit phase (publish_submit_started_at @@ -1917,6 +1922,13 @@ func resolvePublishRecoveryOutcome(submitStarted bool, attempts int) (status str } } +func resolveMonitorRecoveryOutcome(mode desktopTaskRecoveryMode, attempts int) (status string, payloadKind string) { + if mode == desktopTaskRecoveryModeLeaseExpiry || attempts >= 2 { + return "failed", "monitor_final" + } + return "queued", "monitor" +} + func (s *DesktopTaskService) recoverClientTasksOnStartup(ctx context.Context, client *repository.DesktopClient) error { return s.recoverClientTasks(ctx, client, desktopTaskRecoveryModeStartup) } @@ -1950,7 +1962,7 @@ func (s *DesktopTaskService) recoverClientTasks( if err != nil { return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish desktop task recovery payload") } - monitorErrorJSON, monitorReason, _, err := desktopTaskRecoveryPayload(mode, "monitor") + monitorErrorJSON, _, _, err := desktopTaskRecoveryPayload(mode, "monitor") if err != nil { return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload") } @@ -1983,6 +1995,7 @@ func (s *DesktopTaskService) recoverClientTasks( &item.ActiveAttempt, &item.Attempts, &item.SubmitStarted, + &item.MonitorTaskID, ); scanErr != nil { s.logWarn("desktop task recovery scan failed", scanErr, zap.String("client_id", client.ID.String()), zap.String("mode", string(mode))) return response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks") @@ -1998,23 +2011,40 @@ func (s *DesktopTaskService) recoverClientTasks( for index := range recovered { item := &recovered[index] if item.Kind == "monitor" { + finalStatus, monitorPayloadKind := resolveMonitorRecoveryOutcome(mode, item.Attempts) + monitorErrorForTask, monitorReasonForTask, _, payloadErr := desktopTaskRecoveryPayload(mode, monitorPayloadKind) + if payloadErr != nil { + return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload") + } + if finalStatus == "failed" { + if item.MonitorTaskID.Valid { + failedMonitorTaskIDs, failErr := s.failMonitorCollectTasksForDesktopRecovery(ctx, []int64{item.MonitorTaskID.Int64}, monitorErrorForTask) + if failErr != nil { + return failErr + } + if len(failedMonitorTaskIDs) == 0 { + continue + } + } + } if _, execErr := tx.Exec(ctx, ` UPDATE desktop_tasks - SET status = 'queued', - error = $2, + SET status = $2, + error = $3, active_attempt_id = NULL, lease_token_hash = NULL, lease_expires_at = NULL, interrupted_at = COALESCE(interrupted_at, NOW()), - interrupt_reason = $3, - enqueued_at = NOW(), + interrupt_reason = $4, + enqueued_at = CASE WHEN $2 = 'queued' THEN NOW() ELSE enqueued_at END, updated_at = NOW() WHERE desktop_id = $1 - `, item.TaskID, monitorErrorJSON, monitorReason); execErr != nil { + `, item.TaskID, finalStatus, monitorErrorForTask, monitorReasonForTask); execErr != nil { s.logWarn("desktop monitor task recovery update failed", execErr, zap.String("task_id", item.TaskID.String()), zap.String("mode", string(mode))) return response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks") } - item.Status = "queued" + item.Status = finalStatus + item.ErrorJSON = monitorErrorForTask continue } @@ -2080,8 +2110,16 @@ func (s *DesktopTaskService) recoverClientTasks( errorJSON = publishErrorJSON } if item.Kind == "monitor" { - finalStatus = literalStringPtr("aborted") - errorJSON = monitorErrorJSON + if item.Status == "failed" { + finalStatus = literalStringPtr("failed") + } else { + finalStatus = literalStringPtr("aborted") + } + if len(item.ErrorJSON) > 0 { + errorJSON = item.ErrorJSON + } else { + errorJSON = monitorErrorJSON + } } if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{ @@ -2308,6 +2346,318 @@ func (s *DesktopTaskService) RecoverExpiredPublishTasks(ctx context.Context, lim return result, nil } +func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, limit int) (MonitorLeaseRecoveryResult, error) { + var result MonitorLeaseRecoveryResult + if s == nil || s.pool == nil { + return result, nil + } + if limit <= 0 { + limit = 1000 + } + + errorJSON, reason, _, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "monitor_final") + if err != nil { + return result, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload") + } + + tx, err := s.pool.Begin(ctx) + if err != nil { + return result, response.ErrInternal(50196, "desktop_task_recovery_begin_failed", "failed to start desktop task recovery") + } + defer tx.Rollback(ctx) + + rows, err := tx.Query(ctx, expiredMonitorDesktopTasksCandidateSQL(), limit) + if err != nil { + s.logWarn("expired desktop monitor recovery query failed", err) + return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks") + } + defer rows.Close() + + recovered := make([]recoveredDesktopTaskLease, 0) + monitorTaskIDs := make([]int64, 0) + for rows.Next() { + var item recoveredDesktopTaskLease + if scanErr := rows.Scan( + &item.TaskID, + &item.WorkspaceID, + &item.Kind, + &item.Status, + &item.ActiveAttempt, + &item.Attempts, + &item.SubmitStarted, + &item.MonitorTaskID, + ); scanErr != nil { + s.logWarn("expired desktop monitor recovery scan failed", scanErr) + return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks") + } + item.Status = "failed" + item.ErrorJSON = errorJSON + recovered = append(recovered, item) + if item.MonitorTaskID.Valid { + monitorTaskIDs = append(monitorTaskIDs, item.MonitorTaskID.Int64) + } + } + if err := rows.Err(); err != nil { + s.logWarn("expired desktop monitor recovery rows failed", err) + return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks") + } + + failedMonitorTaskIDs, err := s.failMonitorCollectTasksForDesktopRecovery(ctx, monitorTaskIDs, errorJSON) + if err != nil { + return result, err + } + failedMonitorTaskSet := int64Set(failedMonitorTaskIDs) + recoverableDesktopIDs := make([]uuid.UUID, 0, len(recovered)) + recoverable := make([]recoveredDesktopTaskLease, 0, len(recovered)) + for _, item := range recovered { + if !item.MonitorTaskID.Valid { + continue + } + if _, ok := failedMonitorTaskSet[item.MonitorTaskID.Int64]; !ok { + continue + } + recoverableDesktopIDs = append(recoverableDesktopIDs, item.TaskID) + recoverable = append(recoverable, item) + } + if len(recoverableDesktopIDs) == 0 { + return result, nil + } + + updateRows, err := tx.Query(ctx, failExpiredMonitorDesktopTasksSQL(), recoverableDesktopIDs, errorJSON, reason) + if err != nil { + s.logWarn("expired desktop monitor recovery update failed", err) + return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks") + } + defer updateRows.Close() + + updatedDesktopIDs := make(map[uuid.UUID]struct{}, len(recoverableDesktopIDs)) + for updateRows.Next() { + var desktopID uuid.UUID + if scanErr := updateRows.Scan(&desktopID); scanErr != nil { + s.logWarn("expired desktop monitor recovery update scan failed", scanErr) + return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks") + } + updatedDesktopIDs[desktopID] = struct{}{} + } + if err := updateRows.Err(); err != nil { + s.logWarn("expired desktop monitor recovery update rows failed", err) + return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks") + } + + repo := repository.NewDesktopTaskRepository(tx) + for _, item := range recoverable { + if _, ok := updatedDesktopIDs[item.TaskID]; !ok { + continue + } + if !item.ActiveAttempt.Valid { + continue + } + + attemptID, convErr := uuid.FromBytes(item.ActiveAttempt.Bytes[:]) + if convErr != nil { + s.logWarn("expired desktop monitor recovery attempt id decode failed", convErr, zap.String("task_id", item.TaskID.String())) + continue + } + + if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{ + TaskID: item.TaskID, + AttemptID: attemptID, + FinalStatus: literalStringPtr("failed"), + Error: errorJSON, + }); finishErr != nil { + s.logWarn("expired desktop monitor recovery attempt finish failed", finishErr, zap.String("task_id", item.TaskID.String())) + } + } + + if err := tx.Commit(ctx); err != nil { + return result, response.ErrInternal(50199, "desktop_task_recovery_commit_failed", "failed to commit desktop task recovery") + } + + result.Failed = len(updatedDesktopIDs) + + for _, item := range recoverable { + if _, ok := updatedDesktopIDs[item.TaskID]; !ok { + continue + } + task, getErr := s.repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID) + if getErr != nil { + s.logWarn("expired desktop monitor recovery reload after commit failed", getErr, zap.String("task_id", item.TaskID.String())) + continue + } + s.publishTaskEvent(ctx, task, "task_completed") + } + + if result.Failed > 0 && s.logger != nil { + s.logger.Info("expired desktop monitor tasks recovered", + zap.Int("failed", result.Failed), + ) + } + + return result, nil +} + +func expiredMonitorDesktopTasksCandidateSQL() string { + return ` + SELECT + t.desktop_id, + t.workspace_id, + t.kind, + t.status, + t.active_attempt_id, + t.attempts, + (t.publish_submit_started_at IS NOT NULL) AS submit_started, + t.monitor_task_id + FROM desktop_tasks AS t + WHERE t.kind = 'monitor' + AND t.status = 'in_progress' + AND t.monitor_task_id IS NOT NULL + AND t.lease_expires_at IS NOT NULL + AND t.lease_expires_at < NOW() + ORDER BY t.lease_expires_at ASC, t.updated_at ASC, t.desktop_id ASC + LIMIT $1 + FOR UPDATE OF t SKIP LOCKED + ` +} + +func failExpiredMonitorDesktopTasksSQL() string { + return ` + UPDATE desktop_tasks + SET status = 'failed', + error = $2, + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + interrupted_at = COALESCE(interrupted_at, NOW()), + interrupt_reason = COALESCE(interrupt_reason, $3), + updated_at = NOW() + WHERE desktop_id = ANY($1::uuid[]) + AND kind = 'monitor' + AND status = 'in_progress' + RETURNING desktop_id + ` +} + +func (s *DesktopTaskService) failMonitorCollectTasksForDesktopRecovery(ctx context.Context, monitorTaskIDs []int64, errorJSON []byte) ([]int64, error) { + if s == nil || s.monitoringPool == nil || len(monitorTaskIDs) == 0 { + return nil, nil + } + monitorTaskIDs = uniqueInt64s(monitorTaskIDs) + message := desktopTaskRecoveryMessage(errorJSON) + requestPayloadJSON, err := json.Marshal(map[string]any{ + "runtime_error": json.RawMessage(errorJSON), + }) + if err != nil { + return nil, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload") + } + rows, err := s.monitoringPool.Query(ctx, ` + WITH input AS ( + SELECT unnest($1::bigint[]) AS id + ), + updated AS ( + UPDATE monitoring_collect_tasks + SET status = 'failed', + lease_token_hash = NULL, + leased_to_executor = NULL, + leased_at = NULL, + lease_expires_at = NULL, + callback_received_at = COALESCE(callback_received_at, NOW()), + completed_at = COALESCE(completed_at, NOW()), + skip_reason = NULL, + error_message = COALESCE(NULLIF(error_message, ''), $2), + request_payload_json = ( + CASE + WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object' + THEN '{}'::jsonb + ELSE request_payload_json + END + ) || $3::jsonb, + updated_at = NOW() + WHERE id IN (SELECT id FROM input) + AND COALESCE(execution_owner, 'legacy') = 'desktop_tasks' + AND callback_received_at IS NULL + AND status IN ('pending', 'leased', 'expired') + RETURNING id + ), + already_failed AS ( + SELECT t.id + FROM monitoring_collect_tasks t + JOIN input i ON i.id = t.id + WHERE COALESCE(t.execution_owner, 'legacy') = 'desktop_tasks' + AND t.status = 'failed' + AND ( + t.error_message = $2 + OR t.request_payload_json #>> '{runtime_error,source}' = 'desktop_task_lease_expiry' + OR t.request_payload_json #>> '{runtime_error,reason}' = 'lease_expired' + ) + ) + SELECT id FROM updated + UNION + SELECT id FROM already_failed + `, monitorTaskIDs, message, requestPayloadJSON) + if err != nil { + return nil, response.ErrInternal(50041, "task_update_failed", "failed to fail expired monitor desktop tasks") + } + defer rows.Close() + + failedIDs := make([]int64, 0, len(monitorTaskIDs)) + for rows.Next() { + var id int64 + if scanErr := rows.Scan(&id); scanErr != nil { + return nil, response.ErrInternal(50041, "task_update_failed", "failed to parse failed monitor desktop tasks") + } + failedIDs = append(failedIDs, id) + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50041, "task_update_failed", "failed to iterate failed monitor desktop tasks") + } + return failedIDs, nil +} + +func uniqueInt64s(values []int64) []int64 { + if len(values) <= 1 { + return values + } + seen := make(map[int64]struct{}, len(values)) + result := make([]int64, 0, len(values)) + for _, value := range values { + if value == 0 { + continue + } + if _, ok := seen[value]; ok { + continue + } + seen[value] = struct{}{} + result = append(result, value) + } + return result +} + +func int64Set(values []int64) map[int64]struct{} { + result := make(map[int64]struct{}, len(values)) + for _, value := range values { + if value == 0 { + continue + } + result[value] = struct{}{} + } + return result +} + +func desktopTaskRecoveryMessage(errorJSON []byte) string { + const fallback = "desktop monitor task recovered to failed" + if len(errorJSON) == 0 { + return fallback + } + var payload map[string]any + if err := json.Unmarshal(errorJSON, &payload); err != nil { + return fallback + } + if message, ok := payload["message"].(string); ok && strings.TrimSpace(message) != "" { + return strings.TrimSpace(message) + } + return fallback +} + func desktopTaskRecoverySelectQuery(mode desktopTaskRecoveryMode) string { query := ` SELECT @@ -2317,7 +2667,8 @@ func desktopTaskRecoverySelectQuery(mode desktopTaskRecoveryMode) string { status, active_attempt_id, attempts, - publish_submit_started_at IS NOT NULL AS submit_started + publish_submit_started_at IS NOT NULL AS submit_started, + monitor_task_id FROM desktop_tasks WHERE target_client_id = $1 AND workspace_id = $2 @@ -2341,11 +2692,14 @@ func desktopTaskRecoveryPayload(mode desktopTaskRecoveryMode, kind string) ([]by source := "desktop_task_recovery" isPublishFinal := kind == "publish_final" isPublishUncertain := kind == "publish_uncertain" + isMonitorFinal := kind == "monitor_final" if isPublishFinal { message = fmt.Sprintf("desktop publish task exceeded %d attempts during recovery; task has been marked failed", desktopPublishMaxAttempts) } else if isPublishUncertain { message = "desktop publish task may have already been submitted to the platform before the lease was lost; kept as unknown for manual reconcile to avoid duplicate publishing" + } else if isMonitorFinal { + message = "desktop monitor task lost its lease; task has been marked failed to avoid repeated platform retries" } switch mode { @@ -2355,6 +2709,8 @@ func desktopTaskRecoveryPayload(mode desktopTaskRecoveryMode, kind string) ([]by message = fmt.Sprintf("desktop client restarted while the publish task was in progress; task exceeded %d attempts and has been marked failed", desktopPublishMaxAttempts) } else if isPublishUncertain { message = "desktop client restarted while the publish task was mid-submit; kept as unknown for manual reconcile to avoid duplicate publishing" + } else if isMonitorFinal { + message = "desktop client restarted while the monitor task was in progress; task has been marked failed after repeated recovery" } else if kind == "monitor" { message = "desktop client restarted while the monitor task was in progress; task has been re-queued" } else { @@ -2366,6 +2722,8 @@ func desktopTaskRecoveryPayload(mode desktopTaskRecoveryMode, kind string) ([]by message = fmt.Sprintf("desktop client went offline while the publish task was in progress; task exceeded %d attempts and has been marked failed", desktopPublishMaxAttempts) } else if isPublishUncertain { message = "desktop client went offline while the publish task was mid-submit; kept as unknown for manual reconcile to avoid duplicate publishing" + } else if isMonitorFinal { + message = "desktop client went offline while the monitor task was in progress; task has been marked failed after repeated recovery" } else if kind == "monitor" { message = "desktop client went offline while the monitor task was in progress; task has been re-queued" } else { @@ -2377,8 +2735,8 @@ func desktopTaskRecoveryPayload(mode desktopTaskRecoveryMode, kind string) ([]by message = fmt.Sprintf("desktop task lease expired before publish completion; task exceeded %d attempts and has been marked failed", desktopPublishMaxAttempts) } else if isPublishUncertain { message = "desktop task lease expired while the publish task was mid-submit; kept as unknown for manual reconcile to avoid duplicate publishing" - } else if kind == "monitor" { - message = "desktop task lease expired before monitor completion; task has been re-queued" + } else if isMonitorFinal || kind == "monitor" { + message = "desktop task lease expired before monitor completion; task has been marked failed" } else { message = "desktop task lease expired before publish completion; task has been re-queued for retry" } diff --git a/server/internal/tenant/app/desktop_task_service_test.go b/server/internal/tenant/app/desktop_task_service_test.go index b3ff740..25758e3 100644 --- a/server/internal/tenant/app/desktop_task_service_test.go +++ b/server/internal/tenant/app/desktop_task_service_test.go @@ -23,6 +23,7 @@ func TestDesktopTaskRecoverySelectQueryMatchesRecoveredLeaseScanOrder(t *testing "active_attempt_id", "attempts", "publish_submit_started_at", + "monitor_task_id", } if len(gotColumns) != len(wantColumns) { @@ -100,6 +101,27 @@ func TestDesktopTaskRecoveryPayloadPublishFinalFailsAfterMaxAttempts(t *testing. } } +func TestDesktopTaskRecoveryPayloadMonitorLeaseExpiryFails(t *testing.T) { + t.Parallel() + + payload, reason, message, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "monitor_final") + if err != nil { + t.Fatalf("desktopTaskRecoveryPayload returned error: %v", err) + } + if reason != "lease_expired" { + t.Fatalf("reason = %q, want lease_expired", reason) + } + if !strings.Contains(message, "marked failed") { + t.Fatalf("message = %q, want marked failed", message) + } + if strings.Contains(message, "re-queued") { + t.Fatalf("message must not requeue expired monitor tasks: %q", message) + } + if !strings.Contains(string(payload), "desktop_task_lease_expiry") { + t.Fatalf("payload missing source: %s", payload) + } +} + func TestReconcileNonRetryPublishesTaskReconciled(t *testing.T) { t.Parallel() @@ -140,6 +162,35 @@ func TestResolvePublishRecoveryOutcome(t *testing.T) { } } +func TestResolveMonitorRecoveryOutcome(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + mode desktopTaskRecoveryMode + attempts int + wantStatus string + wantKind string + }{ + {"startup first attempt requeues", desktopTaskRecoveryModeStartup, 1, "queued", "monitor"}, + {"disconnect first attempt requeues", desktopTaskRecoveryModeDisconnect, 1, "queued", "monitor"}, + {"startup repeated recovery fails", desktopTaskRecoveryModeStartup, 2, "failed", "monitor_final"}, + {"disconnect repeated recovery fails", desktopTaskRecoveryModeDisconnect, 2, "failed", "monitor_final"}, + {"lease expiry fails immediately", desktopTaskRecoveryModeLeaseExpiry, 1, "failed", "monitor_final"}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + gotStatus, gotKind := resolveMonitorRecoveryOutcome(tc.mode, tc.attempts) + if gotStatus != tc.wantStatus || gotKind != tc.wantKind { + t.Fatalf("resolveMonitorRecoveryOutcome(%q, %d) = (%q, %q), want (%q, %q)", + tc.mode, tc.attempts, gotStatus, gotKind, tc.wantStatus, tc.wantKind) + } + }) + } +} + func TestNormalizeDesktopTaskCompletionStatusForKindFailsMonitorUnknown(t *testing.T) { t.Parallel() @@ -151,6 +202,46 @@ func TestNormalizeDesktopTaskCompletionStatusForKindFailsMonitorUnknown(t *testi } } +func TestExpiredMonitorDesktopTasksCandidateSQLSelectsOnlyExpiredInProgressMonitorTasks(t *testing.T) { + t.Parallel() + + normalized := normalizeSQLForDesktopTaskTest(expiredMonitorDesktopTasksCandidateSQL()) + for _, fragment := range []string{ + "t.kind = 'monitor'", + "t.status = 'in_progress'", + "t.monitor_task_id IS NOT NULL", + "t.lease_expires_at IS NOT NULL", + "t.lease_expires_at < NOW()", + "ORDER BY t.lease_expires_at ASC", + "FOR UPDATE OF t SKIP LOCKED", + } { + if !strings.Contains(normalized, fragment) { + t.Fatalf("expired monitor candidate query missing %q: %s", fragment, normalized) + } + } +} + +func TestFailExpiredMonitorDesktopTasksSQLMarksTerminalFailure(t *testing.T) { + t.Parallel() + + normalized := normalizeSQLForDesktopTaskTest(failExpiredMonitorDesktopTasksSQL()) + for _, fragment := range []string{ + "SET status = 'failed'", + "active_attempt_id = NULL", + "lease_token_hash = NULL", + "lease_expires_at = NULL", + "interrupt_reason = COALESCE(interrupt_reason, $3)", + "WHERE desktop_id = ANY($1::uuid[])", + "AND kind = 'monitor'", + "AND status = 'in_progress'", + "RETURNING desktop_id", + } { + if !strings.Contains(normalized, fragment) { + t.Fatalf("expired monitor update query missing %q: %s", fragment, normalized) + } + } +} + func TestClassifyLeaseErrorDefersQueuedMonitorTask(t *testing.T) { t.Parallel() diff --git a/server/migrations/20260623103000_add_monitor_task_lease_recovery_index.down.sql b/server/migrations/20260623103000_add_monitor_task_lease_recovery_index.down.sql new file mode 100644 index 0000000..f976dff --- /dev/null +++ b/server/migrations/20260623103000_add_monitor_task_lease_recovery_index.down.sql @@ -0,0 +1 @@ +DROP INDEX CONCURRENTLY IF EXISTS idx_desktop_tasks_monitor_lease_recovery; diff --git a/server/migrations/20260623103000_add_monitor_task_lease_recovery_index.up.sql b/server/migrations/20260623103000_add_monitor_task_lease_recovery_index.up.sql new file mode 100644 index 0000000..1a60dda --- /dev/null +++ b/server/migrations/20260623103000_add_monitor_task_lease_recovery_index.up.sql @@ -0,0 +1,6 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_desktop_tasks_monitor_lease_recovery + ON desktop_tasks (lease_expires_at ASC, updated_at ASC, desktop_id ASC) + WHERE kind = 'monitor' + AND status = 'in_progress' + AND monitor_task_id IS NOT NULL + AND lease_expires_at IS NOT NULL; diff --git a/server/migrations_ops/20260623103000_tune_monitoring_lease_recovery.down.sql b/server/migrations_ops/20260623103000_tune_monitoring_lease_recovery.down.sql new file mode 100644 index 0000000..a562fc1 --- /dev/null +++ b/server/migrations_ops/20260623103000_tune_monitoring_lease_recovery.down.sql @@ -0,0 +1,5 @@ +UPDATE ops.scheduler_jobs +SET interval_seconds = 1800, + config = config - 'suppress_empty_success_runs' - 'empty_success_metric_keys', + updated_at = NOW() +WHERE job_key = 'monitoring_lease_recovery'; diff --git a/server/migrations_ops/20260623103000_tune_monitoring_lease_recovery.up.sql b/server/migrations_ops/20260623103000_tune_monitoring_lease_recovery.up.sql new file mode 100644 index 0000000..6abd10d --- /dev/null +++ b/server/migrations_ops/20260623103000_tune_monitoring_lease_recovery.up.sql @@ -0,0 +1,6 @@ +UPDATE ops.scheduler_jobs +SET interval_seconds = 60, + description = '按过期租约索引限批回收超过租约时间仍未回调的监控任务，并释放卡住的桌面监控任务。', + config = config || '{"suppress_empty_success_runs":true,"empty_success_metric_keys":["expired_task_count","failed_desktop_monitor_count"]}'::jsonb, + updated_at = NOW() +WHERE job_key = 'monitoring_lease_recovery';