diff --git a/server/internal/tenant/app/brand_monitoring_cleanup.go b/server/internal/tenant/app/brand_monitoring_cleanup.go index ae31fc2..8e82494 100644 --- a/server/internal/tenant/app/brand_monitoring_cleanup.go +++ b/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, ` - UPDATE monitoring_collect_tasks - SET status = 'skipped', + query += ` AND question_id = ANY($7)` + args = append(args, options.questionIDs) + } + 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()), - skip_reason = $4, - error_message = $5, + interrupted_at = COALESCE(interrupted_at, NOW()), + interrupt_reason = COALESCE(interrupt_reason, $3), updated_at = NOW() - WHERE tenant_id = $1 - AND brand_id = $2 - AND collector_type = $3 - AND business_date = $6::date - AND question_id = ANY($7) - AND status IN ('pending', 'leased', 'received', 'expired') - `, tenantID, brandID, monitoringCollectorType, options.skipReason, options.errorMessage, dateText, options.questionIDs); err != nil { - return response.ErrInternal(50041, "task_skip_failed", "failed to invalidate question monitoring tasks") + FROM target + WHERE t.desktop_id = target.desktop_id + RETURNING t.desktop_id, target.active_attempt_id + ) + SELECT desktop_id, active_attempt_id + FROM updated + `, monitorTaskIDs, errorJSON, options.skipReason) + 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 diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index cfd4a97..0951ff9 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/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() { - 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") + if len(recoverableDesktopIDs) > 0 { + 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() + + 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) - return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks") + + abortedDesktopIDs := make(map[uuid.UUID]struct{}, len(unrecoverableDesktopIDs)) + 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 diff --git a/server/internal/tenant/app/desktop_task_service_test.go b/server/internal/tenant/app/desktop_task_service_test.go index d5479bb..3353f40 100644 --- a/server/internal/tenant/app/desktop_task_service_test.go +++ b/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()