diff --git a/app/controllers/checkout_controller.rb b/app/controllers/checkout_controller.rb
index c83b5e361d..1dbffc71e1 100644
--- a/app/controllers/checkout_controller.rb
+++ b/app/controllers/checkout_controller.rb
@@ -32,20 +32,20 @@ class CheckoutController < Spree::StoreController
 
   helper 'spree/orders'
 
-  rescue_from Spree::Core::GatewayError, with: :rescue_from_spree_gateway_error
-
   def edit
     return handle_redirect_from_stripe if valid_payment_intent_provided?
 
     # This is only required because of spree_paypal_express. If we implement
     # a version of paypal that uses this controller, and more specifically
-    # the #update_failed method, then we can remove this call
+    # the #action_failed method, then we can remove this call
     OrderCheckoutRestart.new(@order).call
+  rescue Spree::Core::GatewayError => e
+    rescue_from_spree_gateway_error(e)
   end
 
   def update
     params_adapter = Checkout::FormDataAdapter.new(permitted_params, @order, spree_current_user)
-    return update_failed unless @order.update(params_adapter.params[:order])
+    return action_failed unless @order.update(params_adapter.params[:order])
 
     fire_event('spree.checkout.update')
 
@@ -54,7 +54,7 @@ class CheckoutController < Spree::StoreController
     rescue_from_spree_gateway_error(e)
   rescue StandardError => e
     flash[:error] = I18n.t("checkout.failed")
-    update_failed(e)
+    action_failed(e)
   end
 
   # Clears the cached order. Required for #current_order to return a new order
@@ -138,14 +138,6 @@ class CheckoutController < Spree::StoreController
     current_order.payments.destroy_all if request.put?
   end
 
-  def rescue_from_spree_gateway_error(error)
-    flash[:error] = t(:spree_gateway_error_flash_for_checkout, error: error.message)
-    respond_to do |format|
-      format.html { render :edit }
-      format.json { render json: { flash: flash.to_hash }, status: :bad_request }
-    end
-  end
-
   def valid_payment_intent_provided?
     return false unless params["payment_intent"]&.starts_with?("pi_")
 
@@ -156,11 +148,10 @@ class CheckoutController < Spree::StoreController
   end
 
   def handle_redirect_from_stripe
-    if advance_order_state(@order) && order_complete?
+    if OrderWorkflow.new(@order).next && order_complete?
       checkout_succeeded
       redirect_to(order_path(@order)) && return
     else
-      flash[:error] = order_error
       checkout_failed
     end
   end
@@ -171,11 +162,9 @@ class CheckoutController < Spree::StoreController
         return if redirect_to_payment_gateway
       end
 
-      @order.select_shipping_method(shipping_method_id) if @order.state == "delivery"
+      next if OrderWorkflow.new(@order).next({ shipping_method_id: shipping_method_id })
 
-      next if advance_order_state(@order)
-
-      return update_failed
+      return action_failed
     end
 
     update_response
@@ -190,15 +179,6 @@ class CheckoutController < Spree::StoreController
     true
   end
 
-  # Perform order.next, guarding against StaleObjectErrors
-  def advance_order_state(order)
-    tries ||= 3
-    order.next
-  rescue ActiveRecord::StaleObjectError
-    retry unless (tries -= 1).zero?
-    false
-  end
-
   def order_error
     if @order.errors.present?
       @order.errors.full_messages.to_sentence
@@ -212,7 +192,7 @@ class CheckoutController < Spree::StoreController
       checkout_succeeded
       update_succeeded_response
     else
-      update_failed(RuntimeError.new("Order not complete after the checkout workflow"))
+      action_failed(RuntimeError.new("Order not complete after the checkout workflow"))
     end
   end
 
@@ -238,19 +218,18 @@ class CheckoutController < Spree::StoreController
     end
   end
 
-  def update_failed(error = RuntimeError.new(order_error))
-    Bugsnag.notify(error)
-
-    flash[:error] = order_error if flash.blank?
-    checkout_failed
-    update_failed_response
+  def action_failed(error = RuntimeError.new(order_error))
+    checkout_failed(error)
+    action_failed_response
   end
 
-  def checkout_failed
+  def checkout_failed(error = RuntimeError.new(order_error))
+    Bugsnag.notify(error)
+    flash[:error] = order_error if flash.blank?
     Checkout::PostCheckoutActions.new(@order).failure
   end
 
