remoteFDB - negotiate common set of client & server features

src/fdb5/remote/Messages.cc

@@ -17,6 +17,42 @@
 
 namespace fdb5::remote {
 
+namespace {
+
+/// Ancillary messages are governed by the feature they belong to, rather than by their own bit.
+constexpr Message governingFeature(const Message msg) {
+    switch (msg) {
+        case Message::DoWipeURIs:
+        case Message::DoWipeUnknowns:
+        case Message::DoWipeFinish:
+        case Message::DoMaskIndexEntries:
+        case Message::DoUnsafeFullWipe:
+            return Message::Wipe;
+        default:
+            return msg;
+    }
+}
+
+}  // namespace
+
+std::string messageMask2String(uint64_t mm) {
+    std::string binary;
+    uint16_t offset = static_cast<uint16_t>(Message::DoWipeURIs) - static_cast<uint16_t>(Message::Flush);
+    for (size_t j = 0; j < offset; j++) {
+        binary = ((mm & 1) ? 'O' : '.') + binary;
+        mm >>= 1;
+    }
+    return binary;
+}
+
+bool enabled(const uint64_t enabledFeatures, const Message msg) {
+    const Message aux = governingFeature(msg);
+    // Check if the message is enabled - we are only interested in messages within the range [Flush, DoWipeURIs)
+    // The others are assumed to be always enabled since are required to establish the connection
+    // n.b. the range checks must come first, toMask() would throw for messages outside it
+    return (aux < Message::Flush || Message::DoWipeURIs <= aux || (enabledFeatures & toMask(aux)));
+}
+
 //----------------------------------------------------------------------------------------------------------------------
 
 std::ostream& operator<<(std::ostream& s, const Message& m) {

src/fdb5/remote/Messages.h

@@ -21,6 +21,7 @@
 #include <cmath>
 #include <cstddef>
 #include <cstdint>
+#include <stdexcept>
 
 #include "eckit/types/FixedString.h"
 
@@ -87,6 +88,23 @@ enum class Message : uint16_t {
 
 std::ostream& operator<<(std::ostream& s, const Message& m);
 
+constexpr uint64_t toMask(const Message msg) {
+    uint16_t offset = static_cast<uint16_t>(msg) - static_cast<uint16_t>(Message::Flush);
+    if (offset >= 64U) {
+        throw std::out_of_range("Message offset exceeds 64-bit mask limit");
+    }
+    return static_cast<uint64_t>(1) << offset;
+}
+
+inline constexpr uint64_t maskOfDefaultFeatures =
+    toMask(Message::Flush) | toMask(Message::Archive) | toMask(Message::Retrieve) | toMask(Message::List) |
+    toMask(Message::Stats) | toMask(Message::Inspect) | toMask(Message::Read) | toMask(Message::Store) |
+    toMask(Message::Axes) | toMask(Message::Exists);
+
+std::string messageMask2String(uint64_t mm);
+
+bool enabled(const uint64_t enabledFeatures, const Message msg);
+
 //----------------------------------------------------------------------------------------------------------------------
 
 // Header used for all messages

src/fdb5/remote/RemoteConfiguration.cc

@@ -1,19 +1,23 @@
 #include "fdb5/remote/RemoteConfiguration.h"
 
 #include <algorithm>
+#include <array>
 
 #include "eckit/config/Configuration.h"
+#include "eckit/config/Resource.h"
 #include "eckit/log/Log.h"
 #include "eckit/serialisation/Stream.h"
 #include "eckit/value/Value.h"
 
 #include "fdb5/LibFdb5.h"
+#include "fdb5/remote/Messages.h"
 
 namespace {
 
-std::vector<int> intersection(std::vector<int>& v1, std::vector<int>& v2) {
+template <class T>
+std::vector<T> intersection(std::vector<T>& v1, std::vector<T>& v2) {
 
-    std::vector<int> v3;
+    std::vector<T> v3;
 
     std::sort(v1.begin(), v1.end());
     std::sort(v2.begin(), v2.end());
@@ -48,6 +52,12 @@ RemoteConfiguration::RemoteConfiguration(const eckit::Configuration& config) {
     else {
         preferSingleConnection_ = std::nullopt;
     }
+
+    enabledFeatures_ = maskOfDefaultFeatures;
+
+    if (config.getBool("wipe", true)) {
+        enabledFeatures_ |= toMask(Message::Wipe);
+    }
 }
 
 RemoteConfiguration::RemoteConfiguration(eckit::Stream& s) {
@@ -91,6 +101,14 @@ RemoteConfiguration::RemoteConfiguration(eckit::Stream& s) {
     else {
         preferSingleConnection_ = std::nullopt;
     }
+    singleConnection_ = numberOfConnections_.size() == 1 && numberOfConnections_[0] == 1;
+
+    if (v.contains("EnabledFeatures")) {
+        enabledFeatures_ = v["EnabledFeatures"];
+    }
+    else {
+        enabledFeatures_ = maskOfDefaultFeatures;
+    }
 }
 
 bool RemoteConfiguration::singleConnection() const {
@@ -104,10 +122,19 @@ eckit::Stream& operator<<(eckit::Stream& s, const RemoteConfiguration& r) {
     if (r.preferSingleConnection_) {
         val["PreferSingleConnection"] = eckit::toValue(r.preferSingleConnection_.value());
     }
+    val["EnabledFeatures"] = eckit::toValue(r.enabledFeatures_);
     s << val;
     return s;
 }
 
+std::ostream& operator<<(std::ostream& s, const RemoteConfiguration& r) {
+    s << "RemoteConfiguration[remoteFieldLocationVersions=" << r.remoteFieldLocationVersions_
+      << ", numberOfConnections=" << r.numberOfConnections_ << ", preferSingleConnection="
+      << (r.preferSingleConnection_ ? std::to_string(*r.preferSingleConnection_) : "nullopt")
+      << ", singleConnection=" << r.singleConnection_ << ", enabledFeatures=" << r.enabledFeatures_ << "]";
+    return s;
+}
+
 RemoteConfiguration RemoteConfiguration::common(RemoteConfiguration& clientConf, RemoteConfiguration& serverConf) {
 
     RemoteConfiguration agreedConf{};
@@ -159,7 +186,11 @@ RemoteConfiguration RemoteConfiguration::common(RemoteConfiguration& clientConf,
         }
     }
 
-    LOG_DEBUG_LIB(LibFdb5) << "Protocol negotiation - NumberOfConnections " << ncSelected << std::endl;
+    agreedConf.enabledFeatures_ = clientConf.enabledFeatures_ & serverConf.enabledFeatures_;
+
+    LOG_DEBUG_LIB(LibFdb5) << "Protocol negotiation" << std::endl
+                           << "  EnabledFeatures     " << messageMask2String(agreedConf.enabledFeatures_) << std::endl
+                           << "  NumberOfConnections " << ncSelected << std::endl;
     agreedConf.numberOfConnections_ = {ncSelected};
     agreedConf.singleConnection_ = (ncSelected == 1);
 

src/fdb5/remote/RemoteConfiguration.h

@@ -10,6 +10,7 @@
 
 #pragma once
 
+#include <cstdint>
 #include <optional>
 #include <string>
 #include <vector>
@@ -56,8 +57,10 @@ class RemoteConfiguration {
     static RemoteConfiguration common(RemoteConfiguration& clientConf, RemoteConfiguration& serverConf);
 
     bool singleConnection() const;
+    uint64_t enabledFeatures() const { return enabledFeatures_; }
 
     friend eckit::Stream& operator<<(eckit::Stream& s, const RemoteConfiguration& r);
+    friend std::ostream& operator<<(std::ostream& s, const RemoteConfiguration& r);
 
 private:
 
@@ -66,6 +69,8 @@ class RemoteConfiguration {
 
     std::optional<bool> preferSingleConnection_;
 
+    uint64_t enabledFeatures_{0};
+
     bool singleConnection_{false};
 };
 

src/fdb5/remote/client/ClientConnection.cc

@@ -201,13 +201,26 @@ RemoteConfiguration ClientConnection::availableFunctionality(const Configuration
 
 //----------------------------------------------------------------------------------------------------------------------
 
+void ClientConnection::checkEnabled(Message msg) const {
+    if (!enabled(agreedFeatures_, msg)) {
+        std::ostringstream ss;
+        ss << "Message " << msg << " not enabled by server: " << controlEndpoint_ << std::endl;
+        ss << "  mask:    " << messageMask2String(toMask(msg)) << std::endl;
+        ss << "  enabled: " << messageMask2String(agreedFeatures_) << std::endl;
+
+        throw RemoteFDBException(ss.str(), controlEndpoint_);
+    }
+}
+
 std::future<Buffer> ClientConnection::controlWrite(const Client& client, const Message msg, const uint32_t requestID,
                                                    const bool /*dataListener*/, const PayloadList payloads) const {
     if (!valid()) {
         throw RemoteFDBException("Connection to " + std::string(controlEndpoint_) + " is no longer valid",
                                  controlEndpoint_);
     }
 
+    checkEnabled(msg);
+
     std::future<Buffer> f;
     {
         std::lock_guard lock(promisesMutex_);
@@ -220,6 +233,9 @@ std::future<Buffer> ClientConnection::controlWrite(const Client& client, const M
 }
 
 void ClientConnection::dataWrite(DataWriteRequest& request) const {
+
+    checkEnabled(request.msg_);
+
     Connection::write(request.msg_, false, request.client_->clientId(), request.id_, request.data_.data(),
                       request.data_.size());
 }
@@ -323,12 +339,12 @@ SessionID ClientConnection::verifyServerStartupResponse() {
     SessionID clientSession(s);
     SessionID serverSession(s);
     net::Endpoint dataEndpoint(s);
-    LocalConfiguration serverFunctionality(s);
+    RemoteConfiguration agreedConf{s};
 
     dataEndpoint_ = dataEndpoint;
 
     LOG_DEBUG_LIB(LibFdb5) << "verifyServerStartupResponse - Received from server " << clientSession << " "
-                           << serverSession << " " << dataEndpoint << std::endl;
+                           << serverSession << " " << agreedConf << " " << dataEndpoint << std::endl;
     if (dataEndpoint_.hostname() != controlEndpoint_.hostname()) {
         Log::warning() << "Data and control interface hostnames do not match. " << dataEndpoint_.hostname()
                        << " /= " << controlEndpoint_.hostname() << std::endl;
@@ -339,9 +355,8 @@ SessionID ClientConnection::verifyServerStartupResponse() {
         ss << "Session ID does not match session received from server: " << sessionID_ << " != " << clientSession;
         throw BadValue(ss.str(), Here());
     }
-    if (serverFunctionality.has("NumberOfConnections") && serverFunctionality.getInt("NumberOfConnections") == 1) {
-        single_ = true;
-    }
+    single_ = agreedConf.singleConnection();
+    agreedFeatures_ = agreedConf.enabledFeatures();
 
     if (single_ && !(dataEndpoint_ == controlEndpoint_)) {
         Log::warning() << "Returned control interface does not match. " << dataEndpoint_ << " /= " << controlEndpoint_

src/fdb5/remote/client/ClientConnection.h

@@ -12,6 +12,7 @@
 
 #include "fdb5/remote/Connection.h"
 #include "fdb5/remote/Messages.h"
+#include "fdb5/remote/RemoteConfiguration.h"
 
 #include "eckit/container/Queue.h"
 #include "eckit/io/Buffer.h"
@@ -40,7 +41,6 @@ namespace fdb5::remote {
 class Client;
 class ClientConnectionRouter;
 class DataWriteRequest;
-class RemoteConfiguration;
 
 //----------------------------------------------------------------------------------------------------------------------
 
@@ -100,6 +100,8 @@ class ClientConnection : protected Connection {
 
     const eckit::net::TCPSocket& dataSocket() const override { return dataClient_; }
 
+    void checkEnabled(Message msg) const;
+
 private:  // members
 
     eckit::SessionID sessionID_;
@@ -131,6 +133,8 @@ class ClientConnection : protected Connection {
     std::mutex dataWriteMutex_;
     std::unique_ptr<eckit::Queue<DataWriteRequest>> dataWriteQueue_;
     std::thread dataWriteThread_;
+
+    uint64_t agreedFeatures_{0};
 };
 
 //----------------------------------------------------------------------------------------------------------------------

src/fdb5/remote/client/ClientConnectionRouter.cc

@@ -2,6 +2,7 @@
 
 #include "fdb5/remote/client/ClientConnection.h"
 
+#include "eckit/config/Configuration.h"
 #include "eckit/exception/Exceptions.h"
 #include "eckit/log/Log.h"
 #include "eckit/net/Endpoint.h"

src/fdb5/remote/client/ReadLimiter.h

@@ -13,8 +13,8 @@
 #include "eckit/io/Buffer.h"
 #include "eckit/serialisation/MemoryStream.h"
 
+#include "fdb5/database/FieldLocation.h"
 #include "fdb5/database/Key.h"
-#include "fdb5/remote/RemoteFieldLocation.h"
 
 #include <cstdint>
 #include <deque>

src/fdb5/remote/server/CatalogueHandler.cc

@@ -101,9 +101,14 @@ Handled CatalogueHandler::handleControl(Message message, uint32_t clientID, uint
 Handled CatalogueHandler::handleControl(Message message, uint32_t clientID, uint32_t requestID,
                                         eckit::Buffer&& payload) {
 
-    static bool wipeEnabled = Resource<bool>("fdbWipeEnabled;$FDB_WIPE_ENABLED", false);
-
     try {
+        if (!enabled(agreedConf_.enabledFeatures(), message)) {
+            std::ostringstream ss;
+            ss << "Unauthorized message: " << message;
+            unauthorised(ss.str(), clientID, requestID);
+            return Handled::Replied;
+        }
+
         switch (message) {
 
             case Message::Schema:  // request catalogue schema
@@ -135,51 +140,27 @@ Handled CatalogueHandler::handleControl(Message message, uint32_t clientID, uint
                 return Handled::Replied;
 
             case Message::Wipe:  // Initial wipe request
-                if (!wipeEnabled) {
-                    unauthorised("Wipe functionality is not enabled", clientID, requestID);
-                    return Handled::Replied;
-                }
                 wipe(clientID, requestID, std::move(payload));
                 return Handled::Yes;
 
             case Message::DoMaskIndexEntries:
-                if (!wipeEnabled) {
-                    unauthorised("Wipe functionality is not enabled", clientID, requestID);
-                    return Handled::Replied;
-                }
                 // doit! We expect DoMaskIndexEntries, doWipeURIs, DoWipeUnknowns and doWipeEmptyDatabase in succession
                 doMaskIndexEntries(clientID, requestID, std::move(payload));
                 return Handled::Yes;
 
             case Message::DoWipeURIs:  // Do the wipe on our currentWipeState
-                if (!wipeEnabled) {
-                    unauthorised("Wipe functionality is not enabled", clientID, requestID);
-                    return Handled::Replied;
-                }
                 doWipeURIs(clientID, requestID, std::move(payload));
                 return Handled::Yes;
 
             case Message::DoWipeFinish:  // Finish wipe by deleting empty DBs
-                if (!wipeEnabled) {
-                    unauthorised("Wipe functionality is not enabled", clientID, requestID);
-                    return Handled::Replied;
-                }
                 doWipeEmptyDatabase(clientID, requestID, std::move(payload));
                 return Handled::Yes;
 
             case Message::DoWipeUnknowns:  // Wipe a set of unknown URIs
-                if (!wipeEnabled) {
-                    unauthorised("Wipe functionality is not enabled", clientID, requestID);
-                    return Handled::Replied;
-                }
                 doWipeUnknowns(clientID, requestID, std::move(payload));
                 return Handled::Yes;
 
             case Message::DoUnsafeFullWipe:  // wipe a full database including its content
-                if (!wipeEnabled) {
-                    unauthorised("Wipe functionality is not enabled", clientID, requestID);
-                    return Handled::Replied;
-                }
                 doUnsafeFullWipe(clientID, requestID, std::move(payload));
                 return Handled::Replied;