processOne processes a single task fetched from the queue. It contains its own recover() so a panic inside q.Handler does NOT kill the worker goroutine — the loop continues with the next task. Returns the sleepDelay to use before fetching the next task, and stop=true if the worker must exit (queue
(ctx context.Context, t task.Task)
| 812 | // Returns the sleepDelay to use before fetching the next task, and stop=true |
| 813 | // if the worker must exit (queue context cancelled). |
| 814 | func (q *TaskQueue) processOne(ctx context.Context, t task.Task) (time.Duration, bool) { |
| 815 | // nextSleepDelay and stop are declared here so the inner closure's deferred |
| 816 | // recover can overwrite them on panic without using named return parameters. |
| 817 | var nextSleepDelay time.Duration |
| 818 | var stop bool |
| 819 | |
| 820 | func() { |
| 821 | defer func() { |
| 822 | if r := recover(); r != nil { |
| 823 | q.logger.Warn("panic recovered in TaskQueue Start", |
| 824 | slog.String(pkg.LogKeyQueue, q.Name), |
| 825 | slog.String(pkg.LogKeyTaskID, t.GetId()), |
| 826 | slog.String(pkg.LogKeyTaskType, string(t.GetType())), |
| 827 | slog.Any(pkg.LogKeyError, r), |
| 828 | slog.String(pkg.LogKeyStack, string(debug.Stack())), |
| 829 | ) |
| 830 | |
| 831 | // Make sure the task is no longer marked as processing and apply |
| 832 | // a backoff so we don't hot-loop on the same panicking task. |
| 833 | t.SetProcessing(false) |
| 834 | t.IncrementFailureCount() |
| 835 | nextSleepDelay = q.ExponentialBackoffFn(t.GetFailureCount()) |
| 836 | q.SetStatus(QueueStatusIdle) |
| 837 | q.SetStatusText(fmt.Sprintf("sleep after panic for %s", nextSleepDelay.String())) |
| 838 | } |
| 839 | }() |
| 840 | |
| 841 | q.withLock(func() { |
| 842 | if q.queueTasksCounter.IsAnyCapReached() { |
| 843 | q.lazydebug("triggering compaction before task processing", func() []any { |
| 844 | return []any{slog.String(pkg.LogKeyQueue, q.Name), slog.String(pkg.LogKeyTaskID, t.GetId()), slog.String(pkg.LogKeyTaskType, string(t.GetType())), slog.Int(pkg.LogKeyQueueLength, q.storage.Length())} |
| 845 | }) |
| 846 | |
| 847 | q.compaction(q.queueTasksCounter.GetReachedCap()) |
| 848 | |
| 849 | q.lazydebug("compaction completed, queue no longer dirty", func() []any { |
| 850 | return []any{slog.String(pkg.LogKeyQueue, q.Name), slog.Int(pkg.LogKeyQueueLengthAfter, q.storage.Length())} |
| 851 | }) |
| 852 | } |
| 853 | }) |
| 854 | |
| 855 | // set that current task is being processed, so we don't merge it with other tasks |
| 856 | t.SetProcessing(true) |
| 857 | |
| 858 | q.lazydebug("queue tasks after wait", func() []any { |
| 859 | return []any{ |
| 860 | slog.String(pkg.LogKeyQueue, q.Name), |
| 861 | slog.String(pkg.LogKeyTasks, q.String()), |
| 862 | } |
| 863 | }) |
| 864 | |
| 865 | q.lazydebug("queue task to handle", func() []any { |
| 866 | return []any{ |
| 867 | slog.String(pkg.LogKeyQueue, q.Name), |
| 868 | slog.String(pkg.LogKeyTaskType, string(t.GetType())), |
| 869 | } |
| 870 | }) |
| 871 |
no test coverage detected