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")  | 
