Enqueue jobs inside a connected_to block

[bubiche]

My rails application connects to multiple databases with the configs below:

database.yml

production:
  primary_shard:
    ...
  secondary_shard:
    ...
  queue:
    ...

config/initializers/database.rb

# This is needed due to us using another gem whose model writes to both primary_shard and secondary_shard and its model inherits from ActiveRecord::Base
ActiveRecord::Base.instance_eval do
    connects_to shards: {
      primary_shard: {
        writing: :primary_shard,
      },
      secondary_shard: {
        writing: :secondary_shard,
      },
    }
end

SolidQueue has been working well, the only time where we cannot enqueue is if we have a block of code like below:

ActiveRecord::Base.connected_to(role: :writing, shard: :secondary_shard) do
  # ...
  SomeJob.perform_later(some_argument)
  # ...
end

I think it is because SolidQueue is trying to insert the job into secondary_shard instead of the queue database. Is there any way to circumvent this besides rewriting the code to call perform_later outside of the connected_to block? SomeJob.perform_later might be called inside a function with the connected_to block so it's quite a big refactor for us.

[rosa]

Ohhh, totally! Let me look into this one to see how to fix, I think it's something Solid Queue should handle. Thanks for the report!

[djpate]

Same issue here. is there a workaround? my entire controller is in a connected_to block?
I tried wrapping the perform_later in another connection switch but no luck.

ActiveRecord::Base.connected_to(role: :writing, shard: :queue) do
  Workers::Foobar.perform_later(account_id: account_id, id: id)
end

[thibaudgg]

Same problem for me, I'm working on switching my multi-tenant app to SQLite (one database per tenant) using sharding, while keep the SolidQueue database separated.

Please let me know how I could help, or if sharing more code would be helpful. I'm happy to help fix the issue as well.

class ApplicationRecord < ActiveRecord::Base
  self.abstract_class = true

  connects_to shards: Tenant.shards.map { |shard|
    [ shard, { writing: shard } ]
  }.to_h
end

Production config:

  config.active_job.queue_adapter = :solid_queue
  config.solid_queue.connects_to = { database: { writing: :queue } }

Tenant switching logic (simplified):

module Tenant
  extend self
  def switch(tenant)
    ActiveRecord::Base.connected_to(shard: tenant.to_sym) do
        yield
    end
  end
  # ...
end

But trying to enqueue jobs gives me that error trace:

Stacktrace

