| 266 | } |
| 267 | |
| 268 | func (q *queue) run(ctx context.Context, t Task) (task.Status, error) { |
| 269 | l := logging.FromContext(ctx) |
| 270 | |
| 271 | // create channel with buffer size 1 to avoid goroutine leak |
| 272 | done := make(chan struct { |
| 273 | err error |
| 274 | next task.Status |
| 275 | }, 1) |
| 276 | panicChan := make(chan interface{}, 1) |
| 277 | startTime := time.Now() |
| 278 | ctx, cancel := context.WithTimeout(ctx, q.maxTaskExecution-t.Executed()) |
| 279 | defer func() { |
| 280 | cancel() |
| 281 | }() |
| 282 | |
| 283 | // run the job |
| 284 | go func() { |
| 285 | // handle panic issue |
| 286 | defer func() { |
| 287 | if p := recover(); p != nil { |
| 288 | q.logger.Error("panic in queue %q: %s", q.name, debug.Stack()) |
| 289 | panicChan <- p |
| 290 | } |
| 291 | }() |
| 292 | |
| 293 | l.Debug("Iteration started.") |
| 294 | next, err := t.Do(ctx) |
| 295 | l.Debug("Iteration ended with err=%s", err) |
| 296 | if err != nil && q.maxRetry-t.Retried() > 0 && !errors.Is(err, CriticalErr) && atomic.LoadInt32(&q.stopFlag) != 1 { |
| 297 | // Retry needed |
| 298 | t.OnRetry(err) |
| 299 | b := &backoff.Backoff{ |
| 300 | Max: q.backoffMaxDuration, |
| 301 | Factor: q.backoffFactor, |
| 302 | } |
| 303 | delay := q.retryDelay |
| 304 | if q.retryDelay == 0 { |
| 305 | delay = b.ForAttempt(float64(t.Retried())) |
| 306 | } |
| 307 | |
| 308 | // Resume after to retry |
| 309 | l.Info("Will be retried in %s", delay) |
| 310 | t.OnSuspend(time.Now().Add(delay).Unix()) |
| 311 | err = nil |
| 312 | next = task.StatusSuspending |
| 313 | } |
| 314 | |
| 315 | done <- struct { |
| 316 | err error |
| 317 | next task.Status |
| 318 | }{err: err, next: next} |
| 319 | }() |
| 320 | |
| 321 | select { |
| 322 | case p := <-panicChan: |
| 323 | panic(p) |
| 324 | case <-ctx.Done(): // timeout reached |
| 325 | return task.StatusError, ctx.Err() |