ensure we we abort operation if timestamp has changed between start and prepare. Also control the lifetime of the keys so they are either destructed or carried over to the next phase (completeMove) to ensure no races for the accounting.

3 files changed, 12 insertions, 8 deletions

diff --git a/searchcore/src/vespa/searchcore/proton/server/bucketmovejobv2.cpp b/searchcore/src/vespa/searchcore/proton/server/bucketmovejobv2.cpp
index 93be0ee0591..49a2062595e 100644
--- a/searchcore/src/vespa/searchcore/proton/server/bucketmovejobv2.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/bucketmovejobv2.cpp
@@ -224,7 +224,7 @@ BucketMoveJobV2::prepareMove(BucketMoverSP mover, std::vector<MoveKey> keys, IDe
 {
     IncOnDestruct countGuard(_executedCount);
     if (_stopped.load(std::memory_order_relaxed)) return;
-    auto moveOps = mover->createMoveOperations(keys);
+    auto moveOps = mover->createMoveOperations(std::move(keys));
     _master.execute(makeLambdaTask([this, mover=std::move(mover), moveOps=std::move(moveOps), onDone=std::move(onDone)]() mutable {
         if (_stopped.load(std::memory_order_relaxed)) return;
         completeMove(std::move(mover), std::move(moveOps), std::move(onDone));
diff --git a/searchcore/src/vespa/searchcore/proton/server/documentbucketmover.cpp b/searchcore/src/vespa/searchcore/proton/server/documentbucketmover.cpp
index 6defd3e7037..662e9e2e920 100644
--- a/searchcore/src/vespa/searchcore/proton/server/documentbucketmover.cpp
+++ b/searchcore/src/vespa/searchcore/proton/server/documentbucketmover.cpp
@@ -22,9 +22,12 @@ typedef IDocumentMetaStore::Iterator Iterator;
 
 BucketMover::GuardedMoveOp
 BucketMover::createMoveOperation(MoveKey &key) {
-    if (_source->lidNeedsCommit(key._lid)) {
-        return {};
-    }
+    if (_source->lidNeedsCommit(key._lid)) return {};
+
+    const RawDocumentMetaData &metaNow = _source->meta_store()->getRawMetaData(key._lid);
+    if (metaNow.getGid() != key._gid) return {};
+    if (metaNow.getTimestamp() != key._timestamp) return {};
+
     Document::SP doc(_source->retriever()->getFullDocument(key._lid));
     if (!doc || doc->getId().getGlobalId() != key._gid)
         return {}; // Failed to retrieve document, removed or changed identity
@@ -86,7 +89,7 @@ BucketMover::getKeysToMove(size_t maxDocsToMove) {
 }
 
 std::vector<BucketMover::GuardedMoveOp>
-BucketMover::createMoveOperations(std::vector<MoveKey> &toMove) {
+BucketMover::createMoveOperations(std::vector<MoveKey> toMove) {
     std::vector<GuardedMoveOp> successfulReads;
     successfulReads.reserve(toMove.size());
     for (MoveKey &key : toMove) {
@@ -138,8 +141,9 @@ DocumentBucketMover::moveDocuments(size_t maxDocsToMove, IMoveOperationLimiter &
         return true;
     }
     auto [keys, done] = _impl->getKeysToMove(maxDocsToMove);
-    auto moveOps = _impl->createMoveOperations(keys);
-    bool allOk = keys.size() == moveOps.size();
+    size_t numKeys = keys.size();
+    auto moveOps = _impl->createMoveOperations(std::move(keys));
+    bool allOk = (numKeys == moveOps.size());
     if (done && allOk) {
         _impl->setBucketDone();
     }
diff --git a/searchcore/src/vespa/searchcore/proton/server/documentbucketmover.h b/searchcore/src/vespa/searchcore/proton/server/documentbucketmover.h
index d3743bebee4..807151b0769 100644
--- a/searchcore/src/vespa/searchcore/proton/server/documentbucketmover.h
+++ b/searchcore/src/vespa/searchcore/proton/server/documentbucketmover.h
@@ -76,7 +76,7 @@ public:
     /// Must be called in master thread
     std::pair<std::vector<MoveKey>, bool> getKeysToMove(size_t maxDocsToMove);
     /// Call from any thread
-    std::vector<GuardedMoveOp> createMoveOperations(std::vector<MoveKey> & toMove);
+    std::vector<GuardedMoveOp> createMoveOperations(std::vector<MoveKey> toMove);
     /// Must be called in master thread
     void moveDocuments(std::vector<GuardedMoveOp> moveOps, IDestructorCallbackSP onDone);
     void moveDocument(MoveOperationUP moveOp, IDestructorCallbackSP onDone);