Substantially (~20-45x) faster unique insertion using unique index

CHANGELOG.md

@@ -14,6 +14,20 @@ go install github.com/riverqueue/river/cmd/river@latest
 river migrate-up --database-url "$DATABASE_URL"
 ```
 
+The migration **includes a new index**. Users with a very large job table may want to consider raising the index separately using `CONCURRENTLY` (which must be run outside of a transaction), then run `river migrate-up` to finalize the process (it will tolerate an index that already exists):
+
+```sql
+ALTER TABLE river_job
+    ADD COLUMN unique_key bytea;
+
+CREATE UNIQUE INDEX CONCURRENTLY river_job_kind_unique_key_idx ON river_job (kind, unique_key) WHERE unique_key IS NOT NULL;
+```
+
+```shell
+go install github.com/riverqueue/river/cmd/river@latest
+river migrate-up --database-url "$DATABASE_URL"
+```
+
 ### Added
 
 - Fully functional driver for `database/sql` for use with packages like Bun and GORM. [PR #351](https://github.com/riverqueue/river/pull/351).

client.go

@@ -65,6 +65,10 @@ type Config struct {
 	// option then it's recommended to leave it unset because the prefix leaves
 	// only 32 bits of number space for advisory lock hashes, so it makes
 	// internally conflicting River-generated keys more likely.
+	//
+	// Advisory locks are currently only used for the fallback/slow path of
+	// unique job insertion where finalized states are included in a ByState
+	// configuration.
 	AdvisoryLockPrefix int32
 
 	// CancelledJobRetentionPeriod is the amount of time to keep cancelled jobs

cmd/river/go.mod

@@ -10,6 +10,8 @@ replace github.com/riverqueue/river/riverdriver/riverdatabasesql => ../../riverd
 
 replace github.com/riverqueue/river/riverdriver/riverpgxv5 => ../../riverdriver/riverpgxv5
 
+replace github.com/riverqueue/river/rivertype => ../../rivertype
+
 require (
 	github.com/jackc/pgx/v5 v5.6.0
 	github.com/lmittmann/tint v1.0.4

driver_test.go

@@ -192,3 +192,62 @@ func BenchmarkDriverRiverPgxV5_Executor(b *testing.B) {
 		})
 	})
 }
+
+func BenchmarkDriverRiverPgxV5Insert(b *testing.B) {
+	ctx := context.Background()
+
+	type testBundle struct {
+		exec riverdriver.Executor
+		tx   pgx.Tx
+	}
+
+	setup := func(b *testing.B) (*riverpgxv5.Driver, *testBundle) {
+		b.Helper()
+
+		var (
+			driver = riverpgxv5.New(nil)
+			tx     = riverinternaltest.TestTx(ctx, b)
+		)
+
+		bundle := &testBundle{
+			exec: driver.UnwrapExecutor(tx),
+			tx:   tx,
+		}
+
+		return driver, bundle
+	}
+
+	b.Run("InsertFast", func(b *testing.B) {
+		_, bundle := setup(b)
+
+		for n := 0; n < b.N; n++ {
+			_, err := bundle.exec.JobInsertFast(ctx, &riverdriver.JobInsertFastParams{
+				EncodedArgs: []byte(`{"encoded": "args"}`),
+				Kind:        "test_kind",
+				MaxAttempts: rivercommon.MaxAttemptsDefault,
+				Priority:    rivercommon.PriorityDefault,
+				Queue:       rivercommon.QueueDefault,
+				State:       rivertype.JobStateAvailable,
+			})
+			require.NoError(b, err)
+		}
+	})
+
+	b.Run("InsertUnique", func(b *testing.B) {
+		_, bundle := setup(b)
+
+		for n := 0; n < b.N; n++ {
+			_, err := bundle.exec.JobInsertUnique(ctx, &riverdriver.JobInsertUniqueParams{
+				JobInsertFastParams: &riverdriver.JobInsertFastParams{
+					EncodedArgs: []byte(`{"encoded": "args"}`),
+					Kind:        "test_kind",
+					MaxAttempts: rivercommon.MaxAttemptsDefault,
+					Priority:    rivercommon.PriorityDefault,
+					Queue:       rivercommon.QueueDefault,
+					State:       rivertype.JobStateAvailable,
+				},
+			})
+			require.NoError(b, err)
+		}
+	})
+}

insert_opts.go

@@ -136,6 +136,11 @@ type UniqueOpts struct {
 	// With this setting, any jobs of the same kind that have been completed or
 	// discarded, but not yet cleaned out by the system, won't count towards the
 	// uniqueness of a new insert.
+	//
+	// Warning: A non-default slice of states in ByState will force the unique
+	// inserter to fall back to a slower insertion path that takes an advisory
+	// lock and performs a look up before insertion.  For best performance, it's
+	// recommended that the default set of states is used.
 	ByState []rivertype.JobState
 }