diff options
Diffstat (limited to 'storage/src/tests')
-rw-r--r-- | storage/src/tests/distributor/distributortest.cpp | 17 | ||||
-rw-r--r-- | storage/src/tests/distributor/getoperationtest.cpp | 47 | ||||
-rw-r--r-- | storage/src/tests/distributor/twophaseupdateoperationtest.cpp | 85 |
3 files changed, 135 insertions, 14 deletions
diff --git a/storage/src/tests/distributor/distributortest.cpp b/storage/src/tests/distributor/distributortest.cpp index e6c617a32e4..94d33e50047 100644 --- a/storage/src/tests/distributor/distributortest.cpp +++ b/storage/src/tests/distributor/distributortest.cpp @@ -174,6 +174,12 @@ struct DistributorTest : Test, DistributorTestUtil { configureDistributor(builder); } + void configure_update_fast_path_restart_enabled(bool enabled) { + ConfigBuilder builder; + builder.restartWithFastUpdatePathIfAllGetTimestampsAreConsistent = enabled; + configureDistributor(builder); + } + void configureMaxClusterClockSkew(int seconds); void sendDownClusterStateCommand(); void replyToSingleRequestBucketInfoCommandWith1Bucket(); @@ -1001,6 +1007,17 @@ TEST_F(DistributorTest, stale_reads_config_is_propagated_to_external_operation_h EXPECT_FALSE(getExternalOperationHandler().concurrent_gets_enabled()); } +TEST_F(DistributorTest, fast_path_on_consistent_gets_config_is_propagated_to_internal_config) { + createLinks(true); + setupDistributor(Redundancy(1), NodeCount(1), "distributor:1 storage:1"); + + configure_update_fast_path_restart_enabled(true); + EXPECT_TRUE(getConfig().update_fast_path_restart_enabled()); + + configure_update_fast_path_restart_enabled(false); + EXPECT_FALSE(getConfig().update_fast_path_restart_enabled()); +} + TEST_F(DistributorTest, concurrent_reads_not_enabled_if_btree_db_is_not_enabled) { createLinks(false); setupDistributor(Redundancy(1), NodeCount(1), "distributor:1 storage:1"); diff --git a/storage/src/tests/distributor/getoperationtest.cpp b/storage/src/tests/distributor/getoperationtest.cpp index 99d7c12551d..72a124d45b4 100644 --- a/storage/src/tests/distributor/getoperationtest.cpp +++ b/storage/src/tests/distributor/getoperationtest.cpp @@ -117,6 +117,13 @@ struct GetOperationTest : Test, DistributorTestUtil { } } + bool last_reply_had_consistent_replicas() { + assert(!_sender.replies().empty()); + auto& msg = *_sender.replies().back(); + assert(msg.getType() == api::MessageType::GET_REPLY); + return dynamic_cast<api::GetReply&>(msg).had_consistent_replicas(); + } + void setClusterState(const std::string& clusterState) { enableDistributorClusterState(clusterState); } @@ -139,22 +146,25 @@ TEST_F(GetOperationTest, simple) { EXPECT_EQ("GetReply(BucketId(0x0000000000000000), id:ns:text/html::uri, " "timestamp 100) ReturnCode(NONE)", _sender.getLastReply()); + EXPECT_TRUE(last_reply_had_consistent_replicas()); } -TEST_F(GetOperationTest, ask_trusted_node_if_bucket_is_inconsistent) { +TEST_F(GetOperationTest, ask_all_checksum_groups_if_inconsistent_even_if_trusted_replica_available) { setClusterState("distributor:1 storage:4"); addNodesToBucketDB(bucketId, "0=100/3/10,1=200/4/12/t"); sendGet(); - ASSERT_EQ("Get => 1", _sender.getCommands(true)); + ASSERT_EQ("Get => 0,Get => 1", _sender.getCommands(true)); - ASSERT_NO_FATAL_FAILURE(replyWithDocument()); + ASSERT_NO_FATAL_FAILURE(sendReply(0, api::ReturnCode::OK, "newauthor", 2)); + ASSERT_NO_FATAL_FAILURE(sendReply(1, api::ReturnCode::OK, "oldauthor", 1)); EXPECT_EQ("GetReply(BucketId(0x0000000000000000), id:ns:text/html::uri, " - "timestamp 100) ReturnCode(NONE)", + "timestamp 2) ReturnCode(NONE)", _sender.getLastReply()); + EXPECT_FALSE(last_reply_had_consistent_replicas()); } TEST_F(GetOperationTest, ask_all_nodes_if_bucket_is_inconsistent) { @@ -174,6 +184,7 @@ TEST_F(GetOperationTest, ask_all_nodes_if_bucket_is_inconsistent) { _sender.getLastReply()); EXPECT_EQ("newauthor", getLastReplyAuthor()); + EXPECT_FALSE(last_reply_had_consistent_replicas()); } TEST_F(GetOperationTest, send_to_all_invalid_copies) { @@ -193,6 +204,7 @@ TEST_F(GetOperationTest, send_to_all_invalid_copies) { _sender.getLastReply()); EXPECT_EQ("newauthor", getLastReplyAuthor()); + EXPECT_FALSE(last_reply_had_consistent_replicas()); } TEST_F(GetOperationTest, send_to_all_invalid_nodes_when_inconsistent) { @@ -235,6 +247,7 @@ TEST_F(GetOperationTest, inconsistent_split) { _sender.getLastReply()); EXPECT_EQ("newauthor", getLastReplyAuthor()); + EXPECT_FALSE(last_reply_had_consistent_replicas()); } TEST_F(GetOperationTest, multi_inconsistent_bucket_not_found) { @@ -252,6 +265,7 @@ TEST_F(GetOperationTest, multi_inconsistent_bucket_not_found) { ASSERT_EQ("GetReply(BucketId(0x0000000000000000), id:ns:text/html::uri, " "timestamp 2) ReturnCode(NONE)", _sender.getLastReply()); + EXPECT_FALSE(last_reply_had_consistent_replicas()); } TEST_F(GetOperationTest, multi_inconsistent_bucket_not_found_deleted) { @@ -271,6 +285,7 @@ TEST_F(GetOperationTest, multi_inconsistent_bucket_not_found_deleted) { ASSERT_EQ("GetReply(BucketId(0x0000000000000000), id:ns:text/html::uri, " "timestamp 3) ReturnCode(NONE)", _sender.getLastReply()); + EXPECT_FALSE(last_reply_had_consistent_replicas()); } TEST_F(GetOperationTest, multi_inconsistent_bucket) { @@ -290,6 +305,7 @@ TEST_F(GetOperationTest, multi_inconsistent_bucket) { _sender.getLastReply()); EXPECT_EQ("newauthor", getLastReplyAuthor()); + EXPECT_FALSE(last_reply_had_consistent_replicas()); } TEST_F(GetOperationTest, multi_inconsistent_bucket_fail) { @@ -312,6 +328,7 @@ TEST_F(GetOperationTest, multi_inconsistent_bucket_fail) { ASSERT_EQ("GetReply(BucketId(0x0000000000000000), id:ns:text/html::uri, " "timestamp 100) ReturnCode(NONE)", _sender.getLastReply()); + EXPECT_FALSE(last_reply_had_consistent_replicas()); } TEST_F(GetOperationTest, return_not_found_when_bucket_not_in_db) { @@ -322,6 +339,7 @@ TEST_F(GetOperationTest, return_not_found_when_bucket_not_in_db) { ASSERT_EQ("GetReply(BucketId(0x0000000000000000), id:ns:text/html::uri, " "timestamp 0) ReturnCode(NONE)", _sender.getLastReply()); + EXPECT_TRUE(last_reply_had_consistent_replicas()); // Nothing in the bucket, so nothing to be inconsistent with. } TEST_F(GetOperationTest, not_found) { @@ -340,8 +358,9 @@ TEST_F(GetOperationTest, not_found) { "timestamp 0) ReturnCode(NONE)", _sender.getLastReply()); - EXPECT_EQ(1, getDistributor().getMetrics().gets[documentapi::LoadType::DEFAULT]. + EXPECT_EQ(1, getDistributor().getMetrics().gets[documentapi::LoadType::DEFAULT]. failures.notfound.getValue()); + EXPECT_TRUE(last_reply_had_consistent_replicas()); } TEST_F(GetOperationTest, resend_on_storage_failure) { @@ -366,6 +385,21 @@ TEST_F(GetOperationTest, resend_on_storage_failure) { ASSERT_EQ("GetReply(BucketId(0x0000000000000000), id:ns:text/html::uri, " "timestamp 100) ReturnCode(NONE)", _sender.getLastReply()); + // Replica had read failure, but they're still in sync. An immutable Get won't change that fact. + EXPECT_TRUE(last_reply_had_consistent_replicas()); +} + +TEST_F(GetOperationTest, storage_failure_of_out_of_sync_replica_is_tracked_as_inconsistent) { + setClusterState("distributor:1 storage:3"); + addNodesToBucketDB(bucketId, "1=100,2=200"); + sendGet(); + ASSERT_EQ("Get => 1,Get => 2", _sender.getCommands(true)); + ASSERT_NO_FATAL_FAILURE(sendReply(0, api::ReturnCode::TIMEOUT, "", 0)); + ASSERT_NO_FATAL_FAILURE(sendReply(1, api::ReturnCode::OK, "newestauthor", 3)); + ASSERT_EQ("GetReply(BucketId(0x0000000000000000), id:ns:text/html::uri, " + "timestamp 3) ReturnCode(NONE)", + _sender.getLastReply()); + EXPECT_FALSE(last_reply_had_consistent_replicas()); } TEST_F(GetOperationTest, resend_on_storage_failure_all_fail) { @@ -390,6 +424,7 @@ TEST_F(GetOperationTest, resend_on_storage_failure_all_fail) { ASSERT_EQ("GetReply(BucketId(0x0000000000000000), id:ns:text/html::uri, " "timestamp 0) ReturnCode(IO_FAILURE)", _sender.getLastReply()); + EXPECT_TRUE(last_reply_had_consistent_replicas()); // Doesn't really matter since operation itself failed } TEST_F(GetOperationTest, send_to_ideal_copy_if_bucket_in_sync) { @@ -408,6 +443,7 @@ TEST_F(GetOperationTest, send_to_ideal_copy_if_bucket_in_sync) { ASSERT_EQ("GetReply(BucketId(0x0000000000000000), id:ns:text/html::uri, " "timestamp 100) ReturnCode(NONE)", _sender.getLastReply()); + EXPECT_TRUE(last_reply_had_consistent_replicas()); } TEST_F(GetOperationTest, multiple_copies_with_failure_on_local_node) { @@ -435,6 +471,7 @@ TEST_F(GetOperationTest, multiple_copies_with_failure_on_local_node) { _sender.getLastReply()); EXPECT_EQ("newestauthor", getLastReplyAuthor()); + EXPECT_TRUE(last_reply_had_consistent_replicas()); } TEST_F(GetOperationTest, can_get_documents_when_all_replica_nodes_retired) { diff --git a/storage/src/tests/distributor/twophaseupdateoperationtest.cpp b/storage/src/tests/distributor/twophaseupdateoperationtest.cpp index df9bf683326..67ef3374633 100644 --- a/storage/src/tests/distributor/twophaseupdateoperationtest.cpp +++ b/storage/src/tests/distributor/twophaseupdateoperationtest.cpp @@ -126,6 +126,10 @@ struct TwoPhaseUpdateOperationTest : Test, DistributorTestUtil { void assertAbortedUpdateReplyWithContextPresent( const DistributorMessageSenderStub& closeSender) const; + void do_test_ownership_changed_between_gets_and_second_phase(Timestamp lowest_get_timestamp, + Timestamp highest_get_timestamp, + Timestamp expected_response_timestamp); + }; TwoPhaseUpdateOperationTest::TwoPhaseUpdateOperationTest() = default; @@ -142,7 +146,9 @@ TwoPhaseUpdateOperationTest::replyToMessage( std::shared_ptr<api::StorageMessage> msg2 = sender.command(index); auto& updatec = dynamic_cast<UpdateCommand&>(*msg2); std::unique_ptr<api::StorageReply> reply(updatec.makeReply()); - static_cast<api::UpdateReply*>(reply.get())->setOldTimestamp(oldTimestamp); + auto& update_reply = dynamic_cast<api::UpdateReply&>(*reply); + update_reply.setOldTimestamp(oldTimestamp); + update_reply.setBucketInfo(api::BucketInfo(0x123, 1, 100)); // Dummy info to avoid invalid info being returned reply->setResult(api::ReturnCode(result, "")); callback.receive(sender, @@ -193,7 +199,7 @@ TwoPhaseUpdateOperationTest::replyToGet( api::ReturnCode::Result result, const std::string& traceMsg) { - auto& get = static_cast<const api::GetCommand&>(*sender.command(index)); + auto& get = dynamic_cast<const api::GetCommand&>(*sender.command(index)); std::shared_ptr<api::StorageReply> reply; if (haveDocument) { @@ -786,7 +792,11 @@ TEST_F(TwoPhaseUpdateOperationTest, safe_path_propagates_mbus_traces_from_replie ASSERT_THAT(trace, HasSubstr("baaa")); } -TEST_F(TwoPhaseUpdateOperationTest, update_fails_if_ownership_changes_between_get_and_put) { +void TwoPhaseUpdateOperationTest::do_test_ownership_changed_between_gets_and_second_phase( + Timestamp lowest_get_timestamp, + Timestamp highest_get_timestamp, + Timestamp expected_response_timestamp) +{ setupDistributor(2, 2, "storage:2 distributor:1"); // Update towards inconsistent bucket invokes safe path. @@ -803,21 +813,31 @@ TEST_F(TwoPhaseUpdateOperationTest, update_fails_if_ownership_changes_between_ge // to a bucket we no longer own. enableDistributorClusterState("storage:2 distributor:1 .0.s:d"); getBucketDatabase().clear(); - replyToGet(*cb, sender, 0, 70); - replyToGet(*cb, sender, 1, 70); - + replyToGet(*cb, sender, 0, lowest_get_timestamp); + replyToGet(*cb, sender, 1, highest_get_timestamp); + // BUCKET_NOT_FOUND is a transient error code which should cause the client // to re-send the operation, presumably to the correct distributor the next // time. + // Timestamp of updated doc varies depending on whether fast or safe path + // was triggered, as the reply is created via different paths. EXPECT_EQ("UpdateReply(id:ns:testdoctype1::1, " "BucketId(0x0000000000000000), " - "timestamp 0, timestamp of updated doc: 70) " + "timestamp 0, timestamp of updated doc: " + std::to_string(expected_response_timestamp) + ") " "ReturnCode(BUCKET_NOT_FOUND, Distributor lost " - "ownership of bucket between executing the read " - "and write phases of a two-phase update operation)", + "ownership of bucket between executing the read " + "and write phases of a two-phase update operation)", sender.getLastReply(true)); } +TEST_F(TwoPhaseUpdateOperationTest, update_fails_if_ownership_changes_between_get_and_put) { + do_test_ownership_changed_between_gets_and_second_phase(70, 71, 71); +} + +TEST_F(TwoPhaseUpdateOperationTest, update_fails_if_ownership_changes_between_get_and_restarted_fast_path_updates) { + do_test_ownership_changed_between_gets_and_second_phase(70, 70, 0); // Timestamps in sync -> Update restart +} + TEST_F(TwoPhaseUpdateOperationTest, safe_path_condition_mismatch_fails_with_tas_error) { setupDistributor(2, 2, "storage:2 distributor:1"); std::shared_ptr<TwoPhaseUpdateOperation> cb( @@ -977,6 +997,53 @@ TEST_F(TwoPhaseUpdateOperationTest, safe_path_close_edge_sends_correct_reply) { assertAbortedUpdateReplyWithContextPresent(closeSender); } +TEST_F(TwoPhaseUpdateOperationTest, safe_path_consistent_get_reply_timestamps_restarts_with_fast_path_if_enabled) { + setupDistributor(2, 2, "storage:2 distributor:1"); + getConfig().set_update_fast_path_restart_enabled(true); + + std::shared_ptr<TwoPhaseUpdateOperation> cb(sendUpdate("0=1/2/3,1=2/3/4")); // Inconsistent replicas. + DistributorMessageSenderStub sender; + cb->start(sender, framework::MilliSecTime(0)); + + Timestamp old_timestamp = 500; + ASSERT_EQ("Get => 0,Get => 1", sender.getCommands(true)); + replyToGet(*cb, sender, 0, old_timestamp); + replyToGet(*cb, sender, 1, old_timestamp); + + ASSERT_EQ("Update => 0,Update => 1", sender.getCommands(true, false, 2)); + replyToMessage(*cb, sender, 2, old_timestamp); + replyToMessage(*cb, sender, 3, old_timestamp); + + EXPECT_EQ("UpdateReply(id:ns:testdoctype1::1, " + "BucketId(0x0000000000000000), " + "timestamp 0, timestamp of updated doc: 500) " + "ReturnCode(NONE)", + sender.getLastReply(true)); + + auto& metrics = getDistributor().getMetrics().updates[documentapi::LoadType::DEFAULT]; + EXPECT_EQ(1, metrics.fast_path_restarts.getValue()); +} + +TEST_F(TwoPhaseUpdateOperationTest, safe_path_consistent_get_reply_timestamps_does_not_restart_with_fast_path_if_disabled) { + setupDistributor(2, 2, "storage:2 distributor:1"); + getConfig().set_update_fast_path_restart_enabled(false); + + std::shared_ptr<TwoPhaseUpdateOperation> cb(sendUpdate("0=1/2/3,1=2/3/4")); // Inconsistent replicas. + DistributorMessageSenderStub sender; + cb->start(sender, framework::MilliSecTime(0)); + + Timestamp old_timestamp = 500; + ASSERT_EQ("Get => 0,Get => 1", sender.getCommands(true)); + replyToGet(*cb, sender, 0, old_timestamp); + replyToGet(*cb, sender, 1, old_timestamp); + + // Should _not_ be restarted with fast path, as it has been config disabled + ASSERT_EQ("Put => 1,Put => 0", sender.getCommands(true, false, 2)); + + auto& metrics = getDistributor().getMetrics().updates[documentapi::LoadType::DEFAULT]; + EXPECT_EQ(0, metrics.fast_path_restarts.getValue()); +} + // XXX currently differs in behavior from content nodes in that updates for // document IDs without explicit doctypes will _not_ be auto-failed on the // distributor. |