-  def update_failed_response
+  def action_failed_response
     respond_to do |format|
       format.html do
         render :edit
@@ -261,6 +240,11 @@ class CheckoutController < Spree::StoreController
     end
   end
 
+  def rescue_from_spree_gateway_error(error)
+    flash[:error] = t(:spree_gateway_error_flash_for_checkout, error: error.message)
+    action_failed(error)
+  end
+
   def permitted_params
     PermittedAttributes::Checkout.new(params).call
   end
diff --git a/app/controllers/spree/admin/orders/customer_details_controller.rb b/app/controllers/spree/admin/orders/customer_details_controller.rb
index 82c0c2347e..3239ffff01 100644
--- a/app/controllers/spree/admin/orders/customer_details_controller.rb
+++ b/app/controllers/spree/admin/orders/customer_details_controller.rb
@@ -23,7 +23,7 @@ module Spree
               @order.associate_user!(Spree.user_class.find_by(email: @order.email))
             end
 
-            AdvanceOrderService.new(@order).call
+            OrderWorkflow.new(@order).complete
 
             @order.shipments.map(&:refresh_rates)
             flash[:success] = Spree.t('customer_details_updated')
diff --git a/app/controllers/spree/admin/orders_controller.rb b/app/controllers/spree/admin/orders_controller.rb
index 8cde99be81..e0b07fbbd9 100644
--- a/app/controllers/spree/admin/orders_controller.rb
+++ b/app/controllers/spree/admin/orders_controller.rb
@@ -35,7 +35,7 @@ module Spree
       def edit
         @order.shipments.map(&:refresh_rates)
 
-        AdvanceOrderService.new(@order).call
+        OrderWorkflow.new(@order).complete
 
         # The payment step shows an error of 'No pending payments'
         # Clearing the errors from the order object will stop this error
diff --git a/app/controllers/spree/admin/payments_controller.rb b/app/controllers/spree/admin/payments_controller.rb
index 408e6e5c81..bfdb4fd2ff 100644
--- a/app/controllers/spree/admin/payments_controller.rb
+++ b/app/controllers/spree/admin/payments_controller.rb
@@ -37,7 +37,7 @@ module Spree
 
             redirect_to admin_order_payments_path(@order)
           else
-            AdvanceOrderService.new(@order).call!
+            OrderWorkflow.new(@order).complete!
 
             flash[:success] = Spree.t(:new_order_completed)
             redirect_to edit_admin_order_url(@order)
diff --git a/app/jobs/subscription_placement_job.rb b/app/jobs/subscription_placement_job.rb
index 8b5df2b159..320c7b199a 100644
--- a/app/jobs/subscription_placement_job.rb
+++ b/app/jobs/subscription_placement_job.rb
@@ -66,7 +66,7 @@ class SubscriptionPlacementJob
   end
 
   def move_to_completion(order)
-    AdvanceOrderService.new(order).call!
+    OrderWorkflow.new(order).complete!
   end
 
   def unavailable_stock_lines_for(order)
