fix memory leak of job cancellation contexts

When remote job cancellation was added, a new cancellable context was
allocated within the producer before the executor is spawned. The cancel
func here was only called if the job was actually cancelled remotely or
via a parent context cancellation, meaning we would slowly leak memory
for every job worked that wasn't cancelled.

Thank you @brandur for pinpointing the issue.

Fixes #239.

Co-Authored-By: Brandur Leach <brandur@brandur.org>

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Fixed a memory leak caused by not always cancelling the context used to enable jobs to be cancelled remotely. [PR #243](https://github.com/riverqueue/river/pull/243).
+
 ## [0.0.23] - 2024-02-29
 
 ### Added

job_executor.go

@@ -138,6 +138,9 @@ func (e *jobExecutor) Cancel() {
 }
 
 func (e *jobExecutor) Execute(ctx context.Context) {
+	// Ensure that the context is cancelled no matter what, or it will leak:
+	defer e.CancelFunc(nil)
+
 	e.start = e.TimeNowUTC()
 	e.stats = &jobstats.JobStatistics{
 		QueueWaitDuration: e.start.Sub(e.JobRow.ScheduledAt),

job_executor_test.go

@@ -174,7 +174,12 @@ func TestJobExecutor_Execute(t *testing.T) {
 			jobRow:            job,
 		}
 
+		// allocate this context just so we can set the CancelFunc:
+		_, cancel := context.WithCancelCause(ctx)
+		t.Cleanup(func() { cancel(nil) })
+
 		executor := baseservice.Init(archetype, &jobExecutor{
+			CancelFunc:             cancel,
 			ClientRetryPolicy:      &retryPolicyNoJitter{},
 			Completer:              bundle.completer,
 			ErrorHandler:           bundle.errorHandler,
@@ -640,6 +645,7 @@ func TestJobExecutor_Execute(t *testing.T) {
 
 		workCtx, cancelFunc := context.WithCancelCause(ctx)
 		executor.CancelFunc = cancelFunc
+		t.Cleanup(func() { cancelFunc(nil) })
 
 		executor.Execute(workCtx)
 		executor.Completer.Wait()

producer.go

@@ -358,6 +358,7 @@ func (p *producer) startNewExecutors(workCtx context.Context, jobs []*rivertype.
 			workUnit = workInfo.workUnitFactory.MakeUnit(job)
 		}
 
+		// jobCancel will always be called by the executor to prevent leaks.
 		jobCtx, jobCancel := context.WithCancelCause(workCtx)
 
 		executor := baseservice.Init(&p.Archetype, &jobExecutor{