CreatePendingTaskRuns creates pending task runs. This operation is idempotent and safe for concurrent calls: - Uses WHERE NOT EXISTS to skip tasks that already have active (PENDING/RUNNING/DONE) task runs - Uses ON CONFLICT DO NOTHING to handle race conditions where two requests try to create the sa
(ctx context.Context, creator string, creates ...*TaskRunMessage)
| 302 | // - Uses ON CONFLICT DO NOTHING to handle race conditions where two requests try to create the same task run |
| 303 | // - The unique constraint on (task_id, attempt) ensures no duplicates |
| 304 | func (s *Store) CreatePendingTaskRuns(ctx context.Context, creator string, creates ...*TaskRunMessage) error { |
| 305 | if len(creates) == 0 { |
| 306 | return nil |
| 307 | } |
| 308 | |
| 309 | var taskUIDs []int64 |
| 310 | var runAts []*time.Time |
| 311 | var projects []string |
| 312 | for _, create := range creates { |
| 313 | taskUIDs = append(taskUIDs, create.TaskUID) |
| 314 | runAts = append(runAts, create.RunAt) |
| 315 | projects = append(projects, create.ProjectID) |
| 316 | } |
| 317 | |
| 318 | // Convert empty string to NULL for system-created task runs |
| 319 | var creatorPtr any |
| 320 | if creator == "" { |
| 321 | creatorPtr = nil |
| 322 | } else { |
| 323 | creatorPtr = creator |
| 324 | } |
| 325 | |
| 326 | // Serialize payload from the first create (all creates in a batch share the same payload). |
| 327 | payloadStr := "{}" |
| 328 | if len(creates) > 0 && creates[0].PayloadProto != nil { |
| 329 | payloadBytes, err := protojson.Marshal(creates[0].PayloadProto) |
| 330 | if err != nil { |
| 331 | return errors.Wrapf(err, "failed to marshal task run payload") |
| 332 | } |
| 333 | if s := string(payloadBytes); s != "" { |
| 334 | payloadStr = s |
| 335 | } |
| 336 | } |
| 337 | |
| 338 | projectID := creates[0].ProjectID |
| 339 | |
| 340 | tx, err := s.GetDB().BeginTx(ctx, nil) |
| 341 | if err != nil { |
| 342 | return errors.Wrapf(err, "failed to begin tx") |
| 343 | } |
| 344 | defer tx.Rollback() |
| 345 | |
| 346 | baseID, err := nextProjectID(ctx, tx, "task_run", projectID) |
| 347 | if err != nil { |
| 348 | return err |
| 349 | } |
| 350 | |
| 351 | // Single query that: |
| 352 | // 1. Assigns per-project IDs using ROW_NUMBER() + baseID |
| 353 | // 2. Filters out tasks with existing PENDING/RUNNING/DONE task runs (idempotent) |
| 354 | // 3. Calculates next attempt for each remaining task |
| 355 | // 4. Inserts task runs |
| 356 | // 5. Uses ON CONFLICT DO NOTHING to handle race conditions |
| 357 | q := qb.Q().Space(` |
| 358 | WITH candidates AS ( |
| 359 | SELECT |
| 360 | (ROW_NUMBER() OVER ()) + ? - 1 AS new_id, |
| 361 | tasks.project, |