fix memory leak on job execution from baseservice

The `BaseService` type was originally only used for long-running
internal services. At some point, we refactored the job execution code
to also make use of this type. That exposed a memory leak: every time a
new `BaseService` is `Init`'d, a new random source is allocated. That
meant for every single job.

This is of course totally unnecessary, and to avoid it we can move the
rand source to the `Archetype`, which is only allocated once per client.

Partially fixes #239.

CHANGELOG.md

@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed a memory leak caused by allocating a new random source on every job execution. Thank you @shawnstephens for reporting ❤️ [PR #240](https://github.com/riverqueue/river/pull/240).
 - Fix a problem where `JobListParams.Queues()` didn't filter correctly based on its arguments. [PR #212](https://github.com/riverqueue/river/pull/212).
 - Fix a problem in `DebouncedChan` where it would fire on its "out" channel too often when it was being signaled continuousy on its "in" channel. This would have caused work to be fetched more often than intended in busy systems. [PR #222](https://github.com/riverqueue/river/pull/222).
 

client.go

@@ -24,6 +24,7 @@ import (
 	"github.com/riverqueue/river/internal/maintenance"
 	"github.com/riverqueue/river/internal/notifier"
 	"github.com/riverqueue/river/internal/rivercommon"
+	"github.com/riverqueue/river/internal/util/randutil"
 	"github.com/riverqueue/river/internal/util/sliceutil"
 	"github.com/riverqueue/river/internal/util/valutil"
 	"github.com/riverqueue/river/internal/workunit"
@@ -433,6 +434,7 @@ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client
 	archetype := &baseservice.Archetype{
 		DisableSleep: config.disableSleep,
 		Logger:       config.Logger,
+		Rand:         randutil.NewCryptoSeededConcurrentSafeRand(),
 		TimeNowUTC:   func() time.Time { return time.Now().UTC() },
 	}
 

internal/baseservice/base_service.go

@@ -30,6 +30,15 @@ type Archetype struct {
 	// Logger is a structured logger.
 	Logger *slog.Logger
 
+	// Rand is a random source safe for concurrent access and seeded with a
+	// cryptographically random seed to ensure good distribution between nodes
+	// and services. The random source itself is _not_ cryptographically secure,
+	// and therefore should not be used anywhere security-related. This is a
+	// deliberate choice because Go's non-crypto rand source is about twenty
+	// times faster, and so far none of our uses of random require cryptographic
+	// secure randomness.
+	Rand *rand.Rand
+
 	// TimeNowUTC returns the current time as UTC. Normally it's implemented as
 	// a call to `time.Now().UTC()`, but may be overridden in tests for time
 	// injection. Services should try to use this function instead of the
@@ -116,7 +125,7 @@ func Init[TService withBaseService](archetype *Archetype, service TService) TSer
 	baseService.DisableSleep = archetype.DisableSleep
 	baseService.Logger = archetype.Logger
 	baseService.Name = reflect.TypeOf(service).Elem().Name()
-	baseService.Rand = randutil.NewCryptoSeededConcurrentSafeRand()
+	baseService.Rand = archetype.Rand
 	baseService.TimeNowUTC = archetype.TimeNowUTC
 
 	return service

internal/baseservice/base_service_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/riverqueue/river/internal/util/randutil"
 	"github.com/stretchr/testify/require"
 )
 
@@ -69,6 +70,7 @@ func archetype() *Archetype {
 	return &Archetype{
 		DisableSleep: true,
 		Logger:       slog.New(slog.NewTextHandler(os.Stdout, nil)),
+		Rand:         randutil.NewCryptoSeededConcurrentSafeRand(),
 		TimeNowUTC:   func() time.Time { return time.Now().UTC() },
 	}
 }

internal/riverinternaltest/riverinternaltest.go

@@ -24,6 +24,7 @@ import (
 	"github.com/riverqueue/river/internal/rivercommon"
 	"github.com/riverqueue/river/internal/riverinternaltest/slogtest" //nolint:depguard
 	"github.com/riverqueue/river/internal/testdb"
+	"github.com/riverqueue/river/internal/util/randutil"
 	"github.com/riverqueue/river/internal/util/valutil"
 )
 
@@ -57,6 +58,7 @@ func BaseServiceArchetype(tb testing.TB) *baseservice.Archetype {
 
 	return &baseservice.Archetype{
 		Logger:     Logger(tb),
+		Rand:       randutil.NewCryptoSeededConcurrentSafeRand(),
 		TimeNowUTC: func() time.Time { return time.Now().UTC() },
 	}
 }

rivermigrate/river_migrate.go

@@ -19,6 +19,7 @@ import (
 	"github.com/riverqueue/river/internal/baseservice"
 	"github.com/riverqueue/river/internal/util/dbutil"
 	"github.com/riverqueue/river/internal/util/maputil"
+	"github.com/riverqueue/river/internal/util/randutil"
 	"github.com/riverqueue/river/internal/util/sliceutil"
 	"github.com/riverqueue/river/riverdriver"
 )
@@ -94,6 +95,7 @@ func New[TTx any](driver riverdriver.Driver[TTx], config *Config) *Migrator[TTx]
 
 	archetype := &baseservice.Archetype{
 		Logger:     logger,
+		Rand:       randutil.NewCryptoSeededConcurrentSafeRand(),
 		TimeNowUTC: func() time.Time { return time.Now().UTC() },
 	}