chore: Extend TieredStorage to support QList::Node

* Add ListNodeId (dbid, ql*, node*) as a new tiering identity type
  alongside KeyRef; introduce PendingId/ReadId variants so OpManager
  can handle both key-level and node-level I/O uniformly.

* Update CancelStash / ReadInternal signatures to accept the generic
  PendingId / ReadId types instead of (dbid, key).

* Add flag `--tiered_experimental_list_support` that is used to enable
  list node tiered storage functionality.

* Added StashListNode / ReadTieredListNode.

Signed-off-by: mkaruza <mario@dragonflydb.io>

Author: mkaruza

src/server/db_slice.cc

@@ -631,7 +631,7 @@ auto DbSlice::FindInternal(const Context& cntx, string_view key, optional<unsign
   // Rationale: we either look it up for reads - and then it's hot, or alternatively,
   // we follow up with modifications, so the pending stash becomes outdated.
   if (pv.HasStashPending()) {
-    owner_->tiered_storage()->CancelStash(cntx.db_index, key, &pv);
+    owner_->tiered_storage()->CancelStash(std::make_pair(cntx.db_index, key), &pv);
   }
 
   // Fetch back cool items
@@ -1564,7 +1564,8 @@ void DbSlice::RemoveOffloadedEntriesFromTieredStorage(absl::Span<const DbIndex>
         if (it->second.IsExternal()) {
           tiered_storage->Delete(index, &it->second);
         } else if (it->second.HasStashPending()) {
-          tiered_storage->CancelStash(index, it->first.GetSlice(&scratch), &it->second);
+          tiered_storage->CancelStash(std::make_pair(index, it->first.GetSlice(&scratch)),
+                                      &it->second);
         }
       });
     } while (cursor);
@@ -1773,7 +1774,7 @@ void DbSlice::PerformDeletionAtomic(const Iterator& del_it, DbTable* table, bool
   if (pv.HasStashPending()) {
     string scratch;
     string_view key = del_it->first.GetSlice(&scratch);
-    shard_owner()->tiered_storage()->CancelStash(table->index, key, &pv);
+    shard_owner()->tiered_storage()->CancelStash(std::make_pair(table->index, key), &pv);
   } else if (pv.IsExternal()) {
     shard_owner()->tiered_storage()->Delete(table->index, &del_it->second);
   }

src/server/hset_family.cc

@@ -199,8 +199,8 @@ OpResult<T> ExecuteRO(Transaction* tx, F&& f) {
         fut.Resolve(f(hw));
       };
 
-      es->tiered_storage()->Read(op_args.db_cntx.db_index, key, pv.GetExternalSlice(), D{},
-                                 std::move(read_cb));
+      es->tiered_storage()->Read(std::make_pair(op_args.db_cntx.db_index, key),
+                                 pv.GetExternalSlice(), D{}, std::move(read_cb));
       return CbVariant<T>{std::move(fut)};
     }
 

src/server/tiered_storage.cc

@@ -20,6 +20,7 @@
 #include "base/flags.h"
 #include "base/logging.h"
 #include "core/detail/listpack_wrap.h"
+#include "core/qlist.h"
 #include "server/db_slice.h"
 #include "server/engine_shard_set.h"
 #include "server/snapshot.h"
@@ -54,6 +55,8 @@ ABSL_FLAG(float, tiered_upload_threshold, 0.1,
 
 ABSL_FLAG(bool, tiered_experimental_hash_support, false, "Experimental hash datatype offloading");
 
+ABSL_FLAG(bool, tiered_experimental_list_support, false, "Experimental list node offloading");
+
 namespace dfly {
 
 using namespace std;
@@ -119,7 +122,10 @@ size_t TieredStorage::StashDescriptor::Serialize(io::MutableBytes buffer) const
           lw, {reinterpret_cast<char*>(buffer.data()), buffer.length()});
     }
     case CompactObj::ExternalRep::LIST_NODE: {
-      // Make compiler happy. It will be implemented in following PR
+      // LIST_NODE uses the string_view pair path (same as STRING).
+      auto strs = std::get<std::array<std::string_view, 2>>(blob);
+      memcpy(buffer.data(), strs[0].data(), strs[0].size());
+      return strs[0].size();
     }
   };
   return 0;
@@ -148,6 +154,32 @@ class TieredStorage::ShardOpManager : public tiering::OpManager {
       ClearStashPending(key);
   }
 
