Skip to content

Commit 93a7c71

Browse files
committed
fix: make prefix rename idempotent for crash-resume in sync and async executors
If a migration crashes mid-rename and is resumed, RENAMENX returns False for already-renamed keys. Previously this was treated as a collision, aborting the entire resume. Now the executor checks EXISTS on both source and destination: if source is gone and destination exists, the key is counted as already-renamed and skipped. Applies to both standalone (RENAMENX) and cluster (DUMP/RESTORE) paths. Includes 7 unit tests covering resume, collision, and edge cases.
1 parent 6f928c0 commit 93a7c71

3 files changed

Lines changed: 285 additions & 22 deletions

File tree

redisvl/migration/async_executor.py

Lines changed: 52 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -289,7 +289,22 @@ async def _rename_keys_standalone(
289289
renamed += 1
290290
successfully_renamed.append(batch_key_pairs[j])
291291
else:
292-
collisions.append(batch_key_pairs[j][1])
292+
old_key, new_key = batch_key_pairs[j]
293+
# If the source is gone and destination exists, this
294+
# key was already renamed in a prior (crashed) run —
295+
# treat it as a successful no-op for idempotent resume.
296+
src_exists = await client.exists(old_key)
297+
dst_exists = await client.exists(new_key)
298+
if not src_exists and dst_exists:
299+
logger.info(
300+
"Key '%s' already renamed to '%s' (prior run), skipping",
301+
old_key,
302+
new_key,
303+
)
304+
renamed += 1
305+
successfully_renamed.append(batch_key_pairs[j])
306+
else:
307+
collisions.append(new_key)
293308
except Exception as e:
294309
logger.warning(f"Error in rename batch: {e}")
295310
raise
@@ -347,18 +362,44 @@ async def _rename_keys_cluster(
347362
if not pairs:
348363
continue
349364

350-
# Phase 1: Check destination keys don't exist (batched)
365+
# Phase 1: Check destination keys don't exist (batched).
366+
# Also check source keys so we can detect already-renamed keys
367+
# from a prior crashed run and skip them for idempotent resume.
351368
check_pipe = client.pipeline(transaction=False)
352-
for _, new_key in pairs:
369+
for old_key, new_key in pairs:
353370
check_pipe.exists(new_key)
354-
exists_results = await check_pipe.execute()
355-
for (_, new_key), exists in zip(pairs, exists_results):
356-
if exists:
357-
raise RuntimeError(
358-
f"Prefix rename aborted after {renamed} successful rename(s): "
359-
f"destination key '{new_key}' already exists. "
360-
f"Remove conflicting keys or choose a different prefix."
361-
)
371+
check_pipe.exists(old_key)
372+
check_results = await check_pipe.execute()
373+
374+
live_pairs = []
375+
for idx, (old_key, new_key) in enumerate(pairs):
376+
dst_exists = check_results[idx * 2]
377+
src_exists = check_results[idx * 2 + 1]
378+
if dst_exists:
379+
if not src_exists:
380+
# Already renamed in a prior run — count and skip.
381+
logger.info(
382+
"Key '%s' already renamed to '%s' (prior run), skipping",
383+
old_key,
384+
new_key,
385+
)
386+
renamed += 1
387+
else:
388+
raise RuntimeError(
389+
f"Prefix rename aborted after {renamed} successful rename(s): "
390+
f"destination key '{new_key}' already exists. "
391+
f"Remove conflicting keys or choose a different prefix."
392+
)
393+
else:
394+
if not src_exists:
395+
logger.warning(
396+
"Key '%s' does not exist and destination '%s' is also missing, skipping",
397+
old_key,
398+
new_key,
399+
)
400+
else:
401+
live_pairs.append((old_key, new_key))
402+
pairs = live_pairs
362403

363404
# Phase 2: DUMP + PTTL all source keys (batched — 1 RTT)
364405
dump_pipe = client.pipeline(transaction=False)

redisvl/migration/executor.py

Lines changed: 52 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -300,7 +300,22 @@ def _rename_keys_standalone(
300300
renamed += 1
301301
successfully_renamed.append(batch_key_pairs[j])
302302
else:
303-
collisions.append(batch_key_pairs[j][1])
303+
old_key, new_key = batch_key_pairs[j]
304+
# If the source is gone and destination exists, this
305+
# key was already renamed in a prior (crashed) run —
306+
# treat it as a successful no-op for idempotent resume.
307+
src_exists = client.exists(old_key)
308+
dst_exists = client.exists(new_key)
309+
if not src_exists and dst_exists:
310+
logger.info(
311+
"Key '%s' already renamed to '%s' (prior run), skipping",
312+
old_key,
313+
new_key,
314+
)
315+
renamed += 1
316+
successfully_renamed.append(batch_key_pairs[j])
317+
else:
318+
collisions.append(new_key)
304319
except Exception as e:
305320
logger.warning(f"Error in rename batch: {e}")
306321
raise
@@ -359,18 +374,44 @@ def _rename_keys_cluster(
359374
if not pairs:
360375
continue
361376

362-
# Phase 1: Check destination keys don't exist (batched)
377+
# Phase 1: Check destination keys don't exist (batched).
378+
# Also check source keys so we can detect already-renamed keys
379+
# from a prior crashed run and skip them for idempotent resume.
363380
check_pipe = client.pipeline(transaction=False)
364-
for _, new_key in pairs:
381+
for old_key, new_key in pairs:
365382
check_pipe.exists(new_key)
366-
exists_results = check_pipe.execute()
367-
for (_, new_key), exists in zip(pairs, exists_results):
368-
if exists:
369-
raise RuntimeError(
370-
f"Prefix rename aborted after {renamed} successful rename(s): "
371-
f"destination key '{new_key}' already exists. "
372-
f"Remove conflicting keys or choose a different prefix."
373-
)
383+
check_pipe.exists(old_key)
384+
check_results = check_pipe.execute()
385+
386+
live_pairs = []
387+
for idx, (old_key, new_key) in enumerate(pairs):
388+
dst_exists = check_results[idx * 2]
389+
src_exists = check_results[idx * 2 + 1]
390+
if dst_exists:
391+
if not src_exists:
392+
# Already renamed in a prior run — count and skip.
393+
logger.info(
394+
"Key '%s' already renamed to '%s' (prior run), skipping",
395+
old_key,
396+
new_key,
397+
)
398+
renamed += 1
399+
else:
400+
raise RuntimeError(
401+
f"Prefix rename aborted after {renamed} successful rename(s): "
402+
f"destination key '{new_key}' already exists. "
403+
f"Remove conflicting keys or choose a different prefix."
404+
)
405+
else:
406+
if not src_exists:
407+
logger.warning(
408+
"Key '%s' does not exist and destination '%s' is also missing, skipping",
409+
old_key,
410+
new_key,
411+
)
412+
else:
413+
live_pairs.append((old_key, new_key))
414+
pairs = live_pairs
374415

375416
# Phase 2: DUMP + PTTL all source keys (batched — 1 RTT)
376417
dump_pipe = client.pipeline(transaction=False)

tests/unit/test_async_migration_executor.py

Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -478,3 +478,184 @@ def test_is_already_quantized_same_width_int8_to_uint8():
478478
vec, expected_dims=128, source_dtype="int8", target_dtype="uint8"
479479
)
480480
assert result is False
481+
482+
483+
# =============================================================================
484+
# Idempotent Resume Rename Tests (sync executor)
485+
# =============================================================================
486+
# These tests validate that crash-resume for prefix renames is idempotent:
487+
# if a key was already renamed in a prior (crashed) run, retrying should
488+
# skip it instead of aborting with a collision error.
489+
490+
491+
class TestIdempotentResumeRenameStandalone:
492+
"""Test _rename_keys_standalone handles already-renamed keys during resume."""
493+
494+
def _make_executor(self):
495+
return MigrationExecutor()
496+
497+
def test_already_renamed_keys_skipped_on_resume(self):
498+
"""Simulate crash-resume: 2 of 3 keys were already renamed.
499+
500+
Before the fix, RENAMENX returning False would be treated as a
501+
collision and raise RuntimeError. After the fix, the executor
502+
checks if src is gone + dst exists and counts it as already done.
503+
"""
504+
executor = self._make_executor()
505+
mock_client = MagicMock()
506+
507+
# Pipeline: RENAMENX returns True for key3 (not yet renamed),
508+
# False for key1 and key2 (already renamed in prior run).
509+
mock_pipe = MagicMock()
510+
mock_pipe.execute.return_value = [False, False, True]
511+
mock_client.pipeline.return_value = mock_pipe
512+
513+
# When executor checks EXISTS for the False results:
514+
# key1: src gone, dst exists → already renamed
515+
# key2: src gone, dst exists → already renamed
516+
def exists_side_effect(key):
517+
already_renamed_srcs = {"old:1", "old:2"}
518+
already_renamed_dsts = {"new:1", "new:2"}
519+
if key in already_renamed_srcs:
520+
return 0 # source gone
521+
if key in already_renamed_dsts:
522+
return 1 # destination exists
523+
return 0
524+
525+
mock_client.exists.side_effect = exists_side_effect
526+
527+
keys = ["old:1", "old:2", "old:3"]
528+
result = executor._rename_keys_standalone(mock_client, keys, "old:", "new:")
529+
530+
# All 3 should count as renamed (2 skipped + 1 actually renamed)
531+
assert result == 3
532+
533+
def test_true_collision_still_raises(self):
534+
"""When source AND destination both exist, it's a real collision → RuntimeError."""
535+
executor = self._make_executor()
536+
mock_client = MagicMock()
537+
538+
mock_pipe = MagicMock()
539+
mock_pipe.execute.return_value = [False] # RENAMENX failed
540+
mock_client.pipeline.return_value = mock_pipe
541+
542+
# Both source and destination exist → true collision
543+
mock_client.exists.side_effect = lambda key: 1
544+
545+
keys = ["old:1"]
546+
with pytest.raises(RuntimeError, match="destination key.*already exist"):
547+
executor._rename_keys_standalone(mock_client, keys, "old:", "new:")
548+
549+
def test_src_and_dst_both_gone_is_collision(self):
550+
"""If RENAMENX fails, src is gone, but dst is ALSO gone → collision error.
551+
552+
This is an anomalous state (key deleted externally?) — we treat it
553+
as a collision rather than silently losing data.
554+
"""
555+
executor = self._make_executor()
556+
mock_client = MagicMock()
557+
558+
mock_pipe = MagicMock()
559+
mock_pipe.execute.return_value = [False]
560+
mock_client.pipeline.return_value = mock_pipe
561+
562+
# src gone, dst also gone
563+
exists_map = {"old:1": 0, "new:1": 0}
564+
mock_client.exists.side_effect = lambda key: exists_map.get(key, 0)
565+
566+
keys = ["old:1"]
567+
with pytest.raises(RuntimeError, match="destination key.*already exist"):
568+
executor._rename_keys_standalone(mock_client, keys, "old:", "new:")
569+
570+
def test_mixed_fresh_and_resumed_keys(self):
571+
"""Mix of fresh renames and already-renamed keys — all succeed."""
572+
executor = self._make_executor()
573+
mock_client = MagicMock()
574+
575+
mock_pipe = MagicMock()
576+
# key1: RENAMENX succeeds
577+
# key2: RENAMENX fails — already renamed (src gone, dst exists)
578+
mock_pipe.execute.return_value = [True, False]
579+
mock_client.pipeline.return_value = mock_pipe
580+
581+
exists_map = {
582+
"old:2": 0, # source gone
583+
"new:2": 1, # destination exists
584+
}
585+
mock_client.exists.side_effect = lambda key: exists_map.get(key, 0)
586+
587+
keys = ["old:1", "old:2"]
588+
result = executor._rename_keys_standalone(mock_client, keys, "old:", "new:")
589+
590+
assert result == 2 # 1 fresh + 1 already-renamed
591+
592+
593+
class TestIdempotentResumeRenameCluster:
594+
"""Test _rename_keys_cluster handles already-renamed keys during resume."""
595+
596+
def _make_executor(self):
597+
return MigrationExecutor()
598+
599+
def test_already_renamed_keys_skipped_on_resume(self):
600+
"""Simulate crash-resume on cluster: keys already renamed are skipped."""
601+
executor = self._make_executor()
602+
mock_client = MagicMock()
603+
604+
# Phase 1 check pipeline: exists(new_key), exists(old_key) for each pair
605+
check_pipe = MagicMock()
606+
# key1: dst exists (1), src gone (0) → already renamed
607+
# key2: dst exists (1), src gone (0) → already renamed
608+
# key3: dst gone (0), src exists (1) → needs rename
609+
check_pipe.execute.return_value = [1, 0, 1, 0, 0, 1]
610+
611+
# Phase 2 dump pipeline for key3 only
612+
dump_pipe = MagicMock()
613+
dump_pipe.execute.return_value = [b"\x00\x01\x02", -1] # dump data, pttl
614+
615+
# Phase 3 restore pipeline
616+
restore_pipe = MagicMock()
617+
restore_pipe.execute.return_value = [True, 1] # RESTORE ok, DEL ok
618+
619+
mock_client.pipeline.side_effect = [check_pipe, dump_pipe, restore_pipe]
620+
621+
keys = ["old:1", "old:2", "old:3"]
622+
result = executor._rename_keys_cluster(mock_client, keys, "old:", "new:")
623+
624+
# 2 already-renamed + 1 fresh = 3
625+
assert result == 3
626+
627+
def test_true_collision_raises_on_cluster(self):
628+
"""When source AND destination both exist on cluster → RuntimeError."""
629+
executor = self._make_executor()
630+
mock_client = MagicMock()
631+
632+
check_pipe = MagicMock()
633+
# key1: dst exists (1), src ALSO exists (1) → true collision
634+
check_pipe.execute.return_value = [1, 1]
635+
mock_client.pipeline.return_value = check_pipe
636+
637+
keys = ["old:1"]
638+
with pytest.raises(RuntimeError, match="destination key.*already exists"):
639+
executor._rename_keys_cluster(mock_client, keys, "old:", "new:")
640+
641+
def test_both_missing_key_skipped_on_cluster(self):
642+
"""Key where both source and destination are gone — warn and skip."""
643+
executor = self._make_executor()
644+
mock_client = MagicMock()
645+
646+
check_pipe = MagicMock()
647+
# key1: dst gone (0), src gone (0) → both missing
648+
check_pipe.execute.return_value = [0, 0]
649+
650+
# Even with no live_pairs, the code still creates dump/restore pipelines
651+
dump_pipe = MagicMock()
652+
dump_pipe.execute.return_value = []
653+
restore_pipe = MagicMock()
654+
655+
mock_client.pipeline.side_effect = [check_pipe, dump_pipe, restore_pipe]
656+
657+
keys = ["old:1"]
658+
result = executor._rename_keys_cluster(mock_client, keys, "old:", "new:")
659+
660+
# Key skipped, nothing renamed
661+
assert result == 0

0 commit comments

Comments
 (0)