diff --git a/app/models/spree/order/checkout.rb b/app/models/spree/order/checkout.rb
new file mode 100644
index 0000000000..ffecf63f23
--- /dev/null
+++ b/app/models/spree/order/checkout.rb
@@ -0,0 +1,191 @@
+# frozen_string_literal: true
+
+module Spree
+  class Order < ActiveRecord::Base
+    module Checkout
+      def self.included(klass)
+        klass.class_eval do
+          class_attribute :next_event_transitions
+          class_attribute :previous_states
+          class_attribute :checkout_flow
+          class_attribute :checkout_steps
+          class_attribute :removed_transitions
+
+          def self.checkout_flow(&block)
+            if block_given?
+              @checkout_flow = block
+              define_state_machine!
+            else
+              @checkout_flow
+            end
+          end
+
+          def self.define_state_machine!
+            self.checkout_steps = {}
+            self.next_event_transitions = []
+            self.previous_states = [:cart]
+            self.removed_transitions = []
+
+            # Build the checkout flow using the checkout_flow defined either
+            # within the Order class, or a decorator for that class.
+            #
+            # This method may be called multiple times depending on if the
+            # checkout_flow is re-defined in a decorator or not.
+            instance_eval(&checkout_flow)
+
+            klass = self
+
+            # To avoid a ton of warnings when the state machine is re-defined
+            StateMachine::Machine.ignore_method_conflicts = true
+            # To avoid multiple occurrences of the same transition being defined
+            # On first definition, state_machines will not be defined
+            state_machines.clear if respond_to?(:state_machines)
+            state_machine :state, initial: :cart do
+              klass.next_event_transitions.each { |t| transition(t.merge(on: :next)) }
+
+              # Persist the state on the order
+              after_transition do |order|
+                order.state = order.state
+                order.save
+              end
+
+              event :cancel do
+                transition to: :canceled, if: :allow_cancel?
+              end
+
+              event :return do
+                transition to: :returned, from: :awaiting_return, unless: :awaiting_returns?
+              end
+
+              event :resume do
+                transition to: :resumed, from: :canceled, if: :allow_resume?
+              end
+
+              event :authorize_return do
+                transition to: :awaiting_return
+              end
+
+              if states[:payment]
+                before_transition to: :complete do |order|
+                  order.process_payments! if order.payment_required?
+                end
+              end
+
+              before_transition from: :cart, do: :ensure_line_items_present
+
+              before_transition to: :delivery, do: :create_proposed_shipments
+              before_transition to: :delivery, do: :ensure_available_shipping_rates
+
+              after_transition to: :complete, do: :finalize!
+              after_transition to: :delivery, do: :create_tax_charge!
+              after_transition to: :resumed,  do: :after_resume
+              after_transition to: :canceled, do: :after_cancel
+            end
+          end
+
+          def self.go_to_state(name, options = {})
+            checkout_steps[name] = options
+            previous_states.each do |state|
+              add_transition({ from: state, to: name }.merge(options))
+            end
+            if options[:if]
+              previous_states << name
+            else
+              self.previous_states = [name]
+            end
+          end
+
+          def self.insert_checkout_step(name, options = {})
+            before = options.delete(:before)
+            after = options.delete(:after) unless before
+            after = checkout_steps.keys.last unless before || after
+
+            cloned_steps = checkout_steps.clone
+            cloned_removed_transitions = removed_transitions.clone
+            checkout_flow do
+              cloned_steps.each_pair do |key, value|
+                go_to_state(name, options) if key == before
+                go_to_state(key, value)
+                go_to_state(name, options) if key == after
+              end
+              cloned_removed_transitions.each do |transition|
+                remove_transition(transition)
+              end
+            end
+          end
+
+          def self.remove_checkout_step(name)
+            cloned_steps = checkout_steps.clone
+            cloned_removed_transitions = removed_transitions.clone
+            checkout_flow do
+              cloned_steps.each_pair do |key, value|
+                go_to_state(key, value) unless key == name
+              end
+              cloned_removed_transitions.each do |transition|
+                remove_transition(transition)
+              end
+            end
+          end
+
+          def self.remove_transition(options = {})
+            removed_transitions << options
+            next_event_transitions.delete(find_transition(options))
+          end
+
+          def self.find_transition(options = {})
+            return nil if options.nil? || !options.include?(:from) || !options.include?(:to)
+
+            next_event_transitions.detect do |transition|
+              transition[options[:from].to_sym] == options[:to].to_sym
+            end
+          end
+
+          def self.next_event_transitions
+            @next_event_transitions ||= []
+          end
+
+          def self.checkout_steps
+            @checkout_steps ||= {}
+          end
+
+          def self.add_transition(options)
+            next_event_transitions << { options.delete(:from) => options.delete(:to) }.
+              merge(options)
+          end
+
+          def checkout_steps
+            steps = self.class.checkout_steps.
+              each_with_object([]) { |(step, options), checkout_steps|
+              next if options.include?(:if) && !options[:if].call(self)
+
+              checkout_steps << step
+            }.map(&:to_s)
+            # Ensure there is always a complete step
+            steps << "complete" unless steps.include?("complete")
+            steps
+          end
+
+          def checkout_step?(step)
+            step.present? ? checkout_steps.include?(step) : false
+          end
+
+          def checkout_step_index(step)
+            checkout_steps.index(step)
+          end
+
+          def self.removed_transitions
+            @removed_transitions ||= []
+          end
+
+          def can_go_to_state?(state)
+            return false unless self.state.present? &&
+                                checkout_step?(state) &&
+                                checkout_step?(self.state)
+
+            checkout_step_index(state) > checkout_step_index(self.state)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/app/services/advance_order_service.rb b/app/services/advance_order_service.rb
deleted file mode 100644
index 1377c36147..0000000000
--- a/app/services/advance_order_service.rb
+++ /dev/null
@@ -1,43 +0,0 @@
-class AdvanceOrderService
-  attr_reader :order
-
-  def initialize(order)
-    @order = order
-  end
-
-  def call
-    advance_order(advance_order_options)
-  end
-
-  def call!
-    advance_order!(advance_order_options)
-  end
-
-  private
-
-  def advance_order_options
-    shipping_method_id = order.shipping_method.id if order.shipping_method.present?
-    { shipping_method_id: shipping_method_id }
-  end
-
-  def advance_order(options)
-    until order.state == "complete"
-      break unless order.next
-
-      after_transition_hook(options)
-    end
-  end
-
-  def advance_order!(options)
-    until order.completed?
-      order.next!
-      after_transition_hook(options)
-    end
-  end
-
-  def after_transition_hook(options)
-    if order.state == "delivery"
-      order.select_shipping_method(options[:shipping_method_id]) if options[:shipping_method_id]
-    end
-  end
-end
diff --git a/app/services/order_workflow.rb b/app/services/order_workflow.rb
new file mode 100644
index 0000000000..b55ed43ce1
--- /dev/null
+++ b/app/services/order_workflow.rb
@@ -0,0 +1,79 @@
+class OrderWorkflow
+  attr_reader :order
+
+  def initialize(order)
+    @order = order
+  end
+
+  def complete
+    advance_order(advance_order_options)
+  end
+
+  def complete!
+    advance_order!(advance_order_options)
+  end
+
+  def next(options = {})
+    result = advance_order_one_step
+
+    after_transition_hook(options)
+
+    result
+  end
+
+  private
+
+  def advance_order_options
+    shipping_method_id = order.shipping_method.id if order.shipping_method.present?
+    { shipping_method_id: shipping_method_id }
+  end
+
+  def advance_order(options)
+    until order.state == "complete"
+      break unless order.next
+
+      after_transition_hook(options)
+    end
+  end
+
+  def advance_order!(options)
+    until order.completed?
+      order.next!
+      after_transition_hook(options)
+    end
+  end
+
+  def advance_order_one_step
+    tries ||= 3
+    order.next
+  rescue ActiveRecord::StaleObjectError
+    retry unless (tries -= 1).zero?
+    false
+  end
+
+  def after_transition_hook(options)
+    if order.state == "delivery"
+      order.select_shipping_method(options[:shipping_method_id]) if options[:shipping_method_id]
+    end
+
+    persist_all_payments if order.state == "payment"
+  end
+
+  # When a payment fails, the order state machine stays in 'payment' and rollbacks all transactions
+  #   This rollback also reverts the payment state from 'failed', 'void' or 'invalid' to 'pending'
+  #   Despite the rollback, the in-memory payment still has the correct state, so we persist it
+  def persist_all_payments
+    order.payments.each do |payment|
+      in_memory_payment_state = payment.state
+      if different_from_db_payment_state?(in_memory_payment_state, payment.id)
+        payment.reload.update(state: in_memory_payment_state)
+      end
+    end
+  end
+
+  # Verifies if the in-memory payment state is different from the one stored in the database
+  #   This is be done without reloading the payment so that in-memory data is not changed
+  def different_from_db_payment_state?(in_memory_payment_state, payment_id)
+    in_memory_payment_state != Spree::Payment.find(payment_id).state
+  end
+end
diff --git a/lib/tasks/sample_data/order_factory.rb b/lib/tasks/sample_data/order_factory.rb
index 29d4cf03c6..9614f685a0 100644
--- a/lib/tasks/sample_data/order_factory.rb
+++ b/lib/tasks/sample_data/order_factory.rb
@@ -45,7 +45,7 @@ class OrderFactory
 
   def create_complete_order
     order = create_cart_order