+  // Clear stash pending flag for list node
+  void ClearStashPending(tiering::ListNodeId id) {
+    stats_.total_cancels++;
+    QList::Node* node = reinterpret_cast<QList::Node*>(std::get<2>(id));
+    node->io_pending = 0;
+    // If stashing failed we need to decrease offloaded nodes count.
+    QList* ql = reinterpret_cast<QList*>(std::get<1>(id));
+    ql->IncrementNumOffloadedNodes(-1);
+  }
+
+  void CancelStash(tiering::KeyRef id, size_t size) {
+    UnblockBackpressure(id, false);
+    // TODO: Don't recompute size estimate, try-delete bin first
+    if (OccupiesWholePages(size)) {
+      CancelPending(id);
+    } else if (auto bin = ts_->bins_->Delete(id.first, id.second); bin) {
+      CancelPending(*bin);
+    }
+  }
+
+  void CancelStash(tiering::ListNodeId id) {
+    QList* ql = reinterpret_cast<QList*>(std::get<1>(id));
+    ql->IncrementNumOffloadedNodes(-1);
+    CancelPending(id);
+  }
+
   DbTableStats* GetDbTableStats(DbIndex dbid) {
     return db_slice_.MutableStats(dbid);
   }
@@ -200,7 +232,7 @@ class TieredStorage::ShardOpManager : public tiering::OpManager {
         break;
       }
       case CompactObj::ExternalRep::LIST_NODE: {
-        // Make compiler happy. It will be implemented in following PR.
+        LOG(DFATAL) << "LIST_NODE should not be uploaded to PrimeValue";
         break;
       }
     };
@@ -239,6 +271,26 @@ class TieredStorage::ShardOpManager : public tiering::OpManager {
       SetExternal({sub_dbid, sub_key}, sub_segment);
   }
 
