Add non-atomic migration mode with automatic down rollback

Support `atomic false` for migrations that should not be wrapped in a
transaction. This avoids holding database locks for the full duration
of large migrations.

When atomic is false:
- A `down` method is required (raises ArgumentError if missing)
- If `up` or `verify!` fails, `down` is called automatically
- If `down` itself fails, warns and re-raises the original error
- `batch`/`find_each` throttle between batches by default
- In atomic mode, throttling is disabled (it only prolongs the lock)

Author: MatheusRich

CHANGELOG.md

@@ -1,5 +1,32 @@
 ## [Unreleased]
 
+- Add `atomic false` mode for non-atomic migrations that don't hold a transaction open
+
+```ruby
+class BackfillDefaultUsername < DataCustoms::Migration
+  atomic false
+
+  def up
+    batch(User.where(username: nil)) do |rel|
+      rel.update_all(username: "guest")
+    end
+  end
+
+  def verify!
+    raise "Failed" if User.exists?(username: nil)
+  end
+
+  def down
+    User.where(username: "guest").update_all(username: nil)
+  end
+end
+```
+
+  - Requires a `down` method (raises `ArgumentError` if missing)
+  - Automatically calls `down` if `up` or `verify!` fails
+  - If `down` itself fails, warns and re-raises the original error
+- Disable throttling by default in atomic mode (`batch`/`find_each` only throttle in non-atomic mode)
+
 ## [0.2.0] - 2026-03-04
 
 - Add `progress.report` for tracking migration progress with a visual bar that updates in place on TTY terminals

README.md

@@ -101,48 +101,49 @@ AddDefaultUsername.run("anonymous")
 
 #### Dealing with large datasets
 
-The migration code runs inside a transaction, so be careful when dealing with
-large datasets, as [it will block the database][blocking] for the duration of
-the transaction.
+By default, the migration code runs inside a transaction. For large datasets,
+this [blocks the database][blocking] for the duration of the migration. To
+avoid this, use `atomic false` to opt out of the transaction.
 
-Data Customs provides a few helpers to make working in batches easier and
-automatically throttles the migration to avoid overwhelming the database.
+Non-atomic migrations require a `down` method that reverts the changes made by
+`up`. Since there is no transaction to roll back, `down` is your rollback
+strategy: if `up` or `verify!` fails, `down` is called automatically. It should
+be idempotent, as the gem does not track which records were changed.
 
 ```ruby
-class AddDefaultUsername < DataCustoms::Migration
+class BackfillDefaultUsername < DataCustoms::Migration
+  atomic false
+
   def up
-    batch(User.where.missing(:username)) do |relation|
-      relation.update_all(username: "guest")
+    batch(User.where(username: nil)) do |rel|
+      rel.update_all(username: "guest")
     end
   end
-end
-```
 
-Or, if you need access to each individual record:
+  def verify!
+    raise "Failed" if User.exists?(username: nil)
+  end
 
-```ruby
-class AddDefaultUsername < DataCustoms::Migration
-  def up
-    find_each(User.where.missing(:username)) do |record|
-      # do something with record
-    end
+  def down
+    User.where(username: "guest").update_all(username: nil)
   end
 end
 ```
 
-For both methods, you can configure the batch size and the pause between batches:
+The `batch` and `find_each` helpers process records using
+[`in_batches`][in_batches], with a short pause between each batch so other
+queries can run in between. Use `find_each` when you need access to individual
+records instead of relations.
+
+Both methods accept `batch_size` and `throttle_seconds` options:
 
 ```ruby
-class LongMigration < DataCustoms::Migration
-  def up
-    batch(records, batch_size: 500, throttle_seconds: 0.1) do |relation|
-      # ...
-    end
+batch(records, batch_size: 500, throttle_seconds: 0.1) do |relation|
+  # ...
+end
 
-    find_each(records, batch_size: 500, throttle_seconds: 0.1) do |record|
-      # ...
-    end
-  end
+find_each(records, batch_size: 500, throttle_seconds: 0.1) do |record|
+  # ...
 end
 ```
 
@@ -274,3 +275,4 @@ We love open source software! See [our other projects][community]. We are
 <!-- END /templates/footer.md -->
 
 [blocking]: https://github.com/ankane/strong_migrations?tab=readme-ov-file#backfilling-data
+[in_batches]: https://api.rubyonrails.org/classes/ActiveRecord/Batches.html#method-i-in_batches

lib/data_customs/migration.rb

@@ -5,6 +5,14 @@ class Migration
     DEFAULT_BATCH_SIZE = 1000
     DEFAULT_THROTTLE = 0.01
 
+    def self.atomic(value)
+      @atomic = value
+    end
+
+    def self.atomic?
+      @atomic != false
+    end
+
     def self.progress(**options)
       @progress_options = options
     end
@@ -14,6 +22,8 @@ def self.progress_options
     end
 
     def self.run(*args, **kwargs)
+      ensure_rollback_strategy!
+
       with_transaction(*args, **kwargs) do |migration|
         migration.run
       end
@@ -22,9 +32,29 @@ def self.run(*args, **kwargs)
       raise e
     end
 
+    def self.ensure_rollback_strategy!
+      return if atomic? || method_defined?(:down)
+
+      raise ArgumentError, "down method is required when running a non-atomic migration"
+    end
+
     def self.with_transaction(*args, **kwargs)
-      ActiveRecord::Base.transaction do
-        yield new(*args, **kwargs)
+      if atomic?
+        ActiveRecord::Base.transaction do
+          yield new(*args, **kwargs)
+        end
+      else
+        migration = new(*args, **kwargs)
+        begin
+          yield migration
+        rescue => e
+          begin
+            migration.down
+          rescue => down_error
+            warn "🛃 down failed: #{down_error.message}"
+          end
+          raise e
+        end
       end
     end
 
@@ -50,7 +80,11 @@ def with_progress_reporter(&block)
 
     def progress = @_progress
 
-    def batch(scope, batch_size: DEFAULT_BATCH_SIZE, throttle_seconds: DEFAULT_THROTTLE)
+    def default_throttle
+      self.class.atomic? ? 0 : DEFAULT_THROTTLE
+    end
+
+    def batch(scope, batch_size: DEFAULT_BATCH_SIZE, throttle_seconds: default_throttle)
       scope.in_batches(of: batch_size) do |relation|
         yield relation
         sleep(throttle_seconds) if throttle_seconds.positive?

spec/data_customs/migration_spec.rb

@@ -215,6 +215,114 @@ def verify! = nil
     end
   end
 
+  describe "non-atomic migration" do
+    it "does not wrap in a transaction (changes persist on failure)" do
+      migration = Class.new(DataCustoms::Migration) do
+        atomic false
+
+        def up
+          TestUser.create!(name: "persisted")
+          raise "Boom"
+        end
+
+        def verify! = nil
+        def down = nil
+      end
+
+      expect { migration.run }.to raise_error("Boom")
+      expect(TestUser.exists?(name: "persisted")).to be true
+    end
+
+    it "calls down when up fails" do
+      migration = Class.new(DataCustoms::Migration) do
+        atomic false
+
+        def up
+          TestUser.create!(name: "to_revert")
+          raise "Boom"
+        end
+
+        def verify! = nil
+
+        def down
+          TestUser.where(name: "to_revert").delete_all
+        end
+      end
+
+      expect { migration.run }.to raise_error("Boom")
+      expect(TestUser.exists?(name: "to_revert")).to be false
+    end
+
+    it "calls down when verify! fails" do
+      migration = Class.new(DataCustoms::Migration) do
+        atomic false
+
+        def up
+          TestUser.create!(name: "to_revert")
+        end
+
+        def verify!
+          raise "Verification failed"
+        end
+
+        def down
+          TestUser.where(name: "to_revert").delete_all
+        end
+      end
+
+      expect { migration.run }.to raise_error("Verification failed")
+      expect(TestUser.exists?(name: "to_revert")).to be false
+    end
+
+    it "warns and re-raises the original error if down also fails" do
+      migration = Class.new(DataCustoms::Migration) do
+        atomic false
+
+        def up = raise "Original error"
+        def verify! = nil
+        def down = raise "Down error"
+      end
+
+      expect { migration.run }
+        .to raise_error("Original error")
+        .and output(/down failed: Down error/).to_stderr
+    end
+
+    it "succeeds without calling down" do
+      migration = Class.new(DataCustoms::Migration) do
+        atomic false
+
+        def up
+          TestUser.create!(name: "kept")
+        end
+
+        def verify!
+          raise "Missing!" unless TestUser.exists?(name: "kept")
+        end
+
+        def down
+          raise "down should not be called"
+        end
+      end
+
+      expect { migration.run }.to(
+        change { TestUser.count }.by(1)
+          .and(output("🛃 Data migration ran successfully!\n").to_stdout)
+      )
+    end
+
+    it "raises ArgumentError if down is not defined" do
+      migration = Class.new(DataCustoms::Migration) do
+        atomic false
+
+        def up = nil
+        def verify! = nil
+      end
+
+      expect { migration.run }.to raise_error(ArgumentError, /down method is required/)
+    end
+  end
+
   describe "helpers" do
     it "batches records" do
       3.times { |i| TestUser.create!(name: "User #{i}") }
@@ -232,20 +340,44 @@ def verify!
           raise "Wrong batches #{@batch_sizes}" if @batch_sizes != [2, 1]
         end
       end
+      expect_any_instance_of(Kernel).not_to receive(:sleep)
+
+      expect { migration.run }.to output("🛃 Data migration ran successfully!\n").to_stdout
+    end
+
+    it "throttles between batches in non-atomic mode" do
+      3.times { |i| TestUser.create!(name: "User #{i}") }
+
+      migration = Class.new(DataCustoms::Migration) do
+        atomic false
+
+        def initialize = @batch_sizes = []
+
+        def up
+          batch(TestUser.all, batch_size: 2) do |relation|
+            @batch_sizes << relation.size
+          end
+        end
+
+        def verify!
+          raise "Wrong batches #{@batch_sizes}" if @batch_sizes != [2, 1]
+        end
+
+        def down = nil
+      end
       expect_any_instance_of(Kernel).to receive(:sleep).exactly(2).times
 
       expect { migration.run }.to output("🛃 Data migration ran successfully!\n").to_stdout
     end
 
     it "finds each record" do
-      allow_any_instance_of(Kernel).to receive(:sleep)
       3.times { |i| TestUser.create!(name: "User #{i}") }
 
       migration = Class.new(DataCustoms::Migration) do
         def initialize = @users = []
 
         def up
-          find_each(TestUser.all, throttle_seconds: -1) do |user|
+          find_each(TestUser.all) do |user|
             @users << user.name
           end
         end