Ensure internal messages are flushed before shutting down RPC subsystem

This moves RPC shutdown from being the _first_ thing that happens
to being the _last_ thing that happens during storage chain shutdown.
To avoid concurrent client requests from the outside reaching internal
components during the flushing phases, the Bouncer component will now
explicitly and immediately reject incoming RPCs after closing and all
replies will be silently swallowed (no one is listening for them at that
point anyway).

1 files changed, 36 insertions, 7 deletions

diff --git a/storage/src/tests/storageserver/bouncertest.cpp b/storage/src/tests/storageserver/bouncertest.cpp
index c41696e1a02..5b7d279537e 100644
--- a/storage/src/tests/storageserver/bouncertest.cpp
+++ b/storage/src/tests/storageserver/bouncertest.cpp
@@ -51,9 +51,10 @@ struct BouncerTest : public Test {
             api::Timestamp timestamp,
             document::BucketSpace bucketSpace);
 
-    void expectMessageBouncedWithRejection();
-    void expectMessageBouncedWithAbort();
-    void expectMessageNotBounced();
+    void expectMessageBouncedWithRejection() const;
+    void expect_message_bounced_with_node_down_abort() const;
+    void expect_message_bounced_with_shutdown_abort() const;
+    void expectMessageNotBounced() const;
 };
 
 BouncerTest::BouncerTest()
@@ -181,7 +182,7 @@ TEST_F(BouncerTest, allow_notify_bucket_change_even_when_distributor_down) {
 }
 
 void
-BouncerTest::expectMessageBouncedWithRejection()
+BouncerTest::expectMessageBouncedWithRejection() const
 {
     ASSERT_EQ(1, _upper->getNumReplies());
     EXPECT_EQ(0, _upper->getNumCommands());
@@ -191,7 +192,7 @@ BouncerTest::expectMessageBouncedWithRejection()
 }
 
 void
-BouncerTest::expectMessageBouncedWithAbort()
+BouncerTest::expect_message_bounced_with_node_down_abort() const
 {
     ASSERT_EQ(1, _upper->getNumReplies());
     EXPECT_EQ(0, _upper->getNumCommands());
@@ -204,7 +205,17 @@ BouncerTest::expectMessageBouncedWithAbort()
 }
 
 void
-BouncerTest::expectMessageNotBounced()
+BouncerTest::expect_message_bounced_with_shutdown_abort() const
+{
+    ASSERT_EQ(1, _upper->getNumReplies());
+    EXPECT_EQ(0, _upper->getNumCommands());
+    auto& reply = dynamic_cast<api::StorageReply&>(*_upper->getReply(0));
+    EXPECT_EQ(api::ReturnCode(api::ReturnCode::ABORTED, "Node is shutting down"), reply.getResult());
+    EXPECT_EQ(0, _lower->getNumCommands());
+}
+
+void
+BouncerTest::expectMessageNotBounced() const
 {
     EXPECT_EQ(size_t(0), _upper->getNumReplies());
     EXPECT_EQ(size_t(1), _lower->getNumCommands());
@@ -296,7 +307,7 @@ TEST_F(BouncerTest, abort_request_when_derived_bucket_space_node_state_is_marked
     auto state = makeClusterStateBundle("distributor:3 storage:3", {{ document::FixedBucketSpaces::default_space(), "distributor:3 storage:3 .2.s:d" }});
     _node->getNodeStateUpdater().setClusterStateBundle(state);
     _upper->sendDown(createDummyFeedMessage(11 * 1000000, document::FixedBucketSpaces::default_space()));
-    expectMessageBouncedWithAbort();
+    expect_message_bounced_with_node_down_abort();
     EXPECT_EQ(1, _manager->metrics().unavailable_node_aborts.getValue());
 
     _upper->reset();
@@ -362,5 +373,23 @@ TEST_F(BouncerTest, operation_with_sufficient_bucket_bits_is_not_rejected) {
     expectMessageNotBounced();
 }
 
+TEST_F(BouncerTest, requests_are_rejected_after_close) {
+    _manager->close();
+    _upper->sendDown(createDummyFeedMessage(11 * 1000000, document::FixedBucketSpaces::default_space()));
+    expect_message_bounced_with_shutdown_abort();
+}
+
+TEST_F(BouncerTest, replies_are_swallowed_after_close) {
+    _manager->close();
+    auto req = createDummyFeedMessage(11 * 1000000, document::FixedBucketSpaces::default_space());
+    auto reply = req->makeReply();
+    _upper->sendDown(std::move(reply));
+
+    EXPECT_EQ(0, _upper->getNumCommands());
+    EXPECT_EQ(0, _upper->getNumReplies());
+    EXPECT_EQ(0, _lower->getNumCommands());
+    EXPECT_EQ(0, _lower->getNumReplies());
+}
+
 } // storage