Fix work id for local workers (#611)

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

Author: ortyomka

core/sector_storage/impl/local_worker.cpp

@@ -908,4 +908,8 @@ namespace fc::sector_storage {
 
     return call_id;
   }
+
+  bool LocalWorker::isLocalWorker() const {
+    return true;
+  }
 }  // namespace fc::sector_storage

core/sector_storage/impl/local_worker.hpp

@@ -97,6 +97,8 @@ namespace fc::sector_storage {
     outcome::result<std::vector<primitives::StoragePath>> getAccessiblePaths()
         override;
 
+    bool isLocalWorker() const override;
+
    private:
     template <typename W, typename R>
     outcome::result<CallId> asyncCall(const SectorRef &sector,

core/sector_storage/impl/remote_worker.cpp

@@ -249,4 +249,8 @@ namespace fc::sector_storage {
                                               AcquireMode mode) {
     return api_.Fetch(sector, file_type, path_type, mode);
   }
+
+  bool RemoteWorker::isLocalWorker() const {
+    return false;
+  }
 }  // namespace fc::sector_storage

core/sector_storage/impl/remote_worker.hpp

@@ -95,6 +95,8 @@ namespace fc::sector_storage {
     outcome::result<std::vector<primitives::StoragePath>> getAccessiblePaths()
         override;
 
+    bool isLocalWorker() const override;
+
    private:
     explicit RemoteWorker(io_context &context);
 

core/sector_storage/impl/scheduler_impl.cpp

@@ -103,12 +103,15 @@ namespace fc::sector_storage {
                 const std::shared_ptr<Worker> &worker)
           -> outcome::result<CallId> {
         OUTCOME_TRY(call_id, old_job(worker));
-        WorkState state;
-        state.id = wid;
-        state.status = WorkStatus::kInProgress;
-        state.call_id = call_id;
-        OUTCOME_TRY(state_raw, codec::cbor::encode(state));
-        OUTCOME_TRY(kv->put(static_cast<Bytes>(wid), std::move(state_raw)));
+        // check that the work running on a remote worker
+        if (not worker->isLocalWorker()) {
+          WorkState state;
+          state.id = wid;
+          state.status = WorkStatus::kInProgress;
+          state.call_id = call_id;
+          OUTCOME_TRY(state_raw, codec::cbor::encode(state));
+          OUTCOME_TRY(kv->put(static_cast<Bytes>(wid), std::move(state_raw)));
+        }
         return std::move(call_id);
       };
     }

core/sector_storage/worker.hpp

@@ -142,6 +142,8 @@ namespace fc::sector_storage {
 
     virtual outcome::result<std::vector<primitives::StoragePath>>
     getAccessiblePaths() = 0;
+
+    virtual bool isLocalWorker() const = 0;
   };
 
   enum class CallErrorCode : uint64_t {

test/core/sector_storage/scheduler_test.cpp

@@ -11,6 +11,7 @@
 #include "sector_storage/schedulder_utils.hpp"
 #include "storage/in_memory/in_memory_storage.hpp"
 #include "testutil/mocks/sector_storage/selector_mock.hpp"
+#include "testutil/mocks/sector_storage/worker_mock.hpp"
 #include "testutil/outcome.hpp"
 
 namespace fc::sector_storage {
@@ -72,9 +73,10 @@ namespace fc::sector_storage {
       scheduler_ = scheduler;
 
       std::unique_ptr<WorkerHandle> worker = std::make_unique<WorkerHandle>();
+      mock_worker_ = std::make_unique<WorkerMock>();
 
       worker_name_ = "worker";
-
+      worker->worker = mock_worker_;
       worker->info = WorkerInfo{
           .hostname = worker_name_,
           .resources = WorkerResources{.physical_memory = uint64_t(1) << 20,
@@ -89,7 +91,7 @@ namespace fc::sector_storage {
 
     std::string worker_name_;
     std::vector<WorkState> states_;
-
+    std::shared_ptr<WorkerMock> mock_worker_;
     std::shared_ptr<boost::asio::io_context> io_;
     RegisteredSealProof seal_proof_type_;
     std::shared_ptr<InMemoryStorage> kv_;
@@ -340,6 +342,9 @@ namespace fc::sector_storage {
       is_first_called = true;
     };
 
+    EXPECT_CALL(*mock_worker_, isLocalWorker())
+        .WillOnce(testing::Return(false));
+
     EXPECT_CALL(
         *selector_,
         is_satisfying(task, seal_proof_type_, workerNameMatcher(worker_name_)))

test/testutil/mocks/sector_storage/worker_mock.hpp

@@ -100,5 +100,7 @@ namespace fc::sector_storage {
                                          const std::vector<UnpaddedPieceSize> &,
                                          const UnpaddedPieceSize &,
                                          int));
+
+    MOCK_CONST_METHOD0(isLocalWorker, bool());
   };
 }  // namespace fc::sector_storage