This is an automated email from the ASF dual-hosted git repository.
git-hulk pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new 4f14f65eb fix(cluster): stop raw key-value slot migration promptly
when canceled (#3714)
4f14f65eb is described below
commit 4f14f65eb4180dcea15e11ea3831a6df4765ee87
Author: jonas <[email protected]>
AuthorDate: Mon Oct 26 09:33:36 2026 +0800
fix(cluster): stop raw key-value slot migration promptly when canceled
(#3714)
### What problem does this PR solve?
Issue Number: none
Kvrocks sets `stop_migration_` to `true` on the source node when:
- `FLUSHDB` or `FLUSHALL` is executed
- the node role changes from master to slave
- `Server::PrepareRestoreDB` waits for the migration to finish before
closing and reopening RocksDB
- the server shuts down and destroys the `SlotMigrator`
This flag stops the current migration task without destroying the slot
migration thread.
The former redis-command migration path checked this flag while
iterating keys, subkeys, sending command pipelines, and syncing WAL. The
raw key-value path only checks it in `finishSuccessfulMigration`.
That prevents a canceled task from being marked as successfully
migrated, but it cannot stop the in-flight data transfer promptly, which
leads to the following problems:
1. Snapshot and WAL data keep being transferred after a flush or a role
change.
2. `PrepareRestoreDB` keeps waiting until the migration stage returns to
`kNone`. For a large slot, RocksDB cannot be closed and reopened for a
long time.
3. After the migration has been canceled, `syncWAL` may still call
`setForbiddenSlotRange`, acquiring the server exclusivity guard and
forbidding writes to the migrating slot.
4. When the migration is rate-limited, cancellation may not take effect
until the whole snapshot and WAL transfer completes.
### What is changed and how it works?
This PR adds `stop_migration_` checks to the main loops of the raw
key-value migration:
- when iterating each metadata key in `sendSnapshot`
- when iterating each subkey of complex data types in `sendSnapshot`
- when iterating each WAL item in `migrateIncrementalDataByRawKV`
- before calling `setForbiddenSlotRange` in `syncWAL`
When `stop_migration_` is found to be `true`, the migration returns
`errMigrationTaskCanceled`.
The existing migration state machine turns this error into a failed
migration and then:
- sets the import status on the destination node to failed
- clears the partially imported data on the destination node
- releases the RocksDB snapshot
- clears the forbidden slot range
- resets the migration stage to `kNone`
The comment in the FLUSHDB slot migration integration test is also
updated to match the new cancellation behavior.
### AI Usage Disclosure
AI tools(Kimi K3) were only used to analyze the code logic of the two
migration paths and to draft this PR description. All code changes were
made manually, and the final changes and technical conclusions have been
reviewed and confirmed by the author.
---------
Co-authored-by: Twice <[email protected]>
Co-authored-by: hulk <[email protected]>
Co-authored-by: shouyan <[email protected]>
---
src/cluster/slot_migrate.cc | 14 ++++++++++++--
tests/gocase/integration/slotmigrate/slotmigrate_test.go | 4 ++--
2 files changed, 14 insertions(+), 4 deletions(-)
diff --git a/src/cluster/slot_migrate.cc b/src/cluster/slot_migrate.cc
index 66228058b..3e7d2b7a0 100644
--- a/src/cluster/slot_migrate.cc
+++ b/src/cluster/slot_migrate.cc
@@ -584,12 +584,14 @@ Status SlotMigrator::sendSnapshot() {
BatchSender batch_sender(*dst_fd_, migrate_batch_size_bytes_,
migrate_batch_bytes_per_sec_);
for (iter.Seek(prefix); iter.Valid(); iter.Next()) {
+ if (stop_migration_) {
+ return {Status::NotOK, std::string(errMigrationTaskCanceled)};
+ }
// Iteration is out of range
auto key_slot_id = ExtractSlotId(iter.Key());
if (!slot_range.Contains(key_slot_id)) {
break;
}
-
auto redis_type = iter.Type();
std::string log_data;
if (redis_type == RedisType::kRedisList) {
@@ -617,6 +619,9 @@ Status SlotMigrator::sendSnapshot() {
}
for (subkey_iter->Seek(); subkey_iter->Valid(); subkey_iter->Next()) {
+ if (stop_migration_) {
+ return {Status::NotOK, std::string(errMigrationTaskCanceled)};
+ }
GET_OR_RET(batch_sender.Put(subkey_iter->ColumnFamilyHandle(),
subkey_iter->Key(), subkey_iter->Value()));
if (batch_sender.IsFull()) {
GET_OR_RET(sendMigrationBatch(&batch_sender));
@@ -671,7 +676,9 @@ Status SlotMigrator::syncWAL() {
wal_begin_seq_ = wal_incremental_seq;
epoch++;
}
-
+ if (stop_migration_) {
+ return {Status::NotOK, std::string(errMigrationTaskCanceled)};
+ }
setForbiddenSlotRange(slot_range_);
wal_incremental_seq = storage_->GetDB()->GetLatestSequenceNumber();
@@ -708,6 +715,9 @@ Status SlotMigrator::migrateIncrementalDataByRawKV(uint64_t
end_seq, BatchSender
engine::WALIterator wal_iter(storage_, slot_range_);
uint64_t start_seq = wal_begin_seq_ + 1;
for (wal_iter.Seek(start_seq); wal_iter.Valid(); wal_iter.Next()) {
+ if (stop_migration_) {
+ return {Status::NotOK, std::string(errMigrationTaskCanceled)};
+ }
if (wal_iter.NextSequenceNumber() > end_seq + 1) {
break;
}
diff --git a/tests/gocase/integration/slotmigrate/slotmigrate_test.go
b/tests/gocase/integration/slotmigrate/slotmigrate_test.go
index 6c82bcf14..3934309ed 100644
--- a/tests/gocase/integration/slotmigrate/slotmigrate_test.go
+++ b/tests/gocase/integration/slotmigrate/slotmigrate_test.go
@@ -242,8 +242,8 @@ func TestSlotMigrateSourceServerFlushedOrKilled(t
*testing.T) {
require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate",
slot, id1).Val())
waitForMigrateState(t, rdb0, slot, SlotMigrationStateStarted)
require.NoError(t, rdb0.FlushDB(ctx).Err())
- // The stop flag set by FLUSHDB is only checked when finishing
the migration,
- // so it takes a while for the rate-limited migration to reach
the failed state.
+ // The stop flag set by FLUSHDB is checked during the migration
process,
+ // it takes a while for the rate-limited migration to reach the
failed state.
waitForMigrateStateInDuration(t, rdb0, slot,
SlotMigrationStateFailed, time.Minute)
})