ActiveRecord::ConnectionNotDefined: No database connection defined for SolidQueue::Record with 'TENANT_A' shard. (ActiveRecord::ConnectionNotDefined)
  from active_record/connection_adapters/abstract/connection_handler.rb:230:in `retrieve_connection_pool'
  from active_record/connection_handling.rb:343:in `connection_pool'
  from active_record/connection_handling.rb:310:in `with_connection'
  from active_record/transactions.rb:410:in `with_transaction_returning_status'
  from active_record/transactions.rb:366:in `save!'
  from active_record/suppressor.rb:56:in `save!'
  from active_record/persistence.rb:55:in `create!'
  from bundle/ruby/3.3.0/bundler/gems/solid_queue-51c75bec01c8/app/models/solid_queue/job.rb:41:in `create_from_active_job'
  from bundle/ruby/3.3.0/bundler/gems/solid_queue-51c75bec01c8/app/models/solid_queue/job.rb:31:in `enqueue'
  from active_job/queue_adapters/solid_queue_adapter.rb:16:in `enqueue'
  from active_job/enqueuing.rb:132:in `raw_enqueue'
  from active_job/enqueue_after_transaction_commit.rb:40:in `raw_enqueue'
  from active_job/enqueuing.rb:117:in `block in enqueue'
  from active_support/callbacks.rb:120:in `block in run_callbacks'
  from active_job/instrumentation.rb:40:in `block in instrument'
  from active_support/notifications.rb:210:in `block in instrument'
  from active_support/notifications/instrumenter.rb:58:in `instrument'
  from sentry/rails/tracing.rb:56:in `instrument'
  from active_support/notifications.rb:210:in `instrument'
  from active_job/instrumentation.rb:39:in `instrument'
  from active_record/railties/job_runtime.rb:18:in `instrument'
  from active_job/instrumentation.rb:21:in `block (2 levels) in <module:Instrumentation>'
  from active_support/callbacks.rb:129:in `instance_exec'
  from active_support/callbacks.rb:129:in `block in run_callbacks'
  from active_support/tagged_logging.rb:143:in `block in tagged'
  from active_support/tagged_logging.rb:38:in `tagged'
  from active_support/tagged_logging.rb:143:in `tagged'
  from active_support/broadcast_logger.rb:241:in `method_missing'
  from active_job/logging.rb:39:in `tag_logger'
  from active_job/logging.rb:28:in `block (2 levels) in <module:Logging>'
  from active_support/callbacks.rb:129:in `instance_exec'
  from active_support/callbacks.rb:129:in `block in run_callbacks'
  from active_support/callbacks.rb:140:in `run_callbacks'
  from active_job/enqueuing.rb:116:in `enqueue'
  from active_job/configured_job.rb:15:in `perform_later'
  from action_mailer/parameterized.rb:148:in `enqueue_delivery'
  from action_mailer/message_delivery.rb:103:in `deliver_later'
  from app/controllers/sessions_controller.rb:26:in `create'
  from action_controller/metal/basic_implicit_render.rb:8:in `send_action'
  from abstract_controller/base.rb:226:in `process_action'
  from action_controller/metal/rendering.rb:193:in `process_action'
  from abstract_controller/callbacks.rb:261:in `block in process_action'
  from active_support/callbacks.rb:120:in `block in run_callbacks'
  from turbo-rails.rb:24:in `with_request_id'
  from bundle/ruby/3.3.0/gems/turbo-rails-2.0.11/app/controllers/concerns/turbo/request_id_tracking.rb:10:in `turbo_tracking_request_id'
  from active_support/callbacks.rb:129:in `block in run_callbacks'
  from action_text/rendering.rb:25:in `with_renderer'
  from action_text/engine.rb:71:in `block (4 levels) in <class:Engine>'
  from active_support/callbacks.rb:129:in `instance_exec'
  from active_support/callbacks.rb:129:in `block in run_callbacks'
  from sentry/rails/controller_transaction.rb:21:in `block in sentry_around_action'
  from sentry/hub.rb:115:in `block in with_child_span'
  from sentry/span.rb:232:in `with_child_span'
  from sentry/hub.rb:113:in `with_child_span'
  from sentry-ruby.rb:499:in `with_child_span'
  from sentry/rails/controller_transaction.rb:18:in `sentry_around_action'
  from active_support/callbacks.rb:129:in `block in run_callbacks'
  from active_support/callbacks.rb:140:in `run_callbacks'
  from abstract_controller/callbacks.rb:260:in `process_action'
  from action_controller/metal/rescue.rb:27:in `process_action'
  from action_controller/metal/instrumentation.rb:76:in `block in process_action'
  from active_support/notifications.rb:210:in `block in instrument'
  from active_support/notifications/instrumenter.rb:58:in `instrument'
  from sentry/rails/tracing.rb:56:in `instrument'
  from active_support/notifications.rb:210:in `instrument'
  from action_controller/metal/instrumentation.rb:75:in `process_action'
  from action_controller/metal/params_wrapper.rb:259:in `process_action'
  from active_record/railties/controller_runtime.rb:39:in `process_action'
  from abstract_controller/base.rb:163:in `process'
  from action_view/rendering.rb:40:in `process'
  from action_controller/metal.rb:252:in `dispatch'
  from action_controller/metal.rb:335:in `dispatch'
  from action_dispatch/routing/route_set.rb:67:in `dispatch'
  from action_dispatch/routing/route_set.rb:50:in `serve'
  from action_dispatch/journey/router.rb:53:in `block in serve'
  from action_dispatch/journey/router.rb:133:in `block in find_routes'
  from action_dispatch/journey/router.rb:126:in `each'
  from action_dispatch/journey/router.rb:126:in `find_routes'
  from action_dispatch/journey/router.rb:34:in `serve'
  from action_dispatch/routing/route_set.rb:908:in `call'
  from lib/tenant/middleware.rb:12:in `block in call'
  from lib/tenant.rb:56:in `block (2 levels) in switch'
  from active_record/connection_handling.rb:214:in `prohibit_shard_swapping'
  from lib/tenant.rb:55:in `block in switch'
  from active_record/connection_handling.rb:398:in `with_role_and_shard'
  from active_record/connection_handling.rb:149:in `connected_to'
  from lib/tenant.rb:54:in `switch'
  from lib/tenant/middleware.rb:12:in `call'
  from rack/tempfile_reaper.rb:20:in `call'
  from rack/etag.rb:29:in `call'
  from rack/conditional_get.rb:43:in `call'
  from rack/head.rb:15:in `call'
  from action_dispatch/http/permissions_policy.rb:38:in `call'
  from action_dispatch/http/content_security_policy.rb:35:in `call'
  from rack/session/abstract/id.rb:272:in `context'
  from rack/session/abstract/id.rb:266:in `call'
  from action_dispatch/middleware/cookies.rb:706:in `call'
  from action_dispatch/middleware/callbacks.rb:31:in `block in call'
  from active_support/callbacks.rb:100:in `run_callbacks'
  from action_dispatch/middleware/callbacks.rb:30:in `call'
  from sentry/rails/rescued_exception_interceptor.rb:14:in `call'
  from action_dispatch/middleware/debug_exceptions.rb:31:in `call'
  from sentry/rack/capture_exceptions.rb:30:in `block (2 levels) in call'
  from sentry/hub.rb:265:in `with_session_tracking'
  from sentry-ruby.rb:412:in `with_session_tracking'
  from sentry/rack/capture_exceptions.rb:21:in `block in call'
  from sentry/hub.rb:59:in `with_scope'
  from sentry-ruby.rb:392:in `with_scope'
  from sentry/rack/capture_exceptions.rb:20:in `call'
  from action_dispatch/middleware/show_exceptions.rb:32:in `call'
  from lograge/rails_ext/rack/logger.rb:18:in `call_app'
  from rails/rack/logger.rb:29:in `call'
  from rails/rack/silence_request.rb:28:in `call'
  from action_dispatch/middleware/remote_ip.rb:96:in `call'
  from request_store/middleware.rb:19:in `call'
  from action_dispatch/middleware/request_id.rb:34:in `call'
  from rack/method_override.rb:28:in `call'
  from rack/runtime.rb:24:in `call'
  from action_dispatch/middleware/executor.rb:16:in `call'
  from action_dispatch/middleware/static.rb:27:in `call'
  from rack/sendfile.rb:114:in `call'
  from action_dispatch/middleware/assume_ssl.rb:24:in `call'
  from rack/cors.rb:102:in `call'
  from rails/engine.rb:535:in `call'
  from rack/status.rb:13:in `call'
  from puma/configuration.rb:272:in `call'
  from puma/request.rb:100:in `block in handle_request'
  from puma/thread_pool.rb:378:in `with_force_shutdown'
  from puma/request.rb:99:in `handle_request'
  from puma/server.rb:464:in `process_client'
  from puma/server.rb:245:in `block in run'
  from puma/thread_pool.rb:155:in `block in spawn_thread'

