audi timestamp (#2151)

Signed-off-by: turuslan <turuslan.devbox@gmail.com>
Co-authored-by: kamilsa <kamilsa16@gmail.com>

cmake/functions.cmake

@@ -86,7 +86,7 @@ function(compile_proto_to_cpp PROTO_LIBRARY_NAME PB_H PB_CC PROTO)
       COMMAND ${GEN_COMMAND}
       ARGS -I${PROJECT_SOURCE_DIR}/core -I${GEN_ARGS} --cpp_out=${SCHEMA_OUT_DIR} ${PROTO_ABS}
       WORKING_DIRECTORY ${CMAKE_BINARY_DIR}
-      DEPENDS ${PROTOBUF_DEPENDS}
+      DEPENDS ${PROTOBUF_DEPENDS} ${PROTO_ABS}
       VERBATIM
   )
 

core/authority_discovery/protobuf/authority_discovery.v2.proto

@@ -6,24 +6,32 @@
 
 syntax = "proto3";
 
-package authority_discovery.v2;
+package authority_discovery_v3;
 
 // First we need to serialize the addresses in order to be able to sign them.
 message AuthorityRecord {
-	// Possibly multiple `MultiAddress`es through which the node can be
+	// Possibly multiple `MultiAddress`es through which the node can be reached.
 	repeated bytes addresses = 1;
+	// Information about the creation time of the record
+	TimestampInfo creation_time = 2;
 }
 
 message PeerSignature {
 	bytes signature = 1;
 	bytes public_key = 2;
 }
 
+// Information regarding the creation data of the record
+message TimestampInfo {
+	// Time since UNIX_EPOCH in nanoseconds, scale encoded
+	bytes timestamp = 1;
+}
+
 // Then we need to serialize the authority record and signature to send them over the wire.
 message SignedAuthorityRecord {
 	bytes record = 1;
 	bytes auth_signature = 2;
 	// Even if there are multiple `record.addresses`, all of them have the same peer id.
-    // Old versions are missing this field. It is optional in order to provide compatibility both ways.
+	// Old versions are missing this field. It is optional in order to provide compatibility both ways.
 	PeerSignature peer_signature = 3;
 }

core/authority_discovery/publisher/address_publisher.cpp

@@ -8,6 +8,7 @@
 
 #include "authority_discovery/protobuf/authority_discovery.v2.pb.h"
 
+#include "authority_discovery/timestamp.hpp"
 #include "crypto/sha/sha256.hpp"
 
 #define _PB_SPAN(f) \
@@ -128,17 +129,23 @@ namespace kagome::authority_discovery {
       OUTCOME_TRY(address2, libp2p::multi::Multiaddress::create(s));
       addresses.emplace(std::move(address2));
     }
-    ::authority_discovery::v2::AuthorityRecord record;
+    ::authority_discovery_v3::AuthorityRecord record;
     for (const auto &address : addresses) {
       PB_SPAN_ADD(record, addresses, address.getBytesAddress());
     }
+    TimestampScale time{std::chrono::nanoseconds{
+        std::chrono::system_clock::now().time_since_epoch()}
+                            .count()};
+    PB_SPAN_SET(*record.mutable_creation_time(),
+                timestamp,
+                scale::encode(time).value());
 
     auto record_pb = pbEncodeVec(record);
     OUTCOME_TRY(signature, ed_crypto_provider_->sign(*libp2p_key_, record_pb));
     OUTCOME_TRY(auth_signature,
                 sr_crypto_provider_->sign(*audi_key, record_pb));
 
-    ::authority_discovery::v2::SignedAuthorityRecord signed_record;
+    ::authority_discovery_v3::SignedAuthorityRecord signed_record;
     PB_SPAN_SET(signed_record, auth_signature, auth_signature);
     PB_SPAN_SET(signed_record, record, record_pb);
     auto &ps = *signed_record.mutable_peer_signature();

core/authority_discovery/query/query_impl.cpp

