From 6c8c99231f640c7fede7a268fb5db026542f1771 Mon Sep 17 00:00:00 2001 From: Brandur Date: Fri, 25 Sep 2026 16:22:48 -0500 Subject: [PATCH] Make SQLite primary keys `AUTOINCREMENT` Without the `AUTOINCREMENT` keyword in SQLite, it'll pick new primary keys based off the largest ID in the table. There's a potential class of bug currently wherein if the latest row is moved out of a table, it may clash with another added row that's been added since and acquired the same ID. Here, introduce migration 008 which adds `AUTOINCREMENT` to SQLite tables. Unfortunately SQLite doesn't allow DDL in place, so we have some pretty ugly SQL here that redefines `river_job` and moves data over. I looked into whether any other tables might need `AUTOINCREMENT`, but the answer appears to be "no". `river_notification` already uses it, and the other tables all use non-automatic primary keys. SQLite for Pro is brand new and not yet announced, so taking the easier route here and not accounting for that in the schema. It's not great, but greatly simplifies things. --- CHANGELOG.md | 6 + riverdriver/river_driver_interface.go | 2 +- .../main/008_job_id_autoincrement.down.sql | 3 + .../main/008_job_id_autoincrement.up.sql | 3 + riverdriver/riverdrivertest/job_insert.go | 14 ++ riverdriver/riverdrivertest/migration.go | 100 +++++++++++++- .../main/008_job_id_autoincrement.down.sql | 3 + .../main/008_job_id_autoincrement.up.sql | 3 + .../riversqlite/internal/dbsqlc/river_job.sql | 3 +- .../migration/main/006_bulk_unique.up.sql | 4 +- ...tbox_sqlite_jsonb_and_sql_cleanup.down.sql | 2 +- ...outbox_sqlite_jsonb_and_sql_cleanup.up.sql | 2 +- .../main/008_job_id_autoincrement.down.sql | 121 +++++++++++++++++ .../main/008_job_id_autoincrement.up.sql | 123 ++++++++++++++++++ 14 files changed, 382 insertions(+), 7 deletions(-) create mode 100644 riverdriver/riverdatabasesql/migration/main/008_job_id_autoincrement.down.sql create mode 100644 riverdriver/riverdatabasesql/migration/main/008_job_id_autoincrement.up.sql create mode 100644 riverdriver/riverpgxv5/migration/main/008_job_id_autoincrement.down.sql create mode 100644 riverdriver/riverpgxv5/migration/main/008_job_id_autoincrement.up.sql create mode 100644 riverdriver/riversqlite/migration/main/008_job_id_autoincrement.down.sql create mode 100644 riverdriver/riversqlite/migration/main/008_job_id_autoincrement.up.sql diff --git a/CHANGELOG.md b/CHANGELOG.md index d8b921f5c..e4e93462d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +⚠️ This release contains a new database migration, version 8, but it only affects SQLite: + +- If you're on Postgres, you can ignore it with no adverse affects. +- If you're on SQLite, it rebuilds `river_job` to add an `AUTOINCREMENT` keyword to the primary key, preventing a possible edge case where generated job IDs could be reused after deletion. It's not necessary to run the migration for River to work, but it's a good idea to get it in when convenient. [PR #1390](https://github.com/riverqueue/river/pull/1390). + ### Added - Added `Config.FetchOnlyKnownKinds` to restrict job fetching to registered worker kinds, including aliases. Clients with different workers can share a queue while leaving unknown jobs available without consuming attempts. Disabled by default; leader election and stuck-job rescue behavior are unchanged. [PR #1396](https://github.com/riverqueue/river/pull/1396). @@ -34,6 +39,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). - Fixed the `Job appears to be stuck` log line reporting the client-level `JobTimeout` instead of the worker-level timeout when a worker overrides `Timeout`. [PR #1394](https://github.com/riverqueue/river/pull/1394). ## [0.47.0] - 2026-09-01 diff --git a/riverdriver/river_driver_interface.go b/riverdriver/river_driver_interface.go index 85dbf2f75..f483c4499 100644 --- a/riverdriver/river_driver_interface.go +++ b/riverdriver/river_driver_interface.go @@ -993,7 +993,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 e60a68c36..a911d8cc0 100644 --- a/riverdriver/riverdrivertest/job_insert.go +++ b/riverdriver/riverdrivertest/job_insert.go @@ -664,6 +664,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 b9c24ef4d..08669f10c 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;