Merge pull request #8882 from vespa-engine/vekterli/add-read-only-support-during-cluster-state-transitions

Add read-only support during cluster state transitions

7 files changed, 543 insertions, 67 deletions

diff --git a/storage/src/tests/distributor/bucketdbupdatertest.cpp b/storage/src/tests/distributor/bucketdbupdatertest.cpp
index b2d554c1e42..9795f5db5dc 100644
--- a/storage/src/tests/distributor/bucketdbupdatertest.cpp
+++ b/storage/src/tests/distributor/bucketdbupdatertest.cpp
@@ -26,6 +26,8 @@ using document::test::makeDocumentBucket;
 using document::test::makeBucketSpace;
 using document::BucketSpace;
 using document::FixedBucketSpaces;
+using document::BucketId;
+using document::Bucket;
 
 namespace storage::distributor {
 
@@ -112,6 +114,14 @@ class BucketDBUpdaterTest : public CppUnit::TestFixture,
     CPPUNIT_TEST(adding_diverging_replica_to_existing_trusted_does_not_remove_trusted);
     CPPUNIT_TEST(batch_update_from_distributor_change_does_not_mark_diverging_replicas_as_trusted);
     CPPUNIT_TEST(global_distribution_hash_falls_back_to_legacy_format_upon_request_rejection);
+    CPPUNIT_TEST(non_owned_buckets_moved_to_read_only_db_on_ownership_change);
+    CPPUNIT_TEST(buckets_no_longer_available_are_not_moved_to_read_only_database);
+    CPPUNIT_TEST(non_owned_buckets_purged_when_read_only_support_is_config_disabled);
+    CPPUNIT_TEST(deferred_activated_state_does_not_enable_state_until_activation_received);
+    CPPUNIT_TEST(read_only_db_cleared_once_pending_state_is_activated);
+    CPPUNIT_TEST(read_only_db_is_populated_even_when_self_is_marked_down);
+    CPPUNIT_TEST(activate_cluster_state_request_with_mismatching_version_returns_actual_version);
+    CPPUNIT_TEST(activate_cluster_state_request_without_pending_transition_passes_message_through);
     CPPUNIT_TEST_SUITE_END();
 
 public:
@@ -123,10 +133,7 @@ protected:
     void testDistributorChangeWithGrouping();
     void testNormalUsageInitializing();
     void testFailedRequestBucketInfo();
-    void testNoResponses();
     void testBitChange();
-    void testInconsistentChecksum();
-    void testAddEmptyNode();
     void testNodeDown();
     void testStorageNodeInMaintenanceClearsBucketsForNode();
     void testNodeDownCopiesGetInSync();
@@ -177,6 +184,14 @@ protected:
     void adding_diverging_replica_to_existing_trusted_does_not_remove_trusted();
     void batch_update_from_distributor_change_does_not_mark_diverging_replicas_as_trusted();
     void global_distribution_hash_falls_back_to_legacy_format_upon_request_rejection();
+    void non_owned_buckets_moved_to_read_only_db_on_ownership_change();
+    void buckets_no_longer_available_are_not_moved_to_read_only_database();
+    void non_owned_buckets_purged_when_read_only_support_is_config_disabled();
+    void deferred_activated_state_does_not_enable_state_until_activation_received();
+    void read_only_db_cleared_once_pending_state_is_activated();
+    void read_only_db_is_populated_even_when_self_is_marked_down();
+    void activate_cluster_state_request_with_mismatching_version_returns_actual_version();
+    void activate_cluster_state_request_without_pending_transition_passes_message_through();
 
     auto &defaultDistributorBucketSpace() { return getBucketSpaceRepo().get(makeBucketSpace()); }
 
@@ -190,13 +205,32 @@ protected:
                         getBucketDBUpdater().getDistributorComponent().getIndex(),
                         clusterStateBundle,
                         "ui"));
-        auto &repo = getBucketSpaceRepo();
-        for (auto &elem : repo) {
-            elem.second->setClusterState(clusterStateBundle.getDerivedClusterState(elem.first));
+        for (auto* repo : {&mutable_repo(), &read_only_repo()}) {
+            for (auto& space : *repo) {
+                space.second->setClusterState(clusterStateBundle.getDerivedClusterState(space.first));
+            }
         }
         return clusterInfo;
     }
 
+    DistributorBucketSpaceRepo& mutable_repo() noexcept { return getBucketSpaceRepo(); }
+    // Note: not calling this "immutable_repo" since it may actually be modified by the pending
+    // cluster state component (just not by operations), so it would not have the expected semantics.
+    DistributorBucketSpaceRepo& read_only_repo() noexcept { return getReadOnlyBucketSpaceRepo(); }
+
+    BucketDatabase& mutable_default_db() noexcept {
+        return mutable_repo().get(FixedBucketSpaces::default_space()).getBucketDatabase();
+    }
+    BucketDatabase& mutable_global_db() noexcept {
+        return mutable_repo().get(FixedBucketSpaces::global_space()).getBucketDatabase();
+    }
+    BucketDatabase& read_only_default_db() noexcept {
+        return read_only_repo().get(FixedBucketSpaces::default_space()).getBucketDatabase();
+    }
+    BucketDatabase& read_only_global_db() noexcept {
+        return read_only_repo().get(FixedBucketSpaces::global_space()).getBucketDatabase();
+    }
+
     static std::string getNodeList(std::vector<uint16_t> nodes, size_t count);
 
     std::string getNodeList(std::vector<uint16_t> nodes);
@@ -210,11 +244,17 @@ protected:
         return messagesPerBucketSpace * _bucketSpaces.size();
     }
 
+    void trigger_completed_but_not_yet_activated_transition(
+            vespalib::stringref initial_state, uint32_t initial_buckets, uint32_t initial_expected_msgs,
+            vespalib::stringref pending_state, uint32_t pending_buckets, uint32_t pending_expected_msgs);
+
 public:
     using OutdatedNodesMap = dbtransition::OutdatedNodesMap;
     void setUp() override {
         createLinks();
         _bucketSpaces = getBucketSpaces();
+        // Disable deferred activation by default (at least for now) to avoid breaking the entire world.
+        getConfig().setAllowStaleReadsDuringClusterStateTransitions(false);
     };
 
     void tearDown() override {
@@ -228,7 +268,7 @@ public:
             uint32_t bucketCount,
             uint32_t invalidBucketCount = 0)
     {
-        RequestBucketInfoReply* sreply = new RequestBucketInfoReply(cmd);
+        auto sreply = std::make_shared<RequestBucketInfoReply>(cmd);
         sreply->setAddress(storageAddress(storageIndex));
 
         api::RequestBucketInfoReply::EntryVector &vec = sreply->getBucketInfo();
@@ -261,7 +301,7 @@ public:
             }
         }
 
-        return std::shared_ptr<api::RequestBucketInfoReply>(sreply);
+        return sreply;
     }
 
     void fakeBucketReply(const lib::ClusterState &state,
@@ -371,8 +411,7 @@ public:
     void setSystemState(const lib::ClusterState& state) {
         const size_t sizeBeforeState = _sender.commands.size();
         getBucketDBUpdater().onSetSystemState(
-                std::shared_ptr<api::SetSystemStateCommand>(
-                        new api::SetSystemStateCommand(state)));
+                std::make_shared<api::SetSystemStateCommand>(state));
         // A lot of test logic has the assumption that all messages sent as a
         // result of cluster state changes will be in increasing index order
         // (for simplicity, not because this is required for correctness).
@@ -381,6 +420,26 @@ public:
         sortSentMessagesByIndex(_sender, sizeBeforeState);
     }
 
