diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a25368..0f1a666 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Add SQLite support to the ActiveRecord and Sequel drivers, including current River JSONB storage, atomic bulk inserts, unique jobs, and notification-outbox writes. [PR #67](https://github.com/riverqueue/riverqueue-ruby/pull/67). + +### Changed + +- Stop pulling in `pg` as a hard dependency of either driver. Applications now select their database adapter by including `pg` for PostgreSQL or `sqlite3` for SQLite. [PR #67](https://github.com/riverqueue/riverqueue-ruby/pull/67). + ## [0.10.1] - 2026-04-09 ### Fixed diff --git a/Gemfile b/Gemfile index d174c52..ffa97ff 100644 --- a/Gemfile +++ b/Gemfile @@ -9,8 +9,10 @@ end group :test do gem "debug" + gem "pg" gem "rspec-core" gem "rspec-expectations" gem "riverqueue-sequel", path: "driver/riverqueue-sequel" gem "simplecov", require: false + gem "sqlite3" end diff --git a/Gemfile.lock b/Gemfile.lock index 027e76e..7bca32c 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -7,7 +7,6 @@ PATH remote: driver/riverqueue-sequel specs: riverqueue-sequel (0.10.1) - pg (> 0, < 1000) sequel (> 0, < 1000) GEM @@ -125,6 +124,8 @@ GEM simplecov_json_formatter (~> 0.1) simplecov-html (0.13.2) simplecov_json_formatter (0.1.4) + sqlite3 (2.9.6-arm64-darwin) + sqlite3 (2.9.6-x86_64-linux-gnu) standard (1.54.0) language_server-protocol (~> 3.17.0.2) lint_roller (~> 1.0) @@ -172,11 +173,13 @@ PLATFORMS DEPENDENCIES debug + pg riverqueue! riverqueue-sequel! rspec-core rspec-expectations simplecov + sqlite3 standard steep diff --git a/driver/riverqueue-activerecord/Gemfile b/driver/riverqueue-activerecord/Gemfile index 3fd0fb5..b3a8848 100644 --- a/driver/riverqueue-activerecord/Gemfile +++ b/driver/riverqueue-activerecord/Gemfile @@ -9,7 +9,9 @@ end group :test do gem "debug" + gem "pg" gem "rspec-core" gem "rspec-expectations" gem "simplecov", require: false + gem "sqlite3" end diff --git a/driver/riverqueue-activerecord/Gemfile.lock b/driver/riverqueue-activerecord/Gemfile.lock index feb3486..3cd7c78 100644 --- a/driver/riverqueue-activerecord/Gemfile.lock +++ b/driver/riverqueue-activerecord/Gemfile.lock @@ -9,7 +9,6 @@ PATH riverqueue-activerecord (0.10.1) activerecord (> 0, < 1000) activesupport (> 0, < 1000) - pg (> 0, < 1000) GEM remote: https://rubygems.org/ @@ -115,6 +114,8 @@ GEM simplecov_json_formatter (~> 0.1) simplecov-html (0.13.2) simplecov_json_formatter (0.1.4) + sqlite3 (2.9.3-arm64-darwin) + sqlite3 (2.9.3-x86_64-linux-gnu) standard (1.54.0) language_server-protocol (~> 3.17.0.2) lint_roller (~> 1.0) @@ -146,11 +147,13 @@ PLATFORMS DEPENDENCIES debug + pg riverqueue! riverqueue-activerecord! rspec-core rspec-expectations simplecov + sqlite3 standard BUNDLED WITH diff --git a/driver/riverqueue-activerecord/README.md b/driver/riverqueue-activerecord/README.md index 265dd01..f5546bb 100644 --- a/driver/riverqueue-activerecord/README.md +++ b/driver/riverqueue-activerecord/README.md @@ -1,8 +1,3 @@ -# River Ruby bindings ActiveRecord driver +# riverqueue-activerecord -A future home for River's Ruby bindings. For now, the [Gem is registered](https://rubygems.org/gems/riverqueue), but nothing else is done. - -``` sh -$ gem build riverqueue-activerecord.gemspec -$ gem push riverqueue-activerecord-0.0.1.gem -``` +See the [ActiveRecord driver documentation](./docs/README.md). diff --git a/driver/riverqueue-activerecord/docs/README.md b/driver/riverqueue-activerecord/docs/README.md index 7d01cf8..a653efb 100644 --- a/driver/riverqueue-activerecord/docs/README.md +++ b/driver/riverqueue-activerecord/docs/README.md @@ -1,22 +1,48 @@ -# riverqueue-sequel [![Build Status](https://github.com/riverqueue/riverqueue-ruby-sequel/workflows/CI/badge.svg)](https://github.com/riverqueue/riverqueue-ruby-sequel/actions) +# riverqueue-activerecord -[ActiveRecord](https://github.com/jeremyevans/sequel) driver for [River](https://github.com/riverqueue/river)'s [`riverqueue` gem for Ruby](https://rubygems.org/gems/riverqueue). +[ActiveRecord](https://guides.rubyonrails.org/active_record_basics.html) driver for [River](https://github.com/riverqueue/river)'s [`riverqueue` gem for Ruby](https://rubygems.org/gems/riverqueue). PostgreSQL and SQLite are supported. -`Gemfile` should contain the core gem and a driver like this one: +Add the core gem and this driver to `Gemfile`: -``` yaml +```ruby gem "riverqueue" -gem "riverqueue-sequel" +gem "riverqueue-activerecord" +``` + +Database adapters are optional dependencies. Add only the adapter used by your +application. + +For PostgreSQL, add `pg` to `Gemfile`: + +```ruby +gem "pg" +``` + +Then initialize a client with ActiveRecord's established connection: + +```ruby +ActiveRecord::Base.establish_connection("postgres://localhost/my_app") +client = River::Client.new(River::Driver::ActiveRecord.new) +``` + +For SQLite, add `sqlite3` to `Gemfile`: + +```ruby +gem "sqlite3" ``` -Initialize a client with: +Then initialize a client with an SQLite connection: ```ruby -DB = ActiveRecord.connect("postgres://...") -client = River::Client.new(River::Driver::ActiveRecord.new(DB)) +ActiveRecord::Base.establish_connection( + adapter: "sqlite3", + database: "storage/river.sqlite3", + timeout: 5_000 +) +client = River::Client.new(River::Driver::ActiveRecord.new) ``` -See also [`rubyqueue`](https://github.com/riverqueue/riverqueue-ruby). +Use current River migrations to create and update a production SQLite database. `River::Driver::ActiveRecord.create_sqlite_schema` creates only the tables needed by this insert-only client and is intended primarily for tests. ## Development diff --git a/driver/riverqueue-activerecord/lib/driver.rb b/driver/riverqueue-activerecord/lib/driver.rb index 9e10462..b31410b 100644 --- a/driver/riverqueue-activerecord/lib/driver.rb +++ b/driver/riverqueue-activerecord/lib/driver.rb @@ -1,13 +1,65 @@ +require "securerandom" + module River::Driver - # Provides a ActiveRecord driver for River. + # Provides an ActiveRecord driver for River that supports both PostgreSQL + # and SQLite. # # Used in conjunction with a River client like: # - # DB = ActiveRecord.connect("postgres://...") - # client = River::Client.new(River::Driver::ActiveRecord.new(DB)) + # ActiveRecord::Base.establish_connection("postgres://...") + # client = River::Client.new(River::Driver::ActiveRecord.new) # class ActiveRecord + SQLITE_CONFLICT_WHERE = <<~SQL.chomp + 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 + SQL + private_constant :SQLITE_CONFLICT_WHERE + + # SQLite 3.45+ may store JSON as binary JSONB. Always project JSON columns + # through json() so this driver can read both the current JSONB format and + # the text JSON used by River migrations through version 006. Cast times to + # text so ActiveRecord doesn't interpret timezone-less SQLite timestamps in + # the process timezone. + SQLITE_JOB_COLUMNS = <<~SQL.chomp + id, + json(args) AS args, + attempt, + CAST(attempted_at AS text) AS attempted_at, + json(attempted_by) AS attempted_by, + CAST(created_at AS text) AS created_at, + json(errors) AS errors, + CAST(finalized_at AS text) AS finalized_at, + kind, + max_attempts, + json(metadata) AS metadata, + priority, + queue, + state, + CAST(scheduled_at AS text) AS scheduled_at, + json(tags) AS tags, + unique_key, + unique_states + SQL + private_constant :SQLITE_JOB_COLUMNS + + SQLITE_UNIQUE_NONCE_KEY = "river:unique_nonce" + private_constant :SQLITE_UNIQUE_NONCE_KEY + def initialize + @is_sqlite = ::ActiveRecord::Base.connection.adapter_name.downcase.include?("sqlite") + # It's Ruby, so we can only define a model after ActiveRecord's established a # connection because it's all dynamic. if !River::Driver::ActiveRecord.const_defined?(:RiverJob) @@ -31,9 +83,84 @@ def errors = {} end end + # Creates the SQLite tables needed by this insert-only client and its tests. + # Applications that run River workers should use River's migrations, which + # create the rest of River's schema as well. + def self.create_sqlite_schema + conn = ::ActiveRecord::Base.connection + + conn.execute <<~SQL + CREATE TABLE IF NOT EXISTS river_job ( + id integer PRIMARY KEY, + args blob NOT NULL DEFAULT (jsonb('{}')), + attempt integer NOT NULL DEFAULT 0, + attempted_at timestamp, + attempted_by blob, + created_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, + errors blob, + finalized_at timestamp, + kind text NOT NULL, + max_attempts integer NOT NULL, + 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')) + ) + SQL + + conn.execute "CREATE INDEX IF NOT EXISTS river_job_kind ON river_job (kind)" + conn.execute <<~SQL + CREATE INDEX IF NOT EXISTS river_job_state_and_finalized_at_index + ON river_job (state, finalized_at) WHERE finalized_at IS NOT NULL + SQL + conn.execute <<~SQL + CREATE INDEX IF NOT EXISTS river_job_prioritized_fetching_index + ON river_job (state, queue, priority, scheduled_at, id) + SQL + conn.execute <<~SQL + CREATE UNIQUE INDEX IF NOT EXISTS river_job_unique_idx ON river_job (unique_key) + WHERE #{SQLITE_CONFLICT_WHERE} + SQL + + conn.execute <<~SQL + CREATE TABLE IF NOT EXISTS river_notification ( + id integer PRIMARY KEY AUTOINCREMENT, + created_at timestamp NOT NULL DEFAULT (datetime('now', 'subsec')), + payload text NOT NULL, + topic text NOT NULL, + CONSTRAINT topic_length CHECK (length(topic) > 0 AND length(topic) < 128) + ) + SQL + conn.execute <<~SQL + CREATE INDEX IF NOT EXISTS river_notification_created_at_idx + ON river_notification (created_at) + SQL + conn.execute <<~SQL + CREATE INDEX IF NOT EXISTS river_notification_topic_id_idx + ON river_notification (topic, id) + SQL + end + def job_get_by_id(id) - data_set = RiverJob.where(id: id) - data_set.first ? to_job_row_from_model(data_set.first) : nil + if @is_sqlite + row = sqlite_job_rows("WHERE id = ? LIMIT 1", [id]).first + row ? sqlite_to_job_row_from_raw(row) : nil + else + data_set = RiverJob.where(id: id) + data_set.first ? to_job_row_from_model(data_set.first) : nil + end end def job_insert(insert_params) @@ -41,8 +168,28 @@ def job_insert(insert_params) end def job_insert_many(insert_params_many) + @is_sqlite ? sqlite_job_insert_many(insert_params_many) : postgres_job_insert_many(insert_params_many) + end + + def job_list + if @is_sqlite + sqlite_job_rows("ORDER BY id").map { |row| sqlite_to_job_row_from_raw(row) } + else + RiverJob.order(:id).all.map { |job| to_job_row_from_model(job) } + end + end + + def rollback_exception + ::ActiveRecord::Rollback + end + + def transaction(&) + ::ActiveRecord::Base.transaction(requires_new: true, &) + end + + private def postgres_job_insert_many(insert_params_many) res = RiverJob.upsert_all( - insert_params_many.map { |param| insert_params_to_hash(param) }, + insert_params_many.map { |param| postgres_insert_params_to_hash(param) }, on_duplicate: Arel.sql("kind = EXCLUDED.kind"), returning: Arel.sql("*, (xmax != 0) AS unique_skipped_as_duplicate"), @@ -53,23 +200,68 @@ def job_insert_many(insert_params_many) # clause. The workaround is to target the index name instead of columns. unique_by: "river_job_unique_idx" ) - to_insert_results(res) + postgres_to_insert_results(res) end - def job_list - data_set = RiverJob.order(:id) - data_set.all.map { |job| to_job_row_from_model(job) } - end + # River's current SQLite driver uses json_each to make a batch a single, + # atomic statement. The JSON columns are converted to SQLite JSONB here, + # matching migration 007 and newer River databases. + private def sqlite_job_insert_many(insert_params_many) + return [] if insert_params_many.empty? - def rollback_exception - ::ActiveRecord::Rollback - end + ::ActiveRecord::Base.transaction(requires_new: true) do + nonce = SecureRandom.hex(8) + jobs = insert_params_many.map { |param| sqlite_insert_params_to_hash(param, nonce) } - def transaction(&) - ::ActiveRecord::Base.transaction(requires_new: true, &) + sql = <<~SQL + INSERT INTO river_job ( + args, + created_at, + kind, + max_attempts, + metadata, + priority, + queue, + scheduled_at, + state, + tags, + unique_key, + unique_states + ) + SELECT + jsonb(json_extract(value, '$.args')), + datetime('now', 'subsec'), + cast(json_extract(value, '$.kind') AS text), + cast(json_extract(value, '$.max_attempts') AS integer), + jsonb(json_extract(value, '$.metadata')), + cast(json_extract(value, '$.priority') AS integer), + cast(json_extract(value, '$.queue') AS text), + coalesce(cast(json_extract(value, '$.scheduled_at') AS text), datetime('now', 'subsec')), + cast(json_extract(value, '$.state') AS text), + jsonb(json_extract(value, '$.tags')), + CASE + WHEN length(cast(json_extract(value, '$.unique_key') AS text)) = 0 THEN NULL + ELSE unhex(cast(json_extract(value, '$.unique_key') AS text)) + END, + nullif(cast(json_extract(value, '$.unique_states') AS integer), 0) + FROM json_each(cast(? AS blob)) + WHERE true + ON CONFLICT (unique_key) WHERE #{SQLITE_CONFLICT_WHERE} + DO UPDATE SET kind = EXCLUDED.kind + RETURNING #{SQLITE_JOB_COLUMNS} + SQL + + rows = ::ActiveRecord::Base.connection.raw_connection.execute(sql, [JSON.dump(jobs)]) + sqlite_notify_insert(insert_params_many) + + rows.map do |row| + metadata = JSON.parse(row["metadata"]) + [sqlite_to_job_row_from_raw(row), metadata[SQLITE_UNIQUE_NONCE_KEY] != nonce] + end + end end - private def insert_params_to_hash(insert_params) + private def postgres_insert_params_to_hash(insert_params) { args: JSON.parse(insert_params.encoded_args), kind: insert_params.kind, @@ -84,7 +276,27 @@ def transaction(&) } end + private def sqlite_insert_params_to_hash(insert_params, nonce) + { + args: JSON.parse(insert_params.encoded_args), + kind: insert_params.kind, + max_attempts: insert_params.max_attempts, + metadata: {SQLITE_UNIQUE_NONCE_KEY => nonce}, + priority: insert_params.priority, + queue: insert_params.queue, + scheduled_at: insert_params.scheduled_at ? format_time(insert_params.scheduled_at) : nil, + state: insert_params.state, + tags: insert_params.tags || [], + unique_key: insert_params.unique_key&.unpack1("H*"), + unique_states: insert_params.unique_states&.to_i(2) + } + end + private def to_job_row_from_model(river_job) + @is_sqlite ? sqlite_to_job_row_from_model(river_job) : postgres_to_job_row_from_model(river_job) + end + + private def postgres_to_job_row_from_model(river_job) # needs to be accessed through values because `errors` is shadowed by both # ActiveRecord and the patch above errors = river_job.attributes["errors"] @@ -120,9 +332,46 @@ def transaction(&) ) end - private def to_insert_results(res) + private def sqlite_to_job_row_from_model(river_job) + row = sqlite_job_rows("WHERE id = ? LIMIT 1", [river_job.id]).first + sqlite_to_job_row_from_raw(row) + end + + private def sqlite_to_job_row_from_raw(row) + errors = row["errors"] ? JSON.parse(row["errors"]) : [] + + River::JobRow.new( + id: row["id"], + args: JSON.parse(row["args"]), + attempt: row["attempt"], + attempted_at: parse_sqlite_time(row["attempted_at"]), + attempted_by: row["attempted_by"] ? JSON.parse(row["attempted_by"]) : nil, + created_at: parse_sqlite_time(row["created_at"]), + errors: errors.map { |e| + River::AttemptError.new( + at: Time.parse(e["at"]), + attempt: e["attempt"], + error: e["error"], + trace: e["trace"] + ) + }, + finalized_at: parse_sqlite_time(row["finalized_at"]), + kind: row["kind"], + max_attempts: row["max_attempts"], + metadata: JSON.parse(row["metadata"]), + priority: row["priority"], + queue: row["queue"], + scheduled_at: parse_sqlite_time(row["scheduled_at"]), + state: row["state"], + tags: JSON.parse(row["tags"]), + unique_key: row["unique_key"]&.to_s, + unique_states: row["unique_states"] ? ::River::UniqueBitmask.to_states(row["unique_states"]) : nil + ) + end + + private def postgres_to_insert_results(res) res.rows.map do |row| - to_job_row_from_raw(row, res.columns, res.column_types) + postgres_to_job_row_from_raw(row, res.columns, res.column_types) end end @@ -132,7 +381,7 @@ def transaction(&) # searched long and hard for a way to have the former type of method return # raw or the latter type of method return a model, but was unable to find # anything. - private def to_job_row_from_raw(row, columns, column_types) + private def postgres_to_job_row_from_raw(row, columns, column_types) river_job = {} row.each_with_index do |val, i| @@ -174,5 +423,42 @@ def transaction(&) river_job["unique_skipped_as_duplicate"] ] end + + private def format_time(time) + time.getutc.round(3).strftime("%Y-%m-%d %H:%M:%S.%3N") + end + + private def parse_sqlite_time(value) + return nil unless value + + value = value.to_s + value += " UTC" unless value.match?(/(?:Z|[+-]\d{2}:?\d{2})\z/) + Time.parse(value).utc + end + + private def sqlite_job_rows(suffix, binds = []) + sql = "SELECT #{SQLITE_JOB_COLUMNS} FROM river_job #{suffix}" + ::ActiveRecord::Base.connection.raw_connection.execute(sql, binds) + end + + private def sqlite_notify_insert(insert_params_many) + queues = insert_params_many + .select { |param| param.state == ::River::JOB_STATE_AVAILABLE } + .map(&:queue) + .uniq + return if queues.empty? + + notifications = queues.map do |queue| + {payload: JSON.dump({queue: queue}), topic: "insert"} + end + + ::ActiveRecord::Base.connection.raw_connection.execute(<<~SQL, [JSON.dump(notifications)]) + INSERT INTO river_notification (payload, topic) + SELECT + json_extract(value, '$.payload'), + json_extract(value, '$.topic') + FROM json_each(cast(? AS blob)) + SQL + end end end diff --git a/driver/riverqueue-activerecord/riverqueue-activerecord.gemspec b/driver/riverqueue-activerecord/riverqueue-activerecord.gemspec index bc14ee5..46b4c3a 100644 --- a/driver/riverqueue-activerecord/riverqueue-activerecord.gemspec +++ b/driver/riverqueue-activerecord/riverqueue-activerecord.gemspec @@ -1,8 +1,8 @@ Gem::Specification.new do |s| s.name = "riverqueue-activerecord" s.version = "0.10.1" - s.summary = "ActiveRecord driver for the River Ruby gem." - s.description = "ActiveRecord driver for the River Ruby gem. Use in conjunction with the riverqueue gem to insert jobs that are worked in Go." + s.summary = "ActiveRecord PostgreSQL and SQLite driver for the River Ruby gem." + s.description = "ActiveRecord PostgreSQL and SQLite driver for the River Ruby gem. Use in conjunction with the riverqueue gem to insert jobs that are worked in Go." s.authors = ["Blake Gentry", "Brandur Leach"] s.email = "brandur@brandur.org" s.files = Dir.glob("lib/**/*") @@ -12,5 +12,4 @@ Gem::Specification.new do |s| # The stupid version bounds are used to silence Ruby's extremely obnoxious warnings. s.add_dependency "activerecord", "> 0", "< 1000" s.add_dependency "activesupport", "> 0", "< 1000" # required for ActiveRecord to load properly - s.add_dependency "pg", "> 0", "< 1000" end diff --git a/driver/riverqueue-activerecord/spec/driver_spec.rb b/driver/riverqueue-activerecord/spec/driver_spec.rb index df4d783..ed9642f 100644 --- a/driver/riverqueue-activerecord/spec/driver_spec.rb +++ b/driver/riverqueue-activerecord/spec/driver_spec.rb @@ -2,241 +2,388 @@ require_relative "../../../spec/driver_shared_examples" RSpec.describe River::Driver::ActiveRecord do - around(:each) { |ex| test_transaction(&ex) } - - let!(:driver) { River::Driver::ActiveRecord.new } - let(:client) { River::Client.new(driver) } - before do if ENV["RIVER_DEBUG"] == "1" || ENV["RIVER_DEBUG"] == "true" ActiveRecord::Base.logger = Logger.new($stdout) end end - it_behaves_like "driver shared examples" + { + "PostgreSQL" => {adapter: :postgres, available: PG_AVAILABLE}, + "SQLite" => {adapter: :sqlite, available: true} + }.each do |name, config| + next unless config[:available] - describe "client inserts" do - it "persists args as a JSON object rather than a JSON string" do - insert_res = client.insert(SimpleArgs.new(job_num: 1)) + context "with #{name}" do + before(:all) do + if config[:adapter] == :sqlite + switch_to_sqlite! + River::Driver::ActiveRecord.create_sqlite_schema + else + switch_to_postgres! + end + end - row = ActiveRecord::Base.connection.exec_query(<<~SQL).first - SELECT args, jsonb_typeof(args) AS args_type - FROM river_job - WHERE id = #{insert_res.job.id} - SQL + after(:all) do + switch_to_postgres! if config[:adapter] == :sqlite && PG_AVAILABLE + end - expect(row["args_type"]).to eq("object") - expect(JSON.parse(row["args"])).to eq({"job_num" => 1}) - end - end + around(:each) { |ex| test_transaction(&ex) } - describe "#to_job_row_from_model" do - it "converts a database record to `River::JobRow` with minimal properties" do - river_job = River::Driver::ActiveRecord::RiverJob.create( - id: 1, - args: {"job_num" => 1}, - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - priority: River::PRIORITY_DEFAULT, - queue: River::QUEUE_DEFAULT, - state: River::JOB_STATE_AVAILABLE - ) - - job_row = driver.send(:to_job_row_from_model, river_job) - - expect(job_row).to be_an_instance_of(River::JobRow) - expect(job_row).to have_attributes( - id: 1, - args: {"job_num" => 1}, - attempt: 0, - attempted_at: nil, - attempted_by: nil, - created_at: be_within(2).of(Time.now.getutc), - finalized_at: nil, - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - priority: River::PRIORITY_DEFAULT, - queue: River::QUEUE_DEFAULT, - scheduled_at: be_within(2).of(Time.now.getutc), - state: River::JOB_STATE_AVAILABLE, - tags: [] - ) - end + let!(:driver) { River::Driver::ActiveRecord.new } + let(:client) { River::Client.new(driver) } - it "converts a database record to `River::JobRow` with all properties" do - now = Time.now - river_job = River::Driver::ActiveRecord::RiverJob.create( - id: 1, - attempt: 1, - attempted_at: now, - attempted_by: ["client1"], - created_at: now, - args: {"job_num" => 1}, - finalized_at: now, - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - priority: River::PRIORITY_DEFAULT, - queue: River::QUEUE_DEFAULT, - scheduled_at: now, - state: River::JOB_STATE_COMPLETED, - tags: ["tag1"], - unique_key: Digest::SHA256.digest("unique_key_str") - ) - - job_row = driver.send(:to_job_row_from_model, river_job) - - expect(job_row).to be_an_instance_of(River::JobRow) - expect(job_row).to have_attributes( - id: 1, - args: {"job_num" => 1}, - attempt: 1, - attempted_at: now.getutc, - attempted_by: ["client1"], - created_at: now.getutc, - finalized_at: now.getutc, - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - priority: River::PRIORITY_DEFAULT, - queue: River::QUEUE_DEFAULT, - scheduled_at: now.getutc, - state: River::JOB_STATE_COMPLETED, - tags: ["tag1"], - unique_key: Digest::SHA256.digest("unique_key_str") - ) - end + it_behaves_like "driver shared examples" + + describe "client inserts" do + it "persists SQLite JSON columns as JSONB objects" do + next unless config[:adapter] == :sqlite + + insert_res = client.insert(SimpleArgs.new(job_num: 1)) + + row = ActiveRecord::Base.connection.exec_query(<<~SQL).first + SELECT + json(args) AS args, + json_type(args) AS args_type, + typeof(args) AS args_storage_type, + typeof(metadata) AS metadata_storage_type, + typeof(tags) AS tags_storage_type, + CAST(created_at AS text) AS created_at, + CAST(scheduled_at AS text) AS scheduled_at + FROM river_job + WHERE id = #{insert_res.job.id} + SQL + + expect(row["args_type"]).to eq("object") + expect(JSON.parse(row["args"])).to eq({"job_num" => 1}) + expect(row.values_at("args_storage_type", "metadata_storage_type", "tags_storage_type")).to eq(["blob", "blob", "blob"]) + expect(row["created_at"]).to match(/\.\d{3}\z/) + expect(row["scheduled_at"]).to match(/\.\d{3}\z/) + expect(insert_res.job.errors).to eq([]) + end + + it "persists PostgreSQL args as a JSON object rather than a JSON string" do + next unless config[:adapter] == :postgres + + insert_res = client.insert(SimpleArgs.new(job_num: 1)) + + row = ActiveRecord::Base.connection.exec_query(<<~SQL).first + SELECT args, jsonb_typeof(args) AS args_type + FROM river_job + WHERE id = #{insert_res.job.id} + SQL + + expect(row["args_type"]).to eq("object") + expect(JSON.parse(row["args"])).to eq({"job_num" => 1}) + end + + it "inserts a SQLite batch atomically" do + next unless config[:adapter] == :sqlite + + expect do + client.insert_many([ + SimpleArgs.new(job_num: 1), + River::InsertManyParams.new( + SimpleArgs.new(job_num: 2), + insert_opts: River::InsertOpts.new(queue: "") + ) + ]) + end.to raise_error(SQLite3::ConstraintException) + + expect(driver.job_list).to be_empty + end + + it "notifies each available SQLite queue once per batch" do + next unless config[:adapter] == :sqlite + + client.insert_many([ + SimpleArgs.new(job_num: 1), + SimpleArgs.new(job_num: 2) + ]) + + rows = ActiveRecord::Base.connection.exec_query(<<~SQL).to_a + SELECT payload, topic FROM river_notification ORDER BY id + SQL + expect(rows).to contain_exactly( + {"payload" => JSON.dump({queue: River::QUEUE_DEFAULT}), "topic" => "insert"} + ) + end + + it "handles an empty SQLite batch" do + next unless config[:adapter] == :sqlite + + expect(driver.job_insert_many([])).to eq([]) + end + + it "defaults a missing SQLite scheduled_at" do + next unless config[:adapter] == :sqlite + + params = River::Driver::JobInsertParams.new( + encoded_args: JSON.dump({job_num: 1}), + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: nil, + state: River::JOB_STATE_AVAILABLE, + tags: [] + ) + + job, = driver.job_insert(params) + expect(job.scheduled_at).to be_within(2).of(Time.now.utc) + end + + it "rounds SQLite timestamps to three fractional digits" do + next unless config[:adapter] == :sqlite - it "with errors" do - now = Time.now.utc - river_job = River::Driver::ActiveRecord::RiverJob.create( - args: {"job_num" => 1}, - errors: [JSON.dump( - { - at: now, + time = Time.utc(2026, 8, 31, 12, 34, 56) + 0.1236 + expect(driver.send(:format_time, time)).to eq("2026-08-31 12:34:56.124") + end + end + + describe "#to_job_row_from_model" do + it "converts a database record to `River::JobRow` with minimal properties" do + if config[:adapter] == :sqlite + ActiveRecord::Base.connection.execute( + ActiveRecord::Base.sanitize_sql_array([ + "INSERT INTO river_job (args, kind, max_attempts) VALUES (?, ?, ?)", + '{"job_num":1}', "simple", River::MAX_ATTEMPTS_DEFAULT + ]) + ) + else + River::Driver::ActiveRecord::RiverJob.create( + id: 1, + args: {"job_num" => 1}, + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + state: River::JOB_STATE_AVAILABLE + ) + end + + river_job = River::Driver::ActiveRecord::RiverJob.first + job_row = driver.send(:to_job_row_from_model, river_job) + + expect(job_row).to be_an_instance_of(River::JobRow) + expect(job_row).to have_attributes( + id: be_a(Integer), + args: {"job_num" => 1}, + attempt: 0, + attempted_at: nil, + attempted_by: nil, + created_at: be_within(2).of(Time.now.getutc), + finalized_at: nil, + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: be_within(2).of(Time.now.getutc), + state: River::JOB_STATE_AVAILABLE, + tags: [] + ) + end + + it "converts a database record to `River::JobRow` with all properties" do + now = Time.now.utc + now_str = (config[:adapter] == :sqlite) ? now.iso8601(3) : now.strftime("%Y-%m-%d %H:%M:%S.%3N") + + if config[:adapter] == :sqlite + # Use raw SQLite connection to avoid binary encoding issues with + # sanitize_sql_array. + ActiveRecord::Base.connection.raw_connection.execute( + "INSERT INTO river_job (attempt, attempted_at, attempted_by, created_at, args, finalized_at, kind, max_attempts, priority, queue, scheduled_at, state, tags, unique_key) VALUES (?, ?, jsonb(?), ?, jsonb(?), ?, ?, ?, ?, ?, ?, ?, jsonb(?), ?)", + [1, now_str, JSON.dump(["client1"]), now_str, '{"job_num":1}', now_str, "simple", + River::MAX_ATTEMPTS_DEFAULT, River::PRIORITY_DEFAULT, River::QUEUE_DEFAULT, + now_str, River::JOB_STATE_COMPLETED, JSON.dump(["tag1"]), + Digest::SHA256.digest("unique_key_str")] + ) + else + River::Driver::ActiveRecord::RiverJob.create( + id: 1, + attempt: 1, + attempted_at: now, + attempted_by: ["client1"], + created_at: now, + args: {"job_num" => 1}, + finalized_at: now, + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: now, + state: River::JOB_STATE_COMPLETED, + tags: ["tag1"], + unique_key: Digest::SHA256.digest("unique_key_str") + ) + end + + river_job = River::Driver::ActiveRecord::RiverJob.first + job_row = driver.send(:to_job_row_from_model, river_job) + + expect(job_row).to be_an_instance_of(River::JobRow) + expect(job_row).to have_attributes( + id: be_a(Integer), + args: {"job_num" => 1}, + attempt: 1, + attempted_at: be_within(2).of(now.getutc), + attempted_by: ["client1"], + created_at: be_within(2).of(now.getutc), + finalized_at: be_within(2).of(now.getutc), + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: be_within(2).of(now.getutc), + state: River::JOB_STATE_COMPLETED, + tags: ["tag1"], + unique_key: Digest::SHA256.digest("unique_key_str") + ) + end + + it "with errors" do + now = Time.now.utc + + if config[:adapter] == :sqlite + ActiveRecord::Base.connection.execute( + ActiveRecord::Base.sanitize_sql_array([ + "INSERT INTO river_job (args, errors, kind, max_attempts, state) VALUES (?, ?, ?, ?, ?)", + '{"job_num":1}', + JSON.dump([{at: now.iso8601, attempt: 1, error: "job failure", trace: "error trace"}]), + "simple", River::MAX_ATTEMPTS_DEFAULT, River::JOB_STATE_AVAILABLE + ]) + ) + else + River::Driver::ActiveRecord::RiverJob.create( + args: {"job_num" => 1}, + errors: [JSON.dump({at: now, attempt: 1, error: "job failure", trace: "error trace"})], + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + state: River::JOB_STATE_AVAILABLE + ) + end + + river_job = River::Driver::ActiveRecord::RiverJob.first + job_row = driver.send(:to_job_row_from_model, river_job) + + expect(job_row.errors.count).to be(1) + expect(job_row.errors[0]).to be_an_instance_of(River::AttemptError) + expect(job_row.errors[0]).to have_attributes( + at: now.floor(0), attempt: 1, error: "job failure", trace: "error trace" - } - )], - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - state: River::JOB_STATE_AVAILABLE - ) - - job_row = driver.send(:to_job_row_from_model, river_job) - - expect(job_row.errors.count).to be(1) - expect(job_row.errors[0]).to be_an_instance_of(River::AttemptError) - expect(job_row.errors[0]).to have_attributes( - at: now.floor(0), - attempt: 1, - error: "job failure", - trace: "error trace" - ) - end - end + ) + end + end - describe "#to_job_row_from_raw" do - it "converts a database record to `River::JobRow` with minimal properties" do - res = River::Driver::ActiveRecord::RiverJob.insert({ - id: 1, - args: {"job_num" => 1}, - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT - }, returning: Arel.sql("*, false AS unique_skipped_as_duplicate")) - - job_row, skipped_as_duplicate = driver.send(:to_job_row_from_raw, res.rows[0], res.columns, res.column_types) - - expect(job_row).to be_an_instance_of(River::JobRow) - expect(job_row).to have_attributes( - id: 1, - args: {"job_num" => 1}, - attempt: 0, - attempted_at: nil, - attempted_by: nil, - created_at: be_within(2).of(Time.now.getutc), - finalized_at: nil, - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - priority: River::PRIORITY_DEFAULT, - queue: River::QUEUE_DEFAULT, - scheduled_at: be_within(2).of(Time.now.getutc), - state: River::JOB_STATE_AVAILABLE, - tags: [] - ) - expect(skipped_as_duplicate).to be(false) - end + # PostgreSQL-only: test the raw row conversion used by upsert_all + next unless config[:adapter] == :postgres - it "converts a database record to `River::JobRow` with all properties" do - now = Time.now - res = River::Driver::ActiveRecord::RiverJob.insert({ - id: 1, - attempt: 1, - attempted_at: now, - attempted_by: ["client1"], - created_at: now, - args: {"job_num" => 1}, - finalized_at: now, - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - priority: River::PRIORITY_DEFAULT, - queue: River::QUEUE_DEFAULT, - scheduled_at: now, - state: River::JOB_STATE_COMPLETED, - tags: ["tag1"], - unique_key: Digest::SHA256.digest("unique_key_str") - }, returning: Arel.sql("*, true AS unique_skipped_as_duplicate")) - - job_row, skipped_as_duplicate = driver.send(:to_job_row_from_raw, res.rows[0], res.columns, res.column_types) - - expect(job_row).to be_an_instance_of(River::JobRow) - expect(job_row).to have_attributes( - id: 1, - args: {"job_num" => 1}, - attempt: 1, - attempted_at: be_within(2).of(now.getutc), - attempted_by: ["client1"], - created_at: be_within(2).of(now.getutc), - finalized_at: be_within(2).of(now.getutc), - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - priority: River::PRIORITY_DEFAULT, - queue: River::QUEUE_DEFAULT, - scheduled_at: be_within(2).of(now.getutc), - state: River::JOB_STATE_COMPLETED, - tags: ["tag1"], - unique_key: Digest::SHA256.digest("unique_key_str") - ) - expect(skipped_as_duplicate).to be(true) - end + describe "#postgres_to_job_row_from_raw" do + it "converts a database record to `River::JobRow` with minimal properties" do + res = River::Driver::ActiveRecord::RiverJob.insert({ + id: 1, + args: {"job_num" => 1}, + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT + }, returning: Arel.sql("*, false AS unique_skipped_as_duplicate")) + + job_row, skipped_as_duplicate = driver.send(:postgres_to_job_row_from_raw, res.rows[0], res.columns, res.column_types) + + expect(job_row).to be_an_instance_of(River::JobRow) + expect(job_row).to have_attributes( + id: 1, + args: {"job_num" => 1}, + attempt: 0, + attempted_at: nil, + attempted_by: nil, + created_at: be_within(2).of(Time.now.getutc), + finalized_at: nil, + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: be_within(2).of(Time.now.getutc), + state: River::JOB_STATE_AVAILABLE, + tags: [] + ) + expect(skipped_as_duplicate).to be(false) + end + + it "converts a database record to `River::JobRow` with all properties" do + now = Time.now + res = River::Driver::ActiveRecord::RiverJob.insert({ + id: 1, + attempt: 1, + attempted_at: now, + attempted_by: ["client1"], + created_at: now, + args: {"job_num" => 1}, + finalized_at: now, + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: now, + state: River::JOB_STATE_COMPLETED, + tags: ["tag1"], + unique_key: Digest::SHA256.digest("unique_key_str") + }, returning: Arel.sql("*, true AS unique_skipped_as_duplicate")) + + job_row, skipped_as_duplicate = driver.send(:postgres_to_job_row_from_raw, res.rows[0], res.columns, res.column_types) + + expect(job_row).to be_an_instance_of(River::JobRow) + expect(job_row).to have_attributes( + id: 1, + args: {"job_num" => 1}, + attempt: 1, + attempted_at: be_within(2).of(now.getutc), + attempted_by: ["client1"], + created_at: be_within(2).of(now.getutc), + finalized_at: be_within(2).of(now.getutc), + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: be_within(2).of(now.getutc), + state: River::JOB_STATE_COMPLETED, + tags: ["tag1"], + unique_key: Digest::SHA256.digest("unique_key_str") + ) + expect(skipped_as_duplicate).to be(true) + end + + it "with errors" do + now = Time.now.utc + res = River::Driver::ActiveRecord::RiverJob.insert({ + args: {"job_num" => 1}, + errors: [JSON.dump( + { + at: now, + attempt: 1, + error: "job failure", + trace: "error trace" + } + )], + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + state: River::JOB_STATE_AVAILABLE + }, returning: Arel.sql("*, false AS unique_skipped_as_duplicate")) + + job_row, skipped_as_duplicate = driver.send(:postgres_to_job_row_from_raw, res.rows[0], res.columns, res.column_types) - it "with errors" do - now = Time.now.utc - res = River::Driver::ActiveRecord::RiverJob.insert({ - args: {"job_num" => 1}, - errors: [JSON.dump( - { - at: now, + expect(job_row.errors.count).to be(1) + expect(job_row.errors[0]).to be_an_instance_of(River::AttemptError) + expect(job_row.errors[0]).to have_attributes( + at: now.floor(0), attempt: 1, error: "job failure", trace: "error trace" - } - )], - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - state: River::JOB_STATE_AVAILABLE - }, returning: Arel.sql("*, false AS unique_skipped_as_duplicate")) - - job_row, skipped_as_duplicate = driver.send(:to_job_row_from_raw, res.rows[0], res.columns, res.column_types) - - expect(job_row.errors.count).to be(1) - expect(job_row.errors[0]).to be_an_instance_of(River::AttemptError) - expect(job_row.errors[0]).to have_attributes( - at: now.floor(0), - attempt: 1, - error: "job failure", - trace: "error trace" - ) - expect(skipped_as_duplicate).to be(false) + ) + expect(skipped_as_duplicate).to be(false) + end + end end end end diff --git a/driver/riverqueue-activerecord/spec/spec_helper.rb b/driver/riverqueue-activerecord/spec/spec_helper.rb index a9892cb..575ba96 100644 --- a/driver/riverqueue-activerecord/spec/spec_helper.rb +++ b/driver/riverqueue-activerecord/spec/spec_helper.rb @@ -1,7 +1,14 @@ require "active_record" require "debug" -ActiveRecord::Base.establish_connection(ENV["TEST_DATABASE_URL"] || "postgres://localhost/river_test") +PG_AVAILABLE = begin + ActiveRecord::Base.establish_connection(ENV["TEST_DATABASE_URL"] || "postgres://localhost/river_test") + ActiveRecord::Base.connection.execute("SELECT 1") + true +rescue => e + warn "PostgreSQL not available, skipping PostgreSQL tests: #{e.message}" + false +end def test_transaction ActiveRecord::Base.transaction do @@ -10,6 +17,14 @@ def test_transaction end end +def switch_to_sqlite! + ActiveRecord::Base.establish_connection(adapter: "sqlite3", database: ":memory:") +end + +def switch_to_postgres! + ActiveRecord::Base.establish_connection(ENV["TEST_DATABASE_URL"] || "postgres://localhost/river_test") +end + require "simplecov" SimpleCov.start do enable_coverage :branch diff --git a/driver/riverqueue-sequel/Gemfile b/driver/riverqueue-sequel/Gemfile index be9359d..ab4e105 100644 --- a/driver/riverqueue-sequel/Gemfile +++ b/driver/riverqueue-sequel/Gemfile @@ -8,7 +8,9 @@ group :development, :test do end group :test do + gem "pg" gem "rspec-core" gem "rspec-expectations" gem "simplecov", require: false + gem "sqlite3" end diff --git a/driver/riverqueue-sequel/Gemfile.lock b/driver/riverqueue-sequel/Gemfile.lock index 6446802..81413f1 100644 --- a/driver/riverqueue-sequel/Gemfile.lock +++ b/driver/riverqueue-sequel/Gemfile.lock @@ -7,7 +7,6 @@ PATH remote: . specs: riverqueue-sequel (0.10.1) - pg (> 0, < 1000) sequel (> 0, < 1000) GEM @@ -63,6 +62,8 @@ GEM simplecov_json_formatter (~> 0.1) simplecov-html (0.13.2) simplecov_json_formatter (0.1.4) + sqlite3 (2.9.3-arm64-darwin) + sqlite3 (2.9.3-x86_64-linux-gnu) standard (1.54.0) language_server-protocol (~> 3.17.0.2) lint_roller (~> 1.0) @@ -87,11 +88,13 @@ PLATFORMS x86_64-linux DEPENDENCIES + pg riverqueue! riverqueue-sequel! rspec-core rspec-expectations simplecov + sqlite3 standard BUNDLED WITH diff --git a/driver/riverqueue-sequel/docs/README.md b/driver/riverqueue-sequel/docs/README.md index e3f6601..f01d46e 100644 --- a/driver/riverqueue-sequel/docs/README.md +++ b/driver/riverqueue-sequel/docs/README.md @@ -1,22 +1,44 @@ -# riverqueue-sequel [![Build Status](https://github.com/riverqueue/riverqueue-ruby-sequel/workflows/CI/badge.svg)](https://github.com/riverqueue/riverqueue-ruby-sequel/actions) +# riverqueue-sequel -[Sequel](https://github.com/jeremyevans/sequel) driver for [River](https://github.com/riverqueue/river)'s [`riverqueue` gem for Ruby](https://rubygems.org/gems/riverqueue). +[Sequel](https://sequel.jeremyevans.net/) driver for [River](https://github.com/riverqueue/river)'s [`riverqueue` gem for Ruby](https://rubygems.org/gems/riverqueue). PostgreSQL and SQLite are supported. -`Gemfile` should contain the core gem and a driver like this one: +Add the core gem and this driver to `Gemfile`: -``` yaml +```ruby gem "riverqueue" gem "riverqueue-sequel" ``` -Initialize a client with: +Database adapters are optional dependencies. Add only the adapter used by your +application. + +For PostgreSQL, add `pg` to `Gemfile`: + +```ruby +gem "pg" +``` + +Then initialize a client with a Sequel database: + +```ruby +db = Sequel.connect("postgres://localhost/my_app") +client = River::Client.new(River::Driver::Sequel.new(db)) +``` + +For SQLite, add `sqlite3` to `Gemfile`: + +```ruby +gem "sqlite3" +``` + +Then initialize a client with an SQLite database: ```ruby -DB = Sequel.connect("postgres://...") -client = River::Client.new(River::Driver::Sequel.new(DB)) +db = Sequel.connect("sqlite://storage/river.sqlite3", timeout: 5_000) +client = River::Client.new(River::Driver::Sequel.new(db)) ``` -See also [`rubyqueue`](https://github.com/riverqueue/riverqueue-ruby). +Use current River migrations to create and update a production SQLite database. `River::Driver::Sequel.create_sqlite_schema` creates only the tables needed by this insert-only client and is intended primarily for tests. ## Development diff --git a/driver/riverqueue-sequel/lib/driver.rb b/driver/riverqueue-sequel/lib/driver.rb index 3b41706..4a59609 100644 --- a/driver/riverqueue-sequel/lib/driver.rb +++ b/driver/riverqueue-sequel/lib/driver.rb @@ -1,21 +1,152 @@ +require "securerandom" + module River::Driver - # Provides a Sequel driver for River. + # Provides a Sequel driver for River that supports both PostgreSQL and SQLite. # # Used in conjunction with a River client like: # # DB = Sequel.connect("postgres://...") # client = River::Client.new(River::Driver::Sequel.new(DB)) # + # Or with SQLite: + # + # DB = Sequel.connect("sqlite://path/to/river.db") + # client = River::Client.new(River::Driver::Sequel.new(DB)) + # class Sequel + SQLITE_CONFLICT_WHERE = <<~SQL.chomp + 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 + SQL + private_constant :SQLITE_CONFLICT_WHERE + + # SQLite 3.45+ may store JSON as binary JSONB. Always project JSON columns + # through json() so this driver can read both the current JSONB format and + # the text JSON used by River migrations through version 006. Cast times to + # text so Sequel doesn't interpret timezone-less SQLite timestamps in the + # process timezone. + SQLITE_JOB_COLUMNS = <<~SQL.chomp + id, + json(args) AS args, + attempt, + CAST(attempted_at AS text) AS attempted_at, + json(attempted_by) AS attempted_by, + CAST(created_at AS text) AS created_at, + json(errors) AS errors, + CAST(finalized_at AS text) AS finalized_at, + kind, + max_attempts, + json(metadata) AS metadata, + priority, + queue, + state, + CAST(scheduled_at AS text) AS scheduled_at, + json(tags) AS tags, + unique_key, + unique_states + SQL + private_constant :SQLITE_JOB_COLUMNS + + SQLITE_UNIQUE_NONCE_KEY = "river:unique_nonce" + private_constant :SQLITE_UNIQUE_NONCE_KEY + def initialize(db) @db = db - @db.extension(:pg_array) - @db.extension(:pg_json) + @is_sqlite = (db.database_type == :sqlite) + + unless @is_sqlite + db.extension(:pg_array) + db.extension(:pg_json) + end + end + + # Creates the SQLite tables needed by this insert-only client and its tests. + # Applications that run River workers should use River's migrations, which + # create the rest of River's schema as well. + def self.create_sqlite_schema(db) + db.run <<~SQL + CREATE TABLE IF NOT EXISTS river_job ( + id integer PRIMARY KEY, + args blob NOT NULL DEFAULT (jsonb('{}')), + attempt integer NOT NULL DEFAULT 0, + attempted_at timestamp, + attempted_by blob, + created_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP, + errors blob, + finalized_at timestamp, + kind text NOT NULL, + max_attempts integer NOT NULL, + 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')) + ) + SQL + + db.run "CREATE INDEX IF NOT EXISTS river_job_kind ON river_job (kind)" + db.run <<~SQL + CREATE INDEX IF NOT EXISTS river_job_state_and_finalized_at_index + ON river_job (state, finalized_at) WHERE finalized_at IS NOT NULL + SQL + db.run <<~SQL + CREATE INDEX IF NOT EXISTS river_job_prioritized_fetching_index + ON river_job (state, queue, priority, scheduled_at, id) + SQL + db.run <<~SQL + CREATE UNIQUE INDEX IF NOT EXISTS river_job_unique_idx ON river_job (unique_key) + WHERE #{SQLITE_CONFLICT_WHERE} + SQL + + db.run <<~SQL + CREATE TABLE IF NOT EXISTS river_notification ( + id integer PRIMARY KEY AUTOINCREMENT, + created_at timestamp NOT NULL DEFAULT (datetime('now', 'subsec')), + payload text NOT NULL, + topic text NOT NULL, + CONSTRAINT topic_length CHECK (length(topic) > 0 AND length(topic) < 128) + ) + SQL + db.run <<~SQL + CREATE INDEX IF NOT EXISTS river_notification_created_at_idx + ON river_notification (created_at) + SQL + db.run <<~SQL + CREATE INDEX IF NOT EXISTS river_notification_topic_id_idx + ON river_notification (topic, id) + SQL end def job_get_by_id(id) - data_set = @db[:river_job].where(id: id) - data_set.first ? to_job_row(data_set.first) : nil + if @is_sqlite + row = sqlite_job_rows("WHERE id = ? LIMIT 1", id).first + row ? sqlite_to_job_row_from_raw(row) : nil + else + data_set = @db[:river_job].where(id: id) + data_set.first ? to_job_row(data_set.first) : nil + end end def job_insert(insert_params) @@ -23,6 +154,26 @@ def job_insert(insert_params) end def job_insert_many(insert_params_array) + @is_sqlite ? sqlite_job_insert_many(insert_params_array) : postgres_job_insert_many(insert_params_array) + end + + def job_list + if @is_sqlite + sqlite_job_rows("ORDER BY id").map { |row| sqlite_to_job_row_from_raw(row) } + else + @db[:river_job].order_by(:id).all.map { |job| to_job_row(job) } + end + end + + def rollback_exception + ::Sequel::Rollback + end + + def transaction(&) + @db.transaction(savepoint: true, &) + end + + private def postgres_job_insert_many(insert_params_array) @db[:river_job] .insert_conflict( target: [:unique_key], @@ -32,24 +183,69 @@ def job_insert_many(insert_params_array) update: {kind: ::Sequel[:excluded][:kind]} ) .returning(::Sequel.lit("*, (xmax != 0) AS unique_skipped_as_duplicate")) - .multi_insert(insert_params_array.map { |p| insert_params_to_hash(p) }) - .map { |row| to_insert_result(row) } + .multi_insert(insert_params_array.map { |p| postgres_insert_params_to_hash(p) }) + .map { |row| [to_job_row(row), row[:unique_skipped_as_duplicate]] } end - def job_list - data_set = @db[:river_job].order_by(:id) - data_set.all.map { |job| to_job_row(job) } - end + # River's current SQLite driver uses json_each to make a batch a single, + # atomic statement. The JSON columns are converted to SQLite JSONB here, + # matching migration 007 and newer River databases. + private def sqlite_job_insert_many(insert_params_array) + return [] if insert_params_array.empty? - def rollback_exception - ::Sequel::Rollback - end + @db.transaction(savepoint: true) do + nonce = SecureRandom.hex(8) + jobs = insert_params_array.map { |param| sqlite_insert_params_to_hash(param, nonce) } - def transaction(&) - @db.transaction(savepoint: true, &) + sql = <<~SQL + INSERT INTO river_job ( + args, + created_at, + kind, + max_attempts, + metadata, + priority, + queue, + scheduled_at, + state, + tags, + unique_key, + unique_states + ) + SELECT + jsonb(json_extract(value, '$.args')), + datetime('now', 'subsec'), + cast(json_extract(value, '$.kind') AS text), + cast(json_extract(value, '$.max_attempts') AS integer), + jsonb(json_extract(value, '$.metadata')), + cast(json_extract(value, '$.priority') AS integer), + cast(json_extract(value, '$.queue') AS text), + coalesce(cast(json_extract(value, '$.scheduled_at') AS text), datetime('now', 'subsec')), + cast(json_extract(value, '$.state') AS text), + jsonb(json_extract(value, '$.tags')), + CASE + WHEN length(cast(json_extract(value, '$.unique_key') AS text)) = 0 THEN NULL + ELSE unhex(cast(json_extract(value, '$.unique_key') AS text)) + END, + nullif(cast(json_extract(value, '$.unique_states') AS integer), 0) + FROM json_each(cast(? AS blob)) + WHERE true + ON CONFLICT (unique_key) WHERE #{SQLITE_CONFLICT_WHERE} + DO UPDATE SET kind = EXCLUDED.kind + RETURNING #{SQLITE_JOB_COLUMNS} + SQL + + rows = @db.fetch(sql, JSON.dump(jobs)).all + sqlite_notify_insert(insert_params_array) + + rows.map do |row| + metadata = JSON.parse(row[:metadata]) + [sqlite_to_job_row_from_raw(row), metadata[SQLITE_UNIQUE_NONCE_KEY] != nonce] + end + end end - private def insert_params_to_hash(insert_params) + private def postgres_insert_params_to_hash(insert_params) { args: insert_params.encoded_args, kind: insert_params.kind, @@ -64,11 +260,32 @@ def transaction(&) } end - private def to_insert_result(result) - [to_job_row(result), result[:unique_skipped_as_duplicate]] + private def sqlite_insert_params_to_hash(insert_params, nonce) + { + args: JSON.parse(insert_params.encoded_args), + kind: insert_params.kind, + max_attempts: insert_params.max_attempts, + metadata: {SQLITE_UNIQUE_NONCE_KEY => nonce}, + priority: insert_params.priority, + queue: insert_params.queue, + scheduled_at: insert_params.scheduled_at ? format_time(insert_params.scheduled_at) : nil, + state: insert_params.state, + tags: insert_params.tags || [], + unique_key: insert_params.unique_key&.unpack1("H*"), + unique_states: insert_params.unique_states&.to_i(2) + } end private def to_job_row(river_job) + if @is_sqlite + row = sqlite_job_rows("WHERE id = ? LIMIT 1", river_job[:id]).first + sqlite_to_job_row_from_raw(row) + else + postgres_to_job_row(river_job) + end + end + + private def postgres_to_job_row(river_job) River::JobRow.new( id: river_job[:id], args: river_job[:args].to_h, @@ -97,5 +314,65 @@ def transaction(&) unique_states: ::River::UniqueBitmask.to_states(river_job[:unique_states]&.to_i(2)) ) end + + private def sqlite_to_job_row_from_raw(river_job) + errors = river_job[:errors] ? JSON.parse(river_job[:errors]) : [] + + River::JobRow.new( + id: river_job[:id], + args: JSON.parse(river_job[:args]), + attempt: river_job[:attempt], + attempted_at: parse_sqlite_time(river_job[:attempted_at]), + attempted_by: river_job[:attempted_by] ? JSON.parse(river_job[:attempted_by]) : nil, + created_at: parse_sqlite_time(river_job[:created_at]), + errors: errors.map { |deserialized_error| + River::AttemptError.new( + at: Time.parse(deserialized_error["at"]), + attempt: deserialized_error["attempt"], + error: deserialized_error["error"], + trace: deserialized_error["trace"] + ) + }, + finalized_at: parse_sqlite_time(river_job[:finalized_at]), + kind: river_job[:kind], + max_attempts: river_job[:max_attempts], + metadata: JSON.parse(river_job[:metadata]), + priority: river_job[:priority], + queue: river_job[:queue], + scheduled_at: parse_sqlite_time(river_job[:scheduled_at]), + state: river_job[:state], + tags: JSON.parse(river_job[:tags]), + unique_key: river_job[:unique_key]&.to_s, + unique_states: river_job[:unique_states] ? ::River::UniqueBitmask.to_states(river_job[:unique_states]) : nil + ) + end + + private def format_time(time) + time.getutc.round(3).strftime("%Y-%m-%d %H:%M:%S.%3N") + end + + private def parse_sqlite_time(value) + return nil unless value + + value = value.to_s + value += " UTC" unless value.match?(/(?:Z|[+-]\d{2}:?\d{2})\z/) + Time.parse(value).utc + end + + private def sqlite_job_rows(suffix, *binds) + @db.fetch("SELECT #{SQLITE_JOB_COLUMNS} FROM river_job #{suffix}", *binds).all + end + + private def sqlite_notify_insert(insert_params_array) + queues = insert_params_array + .select { |param| param.state == ::River::JOB_STATE_AVAILABLE } + .map(&:queue) + .uniq + return if queues.empty? + + @db[:river_notification].multi_insert(queues.map do |queue| + {payload: JSON.dump({queue: queue}), topic: "insert"} + end) + end end end diff --git a/driver/riverqueue-sequel/riverqueue-sequel.gemspec b/driver/riverqueue-sequel/riverqueue-sequel.gemspec index 32ae6ac..95428aa 100644 --- a/driver/riverqueue-sequel/riverqueue-sequel.gemspec +++ b/driver/riverqueue-sequel/riverqueue-sequel.gemspec @@ -1,8 +1,8 @@ Gem::Specification.new do |s| s.name = "riverqueue-sequel" s.version = "0.10.1" - s.summary = "Sequel driver for the River Ruby gem." - s.description = "Sequel driver for the River Ruby gem. Use in conjunction with the riverqueue gem to insert jobs that are worked in Go." + s.summary = "Sequel PostgreSQL and SQLite driver for the River Ruby gem." + s.description = "Sequel PostgreSQL and SQLite driver for the River Ruby gem. Use in conjunction with the riverqueue gem to insert jobs that are worked in Go." s.authors = ["Blake Gentry", "Brandur Leach"] s.email = "brandur@brandur.org" s.files = Dir.glob("lib/**/*") @@ -10,6 +10,5 @@ Gem::Specification.new do |s| s.license = "LGPL-3.0-or-later" # The stupid version bounds are used to silence Ruby's extremely obnoxious warnings. - s.add_dependency "pg", "> 0", "< 1000" s.add_dependency "sequel", "> 0", "< 1000" end diff --git a/driver/riverqueue-sequel/spec/driver_spec.rb b/driver/riverqueue-sequel/spec/driver_spec.rb index 30ddbd3..e1f0c8d 100644 --- a/driver/riverqueue-sequel/spec/driver_spec.rb +++ b/driver/riverqueue-sequel/spec/driver_spec.rb @@ -2,127 +2,323 @@ require_relative "../../../spec/driver_shared_examples" RSpec.describe River::Driver::Sequel do - around(:each) { |ex| test_transaction(&ex) } + if DB + context "with PostgreSQL" do + around(:each) { |ex| test_transaction(&ex) } - let!(:driver) { River::Driver::Sequel.new(DB) } - let(:client) { River::Client.new(driver) } + let!(:driver) { River::Driver::Sequel.new(DB) } + let(:client) { River::Client.new(driver) } - it_behaves_like "driver shared examples" + it_behaves_like "driver shared examples" - describe "client inserts" do - it "persists args as a JSON object rather than a JSON string" do - insert_res = client.insert(SimpleArgs.new(job_num: 1)) + describe "client inserts" do + it "persists args as a JSON object rather than a JSON string" do + insert_res = client.insert(SimpleArgs.new(job_num: 1)) - row = DB.fetch(<<~SQL, insert_res.job.id).first - SELECT args, jsonb_typeof(args) AS args_type - FROM river_job - WHERE id = ? - SQL + row = DB.fetch(<<~SQL, insert_res.job.id).first + SELECT args, jsonb_typeof(args) AS args_type + FROM river_job + WHERE id = ? + SQL - expect(row[:args_type]).to eq("object") - expect(row[:args].to_h).to eq({"job_num" => 1}) + expect(row[:args_type]).to eq("object") + expect(row[:args].to_h).to eq({"job_num" => 1}) + end + end + + describe "#to_job_row (PostgreSQL)" do + it "converts a database record to `River::JobRow` with minimal properties" do + river_job = DB[:river_job].returning.insert_select({ + id: 1, + args: %({"job_num":1}), + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT + }) + + job_row = driver.send(:to_job_row, river_job) + + expect(job_row).to be_an_instance_of(River::JobRow) + expect(job_row).to have_attributes( + id: 1, + args: {"job_num" => 1}, + attempt: 0, + attempted_at: nil, + attempted_by: nil, + created_at: be_within(2).of(Time.now.getutc), + finalized_at: nil, + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: be_within(2).of(Time.now.getutc), + state: River::JOB_STATE_AVAILABLE, + tags: [] + ) + end + + it "converts a database record to `River::JobRow` with all properties" do + now = Time.now + river_job = DB[:river_job].returning.insert_select({ + id: 1, + attempt: 1, + attempted_at: now, + attempted_by: ::Sequel.pg_array(["client1"]), + created_at: now, + args: %({"job_num":1}), + finalized_at: now, + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: now, + state: River::JOB_STATE_COMPLETED, + tags: ::Sequel.pg_array(["tag1"]), + unique_key: ::Sequel.blob(Digest::SHA256.digest("unique_key_str")) + }) + + job_row = driver.send(:to_job_row, river_job) + + expect(job_row).to be_an_instance_of(River::JobRow) + expect(job_row).to have_attributes( + id: 1, + args: {"job_num" => 1}, + attempt: 1, + attempted_at: be_within(2).of(now.getutc), + attempted_by: ["client1"], + created_at: be_within(2).of(now.getutc), + finalized_at: be_within(2).of(now.getutc), + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: be_within(2).of(now.getutc), + state: River::JOB_STATE_COMPLETED, + tags: ["tag1"], + unique_key: Digest::SHA256.digest("unique_key_str") + ) + end + + it "with errors" do + now = Time.now.utc + river_job = DB[:river_job].returning.insert_select({ + args: %({"job_num":1}), + errors: ::Sequel.pg_array([ + ::Sequel.pg_json_wrap({ + at: now, + attempt: 1, + error: "job failure", + trace: "error trace" + }) + ]), + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + state: River::JOB_STATE_AVAILABLE + }) + + job_row = driver.send(:to_job_row, river_job) + + expect(job_row.errors.count).to be(1) + expect(job_row.errors[0]).to be_an_instance_of(River::AttemptError) + expect(job_row.errors[0]).to have_attributes( + at: now.floor(0), + attempt: 1, + error: "job failure", + trace: "error trace" + ) + end + end end end - describe "#to_job_row" do - it "converts a database record to `River::JobRow` with minimal properties" do - river_job = DB[:river_job].returning.insert_select({ - id: 1, - args: %({"job_num":1}), - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT - }) - - job_row = driver.send(:to_job_row, river_job) - - expect(job_row).to be_an_instance_of(River::JobRow) - expect(job_row).to have_attributes( - id: 1, - args: {"job_num" => 1}, - attempt: 0, - attempted_at: nil, - attempted_by: nil, - created_at: be_within(2).of(Time.now.getutc), - finalized_at: nil, - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - priority: River::PRIORITY_DEFAULT, - queue: River::QUEUE_DEFAULT, - scheduled_at: be_within(2).of(Time.now.getutc), - state: River::JOB_STATE_AVAILABLE, - tags: [] - ) - end + next unless SQLITE_DB + + context "with SQLite" do + around(:each) { |ex| sqlite_test_transaction(&ex) } + + let!(:driver) { River::Driver::Sequel.new(SQLITE_DB) } + let(:client) { River::Client.new(driver) } + + it_behaves_like "driver shared examples" - it "converts a database record to `River::JobRow` with all properties" do - now = Time.now - river_job = DB[:river_job].returning.insert_select({ - id: 1, - attempt: 1, - attempted_at: now, - attempted_by: ::Sequel.pg_array(["client1"]), - created_at: now, - args: %({"job_num":1}), - finalized_at: now, - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - priority: River::PRIORITY_DEFAULT, - queue: River::QUEUE_DEFAULT, - scheduled_at: now, - state: River::JOB_STATE_COMPLETED, - tags: ::Sequel.pg_array(["tag1"]), - unique_key: ::Sequel.blob(Digest::SHA256.digest("unique_key_str")) - }) - - job_row = driver.send(:to_job_row, river_job) - - expect(job_row).to be_an_instance_of(River::JobRow) - expect(job_row).to have_attributes( - id: 1, - args: {"job_num" => 1}, - attempt: 1, - attempted_at: be_within(2).of(now.getutc), - attempted_by: ["client1"], - created_at: be_within(2).of(now.getutc), - finalized_at: be_within(2).of(now.getutc), - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - priority: River::PRIORITY_DEFAULT, - queue: River::QUEUE_DEFAULT, - scheduled_at: be_within(2).of(now.getutc), - state: River::JOB_STATE_COMPLETED, - tags: ["tag1"], - unique_key: Digest::SHA256.digest("unique_key_str") - ) + describe "client inserts" do + it "persists JSON columns as JSONB objects" do + insert_res = client.insert(SimpleArgs.new(job_num: 1)) + + row = SQLITE_DB.fetch(<<~SQL, insert_res.job.id).first + SELECT + json(args) AS args, + json_type(args) AS args_type, + typeof(args) AS args_storage_type, + typeof(metadata) AS metadata_storage_type, + typeof(tags) AS tags_storage_type, + CAST(created_at AS text) AS created_at, + CAST(scheduled_at AS text) AS scheduled_at + FROM river_job + WHERE id = ? + SQL + + expect(row[:args_type]).to eq("object") + expect(JSON.parse(row[:args])).to eq({"job_num" => 1}) + expect(row.values_at(:args_storage_type, :metadata_storage_type, :tags_storage_type)).to eq(["blob", "blob", "blob"]) + expect(row[:created_at]).to match(/\.\d{3}\z/) + expect(row[:scheduled_at]).to match(/\.\d{3}\z/) + expect(insert_res.job.errors).to eq([]) + end + + it "inserts a batch atomically" do + expect do + client.insert_many([ + SimpleArgs.new(job_num: 1), + River::InsertManyParams.new( + SimpleArgs.new(job_num: 2), + insert_opts: River::InsertOpts.new(queue: "") + ) + ]) + end.to raise_error(Sequel::CheckConstraintViolation) + + expect(driver.job_list).to be_empty + end + + it "notifies each available queue once per batch" do + client.insert_many([ + SimpleArgs.new(job_num: 1), + SimpleArgs.new(job_num: 2) + ]) + + rows = SQLITE_DB[:river_notification].order(:id).select(:payload, :topic).all + expect(rows).to contain_exactly( + {payload: JSON.dump({queue: River::QUEUE_DEFAULT}), topic: "insert"} + ) + end + + it "handles an empty batch" do + expect(driver.job_insert_many([])).to eq([]) + end + + it "defaults a missing scheduled_at" do + params = River::Driver::JobInsertParams.new( + encoded_args: JSON.dump({job_num: 1}), + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: nil, + state: River::JOB_STATE_AVAILABLE, + tags: [] + ) + + job, = driver.job_insert(params) + expect(job.scheduled_at).to be_within(2).of(Time.now.utc) + end + + it "rounds timestamps to three fractional digits" do + time = Time.utc(2026, 8, 31, 12, 34, 56) + 0.1236 + expect(driver.send(:format_time, time)).to eq("2026-08-31 12:34:56.124") + end end - it "with errors" do - now = Time.now.utc - river_job = DB[:river_job].returning.insert_select({ - args: %({"job_num":1}), - errors: ::Sequel.pg_array([ - ::Sequel.pg_json_wrap({ - at: now, + describe "#to_job_row (SQLite)" do + it "converts a database record to `River::JobRow` with minimal properties" do + SQLITE_DB[:river_job].insert( + args: Sequel.function(:jsonb, %({"job_num":1})), + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT + ) + river_job = SQLITE_DB[:river_job].first + + job_row = driver.send(:to_job_row, river_job) + + expect(job_row).to be_an_instance_of(River::JobRow) + expect(job_row).to have_attributes( + id: be_a(Integer), + args: {"job_num" => 1}, + attempt: 0, + attempted_at: nil, + attempted_by: nil, + created_at: be_within(2).of(Time.now.getutc), + finalized_at: nil, + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: be_within(2).of(Time.now.getutc), + state: River::JOB_STATE_AVAILABLE, + tags: [] + ) + end + + it "converts a database record to `River::JobRow` with all properties" do + now = Time.now.utc + now_str = now.iso8601(3) + + SQLITE_DB[:river_job].insert( + attempt: 1, + attempted_at: now_str, + attempted_by: Sequel.function(:jsonb, JSON.dump(["client1"])), + created_at: now_str, + args: Sequel.function(:jsonb, %({"job_num":1})), + finalized_at: now_str, + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: now_str, + state: River::JOB_STATE_COMPLETED, + tags: Sequel.function(:jsonb, JSON.dump(["tag1"])), + unique_key: ::Sequel.blob(Digest::SHA256.digest("unique_key_str")) + ) + river_job = SQLITE_DB[:river_job].first + + job_row = driver.send(:to_job_row, river_job) + + expect(job_row).to be_an_instance_of(River::JobRow) + expect(job_row).to have_attributes( + id: be_a(Integer), + args: {"job_num" => 1}, + attempt: 1, + attempted_at: be_within(2).of(now.getutc), + attempted_by: ["client1"], + created_at: be_within(2).of(now.getutc), + finalized_at: be_within(2).of(now.getutc), + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + priority: River::PRIORITY_DEFAULT, + queue: River::QUEUE_DEFAULT, + scheduled_at: be_within(2).of(now.getutc), + state: River::JOB_STATE_COMPLETED, + tags: ["tag1"], + unique_key: Digest::SHA256.digest("unique_key_str") + ) + end + + it "with errors" do + now = Time.now.utc + + SQLITE_DB[:river_job].insert( + args: Sequel.function(:jsonb, %({"job_num":1})), + errors: Sequel.function(:jsonb, JSON.dump([{ + at: now.iso8601, attempt: 1, error: "job failure", trace: "error trace" - }) - ]), - kind: "simple", - max_attempts: River::MAX_ATTEMPTS_DEFAULT, - state: River::JOB_STATE_AVAILABLE - }) - - job_row = driver.send(:to_job_row, river_job) - - expect(job_row.errors.count).to be(1) - expect(job_row.errors[0]).to be_an_instance_of(River::AttemptError) - expect(job_row.errors[0]).to have_attributes( - at: now.floor(0), - attempt: 1, - error: "job failure", - trace: "error trace" - ) + }])), + kind: "simple", + max_attempts: River::MAX_ATTEMPTS_DEFAULT, + state: River::JOB_STATE_AVAILABLE + ) + river_job = SQLITE_DB[:river_job].first + + job_row = driver.send(:to_job_row, river_job) + + expect(job_row.errors.count).to be(1) + expect(job_row.errors[0]).to be_an_instance_of(River::AttemptError) + expect(job_row.errors[0]).to have_attributes( + at: now.floor(0), + attempt: 1, + error: "job failure", + trace: "error trace" + ) + end end end end diff --git a/driver/riverqueue-sequel/spec/spec_helper.rb b/driver/riverqueue-sequel/spec/spec_helper.rb index 253997f..d728686 100644 --- a/driver/riverqueue-sequel/spec/spec_helper.rb +++ b/driver/riverqueue-sequel/spec/spec_helper.rb @@ -1,6 +1,19 @@ require "sequel" -DB = Sequel.connect(ENV["TEST_DATABASE_URL"] || "postgres://localhost/river_test") +DB = begin + Sequel.connect(ENV["TEST_DATABASE_URL"] || "postgres://localhost/river_test") +rescue => e + warn "PostgreSQL not available, skipping PostgreSQL tests: #{e.message}" + nil +end + +SQLITE_DB = begin + require "sqlite3" + Sequel.sqlite +rescue LoadError + warn "sqlite3 gem not available, skipping SQLite tests" + nil +end def test_transaction DB.transaction do @@ -9,6 +22,13 @@ def test_transaction end end +def sqlite_test_transaction + SQLITE_DB.transaction do + yield + raise Sequel::Rollback + end +end + require "simplecov" SimpleCov.start do enable_coverage :branch @@ -17,3 +37,5 @@ def test_transaction require "riverqueue" require "riverqueue-sequel" + +River::Driver::Sequel.create_sqlite_schema(SQLITE_DB) if SQLITE_DB