Merge branch 'master' into I575-jobcleaner-deadline

.github/workflows/ci.yaml

@@ -215,7 +215,7 @@ jobs:
     name: lint
     runs-on: ubuntu-latest
     env:
-      GOLANGCI_LINT_VERSION: v1.60.1
+      GOLANGCI_LINT_VERSION: v1.61.0
     permissions:
       contents: read
       # allow read access to pull request. Use with `only-new-issues` option.

CHANGELOG.md

@@ -40,6 +40,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Fixed a panic that'd occur if `StopAndCancel` was invoked before a client was started. [PR #557](https://github.com/riverqueue/river/pull/557).
 - A `PeriodicJobConstructor` should be able to return `nil` `JobArgs` if it wishes to not have any job inserted. However, this was either never working or was broken at some point. It's now fixed. Thanks [@semanser](https://github.com/semanser)! [PR #572](https://github.com/riverqueue/river/pull/572).
+- Fixed a nil pointer exception if `Client.Subscribe` was called when the client had no configured workers (it still, panics with a more instructive error message now). [PR #599](https://github.com/riverqueue/river/pull/599).
 
 ## [0.11.4] - 2024-08-20
 

client.go

@@ -739,7 +739,7 @@ func (c *Client[TTx]) Start(ctx context.Context) error {
 			}
 		}
 
-		c.queues.fetchCtx = fetchCtx
+		c.queues.fetchCtx = fetchCtx //nolint:fatcontext
 		c.queues.workCtx = workCtx
 		c.workCancel = workCancel
 
@@ -912,6 +912,10 @@ type SubscribeConfig struct {
 
 // Special internal variant that lets us inject an overridden size.
 func (c *Client[TTx]) SubscribeConfig(config *SubscribeConfig) (<-chan *Event, func()) {
+	if c.subscriptionManager == nil {
+		panic("created a subscription on a client that will never work jobs (Workers not configured)")
+	}
+
 	return c.subscriptionManager.SubscribeConfig(config)
 }
 

client_test.go

@@ -3893,6 +3893,19 @@ func Test_Client_Subscribe(t *testing.T) {
 
 		require.Empty(t, client.subscriptionManager.subscriptions)
 	})
+
+	// Just make sure this doesn't fail on a nil pointer exception.
+	t.Run("SubscribeOnClientWithoutWorkers", func(t *testing.T) {
+		t.Parallel()
+
+		dbPool := riverinternaltest.TestDB(ctx, t)
+
+		client := newTestClient(t, dbPool, &Config{})
+
+		require.PanicsWithValue(t, "created a subscription on a client that will never work jobs (Workers not configured)", func() {
+			_, _ = client.Subscribe(EventKindJobCompleted)
+		})
+	})
 }
 
 // SubscribeConfig uses all the same code as Subscribe, so these are just a

go.mod