+    void set_cluster_state_bundle(const lib::ClusterStateBundle& state) {
+        const size_t sizeBeforeState = _sender.commands.size();
+        getBucketDBUpdater().onSetSystemState(
+                std::make_shared<api::SetSystemStateCommand>(state));
+        sortSentMessagesByIndex(_sender, sizeBeforeState);
+    }
+
+    bool activate_cluster_state_version(uint32_t version) {
+        return getBucketDBUpdater().onActivateClusterStateVersion(
+                std::make_shared<api::ActivateClusterStateVersionCommand>(version));
+    }
+    
+    void assert_has_activate_cluster_state_reply_with_actual_version(uint32_t version) {
+        CPPUNIT_ASSERT_EQUAL(size_t(1), _sender.replies.size());
+        auto* response = dynamic_cast<api::ActivateClusterStateVersionReply*>(_sender.replies.back().get());
+        CPPUNIT_ASSERT(response != nullptr);
+        CPPUNIT_ASSERT_EQUAL(version, response->actualVersion());
+        _sender.clear();
+    }
+
     void completeBucketInfoGathering(const lib::ClusterState& state,
                                      size_t expectedMsgs,
                                      uint32_t bucketCount = 1,
@@ -586,8 +645,9 @@ public:
 
             OutdatedNodesMap outdatedNodesMap;
             state = PendingClusterState::createForClusterStateChange(
-                    clock, clusterInfo, sender, owner.getBucketSpaceRepo(), cmd, outdatedNodesMap,
-                    api::Timestamp(1));
+                    clock, clusterInfo, sender,
+                    owner.getBucketSpaceRepo(), owner.getReadOnlyBucketSpaceRepo(),
+                    cmd, outdatedNodesMap, api::Timestamp(1));
         }
 
         PendingClusterStateFixture(
@@ -598,23 +658,22 @@ public:
                     owner.createClusterInfo(oldClusterState));
 
             state = PendingClusterState::createForDistributionChange(
-                    clock, clusterInfo, sender, owner.getBucketSpaceRepo(), api::Timestamp(1));
+                    clock, clusterInfo, sender, owner.getBucketSpaceRepo(),
+                    owner.getReadOnlyBucketSpaceRepo(), api::Timestamp(1));
         }
     };
 
-    auto createPendingStateFixtureForStateChange(
+    std::unique_ptr<PendingClusterStateFixture> createPendingStateFixtureForStateChange(
             const std::string& oldClusterState,
             const std::string& newClusterState)
     {
-        return std::make_unique<PendingClusterStateFixture>(
-                *this, oldClusterState, newClusterState);
+        return std::make_unique<PendingClusterStateFixture>(*this, oldClusterState, newClusterState);
     }
 
-    auto createPendingStateFixtureForDistributionChange(
+    std::unique_ptr<PendingClusterStateFixture> createPendingStateFixtureForDistributionChange(
             const std::string& oldClusterState)
     {
-        return std::make_unique<PendingClusterStateFixture>(
-                *this, oldClusterState);
+        return std::make_unique<PendingClusterStateFixture>(*this, oldClusterState);
     }
 };
 
@@ -622,8 +681,8 @@ CPPUNIT_TEST_SUITE_REGISTRATION(BucketDBUpdaterTest);
 
 BucketDBUpdaterTest::BucketDBUpdaterTest()
     : CppUnit::TestFixture(),
-    DistributorTestUtil(),
-    _bucketSpaces()
+      DistributorTestUtil(),
+      _bucketSpaces()
 {
 }
 
@@ -1533,7 +1592,8 @@ BucketDBUpdaterTest::getSentNodesDistributionChanged(
     ClusterInformation::CSP clusterInfo(createClusterInfo(oldClusterState));
     std::unique_ptr<PendingClusterState> state(
             PendingClusterState::createForDistributionChange(
-                    clock, clusterInfo, sender, getBucketSpaceRepo(), api::Timestamp(1)));
+                    clock, clusterInfo, sender, getBucketSpaceRepo(),
+                    getReadOnlyBucketSpaceRepo(), api::Timestamp(1)));
 
     sortSentMessagesByIndex(sender);
 
@@ -1698,8 +1758,8 @@ BucketDBUpdaterTest::testPendingClusterStateReceive()
     OutdatedNodesMap outdatedNodesMap;
     std::unique_ptr<PendingClusterState> state(
             PendingClusterState::createForClusterStateChange(
-                    clock, clusterInfo, sender, getBucketSpaceRepo(), cmd, outdatedNodesMap,
-                    api::Timestamp(1)));
+                    clock, clusterInfo, sender, getBucketSpaceRepo(), getReadOnlyBucketSpaceRepo(),
+                    cmd, outdatedNodesMap, api::Timestamp(1)));
 
     CPPUNIT_ASSERT_EQUAL(messageCount(3), sender.commands.size());
 
@@ -1863,8 +1923,8 @@ BucketDBUpdaterTest::mergeBucketLists(
         ClusterInformation::CSP clusterInfo(createClusterInfo("cluster:d"));
         std::unique_ptr<PendingClusterState> state(
                 PendingClusterState::createForClusterStateChange(
-                        clock, clusterInfo, sender, getBucketSpaceRepo(), cmd, outdatedNodesMap,
-                        beforeTime));
+                        clock, clusterInfo, sender, getBucketSpaceRepo(), getReadOnlyBucketSpaceRepo(),
+                        cmd, outdatedNodesMap, beforeTime));
 
         parseInputData(existingData, beforeTime, *state, includeBucketInfo);
         state->mergeIntoBucketDatabases();
@@ -1882,8 +1942,8 @@ BucketDBUpdaterTest::mergeBucketLists(
         ClusterInformation::CSP clusterInfo(createClusterInfo(oldState.toString()));
         std::unique_ptr<PendingClusterState> state(
                 PendingClusterState::createForClusterStateChange(
-                        clock, clusterInfo, sender, getBucketSpaceRepo(), cmd, outdatedNodesMap,
-                        afterTime));
+                        clock, clusterInfo, sender, getBucketSpaceRepo(), getReadOnlyBucketSpaceRepo(),
+                        cmd, outdatedNodesMap, afterTime));
 
         parseInputData(newData, afterTime, *state, includeBucketInfo);
         state->mergeIntoBucketDatabases();
@@ -2599,4 +2659,192 @@ void BucketDBUpdaterTest::global_distribution_hash_falls_back_to_legacy_format_u
     CPPUNIT_ASSERT_EQUAL(current_hash, new_current_req.getDistributionHash());
 }
 
