(task *Task)
| 135 | } |
| 136 | |
| 137 | func (tm *TaskManager) runTask(task *Task) { |
| 138 | // Update status to running |
| 139 | tm.mu.Lock() |
| 140 | task.Status = StatusRunning |
| 141 | task.StartedAt = time.Now() |
| 142 | tm.mu.Unlock() |
| 143 | |
| 144 | if tm.updateNotify != nil { |
| 145 | go tm.updateNotify() |
| 146 | } |
| 147 | |
| 148 | // Execute operation |
| 149 | upid, err := task.Operation() |
| 150 | if err != nil { |
| 151 | tm.completeTask(task, StatusFailed, err) |
| 152 | return |
| 153 | } |
| 154 | |
| 155 | // If no UPID returned, it was a sync op (unlikely for VM ops but possible) |
| 156 | if upid == "" { |
| 157 | tm.completeTask(task, StatusCompleted, nil) |
| 158 | return |
| 159 | } |
| 160 | |
| 161 | // Store UPID |
| 162 | tm.mu.Lock() |
| 163 | task.UPID = upid |
| 164 | tm.mu.Unlock() |
| 165 | |
| 166 | // Notify consumers as soon as UPID is available so the active queue can |
| 167 | // render it immediately instead of waiting for the next state transition. |
| 168 | if tm.updateNotify != nil { |
| 169 | go tm.updateNotify() |
| 170 | } |
| 171 | |
| 172 | // Poll for completion |
| 173 | ticker := time.NewTicker(2 * time.Second) |
| 174 | defer ticker.Stop() |
| 175 | |
| 176 | for { |
| 177 | select { |
| 178 | case <-tm.stopChan: |
| 179 | return |
| 180 | case <-ticker.C: |
| 181 | client, err := tm.clientResolver(task.TargetNode) |
| 182 | if err != nil { |
| 183 | // Failed to resolve client (maybe node moved or group issue), retry |
| 184 | continue |
| 185 | } |
| 186 | |
| 187 | status, err := client.GetTaskStatus(task.TargetNode, upid) |
| 188 | if err != nil { |
| 189 | // Log error but continue polling? or fail? |
| 190 | // Maybe temporary network error. |
| 191 | continue |
| 192 | } |
| 193 | |
| 194 | if status.Status == "stopped" { |
no test coverage detected