diff --git a/activejob/lib/active_job/base.rb b/activejob/lib/active_job/base.rb
index ff5c69ddc6f..94a1faba3f3 100644
--- a/activejob/lib/active_job/base.rb
+++ b/activejob/lib/active_job/base.rb
@@ -5,6 +5,7 @@ require 'active_job/queue_priority'
 require 'active_job/enqueuing'
 require 'active_job/execution'
 require 'active_job/callbacks'
+require 'active_job/exceptions'
 require 'active_job/logging'
 require 'active_job/translation'
 
@@ -62,6 +63,7 @@ module ActiveJob #:nodoc:
     include Enqueuing
     include Execution
     include Callbacks
+    include Exceptions
     include Logging
     include Translation
 
diff --git a/activejob/lib/active_job/core.rb b/activejob/lib/active_job/core.rb
index f0606b3834d..a9ddd029a83 100644
--- a/activejob/lib/active_job/core.rb
+++ b/activejob/lib/active_job/core.rb
@@ -24,6 +24,9 @@ module ActiveJob
       # ID optionally provided by adapter
       attr_accessor :provider_job_id
 
+      # Number of times this job has been executed (which increments on every retry, like after an exception).
+      attr_accessor :executions
+
       # I18n.locale to be used during the job.
       attr_accessor :locale
     end
@@ -68,6 +71,7 @@ module ActiveJob
       @job_id     = SecureRandom.uuid
       @queue_name = self.class.queue_name
       @priority   = self.class.priority
+      @executions = 0
     end
 
     # Returns a hash with the job data that can safely be passed to the
@@ -79,6 +83,7 @@ module ActiveJob
         'queue_name' => queue_name,
         'priority'   => priority,
         'arguments'  => serialize_arguments(arguments),
+        'executions' => executions + 1,
         'locale'     => I18n.locale.to_s
       }
     end
@@ -109,6 +114,7 @@ module ActiveJob
       self.queue_name           = job_data['queue_name']
       self.priority             = job_data['priority']
       self.serialized_arguments = job_data['arguments']
+      self.executions           = job_data['executions']
       self.locale               = job_data['locale'] || I18n.locale.to_s
     end
 
diff --git a/activejob/lib/active_job/enqueuing.rb b/activejob/lib/active_job/enqueuing.rb
index 9dc3c0fa576..7bc0fe6c3a0 100644
--- a/activejob/lib/active_job/enqueuing.rb
+++ b/activejob/lib/active_job/enqueuing.rb
@@ -1,7 +1,7 @@
 require 'active_job/arguments'
 
 module ActiveJob
-  # Provides behavior for enqueuing and retrying jobs.
+  # Provides behavior for enqueuing jobs.
   module Enqueuing
     extend ActiveSupport::Concern
 
@@ -24,31 +24,6 @@ module ActiveJob
         end
     end
 
-    # Reschedules the job to be re-executed. This is useful in combination
-    # with the +rescue_from+ option. When you rescue an exception from your job
-    # you can ask Active Job to retry performing your job.
-    #
-    # ==== Options
-    # * <tt>:wait</tt> - Enqueues the job with the specified delay
-    # * <tt>:wait_until</tt> - Enqueues the job at the time specified
-    # * <tt>:queue</tt> - Enqueues the job on the specified queue
-    # * <tt>:priority</tt> - Enqueues the job with the specified priority
-    #
-    # ==== Examples
-    #
-    #  class SiteScraperJob < ActiveJob::Base
-    #    rescue_from(ErrorLoadingSite) do
-    #      retry_job queue: :low_priority
-    #    end
-    #
-    #    def perform(*args)
-    #      # raise ErrorLoadingSite if cannot scrape
-    #    end
-    #  end
-    def retry_job(options={})
-      enqueue options
-    end
-
     # Enqueues the job to be performed by the queue adapter.
     #
     # ==== Options