-    AdvanceOrderService.new(order).call
+    OrderWorkflow.new(order).complete
     order
   end
 
diff --git a/spec/controllers/admin/proxy_orders_controller_spec.rb b/spec/controllers/admin/proxy_orders_controller_spec.rb
index 0bc63e3cb8..8d74a8d901 100644
--- a/spec/controllers/admin/proxy_orders_controller_spec.rb
+++ b/spec/controllers/admin/proxy_orders_controller_spec.rb
@@ -77,7 +77,7 @@ describe Admin::ProxyOrdersController, type: :controller do
     before do
       # Processing order to completion
       allow(Spree::OrderMailer).to receive(:cancel_email) { double(:email, deliver: true) }
-      AdvanceOrderService.new(order).call!
+      OrderWorkflow.new(order).complete!
       proxy_order.reload
       proxy_order.cancel
       allow(controller).to receive(:spree_current_user) { user }
diff --git a/spec/controllers/checkout_controller_spec.rb b/spec/controllers/checkout_controller_spec.rb
index ecb18971db..0d285dab08 100644
--- a/spec/controllers/checkout_controller_spec.rb
+++ b/spec/controllers/checkout_controller_spec.rb
@@ -67,10 +67,31 @@ describe CheckoutController, type: :controller do
         allow(order).to receive_message_chain(:insufficient_stock_lines, :empty?).and_return true
       end
 