+namespace {
+
+template <typename Func>
+void for_each_bucket(const BucketDatabase& db, const document::BucketSpace& space, Func&& f) {
+    BucketId last(0);
+    auto e = db.getNext(last);
+    while (e.valid()) {
+        f(space, e);
+        e = db.getNext(e.getBucketId());
+    }
+}
+
+template <typename Func>
+void for_each_bucket(const DistributorBucketSpaceRepo& repo, Func&& f) {
+    for (const auto& space : repo) {
+        for_each_bucket(space.second->getBucketDatabase(), space.first, f);
+    }
+}
+
+}
+
+using ConfigBuilder = vespa::config::content::core::StorDistributormanagerConfigBuilder;
+
+void BucketDBUpdaterTest::non_owned_buckets_moved_to_read_only_db_on_ownership_change() {
+    getConfig().setAllowStaleReadsDuringClusterStateTransitions(true);
+
+    lib::ClusterState initial_state("distributor:1 storage:4"); // All buckets owned by us by definition
+    set_cluster_state_bundle(lib::ClusterStateBundle(initial_state, {}, false)); // Skip activation step for simplicity
+
+    CPPUNIT_ASSERT_EQUAL(messageCount(4), _sender.commands.size());
+    constexpr uint32_t n_buckets = 10;
+    completeBucketInfoGathering(initial_state, messageCount(4), n_buckets);
+    _sender.clear();
+
+    CPPUNIT_ASSERT_EQUAL(size_t(n_buckets), mutable_default_db().size());
+    CPPUNIT_ASSERT_EQUAL(size_t(n_buckets), mutable_global_db().size());
+    CPPUNIT_ASSERT_EQUAL(size_t(0), read_only_default_db().size());
+    CPPUNIT_ASSERT_EQUAL(size_t(0), read_only_global_db().size());
+
+    lib::ClusterState pending_state("distributor:2 storage:4");
+
+    std::unordered_set<Bucket, Bucket::hash> buckets_not_owned_in_pending_state;
+    for_each_bucket(mutable_repo(), [&](const auto& space, const auto& entry) {
+        if (!getBucketDBUpdater().getDistributorComponent()
+                .ownsBucketInState(pending_state, makeDocumentBucket(entry.getBucketId()))) {
+            buckets_not_owned_in_pending_state.insert(Bucket(space, entry.getBucketId()));
+        }
+    });
+    CPPUNIT_ASSERT(!buckets_not_owned_in_pending_state.empty());
+
+    set_cluster_state_bundle(lib::ClusterStateBundle(pending_state, {}, true)); // Now requires activation
+
+    const auto buckets_not_owned_per_space = (buckets_not_owned_in_pending_state.size() / 2); // 2 spaces
+    const auto expected_mutable_buckets = n_buckets - buckets_not_owned_per_space;
+    CPPUNIT_ASSERT_EQUAL(expected_mutable_buckets, mutable_default_db().size());
+    CPPUNIT_ASSERT_EQUAL(expected_mutable_buckets, mutable_global_db().size());
+    CPPUNIT_ASSERT_EQUAL(buckets_not_owned_per_space, read_only_default_db().size());
+    CPPUNIT_ASSERT_EQUAL(buckets_not_owned_per_space, read_only_global_db().size());
+
+    for_each_bucket(read_only_repo(), [&](const auto& space, const auto& entry) {
+        CPPUNIT_ASSERT(buckets_not_owned_in_pending_state.find(Bucket(space, entry.getBucketId()))
+                       != buckets_not_owned_in_pending_state.end());
+    });
+}
+
+void BucketDBUpdaterTest::buckets_no_longer_available_are_not_moved_to_read_only_database() {
+    constexpr uint32_t n_buckets = 10;
+    // No ownership change, just node down. Test redundancy is 2, so removing 2 nodes will
+    // cause some buckets to be entirely unavailable.
+    trigger_completed_but_not_yet_activated_transition("version:1 distributor:1 storage:4", n_buckets, 4,
+                                                       "version:2 distributor:1 storage:4 .0.s:d .1.s:m", n_buckets, 0);
+
+    CPPUNIT_ASSERT_EQUAL(size_t(0), read_only_default_db().size());
+    CPPUNIT_ASSERT_EQUAL(size_t(0), read_only_global_db().size());
+}
+
+void BucketDBUpdaterTest::non_owned_buckets_purged_when_read_only_support_is_config_disabled() {
+    getConfig().setAllowStaleReadsDuringClusterStateTransitions(false);
+
+    lib::ClusterState initial_state("distributor:1 storage:4"); // All buckets owned by us by definition
+    set_cluster_state_bundle(lib::ClusterStateBundle(initial_state, {}, false)); // Skip activation step for simplicity
+
+    CPPUNIT_ASSERT_EQUAL(messageCount(4), _sender.commands.size());
+    constexpr uint32_t n_buckets = 10;
+    completeBucketInfoGathering(initial_state, messageCount(4), n_buckets);
+    _sender.clear();
+
+    // Nothing in read-only DB after first bulk load of buckets.
+    CPPUNIT_ASSERT_EQUAL(size_t(0), read_only_default_db().size());
+    CPPUNIT_ASSERT_EQUAL(size_t(0), read_only_global_db().size());
+
+    lib::ClusterState pending_state("distributor:2 storage:4");
+    setSystemState(pending_state);
+    // No buckets should be moved into read only db after ownership changes.
+    CPPUNIT_ASSERT_EQUAL(size_t(0), read_only_default_db().size());
+    CPPUNIT_ASSERT_EQUAL(size_t(0), read_only_global_db().size());
+}
+
+void BucketDBUpdaterTest::trigger_completed_but_not_yet_activated_transition(
+        vespalib::stringref initial_state_str,
+        uint32_t initial_buckets,
+        uint32_t initial_expected_msgs,
+        vespalib::stringref pending_state_str,
+        uint32_t pending_buckets,
+        uint32_t pending_expected_msgs)
+{
+    getConfig().setAllowStaleReadsDuringClusterStateTransitions(true);
+    lib::ClusterState initial_state(initial_state_str);
+    setSystemState(initial_state);
+    CPPUNIT_ASSERT_EQUAL(messageCount(initial_expected_msgs), _sender.commands.size());
+    completeBucketInfoGathering(initial_state, messageCount(initial_expected_msgs), initial_buckets);
+    _sender.clear();
+
+    lib::ClusterState pending_state(pending_state_str); // Ownership change
+    set_cluster_state_bundle(lib::ClusterStateBundle(pending_state, {}, true));
+    CPPUNIT_ASSERT_EQUAL(messageCount(pending_expected_msgs), _sender.commands.size());
+    completeBucketInfoGathering(pending_state, messageCount(pending_expected_msgs), pending_buckets);
+    _sender.clear();
+}
+
+void BucketDBUpdaterTest::deferred_activated_state_does_not_enable_state_until_activation_received() {
+    constexpr uint32_t n_buckets = 10;
+    trigger_completed_but_not_yet_activated_transition("version:1 distributor:2 storage:4", 0, 4,
+                                                       "version:2 distributor:1 storage:4", n_buckets, 4);
+
+    // Version should not be switched over yet
+    CPPUNIT_ASSERT_EQUAL(uint32_t(1), getDistributor().getClusterStateBundle().getVersion());
+
+    CPPUNIT_ASSERT_EQUAL(uint64_t(0), mutable_default_db().size());
+    CPPUNIT_ASSERT_EQUAL(uint64_t(0), mutable_global_db().size());
+
+    CPPUNIT_ASSERT(!activate_cluster_state_version(2));
+
+    CPPUNIT_ASSERT_EQUAL(uint32_t(2), getDistributor().getClusterStateBundle().getVersion());
+    CPPUNIT_ASSERT_EQUAL(uint64_t(n_buckets), mutable_default_db().size());
+    CPPUNIT_ASSERT_EQUAL(uint64_t(n_buckets), mutable_global_db().size());
+}
+
+void BucketDBUpdaterTest::read_only_db_cleared_once_pending_state_is_activated() {
+    constexpr uint32_t n_buckets = 10;
+    trigger_completed_but_not_yet_activated_transition("version:1 distributor:1 storage:4", n_buckets, 4,
+                                                       "version:2 distributor:2 storage:4", n_buckets, 0);
+    CPPUNIT_ASSERT(!activate_cluster_state_version(2));
+
+    CPPUNIT_ASSERT_EQUAL(uint64_t(0), read_only_default_db().size());
+    CPPUNIT_ASSERT_EQUAL(uint64_t(0), read_only_global_db().size());
+}
+
+void BucketDBUpdaterTest::read_only_db_is_populated_even_when_self_is_marked_down() {
+    constexpr uint32_t n_buckets = 10;
+    trigger_completed_but_not_yet_activated_transition("version:1 distributor:1 storage:4", n_buckets, 4,
+                                                       "version:2 distributor:1 .0.s:d storage:4", n_buckets, 0);
+
+    // State not yet activated, so read-only DBs have got all the buckets we used to have.
+    CPPUNIT_ASSERT_EQUAL(uint64_t(0), mutable_default_db().size());
+    CPPUNIT_ASSERT_EQUAL(uint64_t(0), mutable_global_db().size());
+    CPPUNIT_ASSERT_EQUAL(uint64_t(n_buckets), read_only_default_db().size());
+    CPPUNIT_ASSERT_EQUAL(uint64_t(n_buckets), read_only_global_db().size());
+}
+
+void BucketDBUpdaterTest::activate_cluster_state_request_with_mismatching_version_returns_actual_version() {
+    constexpr uint32_t n_buckets = 10;
+    trigger_completed_but_not_yet_activated_transition("version:4 distributor:1 storage:4", n_buckets, 4,
+                                                       "version:5 distributor:2 storage:4", n_buckets, 0);
+
+    CPPUNIT_ASSERT(activate_cluster_state_version(4)); // Too old version
+    assert_has_activate_cluster_state_reply_with_actual_version(5);
+
+    CPPUNIT_ASSERT(activate_cluster_state_version(6)); // More recent version than what has been observed
+    assert_has_activate_cluster_state_reply_with_actual_version(5);
+}
+
+void BucketDBUpdaterTest::activate_cluster_state_request_without_pending_transition_passes_message_through() {
+    constexpr uint32_t n_buckets = 10;
+    trigger_completed_but_not_yet_activated_transition("version:1 distributor:2 storage:4", 0, 4,
+                                                       "version:2 distributor:1 storage:4", n_buckets, 4);
+    // Activate version 2; no pending cluster state after this.
+    CPPUNIT_ASSERT(!activate_cluster_state_version(2));
+
+    // No pending cluster state for version 3, just passed through to be implicitly bounced by state manager.
+    // Note: state manager is not modelled in this test, so we just check that the message handler returns
+    // false (meaning "didn't take message ownership") and there's no auto-generated reply.
+    CPPUNIT_ASSERT(!activate_cluster_state_version(3));
+    CPPUNIT_ASSERT_EQUAL(size_t(0), _sender.replies.size());
+}
+
+// TODO rename distributor config to imply two phase functionlity explicitly?
+
 }