+  // Finalize stash for a fragments identified by pointer
+  void SetExternal(tiering::ListNodeId id, tiering::DiskSegment segment) {
+    auto* stats = GetDbTableStats(std::get<0>(id));
+
+    stats->tiered_entries++;
+    stats->tiered_used_bytes += segment.length;
+    stats_.total_stashes++;
+
+    QList::Node* node = reinterpret_cast<QList::Node*>(std::get<2>(id));
+    QList* ql = reinterpret_cast<QList*>(std::get<1>(id));
+
+    node->io_pending = 0;
+
+    // Adjust parent QList node malloc size / number of offloaded nodes.
+    ql->AdjustMallocSize(-segment.length);
+    node->SetExternal(segment.offset, segment.length);
+
+    stats->AddTypeMemoryUsage(OBJ_LIST, -segment.length);
+  }
+
   // If any backpressure (throttling) is active, notify that the operation finished
   void UnblockBackpressure(OpManager::KeyRef id, bool result) {
     if (auto node = ts_->stash_backpressure_.extract(id); !node.empty())
@@ -293,10 +345,28 @@ bool TieredStorage::ShardOpManager::NotifyFetched(const OwnedEntryId& id,
                                                   tiering::Decoder* decoder) {
   ++stats_.total_fetches;
 
-  if (id == OwnedEntryId{kFragmentedBin}) {  // Generally we read whole bins only for defrag
-    auto* bdecoder = static_cast<tiering::BareDecoder*>(decoder);
-    Defragment(segment, bdecoder->slice);
-    return true;  // delete
+  if (const auto* key = std::get_if<tiering::ListNodeId>(&id); key) {
+    ++stats_.total_uploads;
+
+    QList* ql = reinterpret_cast<QList*>(std::get<1>(*key));
+    // Adjust malloc size and number of offloded nodes before uploading.
+    ql->AdjustMallocSize(segment.length);
+    ql->IncrementNumOffloadedNodes(-1);
+
+    DbTableStats* stats = GetDbTableStats(std::get<0>(*key));
+    stats->AddTypeMemoryUsage(OBJ_LIST, segment.length);
+
+    // We return false here, because we don't want to delete the value from storage yet.
+    // It will be done in onload_cb callback.
+    return false;
+  }
+
+  if (const auto* i = std::get_if<uintptr_t>(&id); i) {
+    if (*i == kFragmentedBin) {  // Generally we read whole bins only for defrag
+      auto* bdecoder = static_cast<tiering::BareDecoder*>(decoder);
+      Defragment(segment, bdecoder->slice);
+      return true;  // delete
+    }
   }
 
   tiering::Decoder::UploadMetrics metrics = decoder->GetMetrics();
@@ -389,6 +459,7 @@ error_code TieredStorage::Open(string_view base_path) {
 }
 
 void TieredStorage::Close() {
+  is_closed_ = true;
   for (auto& [_, f] : stash_backpressure_)
     f.Resolve(false);
   op_manager_->Close();
@@ -398,17 +469,17 @@ bool TieredStorage::HasModificationPending(tiering::DiskSegment segment) const {
   return op_manager_->HasModificationPending(segment);
 }
 
-void TieredStorage::ReadInternal(DbIndex dbid, std::string_view key,
-                                 const tiering::DiskSegment& segment,
+void TieredStorage::ReadInternal(tiering::ReadId id, const tiering::DiskSegment& segment,
                                  const tiering::Decoder& decoder,
                                  std::function<void(io::Result<tiering::Decoder*>)> cb,
                                  bool read_only) {
   // TODO: improve performance by avoiding one more function wrap
-  op_manager_->Enqueue(KeyRef(dbid, key), segment, decoder, std::move(cb), read_only);
+  op_manager_->Enqueue(std::visit([](auto&& value) -> tiering::PendingId { return value; }, id),
+                       segment, decoder, std::move(cb), read_only);
 }
 
-void TieredStorage::Stash(DbIndex dbid, string_view key, const StashDescriptor& blobs,
-                          BackPressureFuture* backpressure) {
+void TieredStorage::StashPrimeValue(DbIndex dbid, string_view key, const StashDescriptor& blobs,
+                                    BackPressureFuture* backpressure) {
   CHECK(!bins_->IsPending(dbid, key));  // Because has stash pending is false (ShouldStash checks)
 
   size_t est_size = blobs.EstimatedSerializedSize();
@@ -448,6 +519,7 @@ void TieredStorage::Stash(DbIndex dbid, string_view key, const StashDescriptor&
 }
 
 void TieredStorage::Delete(DbIndex dbid, FragmentRef fragment_ref) {
+  DCHECK(!is_closed_);
   DCHECK(!fragment_ref.HasStashPending());
   ++stats_.total_deletes;
 
@@ -460,22 +532,17 @@ void TieredStorage::Delete(DbIndex dbid, FragmentRef fragment_ref) {
   op_manager_->DeleteOffloaded(dbid, segment);
 }
 
-void TieredStorage::CancelStash(DbIndex dbid, std::string_view key,
-                                tiering::FragmentRef fragment_ref) {
+void TieredStorage::CancelStash(tiering::PendingId id, tiering::FragmentRef fragment_ref) {
   DCHECK(fragment_ref.HasStashPending());
-
-  // If any previous write was happening, it has been cancelled
-  if (auto node = stash_backpressure_.extract(make_pair(dbid, key)); !node.empty())
-    std::move(node.mapped()).Resolve(false);
-
-  // TODO: Don't recompute size estimate, try-delete bin first
-  StashDescriptor blobs{fragment_ref.GetSerializationDescr()};
-  size_t size = blobs.EstimatedSerializedSize();
-  if (OccupiesWholePages(size)) {
-    op_manager_->CancelPending(KeyRef(dbid, key));
-  } else if (auto bin = bins_->Delete(dbid, key); bin) {
-    op_manager_->CancelPending(*bin);
-  }
+  DCHECK(std::holds_alternative<KeyRef>(id) || std::holds_alternative<tiering::ListNodeId>(id));
+  std::visit(absl::Overload{[&fragment_ref, this](KeyRef id) {
+                              StashDescriptor blobs{fragment_ref.GetSerializationDescr()};
+                              op_manager_->CancelStash(id, blobs.EstimatedSerializedSize());
+                            },
+                            [this](tiering::ListNodeId id) { op_manager_->CancelStash(id); },
+                            // Make variant exhaustive, but we should never call with this type.
+                            [](uintptr_t) { LOG(DFATAL) << "Invalid id type for CancelStash"; }},
+             id);
   fragment_ref.SetStashPending(false);
 }
 
@@ -532,13 +599,15 @@ void TieredStorage::UpdateFromFlags() {
       .offload_threshold = absl::GetFlag(FLAGS_tiered_offload_threshold),
       .upload_threshold = absl::GetFlag(FLAGS_tiered_upload_threshold),
       .experimental_hash_offload = absl::GetFlag(FLAGS_tiered_experimental_hash_support),
+      .experimental_list_offload = absl::GetFlag(FLAGS_tiered_experimental_list_support),
   };
 }
 
 std::vector<std::string> TieredStorage::GetMutableFlagNames() {
   return base::GetFlagNames(FLAGS_tiered_min_value_size, FLAGS_tiered_experimental_cooling,
                             FLAGS_tiered_storage_write_depth, FLAGS_tiered_offload_threshold,
-                            FLAGS_tiered_upload_threshold, FLAGS_tiered_experimental_hash_support);
+                            FLAGS_tiered_upload_threshold, FLAGS_tiered_experimental_hash_support,
+                            FLAGS_tiered_experimental_list_support);
 }
 
 bool TieredStorage::ShouldOffload() const {
@@ -576,7 +645,7 @@ void TieredStorage::RunOffloading(DbIndex dbid) {
       } else {
         stats_.offloading_stashes++;
         it->second.SetStashPending(true);
-        Stash(dbid, it->first.GetSlice(&tmp), *blobs, nullptr);
+        StashPrimeValue(dbid, it->first.GetSlice(&tmp), *blobs, nullptr);
       }
     }
   };
@@ -640,12 +709,20 @@ auto TieredStorage::ShouldStash(const tiering::FragmentRef& fragment_ref) const
   if (fragment_ref.ObjType() == OBJ_HASH && !config_.experimental_hash_offload)
     return nullopt;
 
+  // For now, list node offloading is conditional
+  if (fragment_ref.ObjType() == OBJ_LIST && !config_.experimental_list_offload)
+    return nullopt;
+
   // Estimate value size
   StashDescriptor blobs{fragment_ref.GetSerializationDescr()};
   size_t estimated_size = blobs.EstimatedSerializedSize();
   if (estimated_size < config_.min_value_size)
     return nullopt;
 
+  // If the fragment is list node we offload only if it occupies whole page.
+  if (fragment_ref.ObjType() == OBJ_LIST && !OccupiesWholePages(estimated_size))
+    return nullopt;
+
   // Limit write depth. TODO: Provide backpressure?
   if (op_manager_->GetStats().pending_stash_cnt >= config_.write_depth_limit) {
     ++stats_.stash_overflow_cnt;
@@ -707,7 +784,35 @@ void StashPrimeValue(DbIndex dbid, std::string_view key, PrimeValue* pv, TieredS
                      BackPressureFuture* backpressure) {
   if (auto blobs = ts->ShouldStash(*pv); blobs) {
     pv->SetStashPending(true);
-    ts->Stash(dbid, key, *blobs, backpressure);
+    ts->StashPrimeValue(dbid, key, *blobs, backpressure);
+  }
+}
+
+bool StashListNode(DbIndex dbid, QList::Node* node, QList* ql, TieredStorage* ts,
+                   BackPressureFuture* backpressure) {
+  if (auto blobs = ts->ShouldStash(*node); blobs) {
+    // Increment before stashing; decremented on failure in `ClearStashPending`
+    ql->IncrementNumOffloadedNodes(1);
+    node->io_pending = 1;
+    ts->StashPartialValue(tiering::ListNodeId{dbid, ql, node}, *blobs, backpressure);
+    return true;
+  }
+  return false;
+}
+
+void TieredStorage::StashPartialValue(tiering::PendingId id, const StashDescriptor& blobs,
+                                      BackPressureFuture* backpressure) {
+  size_t est_size = blobs.EstimatedSerializedSize();
+  DCHECK_GT(est_size, 0u);
+
+  auto serialize = absl::bind_front(&StashDescriptor::Serialize, &blobs);
+
+  error_code ec = op_manager_->PrepareAndStash(id, est_size, serialize);
+  if (ec) {
+    bool to_log = ec != errc::file_too_large && ec != errc::operation_would_block &&
+                  ec != errc::operation_in_progress;
+    LOG_IF(ERROR, to_log) << "Node stash failed: " << ec.message();
+    std::visit([this](const auto& value) { op_manager_->ClearStashPending(value); }, id);
   }
 }
 
@@ -716,7 +821,18 @@ void ReadTiered(DbIndex dbid, std::string_view key, const PrimeValue& value,
   auto cb = [readf = std::move(readf)](io::Result<tiering::StringDecoder*> res) mutable {
     readf(res.transform([](tiering::StringDecoder* d) { return d->GetView(); }));
   };
-  ts->Read(dbid, key, value.GetExternalSlice(), tiering::StringDecoder{value}, std::move(cb));
+  ts->Read(KeyRef{dbid, key}, value.GetExternalSlice(), tiering::StringDecoder{value},
+           std::move(cb));
+}
+
+void ReadTieredListNode(DbIndex dbid, QList::Node* node, QList* ql,
+                        const tiering::DiskSegment& segment,
+                        std::function<void(io::Result<std::string_view>)> readf,
+                        TieredStorage* ts) {
+  auto cb = [readf = std::move(readf)](io::Result<tiering::BareDecoder*> res) mutable {
+    readf(res.transform([](tiering::BareDecoder* d) { return d->slice; }));
+  };
+  ts->Read(tiering::ListNodeId{dbid, ql, node}, segment, tiering::BareDecoder{}, std::move(cb));
 }
 
 template <typename T>
@@ -730,8 +846,8 @@ TieredStorage::TResult<T> ModifyTiered(DbIndex dbid, std::string_view key, const
   auto cb = [future, modf = std::move(modf)](io::Result<tiering::StringDecoder*> res) mutable {
     future.Resolve(res.transform([&modf](auto* d) { return modf(d->Write()); }));
   };
-  ts->Read(dbid, key, value.GetExternalSlice(), tiering::StringDecoder{value}, std::move(cb),
-           false);
+  ts->Read(KeyRef{dbid, key}, value.GetExternalSlice(), tiering::StringDecoder{value},
+           std::move(cb), false);
 
   return future;
 }