-      it "does not redirect" do
-        expect(order_cycle_distributed_variants).to receive(:distributes_order_variants?).with(order).and_return(true)
-        get :edit
-        expect(response).to be_success
+      describe "order variants are distributed in the OC" do
+        before do
+          expect(order_cycle_distributed_variants).to receive(:distributes_order_variants?).with(order).and_return(true)
+        end
+
+        it "does not redirect" do
+          get :edit
+          expect(response).to be_success
+        end
+
+        it "returns a specific flash message when Spree::Core::GatewayError occurs" do
+          order_checkout_restart = double(:order_checkout_restart)
+          allow(OrderCheckoutRestart).to receive(:new) { order_checkout_restart }
+          call_count = 0
+          allow(order_checkout_restart).to receive(:call) do
+            call_count += 1
+            raise Spree::Core::GatewayError.new("Gateway blow up") if call_count == 1
+          end
+
+          spree_post :edit
+
+          expect(response.status).to eq(200)
+          flash_message = I18n.t(:spree_gateway_error_flash_for_checkout, error: "Gateway blow up")
+          expect(flash[:error]).to eq flash_message
+        end
       end
 
       describe "when the order is in payment state and a stripe payment intent is provided" do
@@ -230,6 +251,15 @@ describe CheckoutController, type: :controller do
       expect(response.body).to eq({ errors: {}, flash: { error: I18n.t("checkout.failed") } }.to_json)
     end
 
+    it "returns a specific error on Spree::Core::GatewayError" do
+      allow(order).to receive(:update).and_raise(Spree::Core::GatewayError.new("Gateway blow up"))
+      spree_post :update, format: :json, order: {}
+
+      expect(response.status).to eq(400)
+      flash_message = I18n.t(:spree_gateway_error_flash_for_checkout, error: "Gateway blow up")
+      expect(json_response["flash"]["error"]).to eq flash_message
+    end
+
     describe "stale object handling" do
       it "retries when a stale object error is encountered" do
         allow(OrderCompletionReset).to receive(:new).with(controller, order) { reset_order_service }
@@ -298,7 +328,7 @@ describe CheckoutController, type: :controller do
     end
   end
 
-  describe "#update_failed" do
+  describe "#action_failed" do
     let(:restart_checkout) { instance_double(OrderCheckoutRestart, call: true) }
 
     before do
@@ -312,7 +342,7 @@ describe CheckoutController, type: :controller do
       expect(restart_checkout).to receive(:call)
       expect(controller).to receive(:respond_to)
 
-      controller.send(:update_failed)
+      controller.send(:action_failed)
     end
   end
 end
diff --git a/spec/jobs/subscription_confirm_job_spec.rb b/spec/jobs/subscription_confirm_job_spec.rb
index 603da10a10..96f718efd1 100644
--- a/spec/jobs/subscription_confirm_job_spec.rb
+++ b/spec/jobs/subscription_confirm_job_spec.rb
@@ -19,7 +19,7 @@ describe SubscriptionConfirmJob do
     let(:proxy_orders) { job.send(:unconfirmed_proxy_orders) }
 
     before do