@@ -23,6 +23,8 @@ OUTCOME_CPP_DEFINE_CATEGORY(kagome::authority_discovery, QueryImpl::Error, e) {
       return "Inconsistent peer id";
     case E::INVALID_SIGNATURE:
       return "Invalid signature";
+    case E::KADEMLIA_OUTDATED_VALUE:
+      return "Kademlia outdated value";
   }
   return "unknown error (authority_discovery::QueryImpl::Error)";
 }
@@ -41,7 +43,7 @@ namespace kagome::authority_discovery {
       std::shared_ptr<libp2p::crypto::CryptoProvider> libp2p_crypto_provider,
       std::shared_ptr<libp2p::crypto::marshaller::KeyMarshaller> key_marshaller,
       libp2p::Host &host,
-      std::shared_ptr<libp2p::protocol::kademlia::Kademlia> kademlia,
+      LazySPtr<libp2p::protocol::kademlia::Kademlia> kademlia,
       std::shared_ptr<libp2p::basic::Scheduler> scheduler)
       : block_tree_{std::move(block_tree)},
         authority_discovery_api_{std::move(authority_discovery_api)},
@@ -80,7 +82,7 @@ namespace kagome::authority_discovery {
     std::unique_lock lock{mutex_};
     auto it = auth_to_peer_cache_.find(authority);
     if (it != auth_to_peer_cache_.end()) {
-      return it->second;
+      return it->second.peer;
     }
     return std::nullopt;
   }
@@ -95,11 +97,51 @@ namespace kagome::authority_discovery {
     return std::nullopt;
   }
 
