Method ProcessTask

server-source-code/internal/queue/jobs.go:598–675

ProcessTask implements asynq.Handler.

(ctx context.Context, t *asynq.Task)

// ProcessTask implements asynq.Handler.
func (h *RunPatchHandler) ProcessTask(ctx context.Context, t *asynq.Task) error {
	var p RunPatchPayload
	if err := json.Unmarshal(t.Payload(), &p); err != nil {
		return err
	}

	// Resolve per-context DB when Host is in payload (multi-host mode).
	if h.poolCache != nil && strings.TrimSpace(p.Host) != "" {
		if db, err := h.poolCache.GetOrCreate(ctx, p.Host); err == nil && db != nil {
			ctx = hostctx.WithDB(ctx, db)
		}
	}

	if !h.registry.IsConnected(p.ApiID) {
		// Check if the patch run still exists in the DB (user may have deleted it).
		run, runErr := h.patchRuns.GetByID(ctx, p.PatchRunID)
		if runErr != nil || run == nil {
			h.log.Info("run_patch: patch run deleted or not found, dropping task", "api_id", p.ApiID, "patch_run_id", p.PatchRunID)
			return nil
		}

		// Keep the correct status: pending_validation for dry runs, queued for real runs.
		status := "queued"
		if p.DryRun {
			status = "pending_validation"
		}

		// If status has changed (e.g. cancelled, completed), don't re-queue.
		if run.Status != status && run.Status != "queued" && run.Status != "pending_validation" {
			h.log.Info("run_patch: run status changed, dropping task", "api_id", p.ApiID, "patch_run_id", p.PatchRunID, "status", run.Status)
			return nil
		}

		_ = h.patchRuns.UpdateStatus(ctx, p.PatchRunID, status)

		// Use a deterministic retry task ID so only one retry can exist at a time
		// and the delete handler can cancel it.
		retryTaskID := "patch-run-" + p.PatchRunID + "-retry"
		task, err := NewRunPatchRetryTask(p, retryTaskID)
		if err != nil {
			return err
		}
		_, err = h.queueClient.Enqueue(task, asynq.ProcessIn(5*time.Minute))
		if err != nil {
			// If a task with this ID already exists (pending), that's fine - skip.
			h.log.Debug("run_patch: re-enqueue skipped or failed", "api_id", p.ApiID, "error", err)
		} else {
			h.log.Info("run_patch: agent offline, re-queued in 5m", "api_id", p.ApiID, "patch_run_id", p.PatchRunID)
		}
		return nil
	}

	// Build run_patch payload
	payload := map[string]interface{}{
		"type":         "run_patch",
		"patch_run_id": p.PatchRunID,
		"patch_type":   p.PatchType,
		"dry_run":      p.DryRun,