Extend tiering apis to support fragments

[romange]

Problem

Current tiering APIs are primarily key/value (PrimeValue) oriented and externalize a single serialized value per key (with optional cooling). This works for strings and current hash path, but does not provide a first-class API for partial offloading of multi-element data structures (list segments, stream ranges, etc).

We need to support offloading arbitrary object fragments (not only whole PrimeValue) so commands can fetch/mutate only relevant parts.

Current state (relevant code)

• TieredStorage API is centered on key/value stash/read/delete and PrimeValue state transitions:

// Enqueue read external value with generic decoder.
template <typename D, typename F>
void Read(DbIndex dbid, std::string_view key, const tiering::DiskSegment& segment,
          const D& decoder, F&& f) { ... }

std::optional<StashDescriptor> ShouldStash(const PrimeValue& pv) const;
std::optional<util::fb2::Future<bool>> Stash(DbIndex dbid, std::string_view key,
                                             const StashDescriptor& blobs, bool provide_bp);
void Delete(DbIndex dbid, PrimeValue* value);
void CancelStash(DbIndex dbid, std::string_view key, PrimeValue* value);

• OpManager already supports generic pending IDs and async segment ops:

using PendingId = std::variant<unsigned, KeyRef>;
void Enqueue(PendingId id, DiskSegment segment, const Decoder& decoder, ReadCallback cb);
void DeleteOffloaded(DiskSegment segment);
std::error_code PrepareAndStash(PendingId id, size_t length,
                                const std::function<size_t(io::MutableBytes)>& writer);

• SmallBins exists for sub-page packing but should not be used for this feature path:

// Small bins accumulate small values into larger bins that fill up 4kb pages.
class SmallBins { ... };

Goal

Introduce a fragment-tiering API that allows composite objects to offload/fetch/update independent fragments, with these constraints:

• Fragment offload path is direct-to-segment (no SmallBins).
• Offload only fragments with serialized size >= 2KB.
• Preserve existing whole-value tiering behavior for current users.

Proposed API / design direction

1) New fragment identity

TieredStorage APIs should not work with PrimeValue directly. Instead they should accept a Fragment, like
std::optional<StashDescriptor> ShouldStash(const Fragment& fragment) const;

Fragment is an adaptor interface that wraps variant<PrimeValue,...> to support differrent types like stream listpacks.
TieredStorage does not operate or update on PrimeValue directly.

2) OpManager pending id support for fragment keys

Extend pending ID to track fragment ops unambiguously, while preserving existing key/bin behavior. We do not need to support small bin packing for collections (other non-string types), and assume a segment will contain a single offloaded value. We may just need to extend PendingId with uintptr_t which is an address of the heap based object we are trying to offload. Question: Do we even need to map from PendingId back to the original data-structure?

3) Support fragments in TieredColdRecord (cool_queue_). Is not a blocker to the POC

We may want to cache offloaded items like we do with strings to wait until we are close to memory limits before starting the eviction.

Out of scope

• Full list/stream adapter implementation details.
• Command-level behavior changes (LRANGE, XRANGE, etc).
• RDB/replication support

Acceptance criteria

1. New fragment-tiering APIs exist and are callable independently of PrimeValue externalization.
2. Fragment stashes <2KB are rejected by API contract and Fragment path never invokes SmallBins.
3. Existing string/hash offload behavior remains unchanged.
4. Unit tests cover:
   • successful fragment stash/read/delete
   • <2KB rejection
   • pending op cancellation
   • no regression in existing tiering tests.

[romange]

I will try to simulate here design of offloading rax listpacks (from streams) as a potential use-case in addition to strings/primevalue use-case that we currently support.

1. ShouldStash queries fragment_ref.ShouldStash() and then continues querying its ops_manager.
2. Delete(DbIndex dbid, PrimeValue* value); should become Delete(DbIndex dbid, FragmentRef frag_ref) or something like that. Underneath Delete makes sure that the underlying data-structure is not marked as tiered and of course removes its external associations from the tiering (including cool entries if needed).
3. CancelStash(DbIndex dbid, std::string_view key, PrimeValue* value) should be transformed to CancelStash(DbIndex dbid, std::string_view key, FragmentRef frag_ref), where we cancel the io-request but also mark frag_ref as stash_pending = false.

class FragmentRef {

public:
  
...
variant<PrimeValue*, FragmentOwner>  val_; 
};

How to maintain a tiering state

PrimeValue has io_pending bit as well as 16 byte "external" representation once the offloading process finishes.

With listpack we can not do this directly. We could use several bits in QList::Node for example to maintain equivalent state
but in case of listpacks inside rax it's more complicated.

We have several options for streams:

Use tagged pointers to differentiate between different states:

a) "stash pending" bit + original listpack pointer.
b) "cool" bit + wrapper object with the original listpack and cooling tiered state
c) cold + a wrapper struct with some tiered state (no listpack object).

we could merge a), b), c) into a single wrapper struct but we would still need a tagged bit to differentiate between plain listpacks and a wrapper object. In this case FragmentOwner would contain the information on how to reach the node that contains the tagged pointer in the original data-structure (rax). raxNode seems to be unstable so we may need to store 16 bytes key and rax tree pointer to be able to find and modify data for the operations above.

Adjust listpack format to support tiering internally.

Instead of using tagged pointers - to use listpack as a runtime structure. streams already store a ""master entry" containing "count", "num_deleted", "num-fields" fields inside listpack. we could add more bytes to contain tiered state and always use listpack as a ultimate run-time struct for in-memory/tiered/cool/cold/stash-pending states.

My intuition is that changing listpack format is too intrusive and will lead to various issues around wire format compatibility, blast radius etc. Also it won't work with lists, where such "master" header does not exist at all.

For tagged pointers, FragmentOwner will need to contain enough information to find the container holding the pointer so we could change their state (change tags, wrappers). So for streams FragmentOwner would contain {rax*, char[16] key}
but for list it would contain QList::Node* as QList::Node* is stable pointer unlike raxNode that may move due to insertions/deletions of other items.

Conclusion

After this analysis, I feel more confident that we can start with lists and not streams as a simpler use-case.
QList::Node* has plenty spare bits to support tiering states, and does not require pointer tagging.

FragmentOwner can contain QList::Node* which is also simple. These changes will force us to write "neutral" code inside tiering and support two-usecases. Streams support will be just a specific extension on top of this.

[romange]

Based on the initial thoughts from #6472
we might have the ABA problem for patterns like,
READ KEY -> ASYNC READ started -> DEL KEY -> pending request marked for deletion -> Stash KEY.

I will be surprised if that's the case but we should double check it via op_manager_test that it handles this correctly.
It's unrelated to this task though.

Update:
This PR explains why ABA can not happen with the current logic:
#6789

it is fine to have same ids concurrently with earlier ones being deleted objects with pending reads that are not finalized yet.