diff --git a/CHANGELOG.md b/CHANGELOG.md index 575e069..b7aebbd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,23 @@ All notable changes to this project are documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [v0.7.1] - 2026-07-02 + +Quieter shutdowns: a job dispatch cancelled during shutdown is no longer reported +as an error. + +### Changed + +- A cancelled job (`context.Canceled`) is now treated as an expected interruption + rather than a failure. When a worker function returns a `context.Canceled` error + (directly or wrapped), or a job dispatch returns one because the worker-thread + context was cancelled (e.g. during shutdown), `jobworker` no longer invokes the + `OnError` callback or logs the "Job error" / "Error while dispatching the job" + lines at `ERROR`/`WARN`. The job is still reset and retried without consuming a + retry attempt, exactly as before — only the spurious error reporting is dropped. + Genuine failures and per-job `JobTimeout` (`context.DeadlineExceeded`) are + unaffected and still reported. + ## [v0.7.0] - 2026-06-18 Faster job claiming via a cached prepared statement, a dedicated claim index, @@ -227,6 +244,7 @@ commit; Last release before the worker liveness heartbeat work. See the git history for details of `v0.5.4` and earlier releases. +[v0.7.1]: https://github.com/domonda/go-jobqueue/compare/v0.7.0...v0.7.1 [v0.7.0]: https://github.com/domonda/go-jobqueue/compare/v0.6.0...v0.7.0 [v0.6.0]: https://github.com/domonda/go-jobqueue/compare/v0.5.4...v0.6.0 [v0.5.4]: https://github.com/domonda/go-jobqueue/releases/tag/v0.5.4 diff --git a/VERSION b/VERSION index e7f5d1a..6425bba 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -v0.7.0 \ No newline at end of file +v0.7.1 \ No newline at end of file diff --git a/jobworker/dojob.go b/jobworker/dojob.go index 5749a62..cb444ee 100644 --- a/jobworker/dojob.go +++ b/jobworker/dojob.go @@ -65,25 +65,30 @@ func DoJob(ctx context.Context, job *jobqueue.Job) (err error) { result, jobErr := worker(jobCtx, job) if jobErr != nil { - errorTitle := errs.Root(jobErr).Error() - if nl := strings.IndexByte(errorTitle, '\n'); nl > 0 { - // Only use first line of error message as errorTitle - errorTitle = errorTitle[:nl] - } - errorTitle = strings.TrimSpace(errorTitle) + // A cancelled job (e.g. during shutdown) is an expected interruption: + // doJobAndSaveResultInDB resets and retries it without consuming a retry + // attempt, so don't report it via OnError or log it as an error/warning. + if !errors.Is(jobErr, context.Canceled) { + errorTitle := errs.Root(jobErr).Error() + if nl := strings.IndexByte(errorTitle, '\n'); nl > 0 { + // Only use first line of error message as errorTitle + errorTitle = errorTitle[:nl] + } + errorTitle = strings.TrimSpace(errorTitle) - OnError(jobErr) + OnError(jobErr) - if job.CurrentRetryCount >= job.MaxRetryCount { - log.ErrorfCtx(jobCtx, "Job error: %s", errorTitle). - Any("job", job). - Err(jobErr). - Log() - } else { - log.WarnfCtx(jobCtx, "Job error: %s", errorTitle). - Any("job", job). - Err(jobErr). - Log() + if job.CurrentRetryCount >= job.MaxRetryCount { + log.ErrorfCtx(jobCtx, "Job error: %s", errorTitle). + Any("job", job). + Err(jobErr). + Log() + } else { + log.WarnfCtx(jobCtx, "Job error: %s", errorTitle). + Any("job", job). + Err(jobErr). + Log() + } } job.ErrorMsg.Set(jobErr.Error()) diff --git a/jobworker/dojob_internal_test.go b/jobworker/dojob_internal_test.go new file mode 100644 index 0000000..27b64cf --- /dev/null +++ b/jobworker/dojob_internal_test.go @@ -0,0 +1,69 @@ +package jobworker + +import ( + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/domonda/go-errs" + + "github.com/domonda/go-jobqueue" +) + +// TestDoJobCanceledJobSkipsOnErrorAndLog verifies the cancellation guard in DoJob: +// a worker that returns a context.Canceled error (directly or wrapped) is an +// expected interruption, so DoJob must NOT invoke OnError or log it, yet it must +// still return the error so doJobAndSaveResultInDB can detect the cancellation and +// reset/retry the job. A genuine failure is the control: OnError still fires. +func TestDoJobCanceledJobSkipsOnErrorAndLog(t *testing.T) { + resetWorkerRegistryState(t) + + // OnError is a package-global callback; capture invocations and restore it. + origOnError := OnError + t.Cleanup(func() { OnError = origOnError }) + var onErrorCalls []error + OnError = func(err error) { onErrorCalls = append(onErrorCalls, err) } + + cases := []struct { + name string + workerErr error + wantOnError bool + wantCancel bool + }{ + {"direct context.Canceled", context.Canceled, false, true}, + {"wrapped context.Canceled", errs.Errorf("dialing backend: %w", context.Canceled), false, true}, + {"genuine failure", errs.New("boom"), true, false}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + onErrorCalls = nil + jobType := JobType("cancel-guard-" + tc.name) + Register(jobType, func(context.Context, *jobqueue.Job) (any, error) { + return nil, tc.workerErr + }) + + job := &jobqueue.Job{Type: jobType} + err := DoJob(t.Context(), job) + + // The error is always returned so the caller can act on it. + require.Error(t, err) + assert.ErrorIs(t, err, tc.workerErr) + + if tc.wantCancel { + // The worker-thread guard relies on context.Canceled surviving the + // wrapping DoJob applies to its returned error. + assert.ErrorIs(t, err, context.Canceled, + "cancellation must propagate out of DoJob") + } + + if tc.wantOnError { + assert.Len(t, onErrorCalls, 1, "OnError must fire for a genuine failure") + } else { + assert.Empty(t, onErrorCalls, "OnError must be skipped for a cancelled job") + } + }) + } +} diff --git a/jobworker/workerthreads.go b/jobworker/workerthreads.go index b2c09f0..7bc966b 100644 --- a/jobworker/workerthreads.go +++ b/jobworker/workerthreads.go @@ -175,7 +175,9 @@ func worker(threadIndex int) { for job := nextJob(ctx); job != nil; job = nextJob(ctx) { err := doJobAndSaveResultInDB(ctx, job) - if err != nil { + // A cancelled dispatch is an expected shutdown interruption: + // the job is reset and retried, so don't report it as an error. + if err != nil && !errors.Is(err, context.Canceled) { OnError(err) log.ErrorCtx(ctx, "Error while dispatching the job"). Err(err).