diff --git a/storage/src/tests/distributor/distributortestutil.cpp b/storage/src/tests/distributor/distributortestutil.cpp
index d3496d0c9f6..3f7f2eac63a 100644
--- a/storage/src/tests/distributor/distributortestutil.cpp
+++ b/storage/src/tests/distributor/distributortestutil.cpp
@@ -388,6 +388,16 @@ DistributorTestUtil::getBucketSpaceRepo() const {
     return _distributor->getBucketSpaceRepo();
 }
 
+DistributorBucketSpaceRepo &
+DistributorTestUtil::getReadOnlyBucketSpaceRepo() {
+    return _distributor->getReadOnlyBucketSpaceRepo();
+}
+
+const DistributorBucketSpaceRepo &
+DistributorTestUtil::getReadOnlyBucketSpaceRepo() const {
+    return _distributor->getReadOnlyBucketSpaceRepo();
+}
+
 const lib::Distribution&
 DistributorTestUtil::getDistribution() const {
     return getBucketSpaceRepo().get(makeBucketSpace()).getDistribution();
diff --git a/storage/src/tests/distributor/distributortestutil.h b/storage/src/tests/distributor/distributortestutil.h
index 10cc5eeaca1..420111437d2 100644
--- a/storage/src/tests/distributor/distributortestutil.h
+++ b/storage/src/tests/distributor/distributortestutil.h
@@ -132,6 +132,8 @@ public:
     const BucketDatabase& getBucketDatabase(document::BucketSpace space) const;
     DistributorBucketSpaceRepo &getBucketSpaceRepo();
     const DistributorBucketSpaceRepo &getBucketSpaceRepo() const;
+    DistributorBucketSpaceRepo& getReadOnlyBucketSpaceRepo();
+    const DistributorBucketSpaceRepo& getReadOnlyBucketSpaceRepo() const;
 
     const lib::Distribution& getDistribution() const;
     // "End to end" distribution change trigger, which will invoke the bucket
diff --git a/storage/src/tests/distributor/externaloperationhandlertest.cpp b/storage/src/tests/distributor/externaloperationhandlertest.cpp
index ddf88f50c36..40fe885dcb1 100644
--- a/storage/src/tests/distributor/externaloperationhandlertest.cpp
+++ b/storage/src/tests/distributor/externaloperationhandlertest.cpp
@@ -4,6 +4,7 @@
 #include <vespa/storage/distributor/externaloperationhandler.h>
 #include <vespa/storage/distributor/distributor.h>
 #include <vespa/storage/distributor/distributormetricsset.h>
+#include <vespa/storage/distributor/operations/external/getoperation.h>
 #include <vespa/storageapi/message/persistence.h>
 #include <vespa/document/repo/documenttyperepo.h>
 #include <vespa/document/update/documentupdate.h>
@@ -20,8 +21,11 @@ class ExternalOperationHandlerTest : public CppUnit::TestFixture,
 
     CPPUNIT_TEST_SUITE(ExternalOperationHandlerTest);
     CPPUNIT_TEST(testBucketSplitMask);
-    CPPUNIT_TEST(testOperationRejectedOnWrongDistribution);
-    CPPUNIT_TEST(testOperationRejectedOnPendingWrongDistribution);
+    CPPUNIT_TEST(mutating_operation_wdr_bounced_on_wrong_current_distribution);
+    CPPUNIT_TEST(mutating_operation_busy_bounced_on_wrong_pending_distribution);
+    CPPUNIT_TEST(mutating_operation_busy_bounced_if_no_cluster_state_received_yet);
+    CPPUNIT_TEST(read_only_operation_wdr_bounced_on_wrong_current_distribution);
+    CPPUNIT_TEST(read_only_operation_busy_bounced_if_no_cluster_state_received_yet);
     CPPUNIT_TEST(reject_put_if_not_past_safe_time_point);
     CPPUNIT_TEST(reject_remove_if_not_past_safe_time_point);
     CPPUNIT_TEST(reject_update_if_not_past_safe_time_point);
@@ -37,6 +41,9 @@ class ExternalOperationHandlerTest : public CppUnit::TestFixture,
     CPPUNIT_TEST(concurrent_get_and_mutation_do_not_conflict);
     CPPUNIT_TEST(sequencing_works_across_mutation_types);
     CPPUNIT_TEST(sequencing_can_be_explicitly_config_disabled);
+    CPPUNIT_TEST(gets_are_started_with_mutable_db_outside_transition_period);
+    CPPUNIT_TEST(gets_are_started_with_read_only_db_during_transition_period);
+    CPPUNIT_TEST(gets_are_busy_bounced_during_transition_period_if_stale_reads_disabled);
     CPPUNIT_TEST_SUITE_END();
 
     document::BucketId findNonOwnedUserBucketInState(vespalib::stringref state);
@@ -49,10 +56,13 @@ class ExternalOperationHandlerTest : public CppUnit::TestFixture,
     std::shared_ptr<api::UpdateCommand> makeUpdateCommand(const vespalib::string& doc_type,
                                                           const vespalib::string& id) const;
     std::shared_ptr<api::UpdateCommand> makeUpdateCommand() const;
+    std::shared_ptr<api::UpdateCommand> makeUpdateCommandForUser(uint64_t id) const;
     std::shared_ptr<api::PutCommand> makePutCommand(const vespalib::string& doc_type,
                                                     const vespalib::string& id) const;
     std::shared_ptr<api::RemoveCommand> makeRemoveCommand(const vespalib::string& id) const;
 
+    void verify_busy_bounced_due_to_no_active_state(std::shared_ptr<api::StorageCommand> cmd);
+
     Operation::SP start_operation_verify_not_rejected(std::shared_ptr<api::StorageCommand> cmd);
     void start_operation_verify_rejected(std::shared_ptr<api::StorageCommand> cmd);
 
@@ -80,10 +90,16 @@ class ExternalOperationHandlerTest : public CppUnit::TestFixture,
 
     const vespalib::string _dummy_id{"id:foo:testdoctype1::bar"};
 
+    // Returns an arbitrary bucket not owned in the pending state
+    document::BucketId set_up_pending_cluster_state_transition(bool read_only_enabled);
+
 protected:
     void testBucketSplitMask();
-    void testOperationRejectedOnWrongDistribution();
-    void testOperationRejectedOnPendingWrongDistribution();
+    void mutating_operation_wdr_bounced_on_wrong_current_distribution();
+    void mutating_operation_busy_bounced_on_wrong_pending_distribution();
+    void mutating_operation_busy_bounced_if_no_cluster_state_received_yet();
+    void read_only_operation_wdr_bounced_on_wrong_current_distribution();
+    void read_only_operation_busy_bounced_if_no_cluster_state_received_yet();
     void reject_put_if_not_past_safe_time_point();
     void reject_remove_if_not_past_safe_time_point();
     void reject_update_if_not_past_safe_time_point();
@@ -99,6 +115,9 @@ protected:
     void concurrent_get_and_mutation_do_not_conflict();
     void sequencing_works_across_mutation_types();
     void sequencing_can_be_explicitly_config_disabled();
+    void gets_are_started_with_mutable_db_outside_transition_period();
+    void gets_are_started_with_read_only_db_during_transition_period();
+    void gets_are_busy_bounced_during_transition_period_if_stale_reads_disabled();
 
     void assert_rejection_due_to_unsafe_time(
             std::shared_ptr<api::StorageCommand> cmd);
@@ -220,6 +239,11 @@ ExternalOperationHandlerTest::makeUpdateCommand() const {
     return makeUpdateCommand("testdoctype1", "id:foo:testdoctype1::baz");
 }
 
+std::shared_ptr<api::UpdateCommand>
+ExternalOperationHandlerTest::makeUpdateCommandForUser(uint64_t id) const {
+    return makeUpdateCommand("testdoctype1", vespalib::make_string("id::testdoctype1:n=%" PRIu64 ":bar", id));
+}
+
 std::shared_ptr<api::PutCommand> ExternalOperationHandlerTest::makePutCommand(
         const vespalib::string& doc_type,
         const vespalib::string& id) const {
@@ -233,10 +257,30 @@ std::shared_ptr<api::RemoveCommand> ExternalOperationHandlerTest::makeRemoveComm
 }
 
 void
-ExternalOperationHandlerTest::testOperationRejectedOnWrongDistribution()
+ExternalOperationHandlerTest::mutating_operation_wdr_bounced_on_wrong_current_distribution()
 {
     createLinks();
-    std::string state("distributor:2 storage:2");
+    std::string state("version:1 distributor:2 storage:2");
+    setupDistributor(1, 2, state);
+
+    document::BucketId bucket(findNonOwnedUserBucketInState(state));
+    auto cmd = makeUpdateCommandForUser(bucket.withoutCountBits());
+
+    Operation::SP genOp;
+    CPPUNIT_ASSERT(getExternalOperationHandler().handleMessage(cmd, genOp));
+    CPPUNIT_ASSERT(!genOp.get());
+    CPPUNIT_ASSERT_EQUAL(size_t(1), _sender.replies.size());
+    CPPUNIT_ASSERT_EQUAL(
+            std::string("ReturnCode(WRONG_DISTRIBUTION, "
+                        "version:1 distributor:2 storage:2)"),
+            _sender.replies[0]->getResult().toString());
+}
+
+void
+ExternalOperationHandlerTest::read_only_operation_wdr_bounced_on_wrong_current_distribution()
+{
+    createLinks();
+    std::string state("version:1 distributor:2 storage:2");
     setupDistributor(1, 2, state);
 
     document::BucketId bucket(findNonOwnedUserBucketInState(state));
@@ -248,43 +292,65 @@ ExternalOperationHandlerTest::testOperationRejectedOnWrongDistribution()
     CPPUNIT_ASSERT_EQUAL(size_t(1), _sender.replies.size());
     CPPUNIT_ASSERT_EQUAL(
             std::string("ReturnCode(WRONG_DISTRIBUTION, "
-                        "distributor:2 storage:2)"),
+                        "version:1 distributor:2 storage:2)"),
             _sender.replies[0]->getResult().toString());
 }
 
 void
