feat(migration): make vector backups mandatory for quantization

Quantization migrations now always write a vector backup before
mutating data. The previous opt-out path (--keep-backup / no
--backup-dir) is gone; instead the executor auto-defaults backup_dir
to ./migration_backups when none is supplied and refuses to run if
backup_dir is explicitly set to an empty value.

Backup files are retained on disk after a successful migration so the
operator can audit or roll back; cleanup is now a manual step.

Changes:
- executor / async_executor: auto-default backup_dir, drop keep_backup,
  drop the post-success cleanup step, expose DEFAULT_BACKUP_DIR.
- batch_executor: thread backup_dir / batch_size / num_workers through
  apply_batch and resume_batch so per-index migrations inherit them.
- cli migrate: drop --keep-backup, drop the '--workers > 1 requires
  --backup-dir' guard (backup is now always present), and add
  --backup-dir / --batch-size / --workers to batch-apply and
  batch-resume. Refresh the BGSAVE warnings to point at the automatic
  backup instead.
- docs/migrate-indexes.md: update the backup section, the rollback
  section, and the CLI flag reference to match the new behavior.
- scripts/test_migration_e2e.py, scripts/test_crash_resume_e2e.py:
  drop the keep_backup kwarg from executor.apply() calls.
- tests: add TestMandatoryBackupEnforcement covering the auto-default,
  the empty-string rejection, and that backups are no longer cleaned
  up on success.

Author: nkanu17

docs/user_guide/how_to_guides/migrate-indexes.md

@@ -606,7 +606,7 @@ The migrator will:
 
 **5. On successful completion:**
 
-The backup phase is set to `completed`. By default, backup files are **automatically deleted** after a successful migration. Pass `--keep-backup` to retain them for post-migration auditing or potential rollback.
+The backup phase is set to `completed`. Backup files are **always retained** on disk for post-migration auditing and rollback. Delete them manually from `--backup-dir` once you have verified the migrated data and no longer need a recovery path.
 
 #### Limitations
 
