This commit is contained in:
Stephan D
2026-03-10 12:31:09 +01:00
parent d87e709f43
commit e77d1ab793
287 changed files with 2089 additions and 1550 deletions

View File

@@ -108,7 +108,7 @@ func (s *service) Start(ctx context.Context) {
if runCtx == nil {
runCtx = context.Background()
}
runCtx, s.cancel = context.WithCancel(runCtx)
runCtx, s.cancel = context.WithCancel(runCtx) //nolint:gosec // canceled by Stop; service lifecycle outlives Start scope
for i := 0; i < s.cfg.WorkerConcurrency; i++ {
workerID := "worker-" + strconv.Itoa(i+1)
@@ -143,6 +143,10 @@ func (s *service) runWorker(ctx context.Context, workerID string) {
now := time.Now().UTC()
task, err := s.tasks.LockNextTask(ctx, now, workerID, s.cfg.LockTTL)
if err != nil {
if errors.Is(err, merrors.ErrNoData) {
time.Sleep(s.cfg.WorkerPoll)
continue
}
s.logger.Warn("Failed to lock next task", zap.String("worker_id", workerID), zap.Error(err))
time.Sleep(s.cfg.WorkerPoll)
continue