consume processes all tasks in the queue
()
| 94 | |
| 95 | // consume processes all tasks in the queue |
| 96 | func (tqm *TaskQueueManager) consume() { |
| 97 | // Prevent concurrent consumption |
| 98 | if !tqm.consuming.CompareAndSwap(false, true) { |
| 99 | log.Warn("previous consume still running, skip this round") |
| 100 | return |
| 101 | } |
| 102 | defer tqm.consuming.Store(false) |
| 103 | |
| 104 | tqm.mu.Lock() |
| 105 | |
| 106 | // extract all tasks |
| 107 | tasks := make([]*QueuedTask, 0, len(tqm.queue)) |
| 108 | for _, task := range tqm.queue { |
| 109 | tasks = append(tasks, task) |
| 110 | } |
| 111 | |
| 112 | // clear queue |
| 113 | tqm.queue = make(map[string]*QueuedTask) |
| 114 | |
| 115 | tqm.mu.Unlock() |
| 116 | |
| 117 | if len(tasks) == 0 { |
| 118 | return |
| 119 | } |
| 120 | |
| 121 | log.Infof("consuming task queue: %d tasks", len(tasks)) |
| 122 | |
| 123 | // sort tasks: shallow paths first, then by enqueue time |
| 124 | sort.Slice(tasks, func(i, j int) bool { |
| 125 | if tasks[i].Depth != tasks[j].Depth { |
| 126 | return tasks[i].Depth < tasks[j].Depth |
| 127 | } |
| 128 | return tasks[i].EnqueueAt.Before(tasks[j].EnqueueAt) |
| 129 | }) |
| 130 | |
| 131 | ctx := context.Background() |
| 132 | |
| 133 | // execute tasks in order |
| 134 | for _, task := range tasks { |
| 135 | // Check if there are pending tasks for this parent |
| 136 | tqm.mu.RLock() |
| 137 | pendingTaskUIDs, hasPending := tqm.pendingTasks[task.Parent] |
| 138 | tqm.mu.RUnlock() |
| 139 | |
| 140 | if hasPending && len(pendingTaskUIDs) > 0 { |
| 141 | // Check all pending task statuses |
| 142 | allCompleted := true |
| 143 | for _, taskUID := range pendingTaskUIDs { |
| 144 | taskStatus, err := tqm.m.getTaskStatus(ctx, taskUID) |
| 145 | if err != nil { |
| 146 | log.Errorf("failed to get task status for parent %s (taskUID: %d): %v", task.Parent, taskUID, err) |
| 147 | // If we can't get status, assume it's done and continue checking |
| 148 | continue |
| 149 | } |
| 150 | |
| 151 | // Check if task is still running |
| 152 | if taskStatus == "enqueued" || taskStatus == "processing" { |
| 153 | log.Warnf("skipping task for parent %s: previous task %d still %s", task.Parent, taskUID, taskStatus) |
no test coverage detected