Must check cluster state version in pending state, if present

Checking just the current state's version does not work as expected
if the distributor is currently _within_ a cluster state transition.
In this case, the pending state's version must be used instead.
Failure to do so will make the code erroneously believe the cluster
state has not changed since the start of the operation.

2 files changed, 18 insertions, 2 deletions

diff --git a/storage/src/tests/distributor/check_condition_test.cpp b/storage/src/tests/distributor/check_condition_test.cpp
index d355194ec0d..757a9329ea6 100644
--- a/storage/src/tests/distributor/check_condition_test.cpp
+++ b/storage/src/tests/distributor/check_condition_test.cpp
@@ -242,7 +242,20 @@ TEST_F(CheckConditionTest, check_fails_if_replica_set_changed_between_start_and_
     });
 }
 
-TEST_F(CheckConditionTest, check_fails_if_bucket_ownership_changed_between_start_and_completion) {
+TEST_F(CheckConditionTest, check_fails_if_bucket_ownership_changed_between_start_and_completion_pending_transition_case) {
+    test_cond_with_2_gets_sent([&](auto& cond) {
+        cond.handle_reply(_sender, make_matched_reply(0));
+        simulate_set_pending_cluster_state("version:2 storage:1 distributor:1 .0.s:d"); // technically, no distributors own anything
+        cond.handle_reply(_sender, make_matched_reply(1));
+    }, [&](auto& outcome) {
+        EXPECT_FALSE(outcome.matched_condition());
+        EXPECT_FALSE(outcome.not_found());
+        EXPECT_TRUE(outcome.failed());
+        EXPECT_EQ(outcome.error_code().getResult(), api::ReturnCode::BUCKET_NOT_FOUND);
+    });
+}
+
+TEST_F(CheckConditionTest, check_fails_if_bucket_ownership_changed_between_start_and_completion_completed_transition_case) {
     test_cond_with_2_gets_sent([&](auto& cond) {
         cond.handle_reply(_sender, make_matched_reply(0));
         enable_cluster_state("version:2 storage:1 distributor:1 .0.s:d"); // technically, no distributors own anything
diff --git a/storage/src/vespa/storage/distributor/operations/external/check_condition.cpp b/storage/src/vespa/storage/distributor/operations/external/check_condition.cpp
index c5f4788ec51..0e12e3e3019 100644
--- a/storage/src/vespa/storage/distributor/operations/external/check_condition.cpp
+++ b/storage/src/vespa/storage/distributor/operations/external/check_condition.cpp
@@ -163,7 +163,10 @@ void CheckCondition::handle_internal_get_operation_reply(std::shared_ptr<api::St
                              reply->steal_trace());
             return;
         }
-        const auto state_version_now = _bucket_space.getClusterState().getVersion();
+        auto state_version_now = _bucket_space.getClusterState().getVersion();
+        if (_bucket_space.has_pending_cluster_state()) {
+            state_version_now = _bucket_space.get_pending_cluster_state().getVersion();
+        }
         if ((state_version_now != _cluster_state_version_at_creation_time)
             && (replica_set_changed_after_get_operation()
                 || distributor_no_longer_owns_bucket()))