Add connects_to option to configure different DB/shards

.github/workflows/main.yml

@@ -1,5 +1,5 @@
 name: Build
-on: [push, pull_request]
+on: [push]
 
 jobs:
   tests:
@@ -39,3 +39,5 @@ jobs:
           bin/rails db:setup
       - name: Run tests
         run: bin/rails test
+      - name: Run tests with separate connection
+        run: SEPARATE_CONNECTION=1 bin/rails test

app/models/solid_queue/record.rb

@@ -1,6 +1,8 @@
 module SolidQueue
   class Record < ActiveRecord::Base
     self.abstract_class = true
+
+    connects_to **SolidQueue.connects_to if SolidQueue.connects_to
   end
 end
 

lib/solid_queue.rb

@@ -20,8 +20,7 @@
 
 module SolidQueue
   mattr_accessor :logger, default: ActiveSupport::Logger.new($stdout)
-  mattr_accessor :app_executor
-  mattr_accessor :on_thread_error
+  mattr_accessor :app_executor, :on_thread_error, :connects_to
 
   mattr_accessor :process_heartbeat_interval, default: 60.seconds
   mattr_accessor :process_alive_threshold, default: 5.minutes

test/dummy/config/application.rb

@@ -28,5 +28,9 @@ class Application < Rails::Application
     # config.eager_load_paths << Rails.root.join("extras")
 
     config.active_job.queue_adapter = :solid_queue
+
+    if ENV["SEPARATE_CONNECTION"] && ENV["TARGET_DB"] != "sqlite"
+      config.solid_queue.connects_to = { database: { writing: :primary, reading: :replica } }
+    end
   end
 end

test/dummy/config/database.yml

@@ -5,21 +5,14 @@
 #   gem "mysql2"
 #
 
+<% def database_name_from(name); ENV["TARGET_DB"]=="sqlite" ? "db/#{name}.sqlite3" : name; end %>
+
 <% if ENV["TARGET_DB"] == "sqlite" %>
 default: &default
   adapter: sqlite3
   pool: <%= ENV.fetch("RAILS_MAX_THREADS") { 50 } %>
   retries: 100
 
-development:
-  <<: *default
-  database: db/development.sqlite3
-
-test:
-  <<: *default
-  pool: 20
-  database: db/test.sqlite3
-
 <% elsif ENV["TARGET_DB"] == "postgres" %>
 default: &default
   adapter: postgresql
@@ -30,30 +23,40 @@ default: &default
   port: 55432
   gssencmode: disable # https://github.com/ged/ruby-pg/issues/311
 
-development:
-  <<: *default
-  database: solid_queue_development
-
-test:
-  <<: *default
-  pool: 20
-  database: solid_queue_test
-
 <% else %>
 default: &default
   adapter: mysql2
   username: root
   pool: <%= ENV.fetch("RAILS_MAX_THREADS") { 5 } %>
   host: "127.0.0.1"
   port: 33060
+<% end %>
+
+# development:
+#   <<: *default
+#   database: <%= database_name_from("solid_queue_development") %>
+
+# test:
+#   <<: *default
+#   database: <%= database_name_from("solid_queue_development") %>
+
 
 development:
-  <<: *default
-  database: solid_queue_development
+  primary:
+    <<: *default
+    database: <%= database_name_from("solid_queue_development") %>
+  replica:
+    <<: *default
+    database: <%= database_name_from("solid_queue_development") %>
+    replica: true
 
 test:
-  <<: *default
-  pool: 20
-  database: solid_queue_test
-<% end %>
-
+  primary:
+    <<: *default
+    pool: 20
+    database: <%= database_name_from("solid_queue_test") %>
+  replica:
+    <<: *default
+    pool: 20
+    database: <%= database_name_from("solid_queue_test") %>
+    replica: true

test/integration/concurrency_controls_test.rb

@@ -41,12 +41,12 @@ class ConcurrencyControlsTest < ActiveSupport::TestCase
   test "schedule several conflicting jobs over the same record sequentially" do
     UpdateResultJob.set(wait: 0.2.seconds).perform_later(@result, name: "000", pause: 0.1.seconds)
 
-    ("A".."F").each do |name|
-      SequentialUpdateResultJob.set(wait: 0.2.seconds).perform_later(@result, name: name, pause: 0.2.seconds)
+    ("A".."F").each_with_index do |name, i|
+      SequentialUpdateResultJob.set(wait: (0.2 + i * 0.01).seconds).perform_later(@result, name: name, pause: 0.2.seconds)
     end
 
-    ("G".."K").each do |name|
-      SequentialUpdateResultJob.set(wait: 0.4.seconds).perform_later(@result, name: name)
+    ("G".."K").each_with_index do |name, i|
+      SequentialUpdateResultJob.set(wait: (0.4 + i * 0.01).seconds).perform_later(@result, name: name)
     end
 
     wait_for_jobs_to_finish_for(4.seconds)

test/models/solid_queue/job_test.rb

@@ -1,6 +1,13 @@
 require "test_helper"
 
 class SolidQueue::JobTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  teardown do
+    SolidQueue::Job.destroy_all
+    JobResult.delete_all
+  end
+
   class NonOverlappingJob < ApplicationJob
     limits_concurrency key: ->(job_result, **) { job_result }
 
@@ -108,6 +115,23 @@ class NonOverlappingGroupedJob2 < NonOverlappingJob
     assert blocked_execution.expires_at <= SolidQueue.default_concurrency_control_period.from_now
   end
 
+  if ENV["SEPARATE_CONNECTION"] && ENV["TARGET_DB"] != "sqlite"
+    test "uses a different connection and transaction than the one in use when connects_to is specified" do
+      assert_difference -> { SolidQueue::Job.count } do
+        assert_no_difference -> { JobResult.count } do
+          JobResult.transaction do
+            JobResult.create!(queue_name: "default", value: "this will be rolled back")
+            StoreResultJob.perform_later("enqueued inside a rolled back transaction")
+            raise ActiveRecord::Rollback
+          end
+        end
+      end
+
+      job = SolidQueue::Job.last
+      assert_equal "enqueued inside a rolled back transaction", job.arguments.dig("arguments", 0)
+    end
+  end
+
   private
     def assert_ready(&block)
       assert_difference -> { SolidQueue::Job.count } => +1, -> { SolidQueue::ReadyExecution.count } => +1, &block

test/test_helper.rb

@@ -122,6 +122,6 @@ def reap_processes
     # by the cached queries might have been updated, created or deleted in the forked
     # processes.
     def skip_active_record_query_cache(&block)
-      ActiveRecord::Base.uncached(&block)
+      SolidQueue::Record.uncached(&block)
     end
 end