@@ -688,7 +688,6 @@ rvl migrate validate \
 - `--backup-dir <dir>` : Directory for vector backup files. Enables crash-safe resume and manual rollback. Required when using `--workers` > 1.
 - `--batch-size <N>` : Keys per pipeline batch (default 500). Values 200 to 1000 are typical.
 - `--workers <N>` : Parallel quantization workers (default 1). Each worker opens its own Redis connection. See [Performance](#performance-tuning) for guidance.
-- `--keep-backup` : Retain backup files after a successful migration (default: auto-cleanup).
 
 **Batch-specific flags:**
 - `--pattern` : Glob pattern to match index names (e.g., `*_idx`)
@@ -753,9 +752,10 @@ Backup files are written to the specified directory with this layout:
 **Disk usage:** approximately `num_docs × dims × bytes_per_element`.
 For example, 1M docs with 768-dim float32 vectors ≈ 2.9 GB.
 
-By default, backup files are **automatically deleted** after a successful
-migration. Pass `--keep-backup` to retain them for post-migration auditing
-or potential rollback.
+Backup files are **always retained** on disk after a successful migration
+so they remain available for post-migration auditing and rollback. Delete
+the files manually from the backup directory once you no longer need a
+recovery path.
 
 ### Crash-Safe Resume
 
@@ -808,9 +808,9 @@ rvl index create --schema original_schema.yaml --url redis://localhost:6379
 ```
 
 ```{important}
-Rollback requires that backup files were preserved. Either pass
-`--keep-backup` during migration, or ensure the backup directory was not
-cleaned up. Without backup files, rollback is not possible.
+Rollback requires that the backup directory still contains the original
+backup files. Backups are retained automatically after migration; do not
+delete the directory until you are certain rollback is no longer needed.
 ```
 
 ### Python API for Rollback
@@ -863,7 +863,6 @@ report = executor.apply(
     backup_dir="/tmp/migration_backups",   # enables crash-safe resume
     batch_size=500,                        # keys per pipeline batch
     num_workers=4,                         # parallel quantization workers
-    keep_backup=True,                      # retain backups for rollback
 )
 print(f"Quantized in {report.timings.quantize_duration_seconds}s")
 ```

redisvl/cli/migrate.py

@@ -205,6 +205,8 @@ def plan(self):
         self._print_plan_summary(args.plan_out, plan)
 
     def apply(self):
+        from redisvl.migration.executor import DEFAULT_BACKUP_DIR
+
         parser = argparse.ArgumentParser(
             usage=(
                 "rvl migrate apply --plan <migration_plan.yaml> "
@@ -222,7 +224,11 @@ def apply(self):
         parser.add_argument(
             "--backup-dir",
             dest="backup_dir",
-            help="Directory for vector backup files. Enables crash-safe resume and rollback.",
+            help=(
+                "Directory for vector backup files. Enables crash-safe resume "
+                "and rollback. Defaults to '{}' when quantization is needed. "
+                "Backup is mandatory for quantization migrations."
+            ).format(DEFAULT_BACKUP_DIR),
             default=None,
         )
         parser.add_argument(
@@ -237,16 +243,9 @@ def apply(self):
             dest="num_workers",
             type=int,
             help="Number of parallel workers for quantization (default 1). "
-            "Each worker gets its own Redis connection. Requires --backup-dir.",
+            "Each worker gets its own Redis connection.",
             default=1,
         )
-        parser.add_argument(
-            "--keep-backup",
-            dest="keep_backup",
-            action="store_true",
-            help="Keep backup files after successful migration (default: auto-delete).",
-            default=False,
-        )
         parser.add_argument(
             "--report-out",
             help="Path to write migration_report.yaml",
@@ -269,10 +268,6 @@ def apply(self):
         if args.num_workers < 1:
             parser.error("--workers must be >= 1")
 
-        # Validate --workers > 1 requires --backup-dir
-        if args.num_workers > 1 and args.backup_dir is None:
-            parser.error("--workers > 1 requires --backup-dir")
-
         redis_url = create_redis_url(args)
         plan = load_migration_plan(args.plan)
 
@@ -300,7 +295,6 @@ def apply(self):
                     backup_dir=args.backup_dir,
                     batch_size=args.batch_size,
                     num_workers=args.num_workers,
-                    keep_backup=args.keep_backup,
                 )
             )
         else:
@@ -311,7 +305,6 @@ def apply(self):
                 backup_dir=args.backup_dir,
                 batch_size=args.batch_size,
                 num_workers=args.num_workers,
-                keep_backup=args.keep_backup,
             )
 
         write_migration_report(report, args.report_out)
@@ -516,7 +509,6 @@ def _apply_sync(
         backup_dir: Optional[str] = None,
         batch_size: int = 500,
         num_workers: int = 1,
-        keep_backup: bool = False,
     ):
         """Execute migration synchronously."""
         executor = MigrationExecutor()
@@ -531,7 +523,6 @@ def _apply_sync(
             backup_dir=backup_dir,
             batch_size=batch_size,
             num_workers=num_workers,
-            keep_backup=keep_backup,
         )
 
         self._print_apply_result(report)
@@ -545,7 +536,6 @@ async def _apply_async(
         backup_dir: Optional[str] = None,
         batch_size: int = 500,
         num_workers: int = 1,
-        keep_backup: bool = False,
     ):
         """Execute migration asynchronously (non-blocking for large quantization jobs)."""
         executor = AsyncMigrationExecutor()
@@ -560,7 +550,6 @@ async def _apply_async(
             backup_dir=backup_dir,
             batch_size=batch_size,
             num_workers=num_workers,
-            keep_backup=keep_backup,
         )
 
         self._print_apply_result(report)
@@ -759,10 +748,13 @@ def batch_plan(self):
 
     def batch_apply(self):
         """Execute a batch migration plan with state tracking."""
+        from redisvl.migration.executor import DEFAULT_BACKUP_DIR
+
         parser = argparse.ArgumentParser(
             usage=(
                 "rvl migrate batch-apply --plan <batch_plan.yaml> "
-                "[--state <batch_state.yaml>] [--report-dir <./reports>]"
+                "[--state <batch_state.yaml>] [--report-dir <./reports>] "
+                "[--backup-dir <dir>] [--workers N]"
             )
         )
         parser.add_argument("--plan", help="Path to batch_plan.yaml", required=True)
@@ -781,6 +773,30 @@ def batch_apply(self):
             help="Directory for per-index migration reports",
             default="./reports",
         )