diff --git a/activejob/lib/active_job/exceptions.rb b/activejob/lib/active_job/exceptions.rb
new file mode 100644
index 00000000000..4217b5dc652
--- /dev/null
+++ b/activejob/lib/active_job/exceptions.rb
@@ -0,0 +1,78 @@
+require 'active_job/arguments'
+
+module ActiveJob
+  # Provides behavior for retrying and discarding jobs on exceptions.
+  module Exceptions
+    extend ActiveSupport::Concern
+
+    module ClassMethods
+      # Catch the exception and reschedule job for re-execution after so many seconds, for a specific number of attempts.
+      # 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.
+      #
+      # ==== Options
+      # * <tt>:wait</tt> - Re-enqueues the job with the specified delay in seconds
+      # * <tt>:attempts</tt> - Re-enqueues the job the specified number of times
+      #
+      # ==== Examples
+      #
+      #  class RemoteServiceJob < ActiveJob::Base
+      #    retry_on Net::OpenTimeout, wait: 30.seconds, attempts: 10
+      #
+      #    def perform(*args)
+      #      # Might raise Net::OpenTimeout when the remote service is down
+      #    end
+      #  end
+      def retry_on(exception, wait: 3.seconds, attempts: 5)
+        rescue_from exception do |error|
+          logger.error "Retrying #{self.class} in #{wait} seconds, due to a #{exception}. The original exception was #{error.cause.inspect}."
+          retry_job wait: wait if executions < attempts
+        end
+      end
+
+      # Discard the job with no attempts to retry, if the exception is raised. This is useful when the subject of the job,
+      # like an Active Record, is no longer available, and the job is thus no longer relevant.
+      #
+      # ==== Example
+      #
+      #  class SearchIndexingJob < ActiveJob::Base
+      #    discard_on ActiveJob::DeserializationError
+      #
+      #    def perform(record)
+      #      # Will raise ActiveJob::DeserializationError if the record can't be deserialized
+      #    end
+      #  end
+      def discard_on(exception)
+        rescue_from exception do |error|
+          logger.error "Discarded #{self.class} due to a #{exception}. The original exception was #{error.cause.inspect}."
+        end
+      end
+    end
+
+    # Reschedules the job to be re-executed. This is useful in combination
+    # with the +rescue_from+ option. When you rescue an exception from your job
+    # you can ask Active Job to retry performing your job.
+    #
+    # ==== Options
+    # * <tt>:wait</tt> - Enqueues the job with the specified delay in seconds
+    # * <tt>:wait_until</tt> - Enqueues the job at the time specified
+    # * <tt>:queue</tt> - Enqueues the job on the specified queue
+    # * <tt>:priority</tt> - Enqueues the job with the specified priority
+    #
+    # ==== Examples
+    #
+    #  class SiteScraperJob < ActiveJob::Base
+    #    rescue_from(ErrorLoadingSite) do
+    #      retry_job queue: :low_priority
+    #    end
+    #
+    #    def perform(*args)
+    #      # raise ErrorLoadingSite if cannot scrape
+    #    end
+    #  end
+    def retry_job(options = {})
+      enqueue options
+    end
+  end
+end
diff --git a/activejob/test/cases/exceptions_test.rb b/activejob/test/cases/exceptions_test.rb
new file mode 100644
index 00000000000..42d4dd78a59
--- /dev/null
+++ b/activejob/test/cases/exceptions_test.rb
@@ -0,0 +1,47 @@
+require 'helper'
+require 'jobs/retry_job'
+
+class ExceptionsTest < ActiveSupport::TestCase
+  setup do
+    JobBuffer.clear
+  end
+
+  test "successfully retry job throwing exception against defaults" do
+    RetryJob.perform_later 'SeriousError', 5
+    
+    assert_equal [ 
+      "Raised SeriousError for the 1st time", 
+      "Raised SeriousError for the 2nd time",
+      "Raised SeriousError for the 3rd time",
+      "Raised SeriousError for the 4th time",
+      "Successfully completed job" ], JobBuffer.values
+  end
+
+  test "successfully retry job throwing exception against higher limit" do
+    RetryJob.perform_later 'VerySeriousError', 9
+    assert_equal 9, JobBuffer.values.count
+  end
+  
+  test "failed retry job when exception kept occurring against defaults" do
+    begin
+      RetryJob.perform_later 'SeriousError', 6
+      assert_equal "Raised SeriousError for the 5th time", JobBuffer.last_value
+    rescue SeriousError
+      pass
+    end
+  end
+
+  test "failed retry job when exception kept occurring against higher limit" do
+    begin
+      RetryJob.perform_later 'VerySeriousError', 11
+      assert_equal "Raised VerySeriousError for the 10th time", JobBuffer.last_value
+    rescue SeriousError
+      pass
+    end
+  end
+
+  test "discard job" do
+    RetryJob.perform_later 'NotSeriousError', 2
+    assert_equal "Raised NotSeriousError for the 1st time", JobBuffer.last_value
+  end
+end
diff --git a/activejob/test/jobs/retry_job.rb b/activejob/test/jobs/retry_job.rb
new file mode 100644
index 00000000000..2e05f95bf6a
--- /dev/null
+++ b/activejob/test/jobs/retry_job.rb
@@ -0,0 +1,21 @@
+require_relative '../support/job_buffer'
+require 'active_support/core_ext/integer/inflections'
+
+class SeriousError < StandardError; end
+class VerySeriousError < StandardError; end
+class NotSeriousError < StandardError; end
+
+class RetryJob < ActiveJob::Base
+  retry_on SeriousError
+  retry_on VerySeriousError, wait: 1.second, attempts: 10
+  discard_on NotSeriousError
+
+  def perform(raising, attempts)
+    if executions < attempts
+      JobBuffer.add("Raised #{raising} for the #{executions.ordinalize} time")
+      raise raising.constantize
+    else
+      JobBuffer.add("Successfully completed job")
+    end
+  end
+end