diff options
Diffstat (limited to 'internal/workers')
-rw-r--r-- | internal/workers/worker_msg.go | 21 | ||||
-rw-r--r-- | internal/workers/workers.go | 10 |
2 files changed, 26 insertions, 5 deletions
diff --git a/internal/workers/worker_msg.go b/internal/workers/worker_msg.go index 92180651a..c7dc568d7 100644 --- a/internal/workers/worker_msg.go +++ b/internal/workers/worker_msg.go @@ -19,6 +19,7 @@ package workers import ( "context" + "errors" "codeberg.org/gruf/go-runners" "codeberg.org/gruf/go-structr" @@ -147,9 +148,25 @@ func (w *MsgWorker[T]) process(ctx context.Context) { return } - // Attempt to process popped message type. - if err := w.Process(ctx, msg); err != nil { + // Attempt to process message. + err := w.Process(ctx, msg) + if err != nil { log.Errorf(ctx, "%p: error processing: %v", w, err) + + if errors.Is(err, context.Canceled) && + ctx.Err() != nil { + // In the case of our own context + // being cancelled, push message + // back onto queue for persisting. + // + // Note we specifically check against + // context.Canceled here as it will + // be faster than the mutex lock of + // ctx.Err(), so gives an initial + // faster check in the if-clause. + w.Queue.Push(msg) + break + } } } } diff --git a/internal/workers/workers.go b/internal/workers/workers.go index 4d2b146b6..377a9d899 100644 --- a/internal/workers/workers.go +++ b/internal/workers/workers.go @@ -55,7 +55,8 @@ type Workers struct { // StartScheduler starts the job scheduler. func (w *Workers) StartScheduler() { - _ = w.Scheduler.Start() // false = already running + _ = w.Scheduler.Start() + // false = already running log.Info(nil, "started scheduler") } @@ -82,9 +83,12 @@ func (w *Workers) Start() { log.Infof(nil, "started %d dereference workers", n) } -// Stop will stop all of the contained worker pools (and global scheduler). +// Stop will stop all of the contained +// worker pools (and global scheduler). func (w *Workers) Stop() { - _ = w.Scheduler.Stop() // false = not running + _ = w.Scheduler.Stop() + // false = not running + log.Info(nil, "stopped scheduler") w.Delivery.Stop() log.Info(nil, "stopped delivery workers") |