diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ea9e7c86..6c511112b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +⚠️ This release contains a new database migration, version 8, that only affects SQLite. It rebuilds `river_job` with `AUTOINCREMENT` to prevent automatically generated job IDs from being reused after deletion. The migration is a no-op for PostgreSQL. On SQLite, both migration directions refuse to run if River Pro schema is already installed, to avoid discarding its additions to `river_job`. + ### Added - Added `Config.LeaderElectionDisabled` to let a client work jobs without participating in leader election or running maintenance services. Other eligible clients in the same database and schema continue handling scheduling, retries, periodic enqueueing, rescue, and cleanup. [PR #1382](https://github.com/riverqueue/river/pull/1382). @@ -22,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Improved PostgreSQL job listing performance when filtering by one finalized state (`completed`, `cancelled`, or `discarded`) and sorting by finalized time, including in River UI. [PR #1374](https://github.com/riverqueue/river/pull/1374). - Fixed `JobRescuer` overwriting jobs that complete, leave the running state, or are claimed again by another worker after being fetched for rescue, preserving their state, errors, metadata, and timestamps across PostgreSQL and SQLite drivers. Fixes [#1302](https://github.com/riverqueue/river/issues/1302). [PR #1373](https://github.com/riverqueue/river/pull/1373). - Fixed SQLite notification listeners delivering notifications from before a subscription or from an unsubscribe gap. Notification reads now fetch subscribed topics in bounded batches, and cleanup deletes expired notifications in batches of 10,000 rows (reduced to 1,000 after repeated timeouts), with pauses between batches to reduce write lock contention. [PR #1381](https://github.com/riverqueue/river/pull/1381). +- Fixed SQLite reusing the ID of a deleted job when that job held the largest ID, which could cause an ID observed earlier to refer to an unrelated job later. [PR #1390](https://github.com/riverqueue/river/pull/1390). ## [0.47.0] - 2026-09-01 diff --git a/riverdriver/river_driver_interface.go b/riverdriver/river_driver_interface.go index 49c3067b6..90f4fdba7 100644 --- a/riverdriver/river_driver_interface.go +++ b/riverdriver/river_driver_interface.go @@ -964,7 +964,7 @@ func MigrationLineMainTruncateTables(version int) []string { return []string{"river_job", "river_leader", "river_queue"} case 5, 6: return []string{"river_job", "river_leader", "river_queue", "river_client", "river_client_queue"} - case 0, 7: + case 0, 7, 8: return []string{"river_job", "river_leader", "river_queue", "river_notification"} } diff --git a/riverdriver/riverdatabasesql/migration/main/008_job_id_autoincrement.down.sql b/riverdriver/riverdatabasesql/migration/main/008_job_id_autoincrement.down.sql new file mode 100644 index 000000000..695357bb8 --- /dev/null +++ b/riverdriver/riverdatabasesql/migration/main/008_job_id_autoincrement.down.sql @@ -0,0 +1,3 @@ +-- No-op. PostgreSQL sequences already prevent automatically generated job IDs +-- from being reused. +SELECT 1; diff --git a/riverdriver/riverdatabasesql/migration/main/008_job_id_autoincrement.up.sql b/riverdriver/riverdatabasesql/migration/main/008_job_id_autoincrement.up.sql new file mode 100644 index 000000000..695357bb8 --- /dev/null +++ b/riverdriver/riverdatabasesql/migration/main/008_job_id_autoincrement.up.sql @@ -0,0 +1,3 @@ +-- No-op. PostgreSQL sequences already prevent automatically generated job IDs +-- from being reused. +SELECT 1; diff --git a/riverdriver/riverdrivertest/job_insert.go b/riverdriver/riverdrivertest/job_insert.go index 6b625a0e5..ed432d145 100644 --- a/riverdriver/riverdrivertest/job_insert.go +++ b/riverdriver/riverdrivertest/job_insert.go @@ -583,6 +583,20 @@ func exerciseJobInsert[TTx any](ctx context.Context, t *testing.T, t.Run("JobInsertFull", func(t *testing.T) { t.Parallel() + t.Run("DoesNotReuseAutomaticallyGeneratedID", func(t *testing.T) { + t.Parallel() + + exec, _ := setup(ctx, t) + + job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{}) + + _, err := exec.JobDelete(ctx, &riverdriver.JobDeleteParams{ID: job.ID}) + require.NoError(t, err) + + jobAfter := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{}) + require.Greater(t, jobAfter.ID, job.ID) + }) + t.Run("MinimalArgsWithDefaults", func(t *testing.T) { t.Parallel() diff --git a/riverdriver/riverdrivertest/migration.go b/riverdriver/riverdrivertest/migration.go index 6d96b51c3..bf37e18b1 100644 --- a/riverdriver/riverdrivertest/migration.go +++ b/riverdriver/riverdrivertest/migration.go @@ -73,6 +73,8 @@ func exerciseMigration[TTx any](ctx context.Context, t *testing.T, driver.GetMigrationTruncateTables(riverdriver.MigrationLineMain, 6)) require.Equal(t, expectedLatestTables, driver.GetMigrationTruncateTables(riverdriver.MigrationLineMain, 7)) + require.Equal(t, expectedLatestTables, + driver.GetMigrationTruncateTables(riverdriver.MigrationLineMain, 8)) require.Equal(t, expectedLatestTables, driver.GetMigrationTruncateTables(riverdriver.MigrationLineMain, 0)) }) @@ -144,7 +146,7 @@ func exerciseMigration[TTx any](ctx context.Context, t *testing.T, } }) - t.Run("MigrateDownFromVersionSevenWithJobData", func(t *testing.T) { + t.Run("MigrateDownFromVersionEightWithJobData", func(t *testing.T) { t.Parallel() driver, schema := driverWithSchema(ctx, t, &riverdbtest.TestSchemaOpts{ @@ -171,6 +173,39 @@ func exerciseMigration[TTx any](ctx context.Context, t *testing.T, require.NotZero(t, job.ID) }) + t.Run("MigrateUpFromVersionSevenWithJobData", func(t *testing.T) { + t.Parallel() + + driver, schema := driverWithSchema(ctx, t, &riverdbtest.TestSchemaOpts{ + DisableReuse: true, + LineTargetVersions: map[string]int{ + riverdriver.MigrationLineMain: 7, + }, + }) + exec := driver.GetExecutor() + + job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Schema: schema}) + + migrator, err := rivermigrate.New(driver, &rivermigrate.Config{ + Line: riverdriver.MigrationLineMain, + Logger: riversharedtest.Logger(t), + Schema: schema, + }) + require.NoError(t, err) + + _, err = migrator.Migrate(ctx, rivermigrate.DirectionUp, nil) + require.NoError(t, err) + + job, err = exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: schema}) + require.NoError(t, err) + + _, err = exec.JobDelete(ctx, &riverdriver.JobDeleteParams{ID: job.ID, Schema: schema}) + require.NoError(t, err) + + jobAfter := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Schema: schema}) + require.Greater(t, jobAfter.ID, job.ID) + }) + t.Run("MigrateUpFromVersionSixWithQueueData", func(t *testing.T) { t.Parallel() @@ -215,6 +250,69 @@ func exerciseMigration[TTx any](ctx context.Context, t *testing.T, require.NotZero(t, queue.UpdatedAt) }) + t.Run("MigrateVersionEightRejectsProSchema", func(t *testing.T) { + t.Parallel() + + for _, testCase := range []struct { + name string + sql string + }{ + {"LegacyWorkflow", `CREATE INDEX river_job_workflow_scheduling ON river_job (state)`}, + {"Pro", `CREATE TABLE river_job_sequence (id integer PRIMARY KEY, key text)`}, + {"WorkflowV2", `CREATE TABLE river_workflow (id text PRIMARY KEY)`}, + } { + t.Run(testCase.name, func(t *testing.T) { + t.Parallel() + + for _, direction := range []rivermigrate.Direction{rivermigrate.DirectionDown, rivermigrate.DirectionUp} { + t.Run(string(direction), func(t *testing.T) { + t.Parallel() + + version := 7 + if direction == rivermigrate.DirectionDown { + version = 8 + } + driver, schema := driverWithSchema(ctx, t, &riverdbtest.TestSchemaOpts{ + DisableReuse: true, + LineTargetVersions: map[string]int{ + riverdriver.MigrationLineMain: version, + }, + Lines: []string{riverdriver.MigrationLineMain}, + }) + if driver.DatabaseName() != riverdriver.DatabaseNameSQLite { + t.Skip("SQLite table rebuild") + } + exec := driver.GetExecutor() + job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{Schema: schema}) + + // Deliberately omit Pro migration records, as when applying SQL + // through an external migration tool. + require.NoError(t, exec.Exec(ctx, testCase.sql)) + require.NoError(t, exec.Exec(ctx, `ALTER TABLE river_job ADD COLUMN partition_key text`)) + migrator, err := rivermigrate.New(driver, &rivermigrate.Config{Logger: riversharedtest.Logger(t), Schema: schema}) + require.NoError(t, err) + + _, err = migrator.Migrate(ctx, direction, &rivermigrate.MigrateOpts{MaxSteps: 1}) + require.ErrorContains(t, err, "River SQLite migration 008 cannot run while River Pro schema is installed") + + jobAfter, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: schema}) + require.NoError(t, err) + require.Equal(t, job, jobAfter) + exists, err := exec.ColumnExists(ctx, &riverdriver.ColumnExistsParams{Column: "partition_key", Schema: schema, Table: "river_job"}) + require.NoError(t, err) + require.True(t, exists) + exists, err = exec.IndexExists(ctx, &riverdriver.IndexExistsParams{Index: "river_job_kind", Schema: schema}) + require.NoError(t, err) + require.True(t, exists) + migrations, err := exec.MigrationGetByLine(ctx, &riverdriver.MigrationGetByLineParams{Line: riverdriver.MigrationLineMain, Schema: schema}) + require.NoError(t, err) + require.Len(t, migrations, version) + }) + } + }) + } + }) + type testBundle struct { driver riverdriver.Driver[TTx] } diff --git a/riverdriver/riverpgxv5/migration/main/008_job_id_autoincrement.down.sql b/riverdriver/riverpgxv5/migration/main/008_job_id_autoincrement.down.sql new file mode 100644 index 000000000..695357bb8 --- /dev/null +++ b/riverdriver/riverpgxv5/migration/main/008_job_id_autoincrement.down.sql @@ -0,0 +1,3 @@ +-- No-op. PostgreSQL sequences already prevent automatically generated job IDs +-- from being reused. +SELECT 1; diff --git a/riverdriver/riverpgxv5/migration/main/008_job_id_autoincrement.up.sql b/riverdriver/riverpgxv5/migration/main/008_job_id_autoincrement.up.sql new file mode 100644 index 000000000..695357bb8 --- /dev/null +++ b/riverdriver/riverpgxv5/migration/main/008_job_id_autoincrement.up.sql @@ -0,0 +1,3 @@ +-- No-op. PostgreSQL sequences already prevent automatically generated job IDs +-- from being reused. +SELECT 1; diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql index 1cf46567b..c3f2f02fd 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql @@ -1,5 +1,6 @@ CREATE TABLE river_job ( - id integer PRIMARY KEY, -- SQLite makes this autoincrementing automatically + -- AUTOINCREMENT prevents SQLite from reusing the IDs of deleted jobs. + id integer PRIMARY KEY AUTOINCREMENT, args jsonb NOT NULL DEFAULT (jsonb('{}')), attempt integer NOT NULL DEFAULT 0, attempted_at timestamp, diff --git a/riverdriver/riversqlite/migration/main/006_bulk_unique.up.sql b/riverdriver/riversqlite/migration/main/006_bulk_unique.up.sql index 83e4991ac..528a4680e 100644 --- a/riverdriver/riversqlite/migration/main/006_bulk_unique.up.sql +++ b/riverdriver/riversqlite/migration/main/006_bulk_unique.up.sql @@ -5,7 +5,7 @@ DROP TABLE /* TEMPLATE: schema */river_job; CREATE TABLE /* TEMPLATE: schema */river_job ( - id integer PRIMARY KEY, -- SQLite makes this autoincrementing automatically + id integer PRIMARY KEY, -- SQLite aliases this to ROWID, which may reuse deleted IDs. args blob NOT NULL DEFAULT '{}', attempt integer NOT NULL DEFAULT 0, attempted_at timestamp, @@ -60,4 +60,4 @@ CREATE UNIQUE INDEX /* TEMPLATE: schema */river_job_unique_idx ON river_job (uni WHEN 'running' THEN unique_states & (1 << 6) WHEN 'scheduled' THEN unique_states & (1 << 7) ELSE 0 - END >= 1; \ No newline at end of file + END >= 1; diff --git a/riverdriver/riversqlite/migration/main/007_notification_outbox_sqlite_jsonb_and_sql_cleanup.down.sql b/riverdriver/riversqlite/migration/main/007_notification_outbox_sqlite_jsonb_and_sql_cleanup.down.sql index 24944ad81..1e3bcffb0 100644 --- a/riverdriver/riversqlite/migration/main/007_notification_outbox_sqlite_jsonb_and_sql_cleanup.down.sql +++ b/riverdriver/riversqlite/migration/main/007_notification_outbox_sqlite_jsonb_and_sql_cleanup.down.sql @@ -53,7 +53,7 @@ DROP INDEX /* TEMPLATE: schema */river_job_unique_idx; ALTER TABLE /* TEMPLATE: schema */river_job RENAME TO river_job_old; CREATE TABLE /* TEMPLATE: schema */river_job ( - id integer PRIMARY KEY, -- SQLite makes this autoincrementing automatically + id integer PRIMARY KEY, -- SQLite aliases this to ROWID, which may reuse deleted IDs. args blob NOT NULL DEFAULT '{}', attempt integer NOT NULL DEFAULT 0, attempted_at timestamp, diff --git a/riverdriver/riversqlite/migration/main/007_notification_outbox_sqlite_jsonb_and_sql_cleanup.up.sql b/riverdriver/riversqlite/migration/main/007_notification_outbox_sqlite_jsonb_and_sql_cleanup.up.sql index 511aedeb5..b1ca9479e 100644 --- a/riverdriver/riversqlite/migration/main/007_notification_outbox_sqlite_jsonb_and_sql_cleanup.up.sql +++ b/riverdriver/riversqlite/migration/main/007_notification_outbox_sqlite_jsonb_and_sql_cleanup.up.sql @@ -35,7 +35,7 @@ DROP INDEX /* TEMPLATE: schema */river_job_unique_idx; ALTER TABLE /* TEMPLATE: schema */river_job RENAME TO river_job_old; CREATE TABLE /* TEMPLATE: schema */river_job ( - id integer PRIMARY KEY, -- SQLite makes this autoincrementing automatically + id integer PRIMARY KEY, -- SQLite aliases this to ROWID, which may reuse deleted IDs. args blob NOT NULL DEFAULT (jsonb('{}')), attempt integer NOT NULL DEFAULT 0, attempted_at timestamp, diff --git a/riverdriver/riversqlite/migration/main/008_job_id_autoincrement.down.sql b/riverdriver/riversqlite/migration/main/008_job_id_autoincrement.down.sql new file mode 100644 index 000000000..aad4f366b --- /dev/null +++ b/riverdriver/riversqlite/migration/main/008_job_id_autoincrement.down.sql @@ -0,0 +1,121 @@ +-- Rebuild river_job to restore SQLite's default ROWID allocation behavior. + +-- Rebuilding river_job would discard schema installed by River Pro. Check +-- schema objects instead of migration records to also catch manually applied +-- Pro migrations and the legacy workflow migration line. +CREATE TEMP TABLE river_job_pro_schema_guard ( + id integer NOT NULL +); + +CREATE TEMP TRIGGER river_job_pro_schema_guard_enforce + BEFORE INSERT ON river_job_pro_schema_guard + WHEN EXISTS ( + SELECT 1 + FROM /* TEMPLATE: schema */sqlite_master + WHERE name IN ('river_job_sequence', 'river_job_workflow_scheduling', 'river_workflow') + ) +BEGIN + SELECT RAISE(ABORT, 'River SQLite migration 008 cannot run while River Pro schema is installed'); +END; + +INSERT INTO river_job_pro_schema_guard (id) VALUES (1); + +DROP TRIGGER river_job_pro_schema_guard_enforce; +DROP TABLE river_job_pro_schema_guard; + +DROP INDEX /* TEMPLATE: schema */river_job_kind; +DROP INDEX /* TEMPLATE: schema */river_job_state_and_finalized_at_index; +DROP INDEX /* TEMPLATE: schema */river_job_prioritized_fetching_index; +DROP INDEX /* TEMPLATE: schema */river_job_unique_idx; + +ALTER TABLE /* TEMPLATE: schema */river_job RENAME TO river_job_old; + +CREATE TABLE /* TEMPLATE: schema */river_job ( + id integer PRIMARY KEY, + args blob NOT NULL DEFAULT (jsonb('{}')), + attempt integer NOT NULL DEFAULT 0, + attempted_at timestamp, + attempted_by blob, -- json + created_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, + errors blob, -- json + finalized_at timestamp, + kind text NOT NULL, + max_attempts integer NOT NULL DEFAULT 25, + metadata blob NOT NULL DEFAULT (jsonb('{}')), + priority integer NOT NULL DEFAULT 1, + queue text NOT NULL DEFAULT 'default', + state text NOT NULL DEFAULT 'available', + scheduled_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, + tags blob NOT NULL DEFAULT (jsonb('[]')), + unique_key blob, + unique_states integer, + CONSTRAINT finalized_or_finalized_at_null CHECK ( + (finalized_at IS NULL AND state NOT IN ('cancelled', 'completed', 'discarded')) OR + (finalized_at IS NOT NULL AND state IN ('cancelled', 'completed', 'discarded')) + ), + CONSTRAINT priority_in_range CHECK (priority >= 1 AND priority <= 4), + CONSTRAINT queue_length CHECK (length(queue) > 0 AND length(queue) < 128), + CONSTRAINT kind_length CHECK (length(kind) > 0 AND length(kind) < 128), + CONSTRAINT state_valid CHECK (state IN ('available', 'cancelled', 'completed', 'discarded', 'pending', 'retryable', 'running', 'scheduled')) +); + +INSERT INTO /* TEMPLATE: schema */river_job ( + id, + args, + attempt, + attempted_at, + attempted_by, + created_at, + errors, + finalized_at, + kind, + max_attempts, + metadata, + priority, + queue, + state, + scheduled_at, + tags, + unique_key, + unique_states +) +SELECT + id, + args, + attempt, + attempted_at, + attempted_by, + created_at, + errors, + finalized_at, + kind, + max_attempts, + metadata, + priority, + queue, + state, + scheduled_at, + tags, + unique_key, + unique_states +FROM /* TEMPLATE: schema */river_job_old; + +DROP TABLE /* TEMPLATE: schema */river_job_old; + +CREATE INDEX /* TEMPLATE: schema */river_job_kind ON river_job (kind); +CREATE INDEX /* TEMPLATE: schema */river_job_state_and_finalized_at_index ON river_job (state, finalized_at) WHERE finalized_at IS NOT NULL; +CREATE INDEX /* TEMPLATE: schema */river_job_prioritized_fetching_index ON river_job (state, queue, priority, scheduled_at, id); +CREATE UNIQUE INDEX /* TEMPLATE: schema */river_job_unique_idx ON river_job (unique_key) + WHERE unique_key IS NOT NULL + AND unique_states IS NOT NULL + AND CASE state + WHEN 'available' THEN unique_states & (1 << 0) + WHEN 'cancelled' THEN unique_states & (1 << 1) + WHEN 'completed' THEN unique_states & (1 << 2) + WHEN 'discarded' THEN unique_states & (1 << 3) + WHEN 'pending' THEN unique_states & (1 << 4) + WHEN 'retryable' THEN unique_states & (1 << 5) + WHEN 'running' THEN unique_states & (1 << 6) + WHEN 'scheduled' THEN unique_states & (1 << 7) + ELSE 0 + END >= 1; diff --git a/riverdriver/riversqlite/migration/main/008_job_id_autoincrement.up.sql b/riverdriver/riversqlite/migration/main/008_job_id_autoincrement.up.sql new file mode 100644 index 000000000..c7de15deb --- /dev/null +++ b/riverdriver/riversqlite/migration/main/008_job_id_autoincrement.up.sql @@ -0,0 +1,123 @@ +-- Rebuild river_job so automatically generated IDs are never reused after the +-- job holding the largest ID is deleted. Unlike PostgreSQL sequences, SQLite's +-- default ROWID allocator may otherwise reuse that deleted ID. + +-- Rebuilding river_job would discard schema installed by River Pro. Check +-- schema objects instead of migration records to also catch manually applied +-- Pro migrations and the legacy workflow migration line. +CREATE TEMP TABLE river_job_pro_schema_guard ( + id integer NOT NULL +); + +CREATE TEMP TRIGGER river_job_pro_schema_guard_enforce + BEFORE INSERT ON river_job_pro_schema_guard + WHEN EXISTS ( + SELECT 1 + FROM /* TEMPLATE: schema */sqlite_master + WHERE name IN ('river_job_sequence', 'river_job_workflow_scheduling', 'river_workflow') + ) +BEGIN + SELECT RAISE(ABORT, 'River SQLite migration 008 cannot run while River Pro schema is installed'); +END; + +INSERT INTO river_job_pro_schema_guard (id) VALUES (1); + +DROP TRIGGER river_job_pro_schema_guard_enforce; +DROP TABLE river_job_pro_schema_guard; + +DROP INDEX /* TEMPLATE: schema */river_job_kind; +DROP INDEX /* TEMPLATE: schema */river_job_state_and_finalized_at_index; +DROP INDEX /* TEMPLATE: schema */river_job_prioritized_fetching_index; +DROP INDEX /* TEMPLATE: schema */river_job_unique_idx; + +ALTER TABLE /* TEMPLATE: schema */river_job RENAME TO river_job_old; + +CREATE TABLE /* TEMPLATE: schema */river_job ( + id integer PRIMARY KEY AUTOINCREMENT, + args blob NOT NULL DEFAULT (jsonb('{}')), + attempt integer NOT NULL DEFAULT 0, + attempted_at timestamp, + attempted_by blob, -- json + created_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, + errors blob, -- json + finalized_at timestamp, + kind text NOT NULL, + max_attempts integer NOT NULL DEFAULT 25, + metadata blob NOT NULL DEFAULT (jsonb('{}')), + priority integer NOT NULL DEFAULT 1, + queue text NOT NULL DEFAULT 'default', + state text NOT NULL DEFAULT 'available', + scheduled_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, + tags blob NOT NULL DEFAULT (jsonb('[]')), + unique_key blob, + unique_states integer, + CONSTRAINT finalized_or_finalized_at_null CHECK ( + (finalized_at IS NULL AND state NOT IN ('cancelled', 'completed', 'discarded')) OR + (finalized_at IS NOT NULL AND state IN ('cancelled', 'completed', 'discarded')) + ), + CONSTRAINT priority_in_range CHECK (priority >= 1 AND priority <= 4), + CONSTRAINT queue_length CHECK (length(queue) > 0 AND length(queue) < 128), + CONSTRAINT kind_length CHECK (length(kind) > 0 AND length(kind) < 128), + CONSTRAINT state_valid CHECK (state IN ('available', 'cancelled', 'completed', 'discarded', 'pending', 'retryable', 'running', 'scheduled')) +); + +INSERT INTO /* TEMPLATE: schema */river_job ( + id, + args, + attempt, + attempted_at, + attempted_by, + created_at, + errors, + finalized_at, + kind, + max_attempts, + metadata, + priority, + queue, + state, + scheduled_at, + tags, + unique_key, + unique_states +) +SELECT + id, + args, + attempt, + attempted_at, + attempted_by, + created_at, + errors, + finalized_at, + kind, + max_attempts, + metadata, + priority, + queue, + state, + scheduled_at, + tags, + unique_key, + unique_states +FROM /* TEMPLATE: schema */river_job_old; + +DROP TABLE /* TEMPLATE: schema */river_job_old; + +CREATE INDEX /* TEMPLATE: schema */river_job_kind ON river_job (kind); +CREATE INDEX /* TEMPLATE: schema */river_job_state_and_finalized_at_index ON river_job (state, finalized_at) WHERE finalized_at IS NOT NULL; +CREATE INDEX /* TEMPLATE: schema */river_job_prioritized_fetching_index ON river_job (state, queue, priority, scheduled_at, id); +CREATE UNIQUE INDEX /* TEMPLATE: schema */river_job_unique_idx ON river_job (unique_key) + WHERE unique_key IS NOT NULL + AND unique_states IS NOT NULL + AND CASE state + WHEN 'available' THEN unique_states & (1 << 0) + WHEN 'cancelled' THEN unique_states & (1 << 1) + WHEN 'completed' THEN unique_states & (1 << 2) + WHEN 'discarded' THEN unique_states & (1 << 3) + WHEN 'pending' THEN unique_states & (1 << 4) + WHEN 'retryable' THEN unique_states & (1 << 5) + WHEN 'running' THEN unique_states & (1 << 6) + WHEN 'scheduled' THEN unique_states & (1 << 7) + ELSE 0 + END >= 1;