Remap missing bucket space for DocumentAPI Gets to "Not Found"

Lets legacy routing of Gets to all clusters successfully merge responses
even if a recipient cluster does not have a valid mapping for the requested
bucket space.

3 files changed, 51 insertions, 9 deletions

diff --git a/storage/src/tests/storageserver/communicationmanagertest.cpp b/storage/src/tests/storageserver/communicationmanagertest.cpp
index a271fbccf50..3b72585b076 100644
--- a/storage/src/tests/storageserver/communicationmanagertest.cpp
+++ b/storage/src/tests/storageserver/communicationmanagertest.cpp
@@ -14,6 +14,8 @@
 #include <vespa/documentapi/messagebus/messages/getdocumentmessage.h>
 #include <vespa/vdstestlib/cppunit/macros.h>
 #include <vespa/vespalib/util/stringfmt.h>
+#include <vespa/documentapi/messagebus/messages/removedocumentmessage.h>
+#include <vespa/documentapi/messagebus/messages/getdocumentreply.h>
 
 using document::test::makeDocumentBucket;
 
@@ -27,6 +29,7 @@ struct CommunicationManagerTest : public CppUnit::TestFixture {
     void testRepliesAreDequeuedInFifoOrder();
     void bucket_space_config_can_be_updated_live();
     void unmapped_bucket_space_documentapi_request_returns_error_reply();
+    void unmapped_bucket_space_for_get_documentapi_request_returns_empty_reply();
 
     static constexpr uint32_t MESSAGE_WAIT_TIME_SEC = 60;
 
@@ -51,6 +54,7 @@ struct CommunicationManagerTest : public CppUnit::TestFixture {
     CPPUNIT_TEST(testRepliesAreDequeuedInFifoOrder);
     CPPUNIT_TEST(bucket_space_config_can_be_updated_live);
     CPPUNIT_TEST(unmapped_bucket_space_documentapi_request_returns_error_reply);
+    CPPUNIT_TEST(unmapped_bucket_space_for_get_documentapi_request_returns_empty_reply);
     CPPUNIT_TEST_SUITE_END();
 };
 
@@ -267,13 +271,21 @@ struct CommunicationManagerFixture {
     }
     ~CommunicationManagerFixture();
 
-    std::unique_ptr<documentapi::GetDocumentMessage> documentapi_message_for_space(const char* space) {
-        auto cmd = std::make_unique<documentapi::GetDocumentMessage>(
-                document::DocumentId(vespalib::make_string("id::%s::stuff", space)));
+    template <typename T>
+    std::unique_ptr<T> documentapi_message_for_space(const char *space) {
+        auto cmd = std::make_unique<T>(document::DocumentId(vespalib::make_string("id::%s::stuff", space)));
         // Bind reply handling to our own mock handler
         cmd->pushHandler(reply_handler);
         return cmd;
     }
+
+    std::unique_ptr<documentapi::RemoveDocumentMessage> documentapi_remove_message_for_space(const char *space) {
+        return documentapi_message_for_space<documentapi::RemoveDocumentMessage>(space);
+    }
+
+    std::unique_ptr<documentapi::GetDocumentMessage> documentapi_get_message_for_space(const char *space) {
+        return documentapi_message_for_space<documentapi::GetDocumentMessage>(space);
+    }
 };
 
 CommunicationManagerFixture::~CommunicationManagerFixture() = default;
@@ -298,8 +310,8 @@ void CommunicationManagerTest::bucket_space_config_can_be_updated_live() {
     config.documenttype.emplace_back(doc_type("bar", "global"));
     f.comm_mgr->updateBucketSpacesConfig(config);
 
-    f.comm_mgr->handleMessage(f.documentapi_message_for_space("bar"));
-    f.comm_mgr->handleMessage(f.documentapi_message_for_space("foo"));
+    f.comm_mgr->handleMessage(f.documentapi_remove_message_for_space("bar"));
+    f.comm_mgr->handleMessage(f.documentapi_remove_message_for_space("foo"));
     f.bottom_link->waitForMessages(2, MESSAGE_WAIT_TIME_SEC);
 
     auto cmd1 = f.bottom_link->getCommand(0);
@@ -310,7 +322,7 @@ void CommunicationManagerTest::bucket_space_config_can_be_updated_live() {
 
     config.documenttype[1] = doc_type("bar", "default");
     f.comm_mgr->updateBucketSpacesConfig(config);
-    f.comm_mgr->handleMessage(f.documentapi_message_for_space("bar"));
+    f.comm_mgr->handleMessage(f.documentapi_remove_message_for_space("bar"));
     f.bottom_link->waitForMessages(3, MESSAGE_WAIT_TIME_SEC);
 
     auto cmd3 = f.bottom_link->getCommand(2);
@@ -328,7 +340,7 @@ void CommunicationManagerTest::unmapped_bucket_space_documentapi_request_returns
 
     CPPUNIT_ASSERT_EQUAL(uint64_t(0), f.comm_mgr->metrics().bucketSpaceMappingFailures.getValue());
 
-    f.comm_mgr->handleMessage(f.documentapi_message_for_space("fluff"));
+    f.comm_mgr->handleMessage(f.documentapi_remove_message_for_space("fluff"));
     CPPUNIT_ASSERT_EQUAL(size_t(1), f.reply_handler.replies.size());
     auto& reply = *f.reply_handler.replies[0];
     CPPUNIT_ASSERT(reply.hasErrors());
@@ -337,4 +349,23 @@ void CommunicationManagerTest::unmapped_bucket_space_documentapi_request_returns
     CPPUNIT_ASSERT_EQUAL(uint64_t(1), f.comm_mgr->metrics().bucketSpaceMappingFailures.getValue());
 }
 
+// Legacy DocumentAPI routing protocols will send Gets to _all_ clusters even
+// if they do not contain a particular document type. By sending an empty reply
+// we signal a mergeable "not found" to the sender rather than a non-mergeable
+// fatal error.
+void CommunicationManagerTest::unmapped_bucket_space_for_get_documentapi_request_returns_empty_reply() {
+    CommunicationManagerFixture f;
+
+    BucketspacesConfigBuilder config;
+    config.documenttype.emplace_back(doc_type("foo", "default"));
+    f.comm_mgr->updateBucketSpacesConfig(config);
+
+    f.comm_mgr->handleMessage(f.documentapi_get_message_for_space("fluff"));
+    CPPUNIT_ASSERT_EQUAL(size_t(1), f.reply_handler.replies.size());
+    auto& reply = *f.reply_handler.replies[0];
+    CPPUNIT_ASSERT(!reply.hasErrors());
+    auto& get_reply = dynamic_cast<documentapi::GetDocumentReply&>(reply);
+    CPPUNIT_ASSERT(!get_reply.hasDocument());
+}
+
 } // storage
diff --git a/storage/src/vespa/storage/distributor/pendingclusterstate.cpp b/storage/src/vespa/storage/distributor/pendingclusterstate.cpp
index a08445ca3d2..471ce7e2b27 100644
--- a/storage/src/vespa/storage/distributor/pendingclusterstate.cpp
+++ b/storage/src/vespa/storage/distributor/pendingclusterstate.cpp
@@ -236,6 +236,7 @@ PendingClusterState::onRequestBucketInfoReply(const std::shared_ptr<api::Request
     if (result == api::ReturnCode::Result::ENCODE_ERROR) {
         // Handle failure to encode bucket space due to use of old storage api
         // protocol.  Pretend that request succeeded with no buckets returned.
+        // TODO remove this workaround for Vespa 7
         LOG(debug, "Got ENCODE_ERROR, pretending success with no buckets");
     } else if (!result.success()) {
         framework::MilliSecTime resendTime(_clock);
diff --git a/storage/src/vespa/storage/storageserver/communicationmanager.cpp b/storage/src/vespa/storage/storageserver/communicationmanager.cpp
index 22193d7e246..94a151bcdc1 100644
--- a/storage/src/vespa/storage/storageserver/communicationmanager.cpp
+++ b/storage/src/vespa/storage/storageserver/communicationmanager.cpp
@@ -19,6 +19,7 @@
 
 #include <vespa/log/bufferedlogger.h>
 #include <vespa/document/bucket/fixed_bucket_spaces.h>
+#include <vespa/documentapi/messagebus/messages/getdocumentreply.h>
 
 LOG_SETUP(".communication.manager");
 
@@ -258,8 +259,17 @@ void CommunicationManager::fail_with_unresolvable_bucket_space(
 {
     LOG(debug, "Could not map DocumentAPI message to internal bucket: %s", error_message.c_str());
     MBUS_TRACE(msg->getTrace(), 6, "Communication manager: Failing message as its document type has no known bucket space mapping");
-    std::unique_ptr<mbus::Reply> reply(new mbus::EmptyReply());
-    reply->addError(mbus::Error(documentapi::DocumentProtocol::ERROR_REJECTED, error_message));
+    std::unique_ptr<mbus::Reply> reply;
+    if (msg->getType() == documentapi::DocumentProtocol::MESSAGE_GETDOCUMENT) {
+        // HACK: to avoid breaking legacy routing of GetDocumentMessages to _all_ clusters
+        // regardless of them having a document type or not, we remap missing bucket spaces
+        // to explicit Not Found replies (empty document GetDocumentReply).
+        // TODO remove this workaround for Vespa 7
+        reply = std::make_unique<documentapi::GetDocumentReply>(std::shared_ptr<document::Document>());
+    } else {
+        reply = std::make_unique<mbus::EmptyReply>();
+        reply->addError(mbus::Error(documentapi::DocumentProtocol::ERROR_REJECTED, error_message));
+    }
     msg->swapState(*reply);
     _metrics.bucketSpaceMappingFailures.inc();
     _messageBusSession->reply(std::move(reply));