chore(tiering): Fix asserts around reads for serialization (#5802)

Signed-off-by: Vladislav <[email protected]>
Signed-off-by: Vladislav Oleshko <[email protected]>

Author: dranikpg

src/facade/op_status.h

@@ -33,6 +33,7 @@ enum class OpStatus : uint16_t {
   MEMBER_NOTFOUND,
   INVALID_JSON_PATH,
   INVALID_JSON,
+  IO_ERROR
 };
 
 class OpResultBase {

src/server/generic_family.cc

@@ -551,12 +551,13 @@ OpResult<std::string> OpDump(const OpArgs& op_args, string_view key) {
     const PrimeValue& pv = it->second;
 
     if (pv.IsExternal() && !pv.IsCool()) {
-      util::fb2::Future<io::Result<string>> future =
-          op_args.shard->tiered_storage()->Read(op_args.db_cntx.db_index, key, pv);
+      // TODO: consider moving blocking point to coordinator to avoid stalling shard queue
+      auto res = op_args.shard->tiered_storage()->Read(op_args.db_cntx.db_index, key, pv).Get();
+      if (!res.has_value())
+        return OpStatus::IO_ERROR;
 
-      const io::Result<string>& res = future.Get();
-      CompactObj co(*res, false);  // TODO: handle errors.
-      SerializerBase::DumpObject(co, &sink);
+      // TODO: allow saving string directly without proxy object
+      SerializerBase::DumpObject(PrimeValue{*res, false}, &sink);
     } else {
       SerializerBase::DumpObject(it->second, &sink);
     }

src/server/snapshot.cc

@@ -407,9 +407,15 @@ bool SliceSnapshot::PushSerialized(bool force) {
       // 1. We may block here too frequently, slowing down the process.
       // 2. For small bin values, we issue multiple reads for the same page, creating
       //    read factor amplification that can reach factor of ~60.
-      PrimeValue pv{*entry.value.Get(), false};  // Might block until the future resolves.
+      auto res = entry.value.Get();  // Blocking point
+      if (!res.has_value()) {
+        cntx_->ReportError(make_error_code(errc::io_error),
+                           absl::StrCat("Failed to read ", entry.key.ToString()));
+        return false;
+      }
 
       // TODO: to introduce RdbSerializer::SaveString that can accept a string value directly.
+      PrimeValue pv{*res, false};
       serializer_->SaveEntry(entry.key, pv, entry.expire, entry.mc_flags, entry.dbid);
     } while (!delayed_entries_.empty());