[thibaudgg]

Added a comment on this other ticket (#353), that might help here.

[rosa]

Sorry for the delay on this one, I haven't managed to look into it yet, but it's definitely on my radar. I'll try next week 🤞

[thibaudgg]

Thanks @rosa!

I just gave a tried to this patch on production, and it seems to work like a charm.

# config/initializers/solid_queue_patch.rb
Rails.application.config.to_prepare do
  module CurrentShardPatch
    def current_shard; :queue end
  end

  SolidQueue::Record.send(:extend, CurrentShardPatch)
end

My app is open-source, so you can give it a look if that's helpful to you.
Here some relevant parts:

• https://github.com/csa-admin-org/csa-admin/blob/master/config/initializers/solid_queue_shard.rb
• https://github.com/csa-admin-org/csa-admin/blob/master/lib/tenant.rb#L46-L50
• https://github.com/csa-admin-org/csa-admin/blob/master/app/jobs/concerns/tenant_switcher.rb
• https://github.com/csa-admin-org/csa-admin/blob/master/config/environments/production.rb#L61
• https://github.com/csa-admin-org/csa-admin/blob/master/config/database.yml

[rosa]

Oh, awesome! I'm sure this is going to be super helpful! I'll look into all those next week, too 🙏 Thank you so much!

[hubertjakubiak]

Hi @rosa, just checking in to see if there have been any updates on this issue. I’m still encountering the same problem with enqueuing jobs inside a connected_to block. Would love to hear if there's any progress on a solution. Thanks!

[rosa]

@hubertjakubiak so sorry for the delay! I did look into it but was immediately pulled into other directions before I could figure out anything, but this is definitely high on my list.

[rosa]

Hey @hubertjakubiak, @thibaudgg, @bubiche, I'm back again with something I did find back then, and that just revisited... and it was that things work fine as long as you don't pollute ActiveRecord::Base with the shard configuration. In fact, I couldn't get it to work if I tried to do that.

I configured some sharded records as follows:

database.yml config:

  primary:
    <<: *default
    database: <%= database_name_from("development") %>
  shard_one:
    <<: *default
    database: <%= database_name_from("development_shard_one") %>
    migrations_paths: db/migrate_shards
  shard_two:
    <<: *default
    database: <%= database_name_from("development_shard_two") %>
    migrations_paths: db/migrate_shards
  queue:
    <<: *default
    database: <%= database_name_from("development_queue") %>
    migrations_paths: db/queue_migrate

Then in the app:

class ApplicationRecord < ActiveRecord::Base
  primary_abstract_class
end

class ShardedRecord < ApplicationRecord
  self.abstract_class = true

  connects_to shards: {
    shard_one: { writing: :shard_one },
    shard_two: { writing: :shard_two }
  }
end

class JobResult < ApplicationRecord
end

class ShardedJobResult < ShardedRecord
end

Then, the following doesn't work (this might have been a result of changes to connection pool management since the Rail guides for horizontal sharding were written):

ActiveRecord::Base.connected_to(role: :writing, shard: :shard_two) do
  JobResult.create!
  ShardedJobResult.create!(value: "in shard two")
end

No connection pool for 'ActiveRecord::Base' found for the 'shard_two' shard. (ActiveRecord::ConnectionNotEstablished)

The following works:

ShardedRecord.connected_to(role: :writing, shard: :shard_two) do
  JobResult.create!(value: "right place")
  ShardedJobResult.create!(value: "in shard two")
end

  TRANSACTION (1.5ms)  BEGIN
  JobResult Create (3.4ms)  INSERT INTO `job_results` (`queue_name`, `status`, `value`, `created_at`, `updated_at`) VALUES (NULL, NULL, 'right place', '2024-12-05 20:41:54.697692', '2024-12-05 20:41:54.697692')
  TRANSACTION (4.0ms)  COMMIT
  TRANSACTION (0.9ms)  BEGIN
  ShardedJobResult Create (2.0ms)  INSERT INTO `sharded_job_results` (`value`, `created_at`, `updated_at`) VALUES ('in shard two', '2024-12-05 20:41:54.720816', '2024-12-05 20:41:54.720816')
  TRANSACTION (2.1ms)  COMMIT
=> #<ShardedJobResult:0x000000011fc70c98 id: 2, value: "in shard two", created_at: Thu, 05 Dec 2024 20:41:54.720816000 UTC +00:00, updated_at: Thu, 05 Dec 2024 20:41:54.720816000 UTC +00:00>

And the following also works, with everything created in the corresponding DB:

ShardedRecord.connected_to(role: :writing, shard: :shard_two) do
  JobResult.create!(value: "right place")
  ShardedJobResult.create!(value: "in shard two")
  AddToBufferJob.perform_later(2)
end

  TRANSACTION (1.7ms)  BEGIN
  JobResult Create (4.3ms)  INSERT INTO `job_results` (`queue_name`, `status`, `value`, `created_at`, `updated_at`) VALUES (NULL, NULL, 'right place', '2024-12-05 20:42:53.607259', '2024-12-05 20:42:53.607259')
  TRANSACTION (4.4ms)  COMMIT
  TRANSACTION (1.1ms)  BEGIN
  ShardedJobResult Create (2.3ms)  INSERT INTO `sharded_job_results` (`value`, `created_at`, `updated_at`) VALUES ('in shard two', '2024-12-05 20:42:53.626346', '2024-12-05 20:42:53.626346')
  TRANSACTION (2.4ms)  COMMIT
  TRANSACTION (0.7ms)  BEGIN
  SolidQueue::Job Create (2.3ms)  INSERT INTO `solid_queue_jobs` (`queue_name`, `class_name`, `arguments`, `priority`, `active_job_id`, `scheduled_at`, `finished_at`, `concurrency_key`, `created_at`, `updated_at`) VALUES ('background', 'AddToBufferJob', '{\"job_class\":\"AddToBufferJob\",\"job_id\":\"3f2b5646-5905-43e0-ad97-7fab2c425494\",\"provider_job_id\":null,\"queue_name\":\"background\",\"priority\":null,\"arguments\":[2],\"executions\":0,\"exception_executions\":{},\"locale\":\"en\",\"timezone\":\"UTC\",\"enqueued_at\":\"2024-12-05T20:42:53.666927000Z\",\"scheduled_at\":\"2024-12-05T20:42:53.666125000Z\"}', 0, '3f2b5646-5905-43e0-ad97-7fab2c425494', '2024-12-05 20:42:53.666125', NULL, NULL, '2024-12-05 20:42:53.702207', '2024-12-05 20:42:53.702207')
  TRANSACTION (0.6ms)  SAVEPOINT active_record_1
  SolidQueue::Job Load (1.7ms)  SELECT `solid_queue_jobs`.* FROM `solid_queue_jobs` WHERE `solid_queue_jobs`.`id` = 3 LIMIT 1
  SolidQueue::ReadyExecution Create (0.8ms)  INSERT INTO `solid_queue_ready_executions` (`job_id`, `queue_name`, `priority`, `created_at`) VALUES (3, 'background', 0, '2024-12-05 20:42:53.740160')
  TRANSACTION (0.5ms)  RELEASE SAVEPOINT active_record_1
  TRANSACTION (2.6ms)  COMMIT
Enqueued AddToBufferJob (Job ID: 3f2b5646-5905-43e0-ad97-7fab2c425494) to SolidQueue(background) with arguments: 2

So, I wonder if things would work fine if you switched from using ActiveRecord::Base to the abstract class that actually has the connects_to shard: ... 🤔

    ApplicationRecord.connected_to(shard: tenant.to_sym) do
        yield
    end

I feel I'm missing something here 😳

[thibaudgg]

Hey @rosa, thanks for looking into this and for your detailed answer.

Your suggestion to switch from ActiveRecord::Base to an abstract class is definitely worth considering. However, in my multi-tenant setup, I also rely on models like ActiveStorage::Record (blobs and attachments models) that must connect to the correct tenant shard and inherit directly from ActiveRecord::Base (and not ApplicationRecord). This makes it tricky to apply your proposed solution as is.

It might just be a unique aspect of my environment, where I have no primary shard and every model (besides SolidQueue::Record models) must connect to their respective tenant shards. If you have any other approaches or insights, I’d be happy to explore them.

So far, the patch I mentioned above has been working like a charm on production.

[rosa]

@thibaudgg good point! I think in that case, you could rely on ActiveRecord::Base.connected_to_many, with both your ApplicationRecord and ActiveStorage::Record. Something like

    ActiveRecord::Base.connected_to_many(ApplicationRecord, ActiveStorage::Record, role: :writing, shard: tenant.to_sym) do
        yield
    end

The more I've been looking into the connection handling code, the more I think the guides are wrong/outdated and that ActiveRecord::Base.connected_to should never be used directly 🤔

I'm going to propose a PR to update the guides shortly.

[robacarp]

connected_to_many is what we have found we need to use as well. Unfortunately, it assumes that role is the same for all connections, so we have this type of thing going on in our middleware:

      ActiveRecord::Base.connected_to_many(GlobalDB, role: :reading) do
        ActiveRecord::Base.connected_to_many(CustomerDB, role: :writing, shard: "customer_NN") do
          @app.call(env)
        end
      end

There's a discussion about activestorage and sharded databases too: #53602

[rosa]

Ohhhh, great point @robacarp. Thanks a lot for sharing that. Going to read that carefully and the forum thread linked there. It seems this area has still some open questions 🤔