fix: code-rev round 7 - resume skips key renames, backup filename collisions

Critical fix:
- Sync and async executor resume paths now perform key prefix renames
  after quantize (renames happen after drop in normal path, so they
  may not have completed before crash)

Backup naming:
- Add sha256[:8] hash of index_name to backup filenames to prevent
  collisions between distinct names that sanitize identically
  (e.g., 'a/b' and 'a:b' both become 'a_b' but have different hashes)
- Applied to single-worker, multi-worker, and cleanup paths

Accepted as-is:
- --resume backward compat: intentional, clear error message
- Rollback non-recursive: flat layout matches how backups are written

Tests: 178 passed

Author: nkanu17

redisvl/migration/async_executor.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import time
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, AsyncGenerator, Callable, Dict, List, Optional
@@ -547,7 +548,10 @@ async def apply(
                 .replace("\\", "_")
                 .replace(":", "_")
             )
-            backup_path = str(Path(backup_dir) / f"migration_backup_{safe_name}")
+            name_hash = hashlib.sha256(plan.source.index_name.encode()).hexdigest()[:8]
+            backup_path = str(
+                Path(backup_dir) / f"migration_backup_{safe_name}_{name_hash}"
+            )
             existing_backup = VectorBackup.load(backup_path)
 
             if existing_backup is not None:
@@ -692,6 +696,36 @@ def _notify(step: str, detail: Optional[str] = None) -> None:
                         "quantize",
                         f"done ({docs_quantized:,} docs in {quantize_duration}s)",
                     )
+
+                # Key prefix renames may not have happened before the crash
+                # (they run after index drop in the normal path). Re-apply
+                # idempotently.
+                if has_prefix_change:
+                    resume_keys = []
+                    for batch_keys, _ in existing_backup.iter_batches():
+                        resume_keys.extend(batch_keys)
+                    if resume_keys:
+                        old_prefix = plan.source.keyspace.prefixes[0]
+                        new_prefix = rename_ops.change_prefix
+                        assert new_prefix is not None
+                        _notify("key_rename", "Renaming keys (resume)...")
+                        key_rename_started = time.perf_counter()
+                        renamed_count = await self._rename_keys(
+                            client,
+                            resume_keys,
+                            old_prefix,
+                            new_prefix,
+                            progress_callback=lambda done, total: _notify(
+                                "key_rename", f"{done:,}/{total:,} keys"
+                            ),
+                        )
+                        key_rename_duration = round(
+                            time.perf_counter() - key_rename_started, 3
+                        )
+                        _notify(
+                            "key_rename",
+                            f"done ({renamed_count:,} keys in {key_rename_duration}s)",
+                        )
             else:
                 # Normal (non-resume) path
                 if needs_enumeration:
