Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -669,6 +669,8 @@ end

When a process dies without a clean shutdown (for example, `SIGKILL`ed by the OS or the container runtime because of memory limits), the jobs it was running can't be released back to their queues. Once another process notices the missing heartbeats and prunes the dead process's registration, its in-flight jobs are marked as failed with `SolidQueue::Processes::ProcessPrunedError`. Solid Queue deliberately doesn't retry these automatically: the job itself might be what's killing the process (for example, a job that exhausts the container's memory), and retrying it blindly would just kill the next worker too.

Note that Active Job's `retry_on` and `rescue_from` have no effect on these errors: they only intercept exceptions raised while your job's `perform` method runs, and `ProcessPrunedError` (as well as `ProcessExitError` and `ProcessMissingError`) is never raised inside the job. The process that was running the job is gone by then — the error is recorded directly as a failed execution by a *different* process, after the fact, so there's no job execution left for Active Job's retry machinery to hook into. To retry these jobs, act on the failed executions from the outside instead: manually, via [Mission Control — Jobs](https://github.com/rails/mission_control-jobs), or automatically, with a subscription like the one below.

If you know your jobs are idempotent and want to implement your own recovery policy, you can subscribe to the `fail_many_claimed.solid_queue` event, which includes the error and the affected job IDs in its payload:

```ruby
Expand Down
6 changes: 5 additions & 1 deletion lib/solid_queue/cli.rb
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,11 @@ def start
SolidQueue::Supervisor.start(**options.symbolize_keys)
end

desc :check, "Validates the Solid Queue configuration for the current Rails env without starting anything. Exits non-zero on errors."
desc :check, "Validates the Solid Queue configuration without starting anything. Exits non-zero on errors."
method_option :env, type: :string,
desc: "Environment section of the configuration files to validate (default: the current Rails env). " \
"Lets a CI run in one env validate the section that will be deployed to another, e.g. --env production.",
banner: "ENVIRONMENT"
def check
configuration = SolidQueue::Configuration.new(**options.symbolize_keys)
exit 1 unless configuration.check
Expand Down
31 changes: 25 additions & 6 deletions lib/solid_queue/configuration.rb
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,12 @@ def ensure_valid_recurring_tasks

errors.add(:base, "Invalid recurring tasks:\n#{error_messages.join("\n")}")
end
rescue ActiveRecord::ActiveRecordError
# Validating recurring tasks needs the solid_queue_recurring_tasks schema, and this
# environment has no usable database connection to load it from. Don't fail the whole
# check over that, but don't stay quiet about it either.
warnings.add(:base, "Warning: recurring tasks couldn't be validated because there's no usable " \
"database connection. Run `bin/jobs check` with the Solid Queue database reachable to validate them.")
end

def warn_about_incorrectly_sized_database_pool
Expand Down Expand Up @@ -156,6 +162,7 @@ def default_options
{
mode: ENV["SOLID_QUEUE_SUPERVISOR_MODE"] || :fork,
standalone: true,
env: Rails.env,
config_file: Rails.root.join(ENV["SOLID_QUEUE_CONFIG"] || DEFAULT_CONFIG_FILE_PATH),
recurring_schedule_file: Rails.root.join(ENV["SOLID_QUEUE_RECURRING_SCHEDULE"] || DEFAULT_RECURRING_SCHEDULE_FILE_PATH),
only_work: false,
Expand Down Expand Up @@ -207,8 +214,8 @@ def dispatchers
def schedulers
return [] if skip_recurring_tasks?

if recurring_tasks.any? || dynamic_recurring_tasks_enabled?
[ Process.new(:scheduler, { recurring_tasks: recurring_tasks, **scheduler_options.with_defaults(SCHEDULER_DEFAULTS) }) ]
if recurring_task_definitions.any? || dynamic_recurring_tasks_enabled?
[ Process.new(:scheduler, { recurring_tasks: recurring_task_definitions, **scheduler_options.with_defaults(SCHEDULER_DEFAULTS) }) ]
else
[]
end
Expand All @@ -232,16 +239,24 @@ def dynamic_recurring_tasks_enabled?
scheduler_options.fetch(:dynamic_tasks_enabled, SCHEDULER_DEFAULTS[:dynamic_tasks_enabled])
end

# Raw [ key, options ] pairs, with no database dependency: instantiating RecurringTask
# requires loading its schema, which isn't possible in an environment without a usable
# database connection. The scheduler wraps these in RecurringTask objects when it boots.
def recurring_task_definitions
@recurring_task_definitions ||= recurring_tasks_config.filter_map do |id, options|
[ id, options.merge(static: true) ] if options&.has_key?(:schedule)
end
end

def recurring_tasks
@recurring_tasks ||= recurring_tasks_config.map do |id, options|
RecurringTask.from_configuration(id, **options.merge(static: true)) if options&.has_key?(:schedule)
end.compact
@recurring_tasks ||= recurring_task_definitions.map { |definition| RecurringTask.wrap(definition) }
end

def processes_config
@processes_config ||= config_from \
options.slice(:workers, :dispatchers, :scheduler).presence || options[:config_file],
keys: [ :workers, :dispatchers, :scheduler ],
env: env,
fallback: {
workers: [ WORKER_DEFAULTS ],
dispatchers: [ DISPATCHER_DEFAULTS ],
Expand All @@ -251,10 +266,14 @@ def processes_config

def recurring_tasks_config
@recurring_tasks_config ||= begin
config_from options[:recurring_schedule_file]
config_from options[:recurring_schedule_file], env: env
end
end

def env
options[:env].presence || Rails.env
end

def config_from(file_or_hash, keys: [], fallback: {}, env: Rails.env)
load_config_from(file_or_hash).then do |config|
config = config[env.to_sym] ? config[env.to_sym] : config
Expand Down
25 changes: 25 additions & 0 deletions test/unit/configuration_test.rb
Original file line number Diff line number Diff line change
Expand Up @@ -283,6 +283,31 @@ class ConfigurationTest < ActiveSupport::TestCase
assert configuration.valid?
end

test "validate another environment's section of the config files with the env option" do
# Without env targeting, the production-only section is invisible from the test env
configuration = SolidQueue::Configuration.new(recurring_schedule_file: config_file_path(:recurring_with_production_only))
assert_processes configuration, :scheduler, 0

configuration = SolidQueue::Configuration.new(recurring_schedule_file: config_file_path(:recurring_with_production_only), env: "production")
assert configuration.valid?
assert_processes configuration, :scheduler, 1

scheduler = configuration.configured_processes.detect { |process| process.kind == :scheduler }.instantiate
assert_has_recurring_task scheduler, key: "periodic_store_result", class_name: "StoreResultJob", schedule: "every second"
end

test "check warns instead of erroring when recurring tasks can't be validated without a database" do
configuration = SolidQueue::Configuration.new(recurring_schedule_file: config_file_path(:recurring_with_invalid))
SolidQueue::RecurringTask.stubs(:from_configuration)
.raises(ActiveRecord::StatementInvalid.new("Could not find table 'solid_queue_recurring_tasks'"))

# The scheduler is still detected from the raw configuration, with no database involved
assert_processes configuration, :scheduler, 1

assert configuration.valid?
assert_match /recurring tasks couldn't be validated/, configuration.warnings.full_messages.join
end

test "reports an undersized database pool as a warning rather than an error" do
configuration = SolidQueue::Configuration.new(workers: [ { queues: "background", threads: 50, polling_interval: 10 } ], skip_recurring: true)

Expand Down
Loading