-ExternalOperationHandlerTest::testOperationRejectedOnPendingWrongDistribution()
+ExternalOperationHandlerTest::mutating_operation_busy_bounced_on_wrong_pending_distribution()
 {
     createLinks();
-    std::string current("distributor:2 storage:2");
-    std::string pending("distributor:3 storage:3");
+    std::string current("version:10 distributor:2 storage:2");
+    std::string pending("version:11 distributor:3 storage:3");
     setupDistributor(1, 3, current);
 
     document::BucketId b(findOwned1stNotOwned2ndInStates(current, pending));
 
     // Trigger pending cluster state
-    auto stateCmd = std::make_shared<api::SetSystemStateCommand>(
-            lib::ClusterState(pending));
+    auto stateCmd = std::make_shared<api::SetSystemStateCommand>(lib::ClusterState(pending));
     getBucketDBUpdater().onSetSystemState(stateCmd);
 
-    auto cmd = makeGetCommandForUser(b.withoutCountBits());
+    auto cmd = makeUpdateCommandForUser(b.withoutCountBits());
 
     Operation::SP genOp;
     CPPUNIT_ASSERT(getExternalOperationHandler().handleMessage(cmd, genOp));
     CPPUNIT_ASSERT(!genOp.get());
     CPPUNIT_ASSERT_EQUAL(size_t(1), _sender.replies.size());
-    // Fail back with _pending_ cluster state so client can start trying
-    // correct distributor immediately. If that distributor has not yet
-    // completed processing its pending cluster state, it'll return the
-    // old (current) cluster state, causing the client to bounce between
-    // the two until the pending states have been resolved. This is pretty
-    // much inevitable with the current design.
     CPPUNIT_ASSERT_EQUAL(
-            std::string("ReturnCode(WRONG_DISTRIBUTION, "
-                        "distributor:3 storage:3)"),
+            std::string("ReturnCode(BUSY, Currently pending cluster state transition from version 10 to 11)"),
             _sender.replies[0]->getResult().toString());
 }
 