+  outcome::result<void> QueryImpl::validate(
+      const libp2p::protocol::kademlia::Key &key,
+      const libp2p::protocol::kademlia::Value &value) {
+    std::unique_lock lock{mutex_};
+    auto id = hashToAuth(key);
+    if (not id) {
+      lock.unlock();
+      return kademlia_validator_.validate(key, value);
+    }
+    auto r = add(*id, value);
+    if (not r) {
+      SL_DEBUG(log_, "Can't add: {}", r.error());
+    }
+    return r;
+  }
+
+  outcome::result<size_t> QueryImpl::select(
+      const libp2p::protocol::kademlia::Key &key,
+      const std::vector<libp2p::protocol::kademlia::Value> &values) {
+    std::unique_lock lock{mutex_};
+    auto id = hashToAuth(key);
+    if (not id) {
+      lock.unlock();
+      return kademlia_validator_.select(key, values);
+    }
+    auto it = auth_to_peer_cache_.find(*id);
+    if (it != auth_to_peer_cache_.end()) {
+      auto it_value = std::find(values.begin(), values.end(), it->second.raw);
+      if (it_value != values.end()) {
+        return it_value - values.begin();
+      }
+    }
+    return Error::KADEMLIA_OUTDATED_VALUE;
+  }
+
   outcome::result<void> QueryImpl::update() {
     std::unique_lock lock{mutex_};
     OUTCOME_TRY(
         authorities,
         authority_discovery_api_->authorities(block_tree_->bestBlock().hash));
+    for (auto &id : authorities) {
+      if (not hash_to_auth_.contains(id)) {
+        hash_to_auth_.emplace(crypto::sha256(id), id);
+      }
+    }
     OUTCOME_TRY(local_keys,
                 key_store_->sr25519().getPublicKeys(
                     crypto::KeyTypes::AUTHORITY_DISCOVERY));
@@ -132,6 +174,17 @@ namespace kagome::authority_discovery {
     return outcome::success();
   }
 
+  std::optional<primitives::AuthorityDiscoveryId> QueryImpl::hashToAuth(
+      BufferView key) const {
+    if (auto r = Hash256::fromSpan(key)) {
+      auto it = hash_to_auth_.find(r.value());
+      if (it != hash_to_auth_.end()) {
+        return it->second;
+      }
+    }
+    return std::nullopt;
+  }
+
   void QueryImpl::pop() {
     while (active_ < kMaxActiveRequests) {
       if (queue_.empty()) {
@@ -144,19 +197,11 @@ namespace kagome::authority_discovery {
       common::Buffer hash{crypto::sha256(authority)};
       scheduler_->schedule([=, this, wp{weak_from_this()}] {
         if (auto self = wp.lock()) {
-          std::ignore = kademlia_->getValue(
+          std::ignore = kademlia_.get()->getValue(
               hash, [=, this](outcome::result<std::vector<uint8_t>> res) {
                 std::unique_lock lock{mutex_};
                 --active_;
                 pop();
-                if (res.has_error()) {
-                  SL_DEBUG(log_, "Kademlia can't get value: {}", res.error());
-                  return;
-                }
-                auto r = add(authority, std::move(res.value()));
-                if (not r) {
-                  SL_DEBUG(log_, "Can't add: {}", r.error());
-                }
               });
         }
       });
@@ -167,7 +212,12 @@ namespace kagome::authority_discovery {
       const primitives::AuthorityDiscoveryId &authority,
       outcome::result<std::vector<uint8_t>> _res) {
     OUTCOME_TRY(signed_record_pb, _res);
-    ::authority_discovery::v2::SignedAuthorityRecord signed_record;
+    auto it = auth_to_peer_cache_.find(authority);
+    if (it != auth_to_peer_cache_.end()
+        and signed_record_pb == it->second.raw) {
+      return outcome::success();
+    }
+    ::authority_discovery_v3::SignedAuthorityRecord signed_record;
     if (not signed_record.ParseFromArray(signed_record_pb.data(),
                                          signed_record_pb.size())) {
       return Error::DECODE_ERROR;
@@ -185,13 +235,23 @@ namespace kagome::authority_discovery {
                 crypto::Sr25519Signature::fromSpan(
                     str2byte(signed_record.auth_signature())));
 
-    ::authority_discovery::v2::AuthorityRecord record;
+    ::authority_discovery_v3::AuthorityRecord record;
     if (not record.ParseFromString(signed_record.record())) {
       return Error::DECODE_ERROR;
     }
     if (record.addresses().empty()) {
       return Error::NO_ADDRESSES;
     }
+    std::optional<Timestamp> time{};
+    if (record.has_creation_time()) {
+      OUTCOME_TRY(tmp,
+                  scale::decode<TimestampScale>(
+                      qtils::str2byte(record.creation_time().timestamp())));
+      time = *tmp;
+      if (it != auth_to_peer_cache_.end() and time <= it->second.time) {
+        return outcome::success();
+      }
+    }
     libp2p::peer::PeerInfo peer{std::move(peer_id), {}};
     auto peer_id_str = peer.id.toBase58();
     for (auto &pb : record.addresses()) {
@@ -226,7 +286,9 @@ namespace kagome::authority_discovery {
         peer.id, peer.addresses, libp2p::peer::ttl::kRecentlyConnected);
 
     peer_to_auth_cache_.insert_or_assign(peer.id, authority);
-    auth_to_peer_cache_.insert_or_assign(authority, std::move(peer));
+    auth_to_peer_cache_.insert_or_assign(
+        authority,
+        Authority{std::move(signed_record_pb), time, std::move(peer)});
 
     return outcome::success();
   }

core/authority_discovery/query/query_impl.hpp

@@ -10,28 +10,33 @@
 
 #include "application/app_state_manager.hpp"
 #include "authority_discovery/interval.hpp"
+#include "authority_discovery/timestamp.hpp"
 #include "blockchain/block_tree.hpp"
 #include "crypto/key_store.hpp"
 #include "crypto/sr25519_provider.hpp"
+#include "injector/lazy.hpp"
 #include "log/logger.hpp"
 #include "runtime/runtime_api/authority_discovery_api.hpp"
 
 #include <libp2p/crypto/crypto_provider.hpp>
 #include <libp2p/crypto/key_marshaller.hpp>
 #include <libp2p/host/host.hpp>
+#include <libp2p/protocol/kademlia/impl/validator_default.hpp>
 #include <libp2p/protocol/kademlia/kademlia.hpp>
 #include <mutex>
 #include <random>
 
 namespace kagome::authority_discovery {
   class QueryImpl : public Query,
+                    public libp2p::protocol::kademlia::Validator,
                     public std::enable_shared_from_this<QueryImpl> {
    public:
     enum class Error {
       DECODE_ERROR = 1,
       NO_ADDRESSES,
       INCONSISTENT_PEER_ID,
       INVALID_SIGNATURE,
+      KADEMLIA_OUTDATED_VALUE,
     };
 
     QueryImpl(
@@ -44,7 +49,7 @@ namespace kagome::authority_discovery {
         std::shared_ptr<libp2p::crypto::marshaller::KeyMarshaller>
             key_marshaller,
         libp2p::Host &host,
-        std::shared_ptr<libp2p::protocol::kademlia::Kademlia> kademlia,
+        LazySPtr<libp2p::protocol::kademlia::Kademlia> kademlia,
         std::shared_ptr<libp2p::basic::Scheduler> scheduler);
 
     bool start();
@@ -55,9 +60,25 @@ namespace kagome::authority_discovery {
     std::optional<primitives::AuthorityDiscoveryId> get(
         const libp2p::peer::PeerId &peer_id) const override;
 
+    // interface: Validator
+    outcome::result<void> validate(
+        const libp2p::protocol::kademlia::Key &,
+        const libp2p::protocol::kademlia::Value &) override;
+    outcome::result<size_t> select(
+        const libp2p::protocol::kademlia::Key &,
+        const std::vector<libp2p::protocol::kademlia::Value> &) override;
+
     outcome::result<void> update();
 
    private:
+    struct Authority {
+      Buffer raw;
+      std::optional<Timestamp> time;
+      libp2p::peer::PeerInfo peer;
+    };
+
+    std::optional<primitives::AuthorityDiscoveryId> hashToAuth(
+        BufferView key) const;
     void pop();
     outcome::result<void> add(const primitives::AuthorityDiscoveryId &authority,
                               outcome::result<std::vector<uint8_t>> _res);
@@ -69,13 +90,15 @@ namespace kagome::authority_discovery {
     std::shared_ptr<libp2p::crypto::CryptoProvider> libp2p_crypto_provider_;
     std::shared_ptr<libp2p::crypto::marshaller::KeyMarshaller> key_marshaller_;
     libp2p::Host &host_;
-    std::shared_ptr<libp2p::protocol::kademlia::Kademlia> kademlia_;
+    LazySPtr<libp2p::protocol::kademlia::Kademlia> kademlia_;
     std::shared_ptr<libp2p::basic::Scheduler> scheduler_;
     ExpIncInterval interval_;
 
     mutable std::mutex mutex_;
     std::default_random_engine random_;
-    std::unordered_map<primitives::AuthorityDiscoveryId, libp2p::peer::PeerInfo>
+    libp2p::protocol::kademlia::ValidatorDefault kademlia_validator_;
+    std::unordered_map<Hash256, primitives::AuthorityDiscoveryId> hash_to_auth_;
+    std::unordered_map<primitives::AuthorityDiscoveryId, Authority>
         auth_to_peer_cache_;
     std::unordered_map<libp2p::peer::PeerId, primitives::AuthorityDiscoveryId>
         peer_to_auth_cache_;

core/authority_discovery/timestamp.hpp

@@ -0,0 +1,14 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "scale/big_fixed_integers.hpp"
+
+namespace kagome::authority_discovery {
+  using Timestamp = scale::uint128_t;
+  using TimestampScale = scale::Fixed<Timestamp>;
+}  // namespace kagome::authority_discovery

core/crypto/bandersnatch/vrf.cpp

@@ -6,8 +6,6 @@
 
 #include "crypto/bandersnatch/vrf.hpp"
 
-#include "utils/non_null_dangling.hpp"
-
 namespace kagome::crypto::bandersnatch::vrf {
 
   VrfInput vrf_input(BytesIn domain, BytesIn data) {
@@ -83,10 +81,10 @@ namespace kagome::crypto::bandersnatch::vrf {
 
     auto res = ::bandersnatch_vrf_sign_data(label.data(),
                                             label.size(),
-                                            nonNullDangling(data_ptrs),
-                                            nonNullDangling(data_sizes),
+                                            data_ptrs.data(),
+                                            data_sizes.data(),
                                             data.size(),
-                                            nonNullDangling(input_ptrs),
+                                            input_ptrs.data(),
                                             input_ptrs.size());
     return VrfSignData(res);
   }