@@ -1057,7 +1091,8 @@ def _cleanup_backup_files(self, backup_dir: str, index_name: str) -> None:
         that happen to share the same prefix.
         """
         safe_name = index_name.replace("/", "_").replace("\\", "_").replace(":", "_")
-        base_prefix = f"migration_backup_{safe_name}"
+        name_hash = hashlib.sha256(index_name.encode()).hexdigest()[:8]
+        base_prefix = f"migration_backup_{safe_name}_{name_hash}"
         known_suffixes = (".header", ".data")
         backup_dir_path = Path(backup_dir)
 

redisvl/migration/executor.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import time
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable, Dict, Generator, List, Optional
@@ -615,13 +616,18 @@ def apply(
         backup_path: Optional[str] = None
 
         if backup_dir:
-            # Sanitize index name for filesystem
+            # Sanitize index name for filesystem with hash suffix to avoid
+            # collisions between distinct names that sanitize identically
+            # (e.g., "a/b" and "a:b" both become "a_b").
             safe_name = (
                 plan.source.index_name.replace("/", "_")
                 .replace("\\", "_")
                 .replace(":", "_")
             )
-            backup_path = str(Path(backup_dir) / f"migration_backup_{safe_name}")
+            name_hash = hashlib.sha256(plan.source.index_name.encode()).hexdigest()[:8]
+            backup_path = str(
+                Path(backup_dir) / f"migration_backup_{safe_name}_{name_hash}"
+            )
             existing_backup = VectorBackup.load(backup_path)
 
             if existing_backup is not None:
@@ -793,6 +799,38 @@ def _notify(step: str, detail: Optional[str] = None) -> None:
                         "quantize",
                         f"done ({docs_quantized:,} docs in {quantize_duration}s)",
                     )
+
+                # Key prefix renames may not have happened before the crash
+                # (they run after index drop in the normal path). Re-apply
+                # idempotently — RENAME is a no-op if old == new or key
+                # was already renamed.
+                if has_prefix_change:
+                    # Collect keys from backup to know what to rename
+                    resume_keys = []
+                    for batch_keys, _ in existing_backup.iter_batches():
+                        resume_keys.extend(batch_keys)
+                    if resume_keys:
+                        old_prefix = plan.source.keyspace.prefixes[0]
+                        new_prefix = rename_ops.change_prefix
+                        assert new_prefix is not None
+                        _notify("key_rename", "Renaming keys (resume)...")
+                        key_rename_started = time.perf_counter()
+                        renamed_count = self._rename_keys(
+                            client,
+                            resume_keys,
+                            old_prefix,
+                            new_prefix,
+                            progress_callback=lambda done, total: _notify(
+                                "key_rename", f"{done:,}/{total:,} keys"
+                            ),
+                        )
+                        key_rename_duration = round(
+                            time.perf_counter() - key_rename_started, 3
+                        )
+                        _notify(
+                            "key_rename",
+                            f"done ({renamed_count:,} keys in {key_rename_duration}s)",
+                        )
             else:
                 # Normal (non-resume) path
                 # STEP 1: Enumerate keys BEFORE any modifications
@@ -1147,7 +1185,8 @@ def _cleanup_backup_files(self, backup_dir: str, index_name: str) -> None:
         that happen to share the same prefix.
         """
         safe_name = index_name.replace("/", "_").replace("\\", "_").replace(":", "_")
-        base_prefix = f"migration_backup_{safe_name}"
+        name_hash = hashlib.sha256(index_name.encode()).hexdigest()[:8]
+        base_prefix = f"migration_backup_{safe_name}_{name_hash}"
         # Exact suffixes written by VectorBackup
         known_suffixes = (".header", ".data")
         backup_dir_path = Path(backup_dir)

redisvl/migration/quantize.py

@@ -7,6 +7,7 @@
 using ThreadPoolExecutor (sync) or asyncio.gather (async).
 """
 
+import hashlib
 import logging
 import math
 from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -264,8 +265,9 @@ def multi_worker_quantize(
 
     # Generate backup paths per worker
     safe_name = index_name.replace("/", "_").replace("\\", "_").replace(":", "_")
+    name_hash = hashlib.sha256(index_name.encode()).hexdigest()[:8]
     worker_backup_paths = [
-        str(Path(backup_dir) / f"migration_backup_{safe_name}_worker{i}")
+        str(Path(backup_dir) / f"migration_backup_{safe_name}_{name_hash}_worker{i}")
         for i in range(actual_workers)
     ]
 
@@ -441,8 +443,9 @@ async def async_multi_worker_quantize(
         )
 
     safe_name = index_name.replace("/", "_").replace("\\", "_").replace(":", "_")
+    name_hash = hashlib.sha256(index_name.encode()).hexdigest()[:8]
     worker_backup_paths = [
-        str(Path(backup_dir) / f"migration_backup_{safe_name}_worker{i}")
+        str(Path(backup_dir) / f"migration_backup_{safe_name}_{name_hash}_worker{i}")
         for i in range(actual_workers)
     ]