diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2abf08c2..a22998bf 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,4 +1,5 @@
 ### Master (unreleased)
+- [140](https://github.com/Shopify/job-iteration/pull/140) - Add `JobIteration::DestroyAssociationJob` to be used by Active Record associations with the `dependent: :destroy_async` option
 
 ## v1.3.0 (Oct 7, 2021)
 - [133](https://github.com/Shopify/job-iteration/pull/133) - Moves attributes out of JobIteration::Iteration included block
diff --git a/lib/job-iteration/destroy_association_job.rb b/lib/job-iteration/destroy_association_job.rb
new file mode 100644
index 00000000..f8e0747b
--- /dev/null
+++ b/lib/job-iteration/destroy_association_job.rb
@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+require "active_job"
+require "active_record/destroy_association_async_job"
+
+module JobIteration
+  # Port of https://github.com/rails/rails/blob/main/activerecord/lib/active_record/destroy_association_async_job.rb
+  # (MIT license) but instead of +ActiveRecord::Batches+ this job uses the +Iteration+ API to destroy associated
+  # objects.
+  #
+  # @see https://guides.rubyonrails.org/association_basics.html Using the 'dependent: :destroy_async' option
+  # @see https://guides.rubyonrails.org/configuring.html#configuring-active-record Configuring Active Record
+  #   'destroy_association_async_job' and 'queues.destroy' options
+  class DestroyAssociationJob < ::ActiveJob::Base
+    include(JobIteration::Iteration)
+
+    queue_as do
+      # Compatibility with Rails 7 and 6.1
+      queues = defined?(ActiveRecord.queues) ? ActiveRecord.queues : ActiveRecord::Base.queues
+      queues[:destroy]
+    end
+
+    discard_on(ActiveJob::DeserializationError)
+
+    def build_enumerator(params, cursor:)
+      association_model = params[:association_class].constantize
+      owner_class = params[:owner_model_name].constantize
+      owner = owner_class.find_by(owner_class.primary_key.to_sym => params[:owner_id])
+
+      unless owner_destroyed?(owner, params[:ensuring_owner_was_method])
+        raise ActiveRecord::DestroyAssociationAsyncError, "owner record not destroyed"
+      end
+
+      enumerator_builder.active_record_on_records(
+        association_model.where(params[:association_primary_key_column] => params[:association_ids]),
+        cursor: cursor,
+      )
+    end
+
+    def each_iteration(record, _params)
+      record.destroy
+    end
+
+    private
+
+    def owner_destroyed?(owner, ensuring_owner_was_method)
+      !owner || (ensuring_owner_was_method && owner.public_send(ensuring_owner_was_method))
+    end
+  end
+end
diff --git a/test/test_helper.rb b/test/test_helper.rb
index b7a0b197..802ef5a2 100644
--- a/test/test_helper.rb
+++ b/test/test_helper.rb
@@ -7,6 +7,7 @@
 
 require "job-iteration"
 require "job-iteration/test_helper"
+require "job-iteration/destroy_association_job"
 
 require "globalid"
 require "sidekiq"
@@ -19,6 +20,7 @@
 
 GlobalID.app = "iteration"
 ActiveRecord::Base.include(GlobalID::Identification) # https://github.com/rails/globalid/blob/master/lib/global_id/railtie.rb
+ActiveRecord::Base.destroy_association_async_job = JobIteration::DestroyAssociationJob
 
 module ActiveJob
   module QueueAdapters
@@ -43,6 +45,26 @@ def enqueue_at(job, _delay)
 ActiveJob::Base.queue_adapter = :iteration_test
 
 class Product < ActiveRecord::Base
+  has_many :variants, dependent: :destroy_async
+end
+
+class SoftDeletedProduct < ActiveRecord::Base
+  self.table_name = "products"
+  has_many :variants, foreign_key: "product_id", dependent: :destroy_async, ensuring_owner_was: :deleted?
+
+  def deleted?
+    deleted
+  end
+
+  def destroy
+    update!(deleted: true)
+    run_callbacks(:destroy)
+    run_callbacks(:commit)
+  end
+end
+
+class Variant < ActiveRecord::Base
+  belongs_to :product
 end
 
 host = ENV["USING_DEV"] == "1" ? "job-iteration.railgun" : "localhost"
@@ -67,6 +89,13 @@ class Product < ActiveRecord::Base
 
 ActiveRecord::Base.connection.create_table(Product.table_name, force: true) do |t|
   t.string(:name)
+  t.string(:deleted, default: false)
+  t.timestamps
+end
+
+ActiveRecord::Base.connection.create_table(Variant.table_name, force: true) do |t|
+  t.references(:product)
+  t.string(:color)
   t.timestamps
 end
 
diff --git a/test/unit/destroy_association_job_test.rb b/test/unit/destroy_association_job_test.rb
new file mode 100644
index 00000000..b5108383
--- /dev/null
+++ b/test/unit/destroy_association_job_test.rb
@@ -0,0 +1,48 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+module JobIteration
+  class DestroyAssociationJobTest < IterationUnitTest
+    setup do
+      @product = Product.first
+      ["pink", "red"].each do |color|
+        @product.variants.create!(color: color)
+      end
+    end
+
+    test "destroys the associated records" do
+      @product.destroy!
+
+      assert_difference(->() { Variant.count }, -2) do
+        work_job
+      end
+    end
+
+    test "checks if owner was destroyed using custom method" do
+      @product = SoftDeletedProduct.first
+      @product.destroy!
+
+      assert_difference(->() { Variant.count }, -2) do
+        work_job
+      end
+    end
+
+    test "throw an error if the record is not actually destroyed" do
+      @product.destroy!
+      Product.create!(id: @product.id, name: @product.name)
+
+      assert_raises(ActiveRecord::DestroyAssociationAsyncError) do
+        work_job
+      end
+    end
+
+    private
+
+    def work_job
+      job = ActiveJob::Base.queue_adapter.enqueued_jobs.pop
+      assert_equal(job["job_class"], "JobIteration::DestroyAssociationJob")
+      ActiveJob::Base.execute(job)
+    end
+  end
+end