@@ -6,8 +6,8 @@ toolchain go1.23.0
 
 require (
 	github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa
-	github.com/jackc/pgx/v5 v5.6.0
-	github.com/jackc/puddle/v2 v2.2.1
+	github.com/jackc/pgx/v5 v5.7.1
+	github.com/jackc/puddle/v2 v2.2.2
 	github.com/riverqueue/river/riverdriver v0.11.4
 	github.com/riverqueue/river/riverdriver/riverdatabasesql v0.11.4
 	github.com/riverqueue/river/riverdriver/riverpgxv5 v0.11.4
@@ -17,16 +17,16 @@ require (
 	github.com/stretchr/testify v1.9.0
 	go.uber.org/goleak v1.3.0
 	golang.org/x/sync v0.8.0
-	golang.org/x/text v0.17.0
+	golang.org/x/text v0.18.0
 )
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
-	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/lib/pq v1.10.9 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	golang.org/x/crypto v0.17.0 // indirect
+	golang.org/x/crypto v0.27.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
 

go.sum

@@ -5,12 +5,12 @@ github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa h1:s+4MhCQ6YrzisK6
 github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
-github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
-github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
-github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY=
-github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw=
-github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
-github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.7.1 h1:x7SYsPBYDkHDksogeSmZZ5xzThcTgRz++I5E+ePFUcs=
+github.com/jackc/pgx/v5 v5.7.1/go.mod h1:e7O26IywZZ+naJtWWos6i6fvWK+29etgITqrqHLfoZA=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
 github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -38,12 +38,12 @@ github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsT
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
-golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k=
-golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
+golang.org/x/crypto v0.27.0 h1:GXm2NjJrPaiv/h1tb2UH8QfgC/hOf/+z0p6PT8o1w7A=
+golang.org/x/crypto v0.27.0/go.mod h1:1Xngt8kV6Dvbssa53Ziq6Eqn0HqbZi5Z6R0ZpwQzt70=
 golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ=
 golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
-golang.org/x/text v0.17.0 h1:XtiM5bkSOt+ewxlOE/aE/AKEHibwj/6gvWMl9Rsh0Qc=
-golang.org/x/text v0.17.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
+golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224=
+golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=

go.work.sum

@@ -27,11 +27,13 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.23.0 h1:YfKFowiIMvtgl1UERQoTPPToxltDeZfbj4H7dVUCwmM=
 golang.org/x/sys v0.23.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2 h1:IRJeR9r1pYWsHKTRe/IInb7lYvbBVIqOgsX/u0mbOWY=
 golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE=
 golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY=
 golang.org/x/term v0.23.0 h1:F6D4vR+EHoL9/sWAWgAR1H2DcHr4PareCbAaCo1RpuU=
 golang.org/x/term v0.23.0/go.mod h1:DgV24QBUrK6jhZXl+20l6UWznPlwAHm1Q1mGHtydmSk=
+golang.org/x/term v0.24.0/go.mod h1:lOBK/LVxemqiMij05LGJ0tzNr8xlmwBRJ81PX6wVLH8=
 golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58=
 golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg=

internal/notifier/notifier.go

@@ -319,7 +319,7 @@ func (n *Notifier) listenerUnlisten(ctx context.Context, topic NotificationTopic
 func (n *Notifier) waitOnce(ctx context.Context) error {
 	n.withLock(func() {
 		n.isWaiting = true
-		ctx, n.waitCancel = context.WithCancel(ctx)
+		ctx, n.waitCancel = context.WithCancel(ctx) //nolint:fatcontext
 	})
 	defer n.withLock(func() {
 		n.isWaiting = false

job_list_params.go

@@ -327,7 +327,7 @@ func (p *JobListParams) First(count int) *JobListParams {
 		panic("count must be <= 10000")
 	}
 	paramsCopy := p.copy()
-	paramsCopy.paginationCount = int32(count) //nolint:gosec
+	paramsCopy.paginationCount = int32(count)
 	return paramsCopy
 }
 

queue_list_params.go

@@ -35,6 +35,6 @@ func (p *QueueListParams) First(count int) *QueueListParams {
 		panic("count must be <= 10000")
 	}
 	result := p.copy()
-	result.paginationCount = int32(count) //nolint:gosec
+	result.paginationCount = int32(count)
 	return result
 }

rivershared/levenshtein/levenshtein.go

@@ -62,7 +62,7 @@ func ComputeDistance(str1, str2 string) int {
 
 	// we start from 1 because index 0 is already 0.
 	for i := 1; i < len(distances); i++ {
-		distances[i] = uint16(i)
+		distances[i] = uint16(i) //nolint:gosec
 	}
 
 	// Make a dummy bounds check to prevent the 2 bounds check down below. The
@@ -71,7 +71,7 @@ func ComputeDistance(str1, str2 string) int {
 
 	// fill in the rest
 	for i := 1; i <= lenRuneSlice2; i++ {
-		prev := uint16(i)
+		prev := uint16(i) //nolint:gosec
 		for j := 1; j <= lenRuneSlice1; j++ {
 			current := distances[j-1] // match
 			if runeSlice2[i-1] != runeSlice1[j-1] {

rivershared/levenshtein/levenshtein_test.go

@@ -59,7 +59,7 @@ func TestUnicode(t *testing.T) {
 }
 
 // Benchmarks
-// ----------------------------------------------
+// ----------------------------------------------.
 var sink int //nolint:gochecknoglobals
 
 func BenchmarkSimple(b *testing.B) {

rivershared/startstop/start_stop_test.go

@@ -198,7 +198,7 @@ func TestErrStop(t *testing.T) {
 			return nil
 		}
 
-		workCtx = ctx
+		workCtx = ctx //nolint:fatcontext
 
 		go func() {
 			started()