diff --git a/lib/dry/operation.rb b/lib/dry/operation.rb
index 08c760a..68b1b62 100644
--- a/lib/dry/operation.rb
+++ b/lib/dry/operation.rb
@@ -53,11 +53,11 @@ module Dry
   #
   # Under the hood, the `#call` method is decorated to allow skipping the rest
   # of its execution when a failure is encountered. You can choose to use another
-  # method with {ClassContext#operate_on}:
+  # method with {ClassContext#operate_on} (which also accepts a list of methods):
   #
   # ```ruby
   # class MyOperation < Dry::Operation
-  #   operate_on :run
+  #   operate_on :run # or operate_on :run, :call
   #
   #   def run(input)
   #     attrs = step validate(input)
@@ -70,8 +70,31 @@ module Dry
   # end
   # ```
   #
+  # As you can see, the aforementioned behavior allows you to write your flow
+  # in a linear fashion. Failures are mostly handled locally by each individual
+  # operation. However, you can also define a global failure handler by defining
+  # an `#on_failure` method. It will be called with the wrapped failure value
+  # and, in the case of accepting a second argument, the name of the method that
+  # defined the flow:
+  #
+  # ```ruby
+  # class MyOperation < Dry::Operation
+  #   def call(input)
+  #     attrs = step validate(input)
+  #     user = step persist(attrs)
+  #     step notify(user)
+  #     user
+  #   end
+  #
+  #   def on_failure(user) # or def on_failure(failure_value, method_name)
+  #     log_failure(user)
+  #   end
+  # end
+  # ```
+  #
   # You can opt out altogether of this behavior via {ClassContext#skip_prepending}. If so,
-  # you manually need to wrap your flow within the {#steps} method.
+  # you manually need to wrap your flow within the {#steps} method and manually
+  # handle global failures.
   #
   # ```ruby
   # class MyOperation < Dry::Operation
@@ -83,6 +106,8 @@ module Dry
   #       user = step persist(attrs)
   #       step notify(user)
   #       user
+  #     end.tap do |result|
+  #       log_failure(result.failure) if result.failure?
   #     end
   #   end
   #
diff --git a/lib/dry/operation/class_context/steps_method_prepender.rb b/lib/dry/operation/class_context/steps_method_prepender.rb
index e1a626f..5e71472 100644
--- a/lib/dry/operation/class_context/steps_method_prepender.rb
+++ b/lib/dry/operation/class_context/steps_method_prepender.rb
@@ -1,13 +1,35 @@
 # frozen_string_literal: true
 
+require "dry/operation/errors"
+
 module Dry
   class Operation
     module ClassContext
       # @api private
       class StepsMethodPrepender < Module
-        def initialize(method:)
+        FAILURE_HOOK_METHOD_NAME = :on_failure
+
+        RESULT_HANDLER = lambda do |instance, method, result|
+          return if result.success? ||
+                    !(instance.methods + instance.private_methods).include?(
+                      FAILURE_HOOK_METHOD_NAME
+                    )
+
+          failure_hook = instance.method(FAILURE_HOOK_METHOD_NAME)
+          case failure_hook.arity
+          when 1
+            failure_hook.(result.failure)
+          when 2
+            failure_hook.(result.failure, method)
+          else
+            raise FailureHookArityError.new(hook: failure_hook)
+          end
+        end
+
+        def initialize(method:, result_handler: RESULT_HANDLER)
           super()
           @method = method
+          @result_handler = result_handler
         end
 
         def included(klass)
@@ -18,9 +40,13 @@ def included(klass)
 
         def mod
           @module ||= Module.new.tap do |mod|
-            mod.define_method(@method) do |*args, **kwargs, &block|
-              steps do
-                super(*args, **kwargs, &block)
+            module_exec(@result_handler) do |result_handler|
+              mod.define_method(@method) do |*args, **kwargs, &block|
+                steps do
+                  super(*args, **kwargs, &block)
+                end.tap do |result|
+                  result_handler.(self, __method__, result)
+                end
               end
             end
           end
diff --git a/lib/dry/operation/errors.rb b/lib/dry/operation/errors.rb
index 705b736..5bd30e3 100644
--- a/lib/dry/operation/errors.rb
+++ b/lib/dry/operation/errors.rb
@@ -35,5 +35,15 @@ def initialize(gem:, extension:)
 
     # An error related to an extension
     class ExtensionError < ::StandardError; end
