fix monitor desktop task recovery

server/internal/tenant/app/brand_monitoring_cleanup.go

@@ -4,9 +4,12 @@ import (
 	"context"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgtype"
 
 	"github.com/geo-platform/tenant-api/internal/shared/response"
+	"github.com/geo-platform/tenant-api/internal/tenant/repository"
 )
 
 const (
@@ -70,7 +73,8 @@ func (s *BrandService) cleanupMonitoringDeletion(ctx context.Context, tenantID,
 	}
 
 	businessDay := monitoringBusinessToday()
-	if err := s.skipMonitoringTasksForDeletion(ctx, tx, tenantID, brandID, businessDay, options); err != nil {
+	skippedTaskIDs, err := s.skipMonitoringTasksForDeletion(ctx, tx, tenantID, brandID, businessDay, options)
+	if err != nil {
 		return err
 	}
 
@@ -82,6 +86,10 @@ func (s *BrandService) cleanupMonitoringDeletion(ctx context.Context, tenantID,
 		return response.ErrInternal(50041, "commit_failed", "failed to commit monitoring cleanup")
 	}
 
+	if err := s.abortDesktopMonitorTasksForSkippedMonitoringTasks(ctx, skippedTaskIDs, options); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -130,54 +138,144 @@ func (s *BrandService) markMonitoringQuestionSnapshotsDeleted(ctx context.Contex
 	return nil
 }
 
-func (s *BrandService) skipMonitoringTasksForDeletion(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, businessDay time.Time, options cleanupMonitoringDeletionOptions) error {
+func (s *BrandService) skipMonitoringTasksForDeletion(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, businessDay time.Time, options cleanupMonitoringDeletionOptions) ([]int64, error) {
 	dateText := businessDay.Format("2006-01-02")
+	query := `
+		UPDATE monitoring_collect_tasks
+		SET status = 'skipped',
+		    lease_token_hash = NULL,
+		    leased_to_executor = NULL,
+		    leased_at = NULL,
+		    lease_expires_at = NULL,
+		    completed_at = COALESCE(completed_at, NOW()),
+		    skip_reason = $4,
+		    error_message = $5,
+		    updated_at = NOW()
+		WHERE tenant_id = $1
+		  AND brand_id = $2
+		  AND collector_type = $3
+		  AND business_date = $6::date
+		  AND status IN ('pending', 'leased', 'received', 'expired')
+	`
+	args := []any{tenantID, brandID, monitoringCollectorType, options.skipReason, options.errorMessage, dateText}
 
 	switch {
 	case options.deleteAllByBrand:
-		if _, err := tx.Exec(ctx, `
-			UPDATE monitoring_collect_tasks
-			SET status = 'skipped',
-			    lease_token_hash = NULL,
-			    leased_to_executor = NULL,
-			    leased_at = NULL,
-			    lease_expires_at = NULL,
-			    completed_at = COALESCE(completed_at, NOW()),
-			    skip_reason = $4,
-			    error_message = $5,
-			    updated_at = NOW()
-			WHERE tenant_id = $1
-			  AND brand_id = $2
-			  AND collector_type = $3
-			  AND business_date = $6::date
-			  AND status IN ('pending', 'leased', 'received', 'expired')
-		`, tenantID, brandID, monitoringCollectorType, options.skipReason, options.errorMessage, dateText); err != nil {
-			return response.ErrInternal(50041, "task_skip_failed", "failed to invalidate brand monitoring tasks")
-		}
 	default:
 		if len(options.questionIDs) == 0 {
-			return nil
+			return nil, nil
 		}
-		if _, err := tx.Exec(ctx, `
+		query += ` AND question_id = ANY($7)`
-			UPDATE monitoring_collect_tasks
+		args = append(args, options.questionIDs)
-			SET status = 'skipped',
+	}
+	query += ` RETURNING id`
+
+	rows, err := tx.Query(ctx, query, args...)
+	if err != nil {
+		return nil, response.ErrInternal(50041, "task_skip_failed", "failed to invalidate monitoring tasks")
+	}
+	defer rows.Close()
+
+	taskIDs := make([]int64, 0)
+	for rows.Next() {
+		var taskID int64
+		if scanErr := rows.Scan(&taskID); scanErr != nil {
+			return nil, response.ErrInternal(50041, "task_skip_failed", "failed to parse invalidated monitoring tasks")
+		}
+		taskIDs = append(taskIDs, taskID)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, response.ErrInternal(50041, "task_skip_failed", "failed to iterate invalidated monitoring tasks")
+	}
+
+	return taskIDs, nil
+}
+
+func (s *BrandService) abortDesktopMonitorTasksForSkippedMonitoringTasks(ctx context.Context, monitorTaskIDs []int64, options cleanupMonitoringDeletionOptions) error {
+	if s == nil || s.pool == nil || len(monitorTaskIDs) == 0 {
+		return nil
+	}
+
+	errorJSON, err := marshalOptionalJSON(map[string]any{
+		"reason":  options.skipReason,
+		"message": options.errorMessage,
+		"source":  "monitoring_cleanup",
+	})
+	if err != nil {
+		return response.ErrInternal(50041, "desktop_task_payload_failed", "failed to encode monitoring cleanup desktop task payload")
+	}
+
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return response.ErrInternal(50041, "desktop_task_begin_failed", "failed to start monitoring cleanup desktop task sync")
+	}
+	defer tx.Rollback(ctx)
+
+	rows, err := tx.Query(ctx, `
+		WITH target AS (
+			SELECT desktop_id, active_attempt_id
+			FROM desktop_tasks
+			WHERE kind = 'monitor'
+			  AND monitor_task_id = ANY($1::bigint[])
+			  AND status IN ('queued', 'in_progress', 'unknown')
+			FOR UPDATE
+		),
+		updated AS (
+			UPDATE desktop_tasks AS t
+			SET status = 'aborted',
+			    error = $2,
+			    result = NULL,
+			    active_attempt_id = NULL,
 			    lease_token_hash = NULL,
-			    leased_to_executor = NULL,
-			    leased_at = NULL,
 			    lease_expires_at = NULL,
-			    completed_at = COALESCE(completed_at, NOW()),
+			    interrupted_at = COALESCE(interrupted_at, NOW()),
-			    skip_reason = $4,
+			    interrupt_reason = COALESCE(interrupt_reason, $3),
-			    error_message = $5,
 			    updated_at = NOW()
-			WHERE tenant_id = $1
+			FROM target
-			  AND brand_id = $2
+			WHERE t.desktop_id = target.desktop_id
-			  AND collector_type = $3
+			RETURNING t.desktop_id, target.active_attempt_id
-			  AND business_date = $6::date
+		)
-			  AND question_id = ANY($7)
+		SELECT desktop_id, active_attempt_id
-			  AND status IN ('pending', 'leased', 'received', 'expired')
+		FROM updated
-		`, tenantID, brandID, monitoringCollectorType, options.skipReason, options.errorMessage, dateText, options.questionIDs); err != nil {
+	`, monitorTaskIDs, errorJSON, options.skipReason)
-			return response.ErrInternal(50041, "task_skip_failed", "failed to invalidate question monitoring tasks")
+	if err != nil {
+		return response.ErrInternal(50041, "desktop_task_sync_failed", "failed to sync skipped monitoring desktop tasks")
+	}
+	defer rows.Close()
+
+	type abortedTask struct {
+		desktopID       uuid.UUID
+		activeAttemptID pgtype.UUID
+	}
+	repo := repository.NewDesktopTaskRepository(tx)
+	for rows.Next() {
+		var task abortedTask
+		if scanErr := rows.Scan(&task.desktopID, &task.activeAttemptID); scanErr != nil {
+			return response.ErrInternal(50041, "desktop_task_sync_failed", "failed to parse skipped monitoring desktop task sync")
 		}
+
+		if !task.activeAttemptID.Valid {
+			continue
+		}
+		attemptID, convErr := uuid.FromBytes(task.activeAttemptID.Bytes[:])
+		if convErr != nil {
+			continue
+		}
+		if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
+			TaskID:      task.desktopID,
+			AttemptID:   attemptID,
+			FinalStatus: literalStringPtr("aborted"),
+			Error:       errorJSON,
+		}); finishErr != nil {
+			return response.ErrInternal(50041, "desktop_task_attempt_finish_failed", "failed to finish skipped monitoring desktop task attempt")
+		}
+	}
+	if err := rows.Err(); err != nil {
+		return response.ErrInternal(50041, "desktop_task_sync_failed", "failed to iterate skipped monitoring desktop task sync")
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return response.ErrInternal(50041, "desktop_task_commit_failed", "failed to commit skipped monitoring desktop task sync")
 	}
 
 	return nil

server/internal/tenant/app/desktop_task_service.go

@@ -2012,6 +2012,10 @@ func (s *DesktopTaskService) recoverClientTasks(
 	if err != nil {
 		return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload")
 	}
+	monitorUnrecoverableErrorJSON, monitorUnrecoverableReason, _, err := desktopTaskRecoveryPayload(mode, "monitor_unrecoverable")
+	if err != nil {
+		return 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 {
@@ -2069,7 +2073,9 @@ func (s *DesktopTaskService) recoverClientTasks(
 						return failErr
 					}
 					if len(failedMonitorTaskIDs) == 0 {
-						continue
+						finalStatus = "aborted"
+						monitorErrorForTask = monitorUnrecoverableErrorJSON
+						monitorReasonForTask = monitorUnrecoverableReason
 					}
 				}
 			} else if item.MonitorTaskID.Valid {
@@ -2078,7 +2084,9 @@ func (s *DesktopTaskService) recoverClientTasks(
 					return requeueErr
 				}
 				if len(requeuedMonitorTaskIDs) == 0 {
-					continue
+					finalStatus = "aborted"
+					monitorErrorForTask = monitorUnrecoverableErrorJSON
+					monitorReasonForTask = monitorUnrecoverableReason
 				}
 			}
 			if _, execErr := tx.Exec(ctx, `
@@ -2414,6 +2422,10 @@ func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, lim
 	if err != nil {
 		return result, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload")
 	}
+	unrecoverableErrorJSON, unrecoverableReason, _, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "monitor_unrecoverable")
+	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 {
@@ -2464,39 +2476,70 @@ func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, lim
 	requeuedMonitorTaskSet := int64Set(requeuedMonitorTaskIDs)
 	recoverableDesktopIDs := make([]uuid.UUID, 0, len(recovered))
 	recoverable := make([]recoveredDesktopTaskLease, 0, len(recovered))
+	unrecoverableDesktopIDs := make([]uuid.UUID, 0, len(recovered))
+	unrecoverable := make([]recoveredDesktopTaskLease, 0, len(recovered))
 	for _, item := range recovered {
 		if !item.MonitorTaskID.Valid {
 			continue
 		}
 		if _, ok := requeuedMonitorTaskSet[item.MonitorTaskID.Int64]; !ok {
+			item.Status = "aborted"
+			item.ErrorJSON = unrecoverableErrorJSON
+			unrecoverableDesktopIDs = append(unrecoverableDesktopIDs, item.TaskID)
+			unrecoverable = append(unrecoverable, item)
 			continue
 		}
 		recoverableDesktopIDs = append(recoverableDesktopIDs, item.TaskID)
 		recoverable = append(recoverable, item)
 	}
-	if len(recoverableDesktopIDs) == 0 {
+	if len(recoverableDesktopIDs) == 0 && len(unrecoverableDesktopIDs) == 0 {
 		return result, nil
 	}
 
-	updateRows, err := tx.Query(ctx, requeueExpiredMonitorDesktopTasksSQL(), 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() {
+	if len(recoverableDesktopIDs) > 0 {
-		var desktopID uuid.UUID
+		updateRows, err := tx.Query(ctx, requeueExpiredMonitorDesktopTasksSQL(), recoverableDesktopIDs, errorJSON, reason)
-		if scanErr := updateRows.Scan(&desktopID); scanErr != nil {
+		if err != nil {
-			s.logWarn("expired desktop monitor recovery update scan failed", scanErr)
+			s.logWarn("expired desktop monitor recovery update failed", err)
-			return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks")
+			return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
+		}
+		defer updateRows.Close()
+
+		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")
 		}
-		updatedDesktopIDs[desktopID] = struct{}{}
 	}
-	if err := updateRows.Err(); err != nil {
+
-		s.logWarn("expired desktop monitor recovery update rows failed", err)
+	abortedDesktopIDs := make(map[uuid.UUID]struct{}, len(unrecoverableDesktopIDs))
-		return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks")
+	if len(unrecoverableDesktopIDs) > 0 {
+		abortRows, err := tx.Query(ctx, abortExpiredMonitorDesktopTasksSQL(), unrecoverableDesktopIDs, unrecoverableErrorJSON, unrecoverableReason)
+		if err != nil {
+			s.logWarn("expired desktop monitor unrecoverable update failed", err)
+			return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
+		}
+		defer abortRows.Close()
+
+		for abortRows.Next() {
+			var desktopID uuid.UUID
+			if scanErr := abortRows.Scan(&desktopID); scanErr != nil {
+				s.logWarn("expired desktop monitor unrecoverable update scan failed", scanErr)
+				return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks")
+			}
+			abortedDesktopIDs[desktopID] = struct{}{}
+		}
+		if err := abortRows.Err(); err != nil {
+			s.logWarn("expired desktop monitor unrecoverable update rows failed", err)
+			return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks")
+		}
 	}
 
 	repo := repository.NewDesktopTaskRepository(tx)
@@ -2523,12 +2566,36 @@ func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, lim
 			s.logWarn("expired desktop monitor recovery attempt finish failed", finishErr, zap.String("task_id", item.TaskID.String()))
 		}
 	}
+	for _, item := range unrecoverable {
+		if _, ok := abortedDesktopIDs[item.TaskID]; !ok {
+			continue
+		}
+		if !item.ActiveAttempt.Valid {
+			continue
+		}
+
+		attemptID, convErr := uuid.FromBytes(item.ActiveAttempt.Bytes[:])
+		if convErr != nil {
+			s.logWarn("expired desktop monitor unrecoverable 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("aborted"),
+			Error:       unrecoverableErrorJSON,
+		}); finishErr != nil {
+			s.logWarn("expired desktop monitor unrecoverable 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.Requeued = len(updatedDesktopIDs)
+	result.Failed = len(abortedDesktopIDs)
 
 	for _, item := range recoverable {
 		if _, ok := updatedDesktopIDs[item.TaskID]; !ok {
@@ -2541,10 +2608,22 @@ func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, lim
 		}
 		s.publishTaskEvent(ctx, task, "task_available")
 	}
+	for _, item := range unrecoverable {
+		if _, ok := abortedDesktopIDs[item.TaskID]; !ok {
+			continue
+		}
+		task, getErr := s.repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
+		if getErr != nil {
+			s.logWarn("expired desktop monitor unrecoverable reload after commit failed", getErr, zap.String("task_id", item.TaskID.String()))
+			continue
+		}
+		s.publishTaskEvent(ctx, task, "task_completed")
+	}
 
-	if result.Requeued > 0 && s.logger != nil {
+	if (result.Requeued > 0 || result.Failed > 0) && s.logger != nil {
 		s.logger.Info("expired desktop monitor tasks recovered",
 			zap.Int("requeued", result.Requeued),
+			zap.Int("failed", result.Failed),
 		)
 	}
 
@@ -2594,6 +2673,25 @@ func requeueExpiredMonitorDesktopTasksSQL() string {
 	`
 }
 
+func abortExpiredMonitorDesktopTasksSQL() string {
+	return `
+		UPDATE desktop_tasks
+		SET status = 'aborted',
+		    error = $2,
+		    result = NULL,
+		    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) requeueMonitorCollectTasksForDesktopRecovery(ctx context.Context, monitorTaskIDs []int64, errorJSON []byte) ([]int64, error) {
 	if s == nil || s.monitoringPool == nil || len(monitorTaskIDs) == 0 {
 		return nil, nil

server/internal/tenant/app/desktop_task_service_test.go

@@ -245,6 +245,27 @@ func TestRequeueExpiredMonitorDesktopTasksSQLReturnsTaskToQueue(t *testing.T) {
 	}
 }
 
+func TestAbortExpiredMonitorDesktopTasksSQLTerminatesUnrecoverableTask(t *testing.T) {
+	t.Parallel()
+
+	normalized := normalizeSQLForDesktopTaskTest(abortExpiredMonitorDesktopTasksSQL())
+	for _, fragment := range []string{
+		"SET status = 'aborted'",
+		"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 abort query missing %q: %s", fragment, normalized)
+		}
+	}
+}
+
 func TestClassifyLeaseErrorDefersQueuedMonitorTask(t *testing.T) {
 	t.Parallel()