+void
+ExternalOperationHandlerTest::verify_busy_bounced_due_to_no_active_state(std::shared_ptr<api::StorageCommand> cmd)
+{
+    createLinks();
+    std::string state{}; // No version --> not yet received
+    setupDistributor(1, 2, state);
+
+    Operation::SP genOp;
+    CPPUNIT_ASSERT(getExternalOperationHandler().handleMessage(cmd, genOp));
+    CPPUNIT_ASSERT(!genOp.get());
+    CPPUNIT_ASSERT_EQUAL(size_t(1), _sender.replies.size());
+    CPPUNIT_ASSERT_EQUAL(
+            std::string("ReturnCode(BUSY, No cluster state activated yet)"),
+            _sender.replies[0]->getResult().toString());
+}
+
+// TODO NOT_READY is a more appropriate return code for this case, but must ensure it's
+// handled gracefully and silently through the stack. BUSY is a safe bet until then.
+void
+ExternalOperationHandlerTest::mutating_operation_busy_bounced_if_no_cluster_state_received_yet()
+{
+    verify_busy_bounced_due_to_no_active_state(makeUpdateCommandForUser(12345));
+}
+
+void
+ExternalOperationHandlerTest::read_only_operation_busy_bounced_if_no_cluster_state_received_yet()
+{
+    verify_busy_bounced_due_to_no_active_state(makeGetCommandForUser(12345));
+}
+
 using TimePoint = ExternalOperationHandler::TimePoint;
 using namespace std::literals::chrono_literals;
 