-      AdvanceOrderService.new(order).call!
+      OrderWorkflow.new(order).complete!
     end
 
     it "returns proxy orders that meet all of the criteria" do
@@ -126,7 +126,7 @@ describe SubscriptionConfirmJob do
     let(:order) { proxy_order.initialise_order! }
 
     before do
-      AdvanceOrderService.new(order).call!
+      OrderWorkflow.new(order).complete!
       allow(job).to receive(:send_confirmation_email).and_call_original
       setup_email
       expect(job).to receive(:record_order)
diff --git a/spec/models/spree/order/checkout_spec.rb b/spec/models/spree/order/checkout_spec.rb
index 8eed383c6a..5f49792aa2 100644
--- a/spec/models/spree/order/checkout_spec.rb
+++ b/spec/models/spree/order/checkout_spec.rb
@@ -1,6 +1,306 @@
 require 'spec_helper'
 
 describe Spree::Order do
+  let(:order) { Spree::Order.new }
+
+  context "with default state machine" do
+    let(:transitions) do
+      [
+        { address: :delivery },
+        { delivery: :payment },
+        { payment: :complete },
+        { delivery: :complete }
+      ]
+    end
+
+    it "has the following transitions" do
+      transitions.each do |transition|
+        puts transition.keys.first
+        puts transition.values.first
+        transition = Spree::Order.find_transition(from: transition.keys.first,
+                                                  to: transition.values.first)
+        expect(transition).to_not be_nil
+      end
+    end
+
+    it "does not have a transition from delivery to confirm" do
+      transition = Spree::Order.find_transition(from: :delivery, to: :confirm)
+      expect(transition).to be_nil
+    end
+
+    it '.find_transition when contract was broken' do
+      expect(Spree::Order.find_transition({ foo: :bar, baz: :dog })).to be_falsy
+    end
+
+    it '.remove_transition' do
+      options = { from: transitions.first.keys.first, to: transitions.first.values.first }
+      allow(Spree::Order).to receive(:next_event_transition).and_return([options])
+      expect(Spree::Order.remove_transition(options)).to be_truthy
+    end
+
+    it '.remove_transition when contract was broken' do
+      expect(Spree::Order.remove_transition(nil)).to be_falsy
+    end
+
+    context "#checkout_steps" do
+      context "when payment not required" do
+        before { allow(order).to receive_messages payment_required?: false }
+        specify do
+          expect(order.checkout_steps).to eq %w(address delivery complete)
+        end
+      end
+
+      context "when payment required" do
+        before { allow(order).to receive_messages payment_required?: true }
+        specify do
+          expect(order.checkout_steps).to eq %w(address delivery payment complete)
+        end
+      end
+    end
+
+    it "starts out at cart" do
+      expect(order.state).to eq "cart"
+    end
+
+    it "transitions to address" do
+      order.line_items << FactoryGirl.create(:line_item)
+      order.email = "user@example.com"
+      order.next!
+      expect(order.state).to eq "address"
+    end
+
+    it "cannot transition to address without any line items" do
+      expect(order.line_items).to be_blank
+      expect(lambda { order.next! }).to raise_error(StateMachine::InvalidTransition,
+                                                /#{Spree.t(:there_are_no_items_for_this_order)}/)
+    end
+
+    context "from address" do
+      before do
+        order.state = 'address'
+        allow(order).to receive(:has_available_payment)
+        order.shipments << create(:shipment)
+        order.email = "user@example.com"
+        order.save!
+      end
+
+      it "transitions to delivery" do
+        allow(order).to receive_messages(ensure_available_shipping_rates: true)
+        order.next!
+        expect(order.state).to eq "delivery"
+      end
+
+      context "cannot transition to delivery" do
+        context "if there are no shipping rates for any shipment" do
+          specify do
+            transition = lambda { order.next! }
+            expect(transition).to raise_error(StateMachine::InvalidTransition,
+                                          /#{Spree.t(:items_cannot_be_shipped)}/)
+          end
+        end
+      end
+    end
+
+    context "from delivery" do
+      before do
+        order.state = 'delivery'
+      end
+
+      context "with payment required" do
+        before do
+          allow(order).to receive_messages payment_required?: true
+        end
+
+        it "transitions to payment" do
+          order.next!
+          expect(order.state).to eq 'payment'
+        end
+      end
+
+      context "without payment required" do
+        before do
+          allow(order).to receive_messages payment_required?: false
+        end
+
+        it "transitions to complete" do
+          order.next!
+          expect(order.state).to eq "complete"
+        end
+      end
+    end
+
+    context "from payment" do
+      before do
+        order.state = 'payment'
+      end
+
+     context "when payment is required" do
+        before do
+          allow(order).to receive_messages confirmation_required?: false
+          allow(order).to receive_messages payment_required?: true
+        end
+
+        it "transitions to complete" do
+          expect(order).to receive(:process_payments!).once.and_return true
+          order.next!
+          expect(order.state).to eq "complete"
+        end
+      end
+
+      # Regression test for #2028
+      context "when payment is not required" do
+        before do
+          allow(order).to receive_messages payment_required?: false
+        end
+
+        it "does not call process payments" do
+          expect(order).to_not receive(:process_payments!)
+          order.next!
+          expect(order.state).to eq "complete"
+        end
+      end
+    end
+  end
+
+  context "subclassed order" do
+    # This causes another test above to fail, but fixing this test should make
+    #   the other test pass
+    class SubclassedOrder < Spree::Order
+      checkout_flow do
+        go_to_state :payment
+        go_to_state :complete
+      end
+    end
+
+    it "should only call default transitions once when checkout_flow is redefined" do
+      order = SubclassedOrder.new
+      allow(order).to receive_messages payment_required?: true
+      expect(order).to receive(:process_payments!).once
+      order.state = "payment"
+      order.next!
+      expect(order.state).to eq "complete"
+    end
+  end
+
+  context "re-define checkout flow" do
+    before do
+      @old_checkout_flow = Spree::Order.checkout_flow
+      Spree::Order.class_eval do
+        checkout_flow do
+          go_to_state :payment
+          go_to_state :complete
+        end
+      end
+    end
+
+    after do
+      Spree::Order.checkout_flow(&@old_checkout_flow)
+    end
+
+    it "should not keep old event transitions when checkout_flow is redefined" do
+      expect(Spree::Order.next_event_transitions).to eq [{ cart: :payment }, { payment: :complete }]
+    end
+
+    it "should not keep old events when checkout_flow is redefined" do
+      state_machine = Spree::Order.state_machine
+      expect(state_machine.states.any? { |s| s.name == :address }).to be_falsy
+      known_states = state_machine.events[:next].branches.map(&:known_states).flatten
+      expect(known_states).to_not include(:address)
+      expect(known_states).to_not include(:delivery)
+      expect(known_states).to_not include(:confirm)
+    end
+  end
+
+  # Regression test for #3665
+  context "with only a complete step" do
+    before do
+      @old_checkout_flow = Spree::Order.checkout_flow
+      Spree::Order.class_eval do
+        checkout_flow do
+          go_to_state :complete
+        end
+      end
+    end
+
+    after do
+      Spree::Order.checkout_flow(&@old_checkout_flow)
+    end
+
+    it "does not attempt to process payments" do
+      allow(order).to receive_message_chain(:line_items, :present?).and_return(true)
+      expect(order).to_not receive(:payment_required?)
+      expect(order).to_not receive(:process_payments!)
+      order.next!
+    end
+  end
+
+  context "insert checkout step" do
+    before do
+      @old_checkout_flow = Spree::Order.checkout_flow
+      Spree::Order.class_eval do
+        insert_checkout_step :new_step, before: :address
+      end
+    end
+
+    after do
+      Spree::Order.checkout_flow(&@old_checkout_flow)
+    end
+
+    it "should maintain removed transitions" do
+      transition = Spree::Order.find_transition(from: :delivery, to: :confirm)
+      expect(transition).to be_nil
+    end
+
+    context "before" do
+      before do
+        Spree::Order.class_eval do
+          insert_checkout_step :before_address, before: :address
+        end
+      end
+
+      specify do
+        order = Spree::Order.new
+        expect(order.checkout_steps).to eq %w(new_step before_address address delivery complete)
+      end
+    end
+
+    context "after" do
+      before do
+        Spree::Order.class_eval do
+          insert_checkout_step :after_address, after: :address
+        end
+      end
+
+      specify do
+        order = Spree::Order.new
+        expect(order.checkout_steps).to eq %w(new_step address after_address delivery complete)
+      end
+    end
+  end
+
+  context "remove checkout step" do
+    before do
+      @old_checkout_flow = Spree::Order.checkout_flow
+      Spree::Order.class_eval do
+        remove_checkout_step :address
+      end
+    end
+
+    after do
+      Spree::Order.checkout_flow(&@old_checkout_flow)
+    end
+
+    it "should maintain removed transitions" do
+      transition = Spree::Order.find_transition(from: :delivery, to: :confirm)
+      expect(transition).to be_nil
+    end
+
+    specify do
+      order = Spree::Order.new
+      expect(order.checkout_steps).to eq %w(delivery complete)
+    end
+  end
+
   describe 'event :restart_checkout' do
     let(:order) { create(:order) }
 
diff --git a/spec/services/order_factory_spec.rb b/spec/services/order_factory_spec.rb
index c67cbaaf0e..394de7eaeb 100644
--- a/spec/services/order_factory_spec.rb
+++ b/spec/services/order_factory_spec.rb
@@ -47,7 +47,7 @@ describe OrderFactory do
     end
 
     it "retains address, delivery, and payment attributes until completion of the order" do
-      AdvanceOrderService.new(order).call
+      OrderWorkflow.new(order).complete
 
       order.reload
 
diff --git a/spec/services/order_syncer_spec.rb b/spec/services/order_syncer_spec.rb
index 9990425f6a..2cd6010736 100644
--- a/spec/services/order_syncer_spec.rb
+++ b/spec/services/order_syncer_spec.rb
@@ -409,7 +409,7 @@ describe OrderSyncer do
 
       context "when order is complete" do
         it "does not update the line_item quantities and adds the order to order_update_issues with insufficient stock" do
-          AdvanceOrderService.new(order).call
+          OrderWorkflow.new(order).complete
 
           expect(syncer.sync!).to be true
 
@@ -423,7 +423,7 @@ describe OrderSyncer do
         it "does not update the line_item quantities and adds the order to order_update_issues with out of stock" do
           # this single item available is used when the order is completed below, making the item out of stock
           variant.update_attribute(:on_hand, 1)
-          AdvanceOrderService.new(order).call
+          OrderWorkflow.new(order).complete
 
           expect(syncer.sync!).to be true
 
@@ -507,7 +507,7 @@ describe OrderSyncer do
       end
 
       context "when order is complete" do
-        before { AdvanceOrderService.new(order).call }
+        before { OrderWorkflow.new(order).complete }
 
         it "does not add line_item and adds the order to order_update_issues" do
           expect(syncer.sync!).to be true
diff --git a/spec/services/advance_order_service_spec.rb b/spec/services/order_workflow_spec.rb
similarity index 89%
rename from spec/services/advance_order_service_spec.rb
rename to spec/services/order_workflow_spec.rb
index 58a2fcb03a..5b2b6f0bf5 100644
--- a/spec/services/advance_order_service_spec.rb
+++ b/spec/services/order_workflow_spec.rb
@@ -1,6 +1,6 @@
 require "spec_helper"
 
-describe AdvanceOrderService do
+describe OrderWorkflow do
   let!(:distributor) { create(:distributor_enterprise) }
   let!(:order) do
     create(:order_with_totals_and_distribution, distributor: distributor,
@@ -13,7 +13,7 @@ describe AdvanceOrderService do
 
   it "transitions the order multiple steps" do
     expect(order.state).to eq("cart")
-    service.call
+    service.complete
     order.reload
     expect(order.state).to eq("complete")
   end
@@ -30,7 +30,7 @@ describe AdvanceOrderService do
 
     it "retains delivery method of the order" do
       order.select_shipping_method(shipping_method_b.id)
-      service.call
+      service.complete
       order.reload
       expect(order.shipping_method).to eq(shipping_method_b)
     end
@@ -38,7 +38,7 @@ describe AdvanceOrderService do
 
   context "when raising on error" do
     it "transitions the order multiple steps" do
-      service.call!
+      service.complete!
       order.reload
       expect(order.state).to eq("complete")
     end
@@ -49,7 +49,7 @@ describe AdvanceOrderService do
       end
 
       it "raises error" do
-        expect { service.call! }.to raise_error(StateMachine::InvalidTransition)
+        expect { service.complete! }.to raise_error(StateMachine::InvalidTransition)
       end
     end
   end