+        parser.add_argument(
+            "--backup-dir",
+            dest="backup_dir",
+            help=(
+                "Directory for vector backup files. "
+                f"Defaults to '{DEFAULT_BACKUP_DIR}' when quantization is needed. "
+                "Backup is mandatory for quantization migrations."
+            ),
+            default=None,
+        )
+        parser.add_argument(
+            "--batch-size",
+            dest="batch_size",
+            type=int,
+            help="Keys per pipeline batch (default 500)",
+            default=500,
+        )
+        parser.add_argument(
+            "--workers",
+            dest="num_workers",
+            type=int,
+            help="Number of parallel workers for quantization (default 1).",
+            default=1,
+        )
         parser = add_redis_connection_options(parser)
         args = parser.parse_args(sys.argv[3:])
 
@@ -795,8 +811,7 @@ def batch_apply(self):
          Vector data will be modified. Original precision cannot be recovered.
          To proceed, add --accept-data-loss flag.
 
-         If you need to preserve original vectors, backup your data first:
-           redis-cli BGSAVE"""
+         Vectors will be automatically backed up before quantization."""
             )
             sys.exit(1)
 
@@ -815,16 +830,22 @@ def progress_callback(
             report_dir=args.report_dir,
             redis_url=redis_url,
             progress_callback=progress_callback,
+            backup_dir=args.backup_dir,
+            batch_size=args.batch_size,
+            num_workers=args.num_workers,
         )
 
         self._print_batch_report_summary(report)
 
     def batch_resume(self):
         """Resume an interrupted batch migration."""
+        from redisvl.migration.executor import DEFAULT_BACKUP_DIR
+
         parser = argparse.ArgumentParser(
             usage=(
                 "rvl migrate batch-resume --state <batch_state.yaml> "
-                "[--plan <batch_plan.yaml>] [--retry-failed]"
+                "[--plan <batch_plan.yaml>] [--retry-failed] "
+                "[--backup-dir <dir>]"
             )
         )
         parser.add_argument("--state", help="Path to batch state file", required=True)
@@ -846,6 +867,30 @@ def batch_resume(self):
             help="Directory for per-index migration reports",
             default="./reports",
         )
+        parser.add_argument(
+            "--backup-dir",
+            dest="backup_dir",
+            help=(
+                "Directory for vector backup files. "
+                f"Defaults to '{DEFAULT_BACKUP_DIR}' when quantization is needed. "
+                "Backup is mandatory for quantization migrations."
+            ),
+            default=None,
+        )
+        parser.add_argument(
+            "--batch-size",
+            dest="batch_size",
+            type=int,
+            help="Keys per pipeline batch (default 500)",
+            default=500,
+        )
+        parser.add_argument(
+            "--workers",
+            dest="num_workers",
+            type=int,
+            help="Number of parallel workers for quantization (default 1).",
+            default=1,
+        )
         parser = add_redis_connection_options(parser)
         args = parser.parse_args(sys.argv[3:])
 
@@ -861,8 +906,7 @@ def batch_resume(self):
          Vector data will be modified. Original precision cannot be recovered.
          To proceed, add --accept-data-loss flag.
 
-         If you need to preserve original vectors, backup your data first:
-           redis-cli BGSAVE"""
+         Vectors will be automatically backed up before quantization."""
                 )
                 sys.exit(1)
 
