diff options
Diffstat (limited to 'activejob/lib')
| -rw-r--r-- | activejob/lib/active_job.rb | 2 | ||||
| -rw-r--r-- | activejob/lib/active_job/arguments.rb | 35 | ||||
| -rw-r--r-- | activejob/lib/active_job/base.rb | 2 | ||||
| -rw-r--r-- | activejob/lib/active_job/callbacks.rb | 5 | ||||
| -rw-r--r-- | activejob/lib/active_job/core.rb | 11 | ||||
| -rw-r--r-- | activejob/lib/active_job/enqueuing.rb | 31 | ||||
| -rw-r--r-- | activejob/lib/active_job/exceptions.rb | 21 | ||||
| -rw-r--r-- | activejob/lib/active_job/execution.rb | 2 | ||||
| -rw-r--r-- | activejob/lib/active_job/logging.rb | 12 | ||||
| -rw-r--r-- | activejob/lib/active_job/queue_adapters.rb | 6 | ||||
| -rw-r--r-- | activejob/lib/active_job/queue_adapters/async_adapter.rb | 2 | ||||
| -rw-r--r-- | activejob/lib/active_job/queue_adapters/test_adapter.rb | 10 | ||||
| -rw-r--r-- | activejob/lib/active_job/queue_name.rb | 2 | ||||
| -rw-r--r-- | activejob/lib/active_job/test_helper.rb | 47 |
14 files changed, 139 insertions, 49 deletions
diff --git a/activejob/lib/active_job.rb b/activejob/lib/active_job.rb index 01fab4d918..5f20ef9b9f 100644 --- a/activejob/lib/active_job.rb +++ b/activejob/lib/active_job.rb @@ -1,7 +1,7 @@ # frozen_string_literal: true #-- -# Copyright (c) 2014-2018 David Heinemeier Hansson +# Copyright (c) 2014-2019 David Heinemeier Hansson # # Permission is hereby granted, free of charge, to any person obtaining # a copy of this software and associated documentation files (the diff --git a/activejob/lib/active_job/arguments.rb b/activejob/lib/active_job/arguments.rb index 8dcf588f35..92eb58aaaf 100644 --- a/activejob/lib/active_job/arguments.rb +++ b/activejob/lib/active_job/arguments.rb @@ -14,18 +14,18 @@ module ActiveJob end # Raised when an unsupported argument type is set as a job argument. We - # currently support NilClass, Integer, Float, String, TrueClass, FalseClass, - # BigDecimal, and objects that can be represented as GlobalIDs (ex: Active Record). + # currently support String, Integer, Float, NilClass, TrueClass, FalseClass, + # BigDecimal, Symbol, Date, Time, DateTime, ActiveSupport::TimeWithZone, + # ActiveSupport::Duration, Hash, ActiveSupport::HashWithIndifferentAccess, + # Array or GlobalID::Identification instances, although this can be extended + # by adding custom serializers. # Raised if you set the key for a Hash something else than a string or # a symbol. Also raised when trying to serialize an object which can't be - # identified with a Global ID - such as an unpersisted Active Record model. + # identified with a GlobalID - such as an unpersisted Active Record model. class SerializationError < ArgumentError; end module Arguments extend self - # :nodoc: - PERMITTED_TYPES = [ NilClass, String, Integer, Float, BigDecimal, TrueClass, FalseClass ] - # Serializes a set of arguments. Intrinsic types that can safely be # serialized without mutation are returned as-is. Arrays/Hashes are # serialized element by element. All other types are serialized using @@ -47,6 +47,8 @@ module ActiveJob private # :nodoc: + PERMITTED_TYPES = [ NilClass, String, Integer, Float, BigDecimal, TrueClass, FalseClass ] + # :nodoc: GLOBALID_KEY = "_aj_globalid" # :nodoc: SYMBOL_KEYS_KEY = "_aj_symbol_keys" @@ -62,7 +64,7 @@ module ActiveJob OBJECT_SERIALIZER_KEY, OBJECT_SERIALIZER_KEY.to_sym, WITH_INDIFFERENT_ACCESS_KEY, WITH_INDIFFERENT_ACCESS_KEY.to_sym, ] - private_constant :RESERVED_KEYS + private_constant :PERMITTED_TYPES, :RESERVED_KEYS, :GLOBALID_KEY, :SYMBOL_KEYS_KEY, :WITH_INDIFFERENT_ACCESS_KEY def serialize_argument(argument) case argument @@ -73,14 +75,14 @@ module ActiveJob when Array argument.map { |arg| serialize_argument(arg) } when ActiveSupport::HashWithIndifferentAccess - result = serialize_hash(argument) - result[WITH_INDIFFERENT_ACCESS_KEY] = serialize_argument(true) - result + serialize_indifferent_hash(argument) when Hash symbol_keys = argument.each_key.grep(Symbol).map(&:to_s) result = serialize_hash(argument) result[SYMBOL_KEYS_KEY] = symbol_keys result + when -> (arg) { arg.respond_to?(:permitted?) } + serialize_indifferent_hash(argument.to_h) else Serializers.serialize(argument) end @@ -89,7 +91,7 @@ module ActiveJob def deserialize_argument(argument) case argument when String - GlobalID::Locator.locate(argument) || argument + argument when *PERMITTED_TYPES argument when Array @@ -146,8 +148,17 @@ module ActiveJob end end + def serialize_indifferent_hash(indifferent_hash) + result = serialize_hash(indifferent_hash) + result[WITH_INDIFFERENT_ACCESS_KEY] = serialize_argument(true) + result + end + def transform_symbol_keys(hash, symbol_keys) - hash.transform_keys do |key| + # NOTE: HashWithIndifferentAccess#transform_keys always + # returns stringified keys with indifferent access + # so we call #to_h here to ensure keys are symbolized. + hash.to_h.transform_keys do |key| if symbol_keys.include?(key) key.to_sym else diff --git a/activejob/lib/active_job/base.rb b/activejob/lib/active_job/base.rb index 95b1062701..ed41fac4b8 100644 --- a/activejob/lib/active_job/base.rb +++ b/activejob/lib/active_job/base.rb @@ -40,7 +40,7 @@ module ActiveJob #:nodoc: # Records that are passed in are serialized/deserialized using Global # ID. More information can be found in Arguments. # - # To enqueue a job to be performed as soon as the queueing system is free: + # To enqueue a job to be performed as soon as the queuing system is free: # # ProcessPhotoJob.perform_later(photo) # diff --git a/activejob/lib/active_job/callbacks.rb b/activejob/lib/active_job/callbacks.rb index 334b24fb3b..6b82ea6cab 100644 --- a/activejob/lib/active_job/callbacks.rb +++ b/activejob/lib/active_job/callbacks.rb @@ -29,6 +29,9 @@ module ActiveJob included do define_callbacks :perform define_callbacks :enqueue + + class_attribute :return_false_on_aborted_enqueue, instance_accessor: false, instance_predicate: false + self.return_false_on_aborted_enqueue = false end # These methods will be included into any Active Job object, adding @@ -130,7 +133,7 @@ module ActiveJob set_callback(:enqueue, :after, *filters, &blk) end - # Defines a callback that will get called around the enqueueing + # Defines a callback that will get called around the enqueuing # of the job. # # class VideoProcessJob < ActiveJob::Base diff --git a/activejob/lib/active_job/core.rb b/activejob/lib/active_job/core.rb index 62bb5861bb..4ab62f89b0 100644 --- a/activejob/lib/active_job/core.rb +++ b/activejob/lib/active_job/core.rb @@ -28,6 +28,12 @@ module ActiveJob # Number of times this job has been executed (which increments on every retry, like after an exception). attr_accessor :executions + # Hash that contains the number of times this job handled errors for each specific retry_on declaration. + # Keys are the string representation of the exceptions listed in the retry_on declaration, + # while its associated value holds the number of executions where the corresponding retry_on + # declaration handled one of its listed exceptions. + attr_accessor :exception_executions + # I18n.locale to be used during the job. attr_accessor :locale @@ -75,10 +81,11 @@ module ActiveJob @queue_name = self.class.queue_name @priority = self.class.priority @executions = 0 + @exception_executions = Hash.new(0) end # Returns a hash with the job data that can safely be passed to the - # queueing adapter. + # queuing adapter. def serialize { "job_class" => self.class.name, @@ -88,6 +95,7 @@ module ActiveJob "priority" => priority, "arguments" => serialize_arguments_if_needed(arguments), "executions" => executions, + "exception_executions" => exception_executions, "locale" => I18n.locale.to_s, "timezone" => Time.zone.try(:name) } @@ -126,6 +134,7 @@ module ActiveJob self.priority = job_data["priority"] self.serialized_arguments = job_data["arguments"] self.executions = job_data["executions"] + self.exception_executions = job_data["exception_executions"] self.locale = job_data["locale"] || I18n.locale.to_s self.timezone = job_data["timezone"] || Time.zone.try(:name) end diff --git a/activejob/lib/active_job/enqueuing.rb b/activejob/lib/active_job/enqueuing.rb index 53cb98fc71..ce118c1e8a 100644 --- a/activejob/lib/active_job/enqueuing.rb +++ b/activejob/lib/active_job/enqueuing.rb @@ -9,10 +9,12 @@ module ActiveJob # Includes the +perform_later+ method for job initialization. module ClassMethods - # Push a job onto the queue. The arguments must be legal JSON types - # (+string+, +int+, +float+, +nil+, +true+, +false+, +hash+ or +array+) or - # GlobalID::Identification instances. Arbitrary Ruby objects - # are not supported. + # Push a job onto the queue. By default the arguments must be either String, + # Integer, Float, NilClass, TrueClass, FalseClass, BigDecimal, Symbol, Date, + # Time, DateTime, ActiveSupport::TimeWithZone, ActiveSupport::Duration, + # Hash, ActiveSupport::HashWithIndifferentAccess, Array or + # GlobalID::Identification instances, although this can be extended by adding + # custom serializers. # # Returns an instance of the job class queued with arguments available in # Job#arguments. @@ -46,14 +48,33 @@ module ActiveJob self.scheduled_at = options[:wait_until].to_f if options[:wait_until] self.queue_name = self.class.queue_name_from_part(options[:queue]) if options[:queue] self.priority = options[:priority].to_i if options[:priority] + successfully_enqueued = false + run_callbacks :enqueue do if scheduled_at self.class.queue_adapter.enqueue_at self, scheduled_at else self.class.queue_adapter.enqueue self end + + successfully_enqueued = true + end + + if successfully_enqueued + self + else + if self.class.return_false_on_aborted_enqueue + false + else + ActiveSupport::Deprecation.warn( + "Rails 6.0 will return false when the enqueing is aborted. Make sure your code doesn't depend on it" \ + " returning the instance of the job and set `config.active_job.return_false_on_aborted_enqueue = true`" \ + " to remove the deprecations." + ) + + self + end end - self end end end diff --git a/activejob/lib/active_job/exceptions.rb b/activejob/lib/active_job/exceptions.rb index bc9e168971..48b35c8d05 100644 --- a/activejob/lib/active_job/exceptions.rb +++ b/activejob/lib/active_job/exceptions.rb @@ -9,7 +9,6 @@ module ActiveJob module ClassMethods # Catch the exception and reschedule job for re-execution after so many seconds, for a specific number of attempts. - # The number of attempts includes the total executions of a job, not just the retried executions. # If the exception keeps getting raised beyond the specified number of attempts, the exception is allowed to # bubble up to the underlying queuing system, which may have its own retry mechanism or place it in a # holding queue for inspection. @@ -22,8 +21,7 @@ module ActiveJob # as a computing proc that the number of executions so far as an argument, or as a symbol reference of # <tt>:exponentially_longer</tt>, which applies the wait algorithm of <tt>(executions ** 4) + 2</tt> # (first wait 3s, then 18s, then 83s, etc) - # * <tt>:attempts</tt> - Re-enqueues the job the specified number of times (default: 5 attempts), - # attempts here refers to the total number of times the job is executed, not just retried executions + # * <tt>:attempts</tt> - Re-enqueues the job the specified number of times (default: 5 attempts) # * <tt>:queue</tt> - Re-enqueues the job on a different queue # * <tt>:priority</tt> - Re-enqueues the job with a different priority # @@ -32,21 +30,30 @@ module ActiveJob # class RemoteServiceJob < ActiveJob::Base # retry_on CustomAppException # defaults to 3s wait, 5 attempts # retry_on AnotherCustomAppException, wait: ->(executions) { executions * 2 } + # + # retry_on ActiveRecord::Deadlocked, wait: 5.seconds, attempts: 3 + # retry_on Net::OpenTimeout, Timeout::Error, wait: :exponentially_longer, attempts: 10 # retries at most 10 times for Net::OpenTimeout and Timeout::Error combined + # # To retry at most 10 times for each individual exception: + # # retry_on Net::OpenTimeout, wait: :exponentially_longer, attempts: 10 + # # retry_on Timeout::Error, wait: :exponentially_longer, attempts: 10 + # # retry_on(YetAnotherCustomAppException) do |job, error| # ExceptionNotifier.caught(error) # end - # retry_on ActiveRecord::Deadlocked, wait: 5.seconds, attempts: 3 - # retry_on Net::OpenTimeout, wait: :exponentially_longer, attempts: 10 # # def perform(*args) # # Might raise CustomAppException, AnotherCustomAppException, or YetAnotherCustomAppException for something domain specific # # Might raise ActiveRecord::Deadlocked when a local db deadlock is detected - # # Might raise Net::OpenTimeout when the remote service is down + # # Might raise Net::OpenTimeout or Timeout::Error when the remote service is down # end # end def retry_on(*exceptions, wait: 3.seconds, attempts: 5, queue: nil, priority: nil) rescue_from(*exceptions) do |error| - if executions < attempts + # Guard against jobs that were persisted before we started having individual executions counters per retry_on + self.exception_executions ||= Hash.new(0) + self.exception_executions[exceptions.to_s] += 1 + + if exception_executions[exceptions.to_s] < attempts retry_job wait: determine_delay(wait), queue: queue, priority: priority, error: error else if block_given? diff --git a/activejob/lib/active_job/execution.rb b/activejob/lib/active_job/execution.rb index f5a343311f..e96dbcd4c9 100644 --- a/activejob/lib/active_job/execution.rb +++ b/activejob/lib/active_job/execution.rb @@ -26,7 +26,7 @@ module ActiveJob end end - # Performs the job immediately. The job is not sent to the queueing adapter + # Performs the job immediately. The job is not sent to the queuing adapter # but directly executed by blocking the execution of others until it's finished. # # MyJob.new(*args).perform_now diff --git a/activejob/lib/active_job/logging.rb b/activejob/lib/active_job/logging.rb index 0abee4ed98..416be83c24 100644 --- a/activejob/lib/active_job/logging.rb +++ b/activejob/lib/active_job/logging.rb @@ -93,8 +93,12 @@ module ActiveJob ex = event.payload[:error] wait = event.payload[:wait] - error do - "Retrying #{job.class} in #{wait.inspect} seconds, due to a #{ex&.class.inspect}. The original exception was #{ex&.cause.inspect}." + info do + if ex + "Retrying #{job.class} in #{wait.to_i} seconds, due to a #{ex.class}." + else + "Retrying #{job.class} in #{wait.to_i} seconds." + end end end @@ -103,7 +107,7 @@ module ActiveJob ex = event.payload[:error] error do - "Stopped retrying #{job.class} due to a #{ex.class}, which reoccurred on #{job.executions} attempts. The original exception was #{ex.cause.inspect}." + "Stopped retrying #{job.class} due to a #{ex.class}, which reoccurred on #{job.executions} attempts." end end @@ -112,7 +116,7 @@ module ActiveJob ex = event.payload[:error] error do - "Discarded #{job.class} due to a #{ex.class}. The original exception was #{ex.cause.inspect}." + "Discarded #{job.class} due to a #{ex.class}." end end diff --git a/activejob/lib/active_job/queue_adapters.rb b/activejob/lib/active_job/queue_adapters.rb index 3e3a474fbb..525e79e302 100644 --- a/activejob/lib/active_job/queue_adapters.rb +++ b/activejob/lib/active_job/queue_adapters.rb @@ -3,7 +3,7 @@ module ActiveJob # == Active Job adapters # - # Active Job has adapters for the following queueing backends: + # Active Job has adapters for the following queuing backends: # # * {Backburner}[https://github.com/nesquena/backburner] # * {Delayed Job}[https://github.com/collectiveidea/delayed_job] @@ -52,7 +52,7 @@ module ActiveJob # # No: The adapter will run jobs at the next opportunity and cannot use perform_later. # - # N/A: The adapter does not support queueing. + # N/A: The adapter does not support queuing. # # NOTE: # queue_classic supports job scheduling since version 3.1. @@ -74,7 +74,7 @@ module ActiveJob # # No: Does not allow the priority of jobs to be configured. # - # N/A: The adapter does not support queueing, and therefore sorting them. + # N/A: The adapter does not support queuing, and therefore sorting them. # # ==== Timeout # diff --git a/activejob/lib/active_job/queue_adapters/async_adapter.rb b/activejob/lib/active_job/queue_adapters/async_adapter.rb index ebf6f384e3..53a7e3d53e 100644 --- a/activejob/lib/active_job/queue_adapters/async_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/async_adapter.rb @@ -31,7 +31,7 @@ module ActiveJob # jobs. Since jobs share a single thread pool, long-running jobs will block # short-lived jobs. Fine for dev/test; bad for production. class AsyncAdapter - # See {Concurrent::ThreadPoolExecutor}[https://ruby-concurrency.github.io/concurrent-ruby/Concurrent/ThreadPoolExecutor.html] for executor options. + # See {Concurrent::ThreadPoolExecutor}[https://ruby-concurrency.github.io/concurrent-ruby/master/Concurrent/ThreadPoolExecutor.html] for executor options. def initialize(**executor_options) @scheduler = Scheduler.new(**executor_options) end diff --git a/activejob/lib/active_job/queue_adapters/test_adapter.rb b/activejob/lib/active_job/queue_adapters/test_adapter.rb index f73ad444ba..c134257ebc 100644 --- a/activejob/lib/active_job/queue_adapters/test_adapter.rb +++ b/activejob/lib/active_job/queue_adapters/test_adapter.rb @@ -65,11 +65,17 @@ module ActiveJob def filtered_job_class?(job) if filter - !Array(filter).include?(job.class) + !filter_as_proc(filter).call(job) elsif reject - Array(reject).include?(job.class) + filter_as_proc(reject).call(job) end end + + def filter_as_proc(filter) + return filter if filter.is_a?(Proc) + + ->(job) { Array(filter).include?(job.class) } + end end end end diff --git a/activejob/lib/active_job/queue_name.rb b/activejob/lib/active_job/queue_name.rb index 9dc6bc7f2e..7bb1e35181 100644 --- a/activejob/lib/active_job/queue_name.rb +++ b/activejob/lib/active_job/queue_name.rb @@ -34,7 +34,7 @@ module ActiveJob end included do - class_attribute :queue_name, instance_accessor: false, default: default_queue_name + class_attribute :queue_name, instance_accessor: false, default: -> { self.class.default_queue_name } class_attribute :queue_name_delimiter, instance_accessor: false, default: "_" end diff --git a/activejob/lib/active_job/test_helper.rb b/activejob/lib/active_job/test_helper.rb index 261b68d4f8..f03780b91e 100644 --- a/activejob/lib/active_job/test_helper.rb +++ b/activejob/lib/active_job/test_helper.rb @@ -1,7 +1,6 @@ # frozen_string_literal: true require "active_support/core_ext/class/subclasses" -require "active_support/core_ext/hash/keys" module ActiveJob # Provides helper methods for testing Active Job @@ -107,6 +106,9 @@ module ActiveJob # end # end # + # +:only+ and +:except+ options accepts Class, Array of Class or Proc. When passed a Proc, + # a hash containing the job's class and it's argument are passed as argument. + # # Asserts the number of times a job is enqueued to a specific queue by passing +:queue+ option. # # def test_logging_job @@ -163,6 +165,9 @@ module ActiveJob # end # end # + # +:only+ and +:except+ options accepts Class, Array of Class or Proc. When passed a Proc, + # a hash containing the job's class and it's argument are passed as argument. + # # Asserts that no jobs are enqueued to a specific queue by passing +:queue+ option # # def test_no_logging @@ -197,7 +202,7 @@ module ActiveJob # assert_performed_jobs 2 # end # - # If a block is passed, that block should cause the specified number of + # If a block is passed, asserts that the block will cause the specified number of # jobs to be performed. # # def test_jobs_again @@ -243,6 +248,18 @@ module ActiveJob # end # end # + # A proc may also be specified. When passed a Proc, the job's instance will be passed as argument. + # + # def test_hello_and_logging_jobs + # assert_nothing_raised do + # assert_performed_jobs(1, only: ->(job) { job.is_a?(HelloJob) }) do + # HelloJob.perform_later('jeremy') + # LoggingJob.perform_later('stewie') + # RescueJob.perform_later('david') + # end + # end + # end + # # If the +:queue+ option is specified, # then only the job(s) enqueued to a specific queue will be performed. # @@ -279,7 +296,7 @@ module ActiveJob # end # end # - # If a block is passed, that block should not cause any job to be performed. + # If a block is passed, asserts that the block will not cause any job to be performed. # # def test_jobs_again # assert_no_performed_jobs do @@ -305,6 +322,9 @@ module ActiveJob # end # end # + # +:only+ and +:except+ options accepts Class, Array of Class or Proc. When passed a Proc, + # an instance of the job will be passed as argument. + # # If the +:queue+ option is specified, # then only the job(s) enqueued to a specific queue will not be performed. # @@ -347,7 +367,7 @@ module ActiveJob # end # # - # If a block is passed, that block should cause the job to be + # If a block is passed, asserts that the block will cause the job to be # enqueued with the given arguments. # # def test_assert_enqueued_with @@ -505,6 +525,9 @@ module ActiveJob # assert_performed_jobs 1 # end # + # +:only+ and +:except+ options accepts Class, Array of Class or Proc. When passed a Proc, + # an instance of the job will be passed as argument. + # # If the +:queue+ option is specified, # then only the job(s) enqueued to a specific queue will be performed. # @@ -569,9 +592,9 @@ module ActiveJob job_class = job.fetch(:job) if only - next false unless Array(only).include?(job_class) + next false unless filter_as_proc(only).call(job) elsif except - next false if Array(except).include?(job_class) + next false if filter_as_proc(except).call(job) end if queue @@ -584,6 +607,12 @@ module ActiveJob end end + def filter_as_proc(filter) + return filter if filter.is_a?(Proc) + + ->(job) { Array(filter).include?(job.fetch(:job)) } + end + def enqueued_jobs_with(only: nil, except: nil, queue: nil, &block) jobs_with(enqueued_jobs, only: only, except: except, queue: queue, &block) end @@ -594,8 +623,7 @@ module ActiveJob def flush_enqueued_jobs(only: nil, except: nil, queue: nil) enqueued_jobs_with(only: only, except: except, queue: queue) do |payload| - args = ActiveJob::Arguments.deserialize(payload[:args]) - instantiate_job(payload.merge(args: args)).perform_now + instantiate_job(payload).perform_now queue_adapter.performed_jobs << payload end end @@ -613,7 +641,8 @@ module ActiveJob end def instantiate_job(payload) - job = payload[:job].new(*payload[:args]) + args = ActiveJob::Arguments.deserialize(payload[:args]) + job = payload[:job].new(*args) job.scheduled_at = Time.at(payload[:at]) if payload.key?(:at) job.queue_name = payload[:queue] job |