@@ -292,7 +358,7 @@ void ExternalOperationHandlerTest::assert_rejection_due_to_unsafe_time(
         std::shared_ptr<api::StorageCommand> cmd)
 {
     createLinks();
-    setupDistributor(1, 2, "distributor:1 storage:1");
+    setupDistributor(1, 2, "version:1 distributor:1 storage:1");
     getClock().setAbsoluteTimeInSeconds(9);
     getExternalOperationHandler().rejectFeedBeforeTimeReached(TimePoint(10s));
 
@@ -327,7 +393,7 @@ void ExternalOperationHandlerTest::reject_update_if_not_past_safe_time_point() {
 
 void ExternalOperationHandlerTest::get_not_rejected_by_unsafe_time_point() {
     createLinks();
-    setupDistributor(1, 2, "distributor:1 storage:1");
+    setupDistributor(1, 2, "version:1 distributor:1 storage:1");
     getClock().setAbsoluteTimeInSeconds(9);
     getExternalOperationHandler().rejectFeedBeforeTimeReached(TimePoint(10s));
 
@@ -342,7 +408,7 @@ void ExternalOperationHandlerTest::get_not_rejected_by_unsafe_time_point() {
 
 void ExternalOperationHandlerTest::mutation_not_rejected_when_safe_point_reached() {
     createLinks();
-    setupDistributor(1, 2, "distributor:1 storage:1");
+    setupDistributor(1, 2, "version:1 distributor:1 storage:1");
     getClock().setAbsoluteTimeInSeconds(10);
     getExternalOperationHandler().rejectFeedBeforeTimeReached(TimePoint(10s));
 
@@ -360,7 +426,7 @@ void ExternalOperationHandlerTest::mutation_not_rejected_when_safe_point_reached
 
 void ExternalOperationHandlerTest::set_up_distributor_for_sequencing_test() {
     createLinks();
-    setupDistributor(1, 2, "distributor:1 storage:1");
+    setupDistributor(1, 2, "version:1 distributor:1 storage:1");
 }
 
 Operation::SP ExternalOperationHandlerTest::start_operation_verify_not_rejected(
@@ -486,6 +552,52 @@ void ExternalOperationHandlerTest::sequencing_can_be_explicitly_config_disabled(
     start_operation_verify_not_rejected(makeRemoveCommand(_dummy_id));
 }
 
+void ExternalOperationHandlerTest::gets_are_started_with_mutable_db_outside_transition_period() {
+    createLinks();
+    std::string current = "version:1 distributor:1 storage:3";
+    setupDistributor(1, 3, current);
+    getConfig().setAllowStaleReadsDuringClusterStateTransitions(true);
+
+    document::BucketId b(16, 1234); // Only 1 distributor (us), so doesn't matter
+
+    auto op = start_operation_verify_not_rejected(makeGetCommandForUser(b.withoutCountBits()));
+    auto& get_op = dynamic_cast<GetOperation&>(*op);
+    const auto* expected_space = &getBucketSpaceRepo().get(document::FixedBucketSpaces::default_space());
+    CPPUNIT_ASSERT_EQUAL(expected_space, &get_op.bucketSpace());
+}
+
+document::BucketId ExternalOperationHandlerTest::set_up_pending_cluster_state_transition(bool read_only_enabled) {
+    createLinks();
+    std::string current = "version:123 distributor:2 storage:2";
+    std::string pending = "version:321 distributor:3 storage:3";
+    setupDistributor(1, 3, current);
+    getConfig().setAllowStaleReadsDuringClusterStateTransitions(read_only_enabled);
+
+    // Trigger pending cluster state
+    auto stateCmd = std::make_shared<api::SetSystemStateCommand>(lib::ClusterState(pending));
+    getBucketDBUpdater().onSetSystemState(stateCmd);
+    return findOwned1stNotOwned2ndInStates(current, pending);
+}
+
+void ExternalOperationHandlerTest::gets_are_started_with_read_only_db_during_transition_period() {
+    auto non_owned_bucket = set_up_pending_cluster_state_transition(true);
+
+    auto op = start_operation_verify_not_rejected(makeGetCommandForUser(non_owned_bucket.withoutCountBits()));
+    auto& get_op = dynamic_cast<GetOperation&>(*op);
+    const auto* expected_space = &getReadOnlyBucketSpaceRepo().get(document::FixedBucketSpaces::default_space());
+    CPPUNIT_ASSERT_EQUAL(expected_space, &get_op.bucketSpace());
+}
+
+void ExternalOperationHandlerTest::gets_are_busy_bounced_during_transition_period_if_stale_reads_disabled() {
+    auto non_owned_bucket = set_up_pending_cluster_state_transition(false);
+
+    start_operation_verify_rejected(makeGetCommandForUser(non_owned_bucket.withoutCountBits()));
+    CPPUNIT_ASSERT_EQUAL(
+            std::string("ReturnCode(BUSY, Currently pending cluster state transition from version 123 to 321)"),
+            _sender.replies[0]->getResult().toString());
+
+}
+
 // TODO support sequencing of RemoveLocation? It's a mutating operation, but supporting it with
 // the current approach is not trivial. A RemoveLocation operation covers the _entire_ bucket
 // sub tree under a given location, while the sequencer works on individual GIDs. Mapping the
diff --git a/storage/src/tests/storageserver/bouncertest.cpp b/storage/src/tests/storageserver/bouncertest.cpp
index 27c13a3707e..371c24accbc 100644
--- a/storage/src/tests/storageserver/bouncertest.cpp
+++ b/storage/src/tests/storageserver/bouncertest.cpp
@@ -43,6 +43,7 @@ struct BouncerTest : public CppUnit::TestFixture {
     void outOfBoundsConfigValuesThrowException();
     void abort_request_when_derived_bucket_space_node_state_is_marked_down();
     void client_operations_are_allowed_through_on_cluster_state_down_distributor();
+    void cluster_state_activation_commands_are_not_bounced();
 
     CPPUNIT_TEST_SUITE(BouncerTest);
     CPPUNIT_TEST(testFutureTimestamp);
@@ -57,6 +58,7 @@ struct BouncerTest : public CppUnit::TestFixture {
     CPPUNIT_TEST(outOfBoundsConfigValuesThrowException);
     CPPUNIT_TEST(abort_request_when_derived_bucket_space_node_state_is_marked_down);
     CPPUNIT_TEST(client_operations_are_allowed_through_on_cluster_state_down_distributor);
+    CPPUNIT_TEST(cluster_state_activation_commands_are_not_bounced);
     CPPUNIT_TEST_SUITE_END();
 
     using Priority = api::StorageMessage::Priority;
@@ -368,5 +370,17 @@ void BouncerTest::client_operations_are_allowed_through_on_cluster_state_down_di
     CPPUNIT_ASSERT_EQUAL(uint64_t(0), _manager->metrics().unavailable_node_aborts.getValue());
 }
 
+void BouncerTest::cluster_state_activation_commands_are_not_bounced() {
+    tearDown();
+    setUpAsNode(lib::NodeType::DISTRIBUTOR);
+
+    auto state = makeClusterStateBundle("version:10 distributor:3 .2.s:d storage:3", {}); // Our index (2) is down
+    _node->getNodeStateUpdater().setClusterStateBundle(state);
+
+    auto activate_cmd = std::make_shared<api::ActivateClusterStateVersionCommand>(11);
+    _upper->sendDown(activate_cmd);
+    assertMessageNotBounced();
+}
+
 } // storage
 
diff --git a/storage/src/tests/storageserver/fnet_listener_test.cpp b/storage/src/tests/storageserver/fnet_listener_test.cpp
index 84051041d25..d40b230d725 100644
--- a/storage/src/tests/storageserver/fnet_listener_test.cpp
+++ b/storage/src/tests/storageserver/fnet_listener_test.cpp
@@ -27,6 +27,9 @@ public:
     CPPUNIT_TEST(set_distribution_rpc_is_immediately_failed_if_listener_is_closed);
     CPPUNIT_TEST(overly_large_uncompressed_bundle_size_parameter_returns_rpc_error);
     CPPUNIT_TEST(mismatching_uncompressed_bundle_size_parameter_returns_rpc_error);
+    CPPUNIT_TEST(true_deferred_activation_flag_can_be_roundtrip_encoded);
+    CPPUNIT_TEST(false_deferred_activation_flag_can_be_roundtrip_encoded);
+    CPPUNIT_TEST(activate_cluster_state_version_rpc_enqueues_command_with_version);
     CPPUNIT_TEST_SUITE_END();
 
     void baseline_set_distribution_states_rpc_enqueues_command_with_state_bundle();
@@ -35,6 +38,9 @@ public:
     void set_distribution_rpc_is_immediately_failed_if_listener_is_closed();
     void overly_large_uncompressed_bundle_size_parameter_returns_rpc_error();
     void mismatching_uncompressed_bundle_size_parameter_returns_rpc_error();
+    void true_deferred_activation_flag_can_be_roundtrip_encoded();
+    void false_deferred_activation_flag_can_be_roundtrip_encoded();
+    void activate_cluster_state_version_rpc_enqueues_command_with_version();
 };
 
 CPPUNIT_TEST_SUITE_REGISTRATION(FNetListenerTest);
@@ -54,24 +60,25 @@ struct DummyReturnHandler : FRT_IReturnHandler {
     FNET_Connection* GetConnection() override { return nullptr; }
 };
 
-struct Fixture {
+struct FixtureBase {
     // TODO factor out Slobrok code to avoid need to set up live ports for unrelated tests
     mbus::Slobrok slobrok;
     vdstestlib::DirConfig config;
     MockOperationEnqueuer enqueuer;
     std::unique_ptr<FNetListener> fnet_listener;
-    SlimeClusterStateBundleCodec codec;
     DummyReturnHandler return_handler;
     bool request_is_detached{false};
     FRT_RPCRequest* bound_request{nullptr};
 
-    Fixture() : config(getStandardConfig(true)) {
+    FixtureBase()
+        : config(getStandardConfig(true))
+    {
         config.getConfig("stor-server").set("node_index", "1");
         addSlobrokConfig(config, slobrok);
         fnet_listener = std::make_unique<FNetListener>(enqueuer, config.getConfigId(), 0);
     }
 
-    ~Fixture() {
+    virtual ~FixtureBase() {
         // Must destroy any associated message contexts that may have refs to FRT_Request
         // instance _before_ we destroy the request itself.
         enqueuer._enqueued.clear();
@@ -79,6 +86,12 @@ struct Fixture {
             bound_request->SubRef();
         }
     }
+};
+
+struct SetStateFixture : FixtureBase {
+    SlimeClusterStateBundleCodec codec;
+
+    SetStateFixture() : FixtureBase() {}
 
     void bind_request_params(EncodedClusterStateBundle& encoded_bundle, uint32_t uncompressed_length) {
         bound_request = new FRT_RPCRequest(); // Naked new isn't pretty, but FRT_RPCRequest has internal refcounting
@@ -123,6 +136,10 @@ struct Fixture {
     lib::ClusterStateBundle dummy_baseline_bundle() const {
         return lib::ClusterStateBundle(lib::ClusterState("version:123 distributor:3 storage:3"));
     }
+
+    lib::ClusterStateBundle dummy_baseline_bundle_with_deferred_activation(bool deferred) const {
+        return lib::ClusterStateBundle(lib::ClusterState("version:123 distributor:3 storage:3"), {}, deferred);
+    }
 };
 
 std::shared_ptr<const lib::ClusterState> state_of(vespalib::stringref state) {
@@ -138,17 +155,17 @@ vespalib::string make_compressable_state_string() {
                                  ss.str().data(), ss.str().data());
 }
 
-}
+} // anon namespace
 
 void FNetListenerTest::baseline_set_distribution_states_rpc_enqueues_command_with_state_bundle() {
-    Fixture f;
+    SetStateFixture f;
     auto baseline = f.dummy_baseline_bundle();
 
     f.assert_request_received_and_propagated(baseline);
 }
 
 void FNetListenerTest::set_distribution_states_rpc_with_derived_enqueues_command_with_state_bundle() {
-    Fixture f;
+    SetStateFixture f;
     lib::ClusterStateBundle spaces_bundle(
             lib::ClusterState("version:123 distributor:3 storage:3"),
             {{FixedBucketSpaces::default_space(), state_of("version:123 distributor:3 storage:3 .0.s:d")},
@@ -158,7 +175,7 @@ void FNetListenerTest::set_distribution_states_rpc_with_derived_enqueues_command
 }
 
 void FNetListenerTest::compressed_bundle_is_transparently_uncompressed() {
-    Fixture f;
+    SetStateFixture f;
     auto state_str = make_compressable_state_string();
     lib::ClusterStateBundle compressable_bundle{lib::ClusterState(state_str)};
 
@@ -171,24 +188,73 @@ void FNetListenerTest::compressed_bundle_is_transparently_uncompressed() {
 }
 
 void FNetListenerTest::set_distribution_rpc_is_immediately_failed_if_listener_is_closed() {
-    Fixture f;
+    SetStateFixture f;
     f.create_request(f.dummy_baseline_bundle());
     f.fnet_listener->close();
     f.assert_request_returns_error_response(RPCRequestWrapper::ERR_NODE_SHUTTING_DOWN);
 }
 
 void FNetListenerTest::overly_large_uncompressed_bundle_size_parameter_returns_rpc_error() {
-    Fixture f;
+    SetStateFixture f;
     auto encoded_bundle = f.codec.encode(f.dummy_baseline_bundle());
     f.bind_request_params(encoded_bundle, FNetListener::StateBundleMaxUncompressedSize + 1);
     f.assert_request_returns_error_response(RPCRequestWrapper::ERR_BAD_REQUEST);
 }
 
 void FNetListenerTest::mismatching_uncompressed_bundle_size_parameter_returns_rpc_error() {
-    Fixture f;
+    SetStateFixture f;
     auto encoded_bundle = f.codec.encode(f.dummy_baseline_bundle());
     f.bind_request_params(encoded_bundle, encoded_bundle._buffer->getDataLen() + 100);
     f.assert_request_returns_error_response(RPCRequestWrapper::ERR_BAD_REQUEST);
 }
 
+void FNetListenerTest::true_deferred_activation_flag_can_be_roundtrip_encoded() {
+    SetStateFixture f;
+    f.assert_request_received_and_propagated(f.dummy_baseline_bundle_with_deferred_activation(true));
+
+}
+
+void FNetListenerTest::false_deferred_activation_flag_can_be_roundtrip_encoded() {
+    SetStateFixture f;
+    f.assert_request_received_and_propagated(f.dummy_baseline_bundle_with_deferred_activation(false));
+}
+
+struct ActivateStateFixture : FixtureBase {
+    ActivateStateFixture() : FixtureBase() {}
+
+    void bind_request_params(uint32_t activate_version) {
+        bound_request = new FRT_RPCRequest(); // Naked new isn't pretty, but FRT_RPCRequest has internal refcounting
+        auto* params = bound_request->GetParams();
+        params->AddInt32(activate_version);
+
+        bound_request->SetDetachedPT(&request_is_detached);
+        bound_request->SetReturnHandler(&return_handler);
+    }
+
+    void create_request(uint32_t activate_version) {
+        // Only 1 request allowed per fixture due to lifetime handling snags
+        assert(bound_request == nullptr);
+        bind_request_params(activate_version);
+    }
+
+    void assert_enqueued_operation_has_activate_version(uint32_t version) {
+        CPPUNIT_ASSERT(bound_request != nullptr);
+        CPPUNIT_ASSERT(request_is_detached);
+        CPPUNIT_ASSERT_EQUAL(size_t(1), enqueuer._enqueued.size());
+        auto& state_request = dynamic_cast<const api::ActivateClusterStateVersionCommand&>(*enqueuer._enqueued[0]);
+        CPPUNIT_ASSERT_EQUAL(version, state_request.version());
+    }
+
+    void assert_request_received_and_propagated(uint32_t activate_version) {
+        create_request(activate_version);
+        fnet_listener->RPC_activateClusterStateVersion(bound_request);
+        assert_enqueued_operation_has_activate_version(activate_version);
+    }
+};
+
+void FNetListenerTest::activate_cluster_state_version_rpc_enqueues_command_with_version() {
+    ActivateStateFixture f;
+    f.assert_request_received_and_propagated(1234567);
+}
+
 }
diff --git a/storage/src/tests/storageserver/statemanagertest.cpp b/storage/src/tests/storageserver/statemanagertest.cpp
index 19f414482db..cdf990fa28f 100644
--- a/storage/src/tests/storageserver/statemanagertest.cpp
+++ b/storage/src/tests/storageserver/statemanagertest.cpp
@@ -37,6 +37,7 @@ struct StateManagerTest : public CppUnit::TestFixture {
     void can_explicitly_send_get_node_state_reply();
     void explicit_node_state_replying_without_pending_request_immediately_replies_on_next_request();
     void immediate_node_state_replying_is_tracked_per_controller();
+    void activation_command_is_bounced_with_current_cluster_state_version();
 
     CPPUNIT_TEST_SUITE(StateManagerTest);
     CPPUNIT_TEST(testSystemState);
@@ -45,8 +46,10 @@ struct StateManagerTest : public CppUnit::TestFixture {
     CPPUNIT_TEST(can_explicitly_send_get_node_state_reply);
     CPPUNIT_TEST(explicit_node_state_replying_without_pending_request_immediately_replies_on_next_request);
     CPPUNIT_TEST(immediate_node_state_replying_is_tracked_per_controller);
+    CPPUNIT_TEST(activation_command_is_bounced_with_current_cluster_state_version);
     CPPUNIT_TEST_SUITE_END();
 
+    void force_current_cluster_state_version(uint32_t version);
     void mark_reported_node_state_up();
     void send_down_get_node_state_request(uint16_t controller_index);
     void assert_ok_get_node_state_reply_sent_and_clear();
@@ -101,6 +104,12 @@ StateManagerTest::tearDown() {
     _metricManager.reset();
 }
 
+void StateManagerTest::force_current_cluster_state_version(uint32_t version) {
+    ClusterState state(*_manager->getClusterStateBundle()->getBaselineClusterState());
+    state.setVersion(version);
+    _manager->setClusterStateBundle(lib::ClusterStateBundle(state));
+}
+
 #define GET_ONLY_OK_REPLY(varname) \
 { \
     CPPUNIT_ASSERT_EQUAL(size_t(1), _upper->getNumReplies()); \
@@ -236,9 +245,7 @@ StateManagerTest::testReportedNodeState()
 }
 
 void StateManagerTest::current_cluster_state_version_is_included_in_host_info_json() {
-    ClusterState state(*_manager->getClusterStateBundle()->getBaselineClusterState());
-    state.setVersion(123);
-    _manager->setClusterStateBundle(lib::ClusterStateBundle(state));
+    force_current_cluster_state_version(123);
 
     std::string nodeInfoString(_manager->getNodeInfo());
     vespalib::Memory goldenMemory(nodeInfoString);
@@ -343,4 +350,21 @@ void StateManagerTest::immediate_node_state_replying_is_tracked_per_controller()
     CPPUNIT_ASSERT_EQUAL(size_t(0), _upper->getNumReplies());
 }
 
+void StateManagerTest::activation_command_is_bounced_with_current_cluster_state_version() {
+    force_current_cluster_state_version(12345);
+
+    auto cmd = std::make_shared<api::ActivateClusterStateVersionCommand>(12340);
+    cmd->setTimeout(10000000);
+    cmd->setSourceIndex(0);
+    _upper->sendDown(cmd);
+
+    CPPUNIT_ASSERT_EQUAL(size_t(1), _upper->getNumReplies());
+    std::shared_ptr<api::StorageReply> reply;
+    GET_ONLY_OK_REPLY(reply); // Implicitly clears messages from _upper
+    CPPUNIT_ASSERT_EQUAL(api::MessageType::ACTIVATE_CLUSTER_STATE_VERSION_REPLY, reply->getType());
+    auto& activate_reply = dynamic_cast<api::ActivateClusterStateVersionReply&>(*reply);
+    CPPUNIT_ASSERT_EQUAL(uint32_t(12340), activate_reply.activateVersion());
+    CPPUNIT_ASSERT_EQUAL(uint32_t(12345), activate_reply.actualVersion());
+}
+
 } // storage