+
+    # Defined failure hook has wrong arity
+    class FailureHookArityError < ::StandardError
+      def initialize(hook:)
+        super <<~MSG
+          ##{hook.name} must accept 1 (failure) or 2 (failure, method name) \
+          arguments, but its arity is #{hook.arity}
+        MSG
+      end
+    end
   end
 end
diff --git a/spec/integration/operations_spec.rb b/spec/integration/operations_spec.rb
index ce0a687..012f417 100644
--- a/spec/integration/operations_spec.rb
+++ b/spec/integration/operations_spec.rb
@@ -70,6 +70,165 @@ def add_one(x) = Success(x + 1)
     ).to eq(Success(2))
   end
 
+  context "#on_failure" do
+    it "is called when prepending if a failure is returned" do
+      klass = Class.new(Dry::Operation) do
+        attr_reader :failure
+
+        def initialize
+          super
+          @failure = nil
+        end
+
+        def call(x)
+          step divide_by_zero(x)
+        end
+
+        def divide_by_zero(_x) = Failure(:not_possible)
+
+        def on_failure(failure)
+          @failure = failure
+        end
+      end
+      instance = klass.new
+
+      instance.(1)
+
+      expect(
+        instance.failure
+      ).to be(:not_possible)
+    end
+
+    it "isn't called if a success is returned" do
+      klass = Class.new(Dry::Operation) do
+        attr_reader :failure
+
+        def initialize
+          super
+          @failure = nil
+        end
+
+        def call(x)
+          step add_one(x)
+        end
+
+        def add_one(x) = Success(x + 1)
+
+        def on_failure(failure)
+          @failure = failure
+        end
+      end
+      instance = klass.new
+
+      instance.(1)
+
+      expect(
+        instance.failure
+      ).to be(nil)
+    end
+
+    it "is given the prepended method name when it accepts a second argument" do
+      klass = Class.new(Dry::Operation) do
+        attr_reader :method_name
+
+        def initialize
+          super
+          @method_name = nil
+        end
+
+        def call(x)
+          step divide_by_zero(x)
+        end
+
+        def divide_by_zero(_x) = Failure(:not_possible)
+
+        def on_failure(_failure, method_name)
+          @method_name = method_name
+        end
+      end
+      instance = klass.new
+
+      instance.(1)
+
+      expect(
+        instance.method_name
+      ).to be(:call)
+    end
+
+    it "has its arity checked and a meaningful error is raised when not conforming" do
+      klass = Class.new(Dry::Operation) do
+        def call(x)
+          step divide_by_zero(x)
+        end
+
+        def divide_by_zero(_x) = Failure(:not_possible)
+
+        def on_failure(_failure, _method_name, _unknown); end
+      end
+
+      expect { klass.new.(1) }.to raise_error(Dry::Operation::FailureHookArityError, /arity is 3/)
+    end
+
+    it "can be defined in a parent class" do
+      klass = Class.new(Dry::Operation) do
+        attr_reader :failure
+
+        def initialize
+          super
+          @failure = nil
+        end
+
+        def on_failure(failure)
+          @failure = failure
+        end
+      end
+      qlass = Class.new(klass) do
+        def call(x)
+          step divide_by_zero(x)
+        end
+
+        def divide_by_zero(_x) = Failure(:not_possible)
+      end
+      instance = qlass.new
+
+      instance.(1)
+
+      expect(
+        instance.failure
+      ).to be(:not_possible)
+    end
+
+    it "can be a private method" do
+      klass = Class.new(Dry::Operation) do
+        attr_reader :failure
+
+        def initialize
+          super
+          @failure = nil
+        end
+
+        def call(x)
+          step divide_by_zero(x)
+        end
+
+        def divide_by_zero(_x) = Failure(:not_possible)
+
+        private
+
+        def on_failure(failure)
+          @failure = failure
+        end
+      end
+      instance = klass.new
+
+      instance.(1)
+
+      expect(
+        instance.failure
+      ).to be(:not_possible)
+    end
+  end
+
   context ".operate_on" do
     it "allows prepending around a method other than #call" do
       klass = Class.new(Dry::Operation) do