Update operation metrics for delayed or chained merge handler replies.

1 files changed, 76 insertions, 2 deletions

diff --git a/storage/src/tests/persistence/apply_bucket_diff_state_test.cpp b/storage/src/tests/persistence/apply_bucket_diff_state_test.cpp
index cdf7ac817fa..ec57d775f43 100644
--- a/storage/src/tests/persistence/apply_bucket_diff_state_test.cpp
+++ b/storage/src/tests/persistence/apply_bucket_diff_state_test.cpp
@@ -3,6 +3,7 @@
 #include <vespa/storage/persistence/apply_bucket_diff_state.h>
 #include <vespa/storage/persistence/merge_bucket_info_syncer.h>
 #include <vespa/storage/persistence/filestorage/merge_handler_metrics.h>
+#include <vespa/storageapi/message/bucket.h>
 #include <vespa/document/base/documentid.h>
 #include <vespa/document/bucket/bucketid.h>
 #include <vespa/document/bucket/bucketidfactory.h>
@@ -10,6 +11,8 @@
 #include <vespa/metrics/metricset.h>
 #include <vespa/persistence/spi/result.h>
 #include <vespa/storageframework/defaultimplementation/clock/fakeclock.h>
+#include <tests/common/message_sender_stub.h>
+#include <tests/persistence/persistencetestutils.h>
 #include <gtest/gtest.h>
 
 using document::DocumentId;
@@ -73,22 +76,25 @@ void push_bad(ApplyBucketDiffState &state)
 
 }
 
-class ApplyBucketDiffStateTestBase : public ::testing::Test
+class ApplyBucketDiffStateTestBase : public PersistenceTestUtils
 {
 public:
     uint32_t                   sync_count;
     DummyMergeBucketInfoSyncer syncer;
     metrics::MetricSet         merge_handler_metrics_owner;
     MergeHandlerMetrics        merge_handler_metrics;
+    FileStorThreadMetrics::Op  op_metrics;
     framework::defaultimplementation::FakeClock clock;
+    MessageSenderStub          message_sender;
     MonitoredRefCount          monitored_ref_count;
 
     ApplyBucketDiffStateTestBase()
-        : ::testing::Test(),
+        : PersistenceTestUtils(),
           sync_count(0u),
           syncer(sync_count),
           merge_handler_metrics_owner("owner", {}, "owner"),
           merge_handler_metrics(&merge_handler_metrics_owner),
+          op_metrics("op", "op", &merge_handler_metrics_owner),
           clock(),
           monitored_ref_count()
     {
@@ -99,6 +105,13 @@ public:
     std::shared_ptr<ApplyBucketDiffState> make_state() {
         return ApplyBucketDiffState::create(syncer, merge_handler_metrics, clock, spi::Bucket(dummy_document_bucket), RetainGuard(monitored_ref_count));
     }
+
+    MessageTracker::UP
+    create_tracker(std::shared_ptr<api::StorageMessage> cmd, document::Bucket bucket) {
+        return MessageTracker::createForTesting(framework::MilliSecTimer(clock), getEnv(),
+                                                message_sender, NoBucketLock::make(bucket), std::move(cmd));
+    }
+
 };
 
 ApplyBucketDiffStateTestBase::~ApplyBucketDiffStateTestBase() = default;
@@ -128,8 +141,44 @@ public:
         check_failure("Failed put for id::test::1 in Bucket(0x0000000000000010): Result(5, write blocked)");
     }
 
+    void test_delayed_reply(bool failed, bool async_failed, bool chained_reply);
+
 };
 
+void
+ApplyBucketDiffStateTest::test_delayed_reply(bool failed, bool async_failed, bool chained_reply)
+{
+    auto cmd = std::make_shared<api::MergeBucketCommand>(dummy_document_bucket, std::vector<api::MergeBucketCommand::Node>{}, 0);
+    std::shared_ptr<api::StorageReply> reply = cmd->makeReply();
+    auto tracker = create_tracker(cmd, dummy_document_bucket);
+    if (failed) {
+        reply->setResult(api::ReturnCode::Result::INTERNAL_FAILURE);
+    }
+    tracker->setMetric(op_metrics);
+    tracker->setReply(reply);
+    if (chained_reply) {
+        state->set_delayed_reply(std::move(tracker), message_sender, &op_metrics, framework::MilliSecTimer(clock), std::move(reply));
+    } else {
+        state->set_delayed_reply(std::move(tracker), std::move(reply));
+    }
+    clock.addMilliSecondsToTime(16);
+    if (async_failed) {
+        push_bad(*state);
+    }
+    state.reset();
+    if (failed || async_failed) {
+        EXPECT_EQ(0.0, op_metrics.latency.getLast());
+        EXPECT_EQ(0, op_metrics.latency.getCount());
+        EXPECT_EQ(1, op_metrics.failed.getValue());
+    } else {
+        EXPECT_EQ(16.0, op_metrics.latency.getLast());
+        EXPECT_EQ(1, op_metrics.latency.getCount());
+        EXPECT_EQ(0, op_metrics.failed.getValue());
+    }
+    ASSERT_EQ(1, message_sender.replies.size());
+    EXPECT_NE(failed || async_failed, std::dynamic_pointer_cast<api::MergeBucketReply>(message_sender.replies.front())->getResult().success());
+}
+
 TEST_F(ApplyBucketDiffStateTest, ok_results_can_be_checked)
 {
     push_ok(*state);
@@ -203,4 +252,29 @@ TEST_F(ApplyBucketDiffStateTest, total_latency_is_updated)
     EXPECT_EQ(1, merge_handler_metrics.mergeLatencyTotal.getCount());
 }
 
+TEST_F(ApplyBucketDiffStateTest, delayed_ok_reply)
+{
+    test_delayed_reply(false, false, false);
+}
+
+TEST_F(ApplyBucketDiffStateTest, delayed_failed_reply)
+{
+    test_delayed_reply(true, false, false);
+}
+
+TEST_F(ApplyBucketDiffStateTest, delayed_ok_chained_reply)
+{
+    test_delayed_reply(false, false, true);
+}
+
+TEST_F(ApplyBucketDiffStateTest, delayed_failed_chained_reply)
+{
+    test_delayed_reply(true, false, true);
+}
+
+TEST_F(ApplyBucketDiffStateTest, delayed_async_failed_reply)
+{
+    test_delayed_reply(false, true, false);
+}
+
 }