failTaskRunsForHA fails all PENDING and AVAILABLE task runs because HA is not licensed.
(ctx context.Context, haErr error)
| 71 | |
| 72 | // failTaskRunsForHA fails all PENDING and AVAILABLE task runs because HA is not licensed. |
| 73 | func (s *Scheduler) failTaskRunsForHA(ctx context.Context, haErr error) { |
| 74 | taskRuns, err := s.store.ListTaskRuns(ctx, &store.FindTaskRunMessage{ |
| 75 | Status: new([]storepb.TaskRun_Status{storepb.TaskRun_PENDING, storepb.TaskRun_AVAILABLE}), |
| 76 | }) |
| 77 | if err != nil { |
| 78 | slog.Error("failed to list task runs for HA limit check", log.BBError(err)) |
| 79 | return |
| 80 | } |
| 81 | if len(taskRuns) == 0 { |
| 82 | return |
| 83 | } |
| 84 | |
| 85 | // Track affected plans to send failure webhooks. |
| 86 | // Key by (projectID, planID) to match ClaimPipelineFailureNotification's dedup key. |
| 87 | type planKey struct { |
| 88 | projectID string |
| 89 | planID int64 |
| 90 | } |
| 91 | affectedPlans := map[planKey]string{} // value = first environment seen |
| 92 | |
| 93 | for _, taskRun := range taskRuns { |
| 94 | if _, err := s.store.UpdateTaskRunStatus(ctx, &store.TaskRunStatusPatch{ |
| 95 | ID: taskRun.ID, |
| 96 | ProjectID: taskRun.ProjectID, |
| 97 | Status: storepb.TaskRun_FAILED, |
| 98 | AllowedStatuses: []storepb.TaskRun_Status{ |
| 99 | storepb.TaskRun_PENDING, |
| 100 | storepb.TaskRun_AVAILABLE, |
| 101 | }, |
| 102 | ResultProto: &storepb.TaskRunResult{ |
| 103 | Detail: haErr.Error(), |
| 104 | }, |
| 105 | }); err != nil { |
| 106 | if common.ErrorCode(err) == common.Conflict { |
| 107 | continue |
| 108 | } |
| 109 | slog.Error("failed to fail task run for HA limit", |
| 110 | slog.Int64("taskRunID", taskRun.ID), |
| 111 | log.BBError(err), |
| 112 | ) |
| 113 | continue |
| 114 | } |
| 115 | key := planKey{projectID: taskRun.ProjectID, planID: taskRun.PlanUID} |
| 116 | if _, ok := affectedPlans[key]; !ok { |
| 117 | affectedPlans[key] = taskRun.Environment |
| 118 | } |
| 119 | } |
| 120 | |
| 121 | slog.Warn("Failed task runs due to HA license restriction", slog.Int64("count", int64(len(taskRuns))), log.BBError(haErr)) |
| 122 | |
| 123 | // Send PIPELINE_FAILED webhook for each affected plan. |
| 124 | for key, environment := range affectedPlans { |
| 125 | claimed, err := s.store.ClaimPipelineFailureNotification(ctx, key.projectID, key.planID) |
| 126 | if err != nil { |
| 127 | slog.Error("failed to claim pipeline failure notification", log.BBError(err)) |
| 128 | continue |
| 129 | } |
| 130 | if !claimed { |
no test coverage detected