@@ -880,6 +924,9 @@ def progress_callback(
             report_dir=args.report_dir,
             redis_url=redis_url,
             progress_callback=progress_callback,
+            backup_dir=args.backup_dir,
+            batch_size=args.batch_size,
+            num_workers=args.num_workers,
         )
 
         self._print_batch_report_summary(report)

redisvl/migration/__init__.py

@@ -14,7 +14,7 @@
 from redisvl.migration.async_validation import AsyncMigrationValidator
 from redisvl.migration.batch_executor import BatchMigrationExecutor
 from redisvl.migration.batch_planner import BatchMigrationPlanner
-from redisvl.migration.executor import MigrationExecutor
+from redisvl.migration.executor import DEFAULT_BACKUP_DIR, MigrationExecutor
 from redisvl.migration.models import BatchPlan, BatchState, SchemaPatch
 from redisvl.migration.planner import MigrationPlanner
 from redisvl.migration.validation import MigrationValidator
@@ -28,6 +28,7 @@
     "BatchMigrationPlanner",
     "BatchPlan",
     "BatchState",
+    "DEFAULT_BACKUP_DIR",
     "MigrationExecutor",
     "MigrationPlanner",
     "MigrationValidator",

redisvl/migration/async_executor.py

@@ -15,6 +15,7 @@
 from redisvl.index import AsyncSearchIndex
 from redisvl.migration.async_planner import AsyncMigrationPlanner
 from redisvl.migration.async_validation import AsyncMigrationValidator
+from redisvl.migration.executor import DEFAULT_BACKUP_DIR
 from redisvl.migration.models import (
     MigrationBenchmarkSummary,
     MigrationPlan,
@@ -549,7 +550,6 @@ async def apply(
         backup_dir: Optional[str] = None,
         batch_size: int = 500,
         num_workers: int = 1,
-        keep_backup: bool = False,
     ) -> MigrationReport:
         """Apply a migration plan asynchronously.
 
@@ -575,8 +575,6 @@ async def apply(
             num_workers: Parallel quantization workers (default 1). For
                 low-dimensional vectors (≤ 256 dims) a single worker is
                 often fastest. Diminishing returns above 4–8 workers.
-            keep_backup: Retain backup files after success (default
-                ``False``).
         """
         started_at = timestamp_utc()
         started = time.perf_counter()
@@ -696,6 +694,28 @@ async def apply(
             for change in datatype_changes.values()
         )
 
+        # Auto-default backup_dir when quantization is needed and no dir
+        # was provided.  This ensures vector data is always backed up
+        # before destructive in-place mutations.
+        if needs_quantization and backup_dir is None:
+            backup_dir = DEFAULT_BACKUP_DIR
+            logger.info(
+                "Quantization detected — using default backup directory: %s",
+                backup_dir,
+            )
+
+        # MANDATORY BACKUP ENFORCEMENT: After auto-defaulting, backup_dir
+        # must be set for any quantization migration.  This is a hard safety
+        # check — quantization without backup is never allowed.
+        if needs_quantization and not backup_dir:
+            raise ValueError(
+                "Vector backup is mandatory for quantization migrations. "
+                "A backup directory must be provided via --backup-dir or the "
+                f"default '{DEFAULT_BACKUP_DIR}' must be writable. "
+                "Quantization without backup is not allowed to prevent "
+                "irreversible data loss."
+            )
+
         if backup_dir and has_same_width_quantization:
             report.validation.errors.append(
                 "Crash-safe resume is not supported for same-width datatype "
@@ -1140,10 +1160,6 @@ def _index_progress(indexed: int, total: int, pct: float) -> None:
         finally:
             report.finished_at = timestamp_utc()
 
-        # Auto-cleanup backup files on success
-        if backup_dir and not keep_backup and report.result == "succeeded":
-            self._cleanup_backup_files(backup_dir, plan.source.index_name)
-
         return report
 
     def _cleanup_backup_files(self, backup_dir: str, index_name: str) -> None: