Method Start

pkg/task/queue/task_queue.go:756–806

(ctx context.Context)

}

func (q *TaskQueue) Start(ctx context.Context) {
	if q.started.Load() {
		return
	}

	if q.Handler == nil {
		log.Error("should set handler before start in queue", slog.String(pkg.LogKeyName, q.Name))
		q.SetStatus(QueueStatusNoHandlerSet)
		return
	}

	// Start metrics monitoring goroutine
	q.startMetricsMonitoring()

	go func() {
		// Outer recover is a last-resort safety net: it prevents the whole worker
		// goroutine from being torn down if a panic escapes processOne (e.g. from
		// q.compaction or q.waitForTask). The main per-task recovery happens
		// inside processOne so that the for-loop keeps running after a panic.
		defer func() {
			if r := recover(); r != nil {
				q.logger.Error("panic escaped TaskQueue worker, queue stopped",
					slog.String(pkg.LogKeyQueue, q.Name),
					slog.Any(pkg.LogKeyError, r),
					slog.String(pkg.LogKeyStack, string(debug.Stack())),
				)
				q.SetStatus(QueueStatusStop)
			}
		}()

		q.SetStatus(QueueStatusIdle)
		var sleepDelay time.Duration
		for {
			q.logger.Debug("queue: wait for task", slog.String(pkg.LogKeyQueue, q.Name), slog.Duration(pkg.LogKeySleepDelay, sleepDelay))
			t := q.waitForTask(sleepDelay)
			if t == nil {
				q.SetStatus(QueueStatusStop)
				q.logger.Info("queue stopped", slog.String(pkg.LogKeyName, q.Name))
				return
			}

			nextSleepDelay, stop := q.processOne(ctx, t)
			if stop {
				return
			}
			sleepDelay = nextSleepDelay
		}
	}()

	q.started.Store(true)
}

// processOne processes a single task fetched from the queue.
// It contains its own recover() so a panic inside q.Handler does NOT kill