Feature/market data store (#476)

* SectorBlocks migration to Datastore

Signed-off-by: elestrias <[email protected]>

* Code upgrade, new test for duplicates

Signed-off-by: elestrias <[email protected]>

* sectorblocks pr fixes, test fixes

* Fixes

Signed-off-by: elestrias <[email protected]>

* logger update

Signed-off-by: elestrias <[email protected]>

Author: Elestrias

core/api/storage_miner/storage_api.hpp

@@ -58,6 +58,7 @@ namespace fc::api {
     PaddedPieceSize offset;
     PaddedPieceSize length;
   };
+  CBOR_TUPLE(PieceLocation, sector_number, offset, length)
 
   // TODO(ortyomka): [FIL-421] implement it
   struct ApiSectorInfo {

core/miner/main/main.cpp

@@ -452,7 +452,8 @@ namespace fc {
         prefixed("stored_ask/"), napi, *config.actor)};
     auto piece_storage{std::make_shared<storage::piece::PieceStorageImpl>(
         prefixed("storage_provider/"))};
-    auto sector_blocks{std::make_shared<sectorblocks::SectorBlocksImpl>(miner)};
+    auto sector_blocks{std::make_shared<sectorblocks::SectorBlocksImpl>(
+        miner, prefixed("sealedblocks/"))};
     auto chain_events{std::make_shared<ChainEventsImpl>(
         napi, ChainEventsImpl::IsDealPrecommited{})};
     OUTCOME_TRY(chain_events->init());

core/sectorblocks/blocks.hpp

@@ -34,6 +34,7 @@ namespace fc::sectorblocks {
 
   enum class SectorBlocksError {
     kNotFoundDeal = 1,
+    kDealAlreadyExist = 2,
   };
 
 }  // namespace fc::sectorblocks

core/sectorblocks/impl/blocks_impl.cpp

@@ -4,10 +4,18 @@
  */
 
 #include "sectorblocks/impl/blocks_impl.hpp"
+#include "adt/uvarint_key.hpp"
+#include "codec/cbor/cbor_codec.hpp"
 
 namespace fc::sectorblocks {
-  SectorBlocksImpl::SectorBlocksImpl(std::shared_ptr<Miner> miner)
-      : miner_{miner} {}
+  using adt::UvarintKeyer;
+  using codec::cbor::decode;
+  using codec::cbor::encode;
+
+  SectorBlocksImpl::SectorBlocksImpl(
+      std::shared_ptr<Miner> miner,
+      std::shared_ptr<DataStore> datastore)
+      : miner_(std::move(miner)), storage_(std::move(datastore)) {}
 
   outcome::result<PieceAttributes> SectorBlocksImpl::addPiece(
       UnpaddedPieceSize size,
@@ -28,21 +36,42 @@ namespace fc::sectorblocks {
                                                    PaddedPieceSize offset,
                                                    UnpaddedPieceSize size) {
     std::lock_guard lock(mutex_);
-    storage_[deal_id].push_back(PieceLocation{
+
+    const Buffer key = Buffer(UvarintKeyer::encode(deal_id));
+    std::vector<PieceLocation> new_data;
+    const auto new_piece = PieceLocation{
         .sector_number = sector,
         .offset = offset,
         .length = size.padded(),
-    });
-    return outcome::success();
+    };
+
+    if (storage_->contains(key)) {
+      OUTCOME_TRY(stored_data, storage_->get(key));
+      OUTCOME_TRY(decoded_out, decode<std::vector<PieceLocation>>(stored_data));
+      if (find(decoded_out.begin(), decoded_out.end(), new_piece)
+          == decoded_out.end()) {
+        decoded_out.push_back(new_piece);
+        new_data = std::move(decoded_out);
+      } else {
+        return SectorBlocksError::kDealAlreadyExist;
+      }
+    } else {
+      new_data = {new_piece};
+    }
+
+    OUTCOME_TRY(encoded_in, encode(new_data));
+    return storage_->put(key, encoded_in);
   }
 
   outcome::result<std::vector<PieceLocation>> SectorBlocksImpl::getRefs(
       DealId deal_id) const {
-    auto refs = storage_.find(deal_id);
-    if (refs == storage_.end()) {
-      return SectorBlocksError::kNotFoundDeal;
+    const Buffer key = Buffer(UvarintKeyer::encode(deal_id));
+    if (storage_->contains(key)) {
+      OUTCOME_TRY(stored_data, storage_->get(key));
+      OUTCOME_TRY(decoded_out, decode<std::vector<PieceLocation>>(stored_data));
+      return std::move(decoded_out);
     }
-    return refs->second;
+    return SectorBlocksError::kNotFoundDeal;
   }
 
   std::shared_ptr<Miner> SectorBlocksImpl::getMiner() const {
@@ -55,7 +84,10 @@ OUTCOME_CPP_DEFINE_CATEGORY(fc::sectorblocks, SectorBlocksError, e) {
   switch (e) {
     case (SectorBlocksError::kNotFoundDeal):
       return "SectorBlocks: not found";
+    case(SectorBlocksError::kDealAlreadyExist):
+      return "SectorBlocks: piece already exist in provided deal";
     default:
       return "SectorBlocks: unknown error";
+
   }
 }

core/sectorblocks/impl/blocks_impl.hpp

@@ -6,14 +6,17 @@
 #pragma once
 
 #include "sectorblocks/blocks.hpp"
+#include "storage/buffer_map.hpp"
 
 namespace fc::sectorblocks {
   using primitives::SectorNumber;
   using primitives::piece::PaddedPieceSize;
+  using DataStore = storage::PersistentBufferMap;
 
   class SectorBlocksImpl : public SectorBlocks {
    public:
-    SectorBlocksImpl(std::shared_ptr<Miner> miner);
+    SectorBlocksImpl(std::shared_ptr<Miner> miner,
+                     std::shared_ptr<DataStore> datastore);
 
     outcome::result<PieceAttributes> addPiece(
         UnpaddedPieceSize size,
@@ -31,11 +34,12 @@ namespace fc::sectorblocks {
                                    PaddedPieceSize offset,
                                    UnpaddedPieceSize size);
 
+    // TODO(@Elestrias):[FIL-423] Make deletion of expired deal associations;
+
     std::shared_ptr<Miner> miner_;
 
     std::mutex mutex_;
-    std::map<DealId, std::vector<PieceLocation>>
-        storage_;  // TODO(ortyomka): [FIL-353] change to DataStore
+    std::shared_ptr<DataStore> storage_;
   };
 
 }  // namespace fc::sectorblocks

test/core/sectorblocks/CMakeLists.txt

@@ -9,4 +9,5 @@ addtest(sectorblocks_test
 target_link_libraries(sectorblocks_test
         base_fs_test
         sectorblocks
+        in_memory_storage
         )

test/core/sectorblocks/sectorblocks_test.cpp

@@ -8,24 +8,27 @@
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
+#include "storage/in_memory/in_memory_storage.hpp"
 #include "testutil/mocks/miner/miner_mock.hpp"
 #include "testutil/outcome.hpp"
 
 namespace fc::sectorblocks {
   using api::PieceLocation;
   using miner::MinerMock;
+  using storage::InMemoryStorage;
   using testing::_;
 
   class SectorBlocksTest : public ::testing::Test {
    protected:
     void SetUp() override {
       miner_ = std::make_unique<MinerMock>();
-
-      sector_blocks_ = std::make_unique<SectorBlocksImpl>(miner_);
+      datastore_ = std::make_shared<InMemoryStorage>();
+      sector_blocks_ = std::make_unique<SectorBlocksImpl>(miner_, datastore_);
     }
 
     std::shared_ptr<MinerMock> miner_;
     std::shared_ptr<SectorBlocks> sector_blocks_;
+    std::shared_ptr<InMemoryStorage> datastore_;
   };
 
   /**
@@ -90,4 +93,38 @@ namespace fc::sectorblocks {
     ASSERT_THAT(refs, testing::ElementsAre(result_ref));
   }
 
+  /**
+   * @given  sectorblocks, deal, size, and path
+   * @when try to add two duplicate pieces to the same deal_id
+   * @then EXPECT_OUTCOME_ERROR
+   */
+  TEST_F(SectorBlocksTest, DuplicatePiece) {
+    const DealInfo deal{
+        .publish_cid = boost::none,
+        .deal_id = 1,
+        .deal_schedule =
+            {
+                .start_epoch = 10,
+                .end_epoch = 11,
+            },
+        .is_keep_unsealed = false,
+    };
+    const UnpaddedPieceSize size(127);
+    const std::string path = "/some/temp/path";
+
+    const PieceAttributes piece{
+        .sector = 1,
+        .offset = PaddedPieceSize(0),
+        .size = UnpaddedPieceSize(127),
+    };
+
+    EXPECT_CALL(*miner_, doAddPieceToAnySector(size, _, deal))
+        .WillOnce(testing::Return(outcome::success(piece)))
+        .WillOnce(testing::Return(outcome::success(piece)));
+
+    EXPECT_OUTCOME_EQ(sector_blocks_->addPiece(size, path, deal), piece);
+    EXPECT_OUTCOME_ERROR(SectorBlocksError::kDealAlreadyExist,
+                         sector_blocks_->addPiece(size, path, deal));
+  }
+
 }  // namespace fc::sectorblocks