Database

Database Rollback Testing: Ensuring Your Migrations Can Be Undone

HelpMeTest

HelpMeTest

— 9 min read
Database Rollback Testing: Ensuring Your Migrations Can Be Undone
  • Rollback scripts are code — they need their own tests, not just "we checked it in staging once"
  • Flyway Teams has undo migrations; Liquibase has rollback — both must be explicitly tested
  • Snapshot-based rollback (restore from backup) is different from script-based rollback — test both
  • Data preservation tests verify that rollback does not lose data written between migration and rollback
  • Production rollback drills find the gaps that staging tests miss — run them on a production clone quarterly
  • CI pipeline should run: migrate → write test data → rollback → verify data + schema

Rollback scripts are the seatbelts of database migrations. Everyone knows they should exist. Almost nobody tests them until the moment they need them, which is also the moment they discover the rollback doesn't work.

A migration rollback test answers one question: "If we run this migration and then need to undo it, does the database return to a functional prior state without data loss?" Most teams answer this question by reading the rollback script and nodding. The right answer is to execute it against real data and verify the outcome.

Why Rollback Scripts Fail Silently

A rollback script can be syntactically valid, execute without errors, and still leave the database in a broken state. Common failure modes:

  • DROP TABLE in the forward migration, but no data was backed up. Rollback can recreate the table structure but can't restore the rows.
  • Rollback drops a column added by the migration, but the column has a NOT NULL constraint. The drop fails because there's a constraint dependency.
  • Rollback removes a column that a newly-deployed application version now requires. The migration was rolled back but the app wasn't.
  • Rollback script was written for the schema at the time the migration was authored, but subsequent migrations changed the schema. The rollback now operates on a different table structure.

None of these are caught by reading the script. All of them are caught by running it against a properly seeded database.

Flyway Undo Migrations

Flyway Teams (paid) supports undo migrations as U{version}__{description}.sql files alongside the corresponding V{version}__ forward migration.

db/migrations/
  V1__create_users.sql
  U1__drop_users.sql
  V2__add_email_address_column.sql
  U2__drop_email_address_column.sql
  V3__create_email_index.sql
  U3__drop_email_index.sql

Example forward and undo pair for adding a column:

-- V4__rename_email_to_email_address.sql (expand/contract phase 1)
ALTER TABLE users ADD COLUMN email_address VARCHAR(255);
UPDATE users SET email_address = email;
-- U4__revert_email_address.sql
-- Data in email_address column will be lost if email was removed
-- This undo only makes sense if we're in the expand phase (email still exists)
ALTER TABLE users DROP COLUMN IF EXISTS email_address;

Testing Flyway Undo

# test_flyway_undo.py
import subprocess
import psycopg2
import pytest

def run_flyway(command, target_version=None):
    cmd = ["flyway", command]
    if target_version:
        cmd += [f"-target={target_version}"]
    result = subprocess.run(cmd, capture_output=True, text=True)
    assert result.returncode == 0, f"Flyway failed:\n{result.stderr}"
    return result

def test_undo_v4_column_addition(db_conn, flyway_env):
    """Undo V4 should drop email_address column and restore schema to V3 state."""
    # Apply through V4
    run_flyway("migrate", target_version="4")

    # Seed data in the new column
    with db_conn.cursor() as cur:
        cur.execute(
            "INSERT INTO users (email, email_address) VALUES (%s, %s)",
            ("alice@example.com", "alice@example.com")
        )
    db_conn.commit()

    # Undo V4
    run_flyway("undo", target_version="3")

    # Verify schema is back to V3
    with db_conn.cursor() as cur:
        cur.execute("""
            SELECT column_name FROM information_schema.columns
            WHERE table_name = 'users'
        """)
        columns = [row[0] for row in cur.fetchall()]

    assert 'email_address' not in columns, "email_address column should be gone after undo"
    assert 'email' in columns, "email column must still exist after undo"

def test_undo_does_not_lose_original_data(db_conn, flyway_env):
    """Rows that existed before the migration must still exist after undo."""
    # Apply V3 (baseline), seed data
    run_flyway("migrate", target_version="3")
    with db_conn.cursor() as cur:
        cur.execute("INSERT INTO users (email) VALUES (%s)", ("bob@example.com",))
    db_conn.commit()

    # Apply V4
    run_flyway("migrate", target_version="4")

    # Undo V4
    run_flyway("undo", target_version="3")

    # Bob's row must still exist
    with db_conn.cursor() as cur:
        cur.execute("SELECT email FROM users WHERE email = %s", ("bob@example.com",))
        row = cur.fetchone()
    assert row is not None, "Pre-migration data must survive undo"
    assert row[0] == "bob@example.com"

def test_undo_is_idempotent(db_conn, flyway_env):
    """Running undo twice should not error."""
    run_flyway("migrate", target_version="4")
    run_flyway("undo", target_version="3")
    # Second undo — already at V3, should be a no-op
    run_flyway("undo", target_version="3")

    with db_conn.cursor() as cur:
        cur.execute("""
            SELECT version FROM flyway_schema_history
            ORDER BY installed_rank DESC LIMIT 1
        """)
        current = cur.fetchone()[0]
    assert current == "3"

Liquibase Rollback Testing

Liquibase has built-in rollback support at both the tag and count level. Every changeset should define a rollback block:

<!-- changelog.xml -->
<changeSet id="4" author="dev">
    <addColumn tableName="users">
        <column name="email_address" type="VARCHAR(255)"/>
    </addColumn>
    <rollback>
        <dropColumn tableName="users" columnName="email_address"/>
    </rollback>
</changeSet>

For complex changes where auto-rollback isn't possible, write it explicitly:

<changeSet id="5" author="dev">
    <sql>UPDATE users SET email_address = email WHERE email_address IS NULL</sql>
    <rollback>
        <!-- Data cannot be restored once overwritten — document this explicitly -->
        <sql>-- WARNING: email_address values set by this changeset cannot be reverted
             -- Rollback only removes the column if we're in expand phase
             SELECT 1; -- intentional no-op
        </sql>
    </rollback>
</changeSet>

Testing Liquibase Rollback

# Roll back to a specific tag
liquibase tag before-v4
liquibase update --changeLogFile changelog.xml
# ... run tests ...
liquibase rollback before-v4

# Roll back a specific number of changesets
liquibase rollbackCount 2
# test_liquibase_rollback.py
import subprocess
import pytest

def liquibase_cmd(*args):
    result = subprocess.run(
        ["liquibase"] + list(args),
        capture_output=True, text=True
    )
    if result.returncode != 0:
        raise RuntimeError(f"Liquibase error:\n{result.stderr}")
    return result.stdout

def test_rollback_count_2(db_conn):
    """Rolling back 2 changesets should restore schema to pre-4/5 state."""
    liquibase_cmd("update")

    # Verify we're at latest
    columns_after = get_columns(db_conn, 'users')
    assert 'email_address' in columns_after

    # Roll back 2 changesets
    liquibase_cmd("rollbackCount", "2")

    columns_after_rollback = get_columns(db_conn, 'users')
    assert 'email_address' not in columns_after_rollback

def test_rollback_to_tag(db_conn):
    """Tag-based rollback should land exactly at the tagged state."""
    liquibase_cmd("tag", "v3-state")
    liquibase_cmd("update")
    liquibase_cmd("rollback", "v3-state")

    # Verify schema matches v3 state exactly
    columns = get_columns(db_conn, 'users')
    assert columns == {'id', 'email', 'created_at'}  # exact set expected at v3

def get_columns(db_conn, table_name):
    with db_conn.cursor() as cur:
        cur.execute("""
            SELECT column_name FROM information_schema.columns
            WHERE table_name = %s
        """, [table_name])
        return {row[0] for row in cur.fetchall()}

Snapshot-Based Rollback Testing

Script-based rollback (Flyway undo, Liquibase rollback) fails when:

  • The forward migration dropped a table or column with data
  • The rollback script has a bug
  • Downstream constraints block the rollback DDL

Snapshot-based rollback bypasses these problems by restoring a point-in-time database backup. It's the nuclear option and must also be tested.

The Snapshot Rollback Test

# test_snapshot_rollback.py
import subprocess
import time

def pg_dump(source_db, output_file):
    subprocess.run(
        ["pg_dump", "--format=custom", "--no-acl", "--no-owner",
         "-d", source_db, "-f", output_file],
        check=True
    )

def pg_restore(dump_file, target_db):
    subprocess.run(
        ["pg_restore", "--no-acl", "--no-owner",
         "-d", target_db, dump_file],
        check=True
    )

def test_snapshot_restore_preserves_data(db_conn, tmp_path):
    snapshot_path = tmp_path / "pre_migration.dump"

    # Seed baseline data
    with db_conn.cursor() as cur:
        cur.execute("INSERT INTO users (email) VALUES (%s)", ("alice@example.com",))
    db_conn.commit()

    # Take snapshot
    pg_dump("test_db", str(snapshot_path))

    # Run migration (destructive change — drop a column)
    with db_conn.cursor() as cur:
        cur.execute("ALTER TABLE users DROP COLUMN email")
    db_conn.commit()

    # Verify data is gone
    with db_conn.cursor() as cur:
        columns = get_columns(db_conn, 'users')
    assert 'email' not in columns

    # Restore snapshot to a recovery DB
    pg_restore(str(snapshot_path), "recovery_db")

    # Verify recovery DB has original data
    recovery_conn = psycopg2.connect("dbname=recovery_db")
    with recovery_conn.cursor() as cur:
        cur.execute("SELECT email FROM users WHERE email = %s", ("alice@example.com",))
        row = cur.fetchone()
    assert row is not None
    assert row[0] == "alice@example.com"

def test_snapshot_restore_timing(tmp_path):
    """Measure how long restore takes — must fit in your RTO."""
    snapshot_path = tmp_path / "timing_test.dump"
    pg_dump("production_clone_db", str(snapshot_path))

    start = time.time()
    pg_restore(str(snapshot_path), "restore_timing_db")
    elapsed = time.time() - start

    # RTO target: restore must complete within 10 minutes
    assert elapsed < 600, f"Restore took {elapsed:.0f}s — exceeds 10-minute RTO"

Data Preservation Tests

The hardest rollback scenario: data was written between the migration and the rollback decision. You need to know what happens to that data.

# test_data_preservation.py
import pytest

def test_data_written_after_migration_survives_rollback(db_conn, flyway_env):
    """Data inserted into a NEW column (created by migration) cannot survive rollback.
    This test documents and validates the expected data loss.
    """
    run_flyway("migrate", target_version="4")

    # Write data to the new column
    with db_conn.cursor() as cur:
        cur.execute(
            "INSERT INTO users (email_address) VALUES (%s) RETURNING id",
            ("post_migration@example.com",)
        )
        new_user_id = cur.fetchone()[0]
    db_conn.commit()

    # Roll back V4 — this DROPS email_address
    run_flyway("undo", target_version="3")

    # The user row should still exist (in the users table)
    with db_conn.cursor() as cur:
        cur.execute("SELECT id FROM users WHERE id = %s", (new_user_id,))
        row = cur.fetchone()

    # Document the outcome: row exists but email_address data is GONE
    assert row is not None, "The user row must still exist after rollback"
    # email_address column no longer exists — querying it would raise an error
    # This is expected and documented: post-migration email_address data is NOT recoverable via script rollback
    # Use snapshot restore if data recovery is required

def test_pre_migration_data_fully_preserved(db_conn, flyway_env):
    """Data that existed before migration must be 100% intact after rollback."""
    run_flyway("migrate", target_version="3")

    # Seed pre-migration data
    with db_conn.cursor() as cur:
        cur.executemany(
            "INSERT INTO users (email) VALUES (%s)",
            [("user1@example.com",), ("user2@example.com",), ("user3@example.com",)]
        )
    db_conn.commit()
    pre_count = get_count(db_conn, 'users')

    # Migrate
    run_flyway("migrate", target_version="4")

    # Undo
    run_flyway("undo", target_version="3")

    # All pre-migration rows must be intact
    post_count = get_count(db_conn, 'users')
    assert post_count == pre_count, f"Row count changed: {pre_count} → {post_count}"

    with db_conn.cursor() as cur:
        cur.execute("SELECT email FROM users ORDER BY email")
        emails = [row[0] for row in cur.fetchall()]
    assert emails == ["user1@example.com", "user2@example.com", "user3@example.com"]

Production Rollback Drills

Staging tests catch rollback script bugs. Production drills catch operational gaps: the DBA who's on-call doesn't know the procedure, the rollback script assumes an environment variable that isn't set in production, the snapshot restore target has insufficient disk space.

Run drills quarterly on a production clone:

#!/bin/bash
# production_rollback_drill.sh
# Run against a production clone, not production itself

set -e

CLONE_DB="prod_clone_$(date +%Y%m%d)"
SNAPSHOT_PATH="/backups/drill_snapshot.dump"

echo "=== DRILL START: $(date) ==="

# Step 1: Restore production snapshot to clone
echo "Restoring production snapshot..."
pg_restore --no-acl --no-owner -d "$CLONE_DB" "$SNAPSHOT_PATH"

# Step 2: Record baseline state
BASELINE_ROW_COUNT=$(psql -d "$CLONE_DB" -t -c "SELECT COUNT(*) FROM users;")
BASELINE_SCHEMA=$(psql -d "$CLONE_DB" -t -c "
    SELECT column_name, data_type
    FROM information_schema.columns
    WHERE table_name = 'users'
    ORDER BY ordinal_position;
")

echo "Baseline: $BASELINE_ROW_COUNT rows"

# Step 3: Apply pending migrations
echo "Applying migrations..."
time flyway -url="jdbc:postgresql://localhost/$CLONE_DB" migrate

# Step 4: Simulate production write load (5 minutes of writes)
echo "Simulating post-migration writes..."
python scripts/load_simulation.py --db "$CLONE_DB" --duration 300

POST_MIGRATION_COUNT=$(psql -d "$CLONE_DB" -t -c "SELECT COUNT(*) FROM users;")
echo "Post-migration row count: $POST_MIGRATION_COUNT"

# Step 5: Execute rollback
echo "Executing rollback..."
time flyway -url="jdbc:postgresql://localhost/$CLONE_DB" undo

# Step 6: Verify schema restored
POST_ROLLBACK_SCHEMA=$(psql -d "$CLONE_DB" -t -c "
    SELECT column_name, data_type
    FROM information_schema.columns
    WHERE table_name = 'users'
    ORDER BY ordinal_position;
")

if [ "$BASELINE_SCHEMA" != "$POST_ROLLBACK_SCHEMA" ]; then
    echo "DRILL FAILED: Schema mismatch after rollback"
    echo "Expected: $BASELINE_SCHEMA"
    echo "Got: $POST_ROLLBACK_SCHEMA"
    exit 1
fi

# Step 7: Verify baseline data preserved
POST_ROLLBACK_BASELINE=$(psql -d "$CLONE_DB" -t -c "
    SELECT COUNT(*) FROM users WHERE created_at < NOW() - INTERVAL '1 hour';
")
if [ "$POST_ROLLBACK_BASELINE" -lt "$BASELINE_ROW_COUNT" ]; then
    echo "DRILL FAILED: Baseline data lost. Expected $BASELINE_ROW_COUNT, got $POST_ROLLBACK_BASELINE"
    exit 1
fi

echo "=== DRILL PASSED: $(date) ==="
echo "Baseline rows preserved: $POST_ROLLBACK_BASELINE / $BASELINE_ROW_COUNT"

Record drill results: timestamp, migration version, duration of rollback, data preserved/lost. This becomes your evidence that rollback is operational.

CI Pipeline for Rollback Testing

Every migration PR should trigger a rollback test as part of CI, not just as a quarterly drill:

# .github/workflows/migration-rollback.yml
name: Migration Rollback Test

on:
  pull_request:
    paths:
      - 'db/migrations/**'

jobs:
  rollback-test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_PASSWORD: test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s

    steps:
      - uses: actions/checkout@v4

      - name: Apply migrations up to the previous version
        run: flyway migrate -target=${{ env.PREV_VERSION }}

      - name: Seed test data representing production-like state
        run: python scripts/seed_test_data.py --rows 10000

      - name: Record pre-migration state
        run: python scripts/capture_schema_snapshot.py --output pre_migration.json

      - name: Apply new migration
        run: flyway migrate

      - name: Write post-migration test data
        run: python scripts/write_post_migration_data.py

      - name: Execute rollback
        run: flyway undo -target=${{ env.PREV_VERSION }}

      - name: Verify schema matches pre-migration snapshot
        run: python scripts/compare_schema_snapshot.py --expected pre_migration.json

      - name: Verify pre-migration data preserved
        run: python scripts/verify_data_preservation.py

      - name: Report data loss (expected or unexpected)
        run: python scripts/data_loss_report.py

The data_loss_report.py step is important: some data loss after rollback is expected and documented (data written to a new column that the rollback drops). The test should distinguish between expected loss (documented in the migration) and unexpected loss (pre-existing data corrupted).

Rollback Testing Checklist

Before merging any migration PR:

  • Forward migration runs cleanly on production-sized data
  • Undo/rollback script runs without errors
  • Schema after rollback matches the expected prior state (verified programmatically, not visually)
  • Data written before the migration is 100% preserved after rollback
  • Data written after the migration and lost during rollback is documented in the migration file
  • Rollback timing measured — fits within your recovery time objective
  • Undo script tested for idempotency (running it twice doesn't error)
  • CI pipeline executes the full migrate → seed → rollback → verify sequence

The last item is the one most teams skip. Running a rollback test manually once, when you write the migration, isn't enough. CI running it on every PR is the only way to catch regressions where a later migration breaks an earlier rollback script.

Rollback scripts that aren't tested are not rollback scripts. They're comments.

Read more

Start now free