refactor(object_store): lazy fetch returns PayloadView (#103)

* refactor(object_store): lazy fetch returns PayloadView; payload slot uses std::any

The lazy resolver registered with ObjectStore::pushLazy now returns a
PayloadView (Span + BufferAnchor) instead of a std::vector<uint8_t>. This
lets producers hand off bytes they already hold behind a shared_ptr (e.g.
a streaming buffer being reused across stores) without copying — the
returned anchor extends the buffer's lifetime through the read.

For producers whose only payload is a plain std::vector<uint8_t>, the new
sdk::makePayloadView() helper in pj_base wraps the vector into a
shared_ptr (which becomes both owner and type-erased anchor) and returns
a PayloadView pointing at its contents. This is also the contract
resolveEntry() relies on: it recovers the shared_ptr<const vector> from
the anchor via static_pointer_cast.

ObjectEntry::payload now stores a std::any instead of a
std::variant<shared_ptr, std::function>. The dispatch in
ObjectStore::resolveEntry uses std::any_cast for each branch. Rationale:
the variant alternatives are not part of any public discriminator
(callers go through pushOwned/pushLazy), so the variant tag was carrying
no semantic value — std::any keeps the storage slot flexible without
constraining future producer shapes.

* refactor(object_store): preserve BufferAnchor end-to-end; restore typed variant

Builds on the prior commit's PayloadView-returning pushLazy signature, but
fixes two anchor-discarding bugs and restores a typed payload variant.

Bugs in the prior implementation:

1. resolveEntry did static_pointer_cast<const vector<uint8_t>>(pv.anchor),
   forcing every producer's anchor to be exactly a shared_ptr<vector>. Any
   other anchor type (chunk cache, mmap, parquet column slice) was UB. This
   negated the entire point of BufferAnchor = shared_ptr<const void>.

2. resolveEntry discarded PayloadView::bytes (the Span) and returned the
   whole anchor's vector. Producers can no longer publish a sub-range of
   a larger backing buffer — which is the canonical zero-copy use case
   (one decompressed MCAP chunk anchored once, many message Spans into it).

Both bugs traced to a single root cause: ResolvedObjectEntry::data was still
shared_ptr<const vector<uint8_t>>, so resolution had to materialize a vector
somehow. Fixed by making ResolvedObjectEntry carry {BufferAnchor anchor,
Span<const uint8_t> view} — type-erased anchor + producer-published span.
No static_pointer_cast anywhere.

Variant + named aliases:

ObjectEntry::payload returns to std::variant; the prior std::any traded
compile-time exhaustiveness for nothing (still two alternatives, still
typed access via the cast). Two named aliases capture the two payload
shapes at the type level:

  using SharedBuffer = std::shared_ptr<const std::vector<uint8_t>>;
  using LazyCallback = std::function<sdk::PayloadView()>;

Trampolines (plugin_data_host.cpp):

ObjectBytesBox now holds {BufferAnchor, Span}; toolboxObjectGetBytes
returns view.data()/view.size() — no shared_ptr<vector> in the read path.

Misc consistency:

- pushOwned/pushLazy now return Status (alias for Expected<void, string>);
  the one other Expected<void, string> use at service_registry_builder.hpp
  also switched.
- PayloadView(shared_ptr<vector>) constructor takes shared_ptr<const vector>
  to match SharedBuffer and makePayloadView.

Tests:

- All resolved->data accessors migrated to resolved->view / resolved->anchor
  across object_store_test, plugin_data_host_object_test,
  plugin_parser_object_write_test.
- Two regression tests added:
    PushLazyPreservesAnchorType — anchor is shared_ptr<TestBuffer> (not
      vector). Bug-1 regression: ASAN would catch the prior static cast.
    PushLazyHonorsSpanSubview — anchor is a 100-byte vector with Span
      [20,30). Bug-2 regression: verifies view.data()==chunk+20.

./build.sh --debug && ./test.sh → 62/62 passing under ASAN.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* refactor(object_store): consolidate ResolvedObjectEntry/ObjectBytesBox on PayloadView

ResolvedObjectEntry and ObjectBytesBox both held an inline {anchor, view}
pair — the same shape as sdk::PayloadView. Collapse both onto a single
PayloadView field; one named point of truth for "bytes + their lifetime
anchor" across producer, store, and consumer-handle layers.

Also tightens the doc comments touched in the prior commit.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* chore: bump version to 0.5.0

Source-incompatible changes in pj_datastore/object_store.hpp:
- ObjectEntry::payload variant tag types
- pushLazy fetch signature (returns PayloadView)
- ResolvedObjectEntry field layout (carries a PayloadView)

Per the pre-1.0 convention, this warrants a MINOR bump.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* docs(object_store): sync design doc with PayloadView refactor

OBJECT_STORE_DESIGN.md still described the pre-refactor API: a
vector-returning lazy callback and a ResolvedObjectEntry carrying a
shared_ptr<const vector<uint8_t>> data field. Bring it in line with the
code already on this branch:

- ObjectEntry::payload is std::variant<SharedBuffer, LazyCallback>;
  document the two named aliases.
- pushLazy's callable returns sdk::PayloadView (Span + type-erased
  BufferAnchor), enabling zero-copy sub-range views; the store never
  copies on resolve.
- ResolvedObjectEntry carries an sdk::PayloadView; bytes/anchor replace
  the old data field, and an empty anchor means "no bytes".

Docs-only; no behavior change.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Davide Faconti <davide.faconti@gmail.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 3 people

CMakeLists.txt

@@ -202,7 +202,7 @@ endif()
 if(PJ_INSTALL_SDK)
   include(CMakePackageConfigHelpers)
 
-  set(PJ_PACKAGE_VERSION "0.4.2")
+  set(PJ_PACKAGE_VERSION "0.5.0")
   set(PJ_PACKAGE_CMAKE_DIR ${CMAKE_INSTALL_LIBDIR}/cmake/plotjuggler_core)
 
   install(EXPORT plotjuggler_coreTargets

conanfile.py

@@ -7,7 +7,7 @@
   plugin_sdk   — umbrella for plugin authors (base + dialog SDK + parser SDK)
   plugin_host  — umbrella for host loaders (data_source/parser/toolbox/dialog)
 
-A consuming Conan recipe declares e.g. `plotjuggler_core/0.4.2` and then:
+A consuming Conan recipe declares e.g. `plotjuggler_core/0.5.0` and then:
 
     find_package(plotjuggler_core REQUIRED COMPONENTS plugin_sdk)
     target_link_libraries(my_plugin PRIVATE plotjuggler_core::plugin_sdk)
@@ -27,7 +27,7 @@
 
 class PlotjugglerCoreConan(ConanFile):
     name = "plotjuggler_core"
-    version = "0.4.2"
+    version = "0.5.0"
     # Apache-2.0 covers pj_base + pj_plugins (the plugin-facing SDK);
     # MPL-2.0 covers pj_datastore (the storage engine). See LICENSE.
     license = "Apache-2.0 AND MPL-2.0"

pj_base/include/pj_base/buffer_anchor.hpp

@@ -15,6 +15,8 @@
 
 #include <cstdint>
 #include <memory>
+#include <utility>
+#include <vector>
 
 #include "pj_base/span.hpp"
 
@@ -44,10 +46,22 @@ struct PayloadView {
 
   PayloadView(Span<const uint8_t> bytes_, BufferAnchor anchor_) : bytes(bytes_), anchor(std::move(anchor_)) {}
 
-  PayloadView(std::shared_ptr<std::vector<uint8_t>> buffer)
+  PayloadView(std::shared_ptr<const std::vector<uint8_t>> buffer)
       : bytes(buffer ? Span<const uint8_t>(buffer->data(), buffer->size()) : Span<const uint8_t>()),
         anchor(std::move(buffer)) {}
 };
 
+/// Wrap a fresh vector as both anchor and Span. Use when the producer has
+/// no natural anchor (e.g. raw bytes from a C-ABI fetch); when an upstream
+/// allocation already exists, construct PayloadView directly to avoid this
+/// helper's copy.
+inline PayloadView makePayloadView(std::vector<uint8_t> bytes) {
+  auto shared = std::make_shared<const std::vector<uint8_t>>(std::move(bytes));
+  return PayloadView{
+      Span<const uint8_t>{shared->data(), shared->size()},
+      BufferAnchor{shared},
+  };
+}
+
 }  // namespace sdk
 }  // namespace PJ

pj_datastore/docs/OBJECT_STORE_DESIGN.md

@@ -30,11 +30,14 @@ struct ObjectTopicDescriptor {
   std::string metadata_json;
 };
 
+// Eager payload: store-owned bytes, counted against the retention budget.
+using SharedBuffer = std::shared_ptr<const std::vector<uint8_t>>;
+// Lazy payload: idempotent fetcher returning a view + ownership anchor.
+using LazyCallback = std::function<sdk::PayloadView()>;
+
 struct ObjectEntry {
   Timestamp timestamp;
-  std::variant<
-      std::shared_ptr<const std::vector<uint8_t>>,
-      std::function<std::vector<uint8_t>()>> payload;
+  std::variant<SharedBuffer, LazyCallback> payload;
 };
 
 struct RetentionBudget {
@@ -55,9 +58,13 @@ order. Equal timestamps are allowed. Out-of-order writes fail.
 shared buffer owned by the store. Owned entries contribute to `memoryUsage()`.
 
 `pushLazy(id, timestamp, fetch)` stores a callable instead of bytes. The callable
-is invoked on each read and returns a fresh byte vector. Lazy entries do not
-contribute to `memoryUsage()` because the store does not retain the bytes between
-reads.
+is invoked on each read and returns a `sdk::PayloadView` — a `Span<const uint8_t>`
+paired with a type-erased `BufferAnchor` that keeps those bytes alive for as long
+as the resolved view is held. The producer anchors on whatever already owns the
+bytes (a decompressed chunk, an mmap, or a fresh allocation via
+`sdk::makePayloadView`), so the store never copies on resolve. Lazy entries do not
+contribute to `memoryUsage()` because the store retains the callable, not the
+fetched bytes.
 
 Both write paths apply the topic retention budget after the new entry is
 inserted.
@@ -80,12 +87,14 @@ Resolved entries contain:
 ```cpp
 struct ResolvedObjectEntry {
   Timestamp timestamp;
-  std::shared_ptr<const std::vector<uint8_t>> data;
+  sdk::PayloadView payload;  // { Span<const uint8_t> bytes; BufferAnchor anchor; }
 };
 ```
 
-The shared pointer keeps resolved bytes alive independently of later store
-mutation.
+`payload.bytes` is the resolved view; `payload.anchor` keeps those bytes alive
+independently of later store mutation (eviction, removal, or `clear()`). For an
+owned entry the anchor is the store's `SharedBuffer`; for a lazy entry it is
+whatever the fetcher anchored on. An empty `anchor` means "no bytes".
 
 ## Retention
 

pj_datastore/include/pj_datastore/object_store.hpp

@@ -16,7 +16,9 @@
 #include <variant>
 #include <vector>
 
+#include "pj_base/buffer_anchor.hpp"
 #include "pj_base/expected.hpp"
+#include "pj_base/span.hpp"
 #include "pj_base/types.hpp"
 
 namespace PJ {
@@ -38,14 +40,21 @@ struct ObjectTopicDescriptor {
   std::string metadata_json;
 };
 
+/// Eager payload: store-owned bytes, counted against the retention budget.
+using SharedBuffer = std::shared_ptr<const std::vector<uint8_t>>;
+
+/// Lazy payload: idempotent, thread-safe fetcher returning bytes + anchor.
+/// Invoked on every read; bytes are not counted against the retention budget.
+using LazyCallback = std::function<sdk::PayloadView()>;
+
 struct ObjectEntry {
   Timestamp timestamp = 0;
-  std::variant<std::shared_ptr<const std::vector<uint8_t>>, std::function<std::vector<uint8_t>()>> payload;
+  std::variant<SharedBuffer, LazyCallback> payload;
 };
 
 struct ResolvedObjectEntry {
   Timestamp timestamp = 0;
-  std::shared_ptr<const std::vector<uint8_t>> data;
+  sdk::PayloadView payload;
 };
 
 struct RetentionBudget {
@@ -106,10 +115,12 @@ class ObjectStore {
 
   // --- Write ---
 
-  Expected<void, std::string> pushOwned(ObjectTopicId id, Timestamp timestamp, std::vector<uint8_t> payload);
+  Status pushOwned(ObjectTopicId id, Timestamp timestamp, std::vector<uint8_t> payload);
 
-  Expected<void, std::string> pushLazy(
-      ObjectTopicId id, Timestamp timestamp, std::function<std::vector<uint8_t>()> fetch);
+  // Fetcher runs on every read. Producers anchor on whatever owns the bytes
+  // (chunk cache, mmap, fresh allocation); the store never copies — it just
+  // retains the anchor through PayloadView.
+  Status pushLazy(ObjectTopicId id, Timestamp timestamp, LazyCallback fetch);
 
   // --- Read ---
 

pj_datastore/src/object_store.cpp

@@ -67,8 +67,7 @@ std::vector<ObjectTopicId> ObjectStore::listTopics(DatasetId dataset_id) const {
 
 // --- Write ---
 
-Expected<void, std::string> ObjectStore::pushOwned(
-    ObjectTopicId id, Timestamp timestamp, std::vector<uint8_t> payload) {
+Status ObjectStore::pushOwned(ObjectTopicId id, Timestamp timestamp, std::vector<uint8_t> payload) {
   std::shared_lock store_lock(store_mutex_);
   auto* series = findSeries(id);
   if (series == nullptr) {
@@ -94,8 +93,7 @@ Expected<void, std::string> ObjectStore::pushOwned(
   return {};
 }
 
-Expected<void, std::string> ObjectStore::pushLazy(
-    ObjectTopicId id, Timestamp timestamp, std::function<std::vector<uint8_t>()> fetch) {
+Status ObjectStore::pushLazy(ObjectTopicId id, Timestamp timestamp, LazyCallback fetch) {
   std::shared_lock store_lock(store_mutex_);
   auto* series = findSeries(id);
   if (series == nullptr) {
@@ -306,11 +304,15 @@ ResolvedObjectEntry ObjectStore::resolveEntry(const ObjectEntry& entry) {
   ResolvedObjectEntry resolved;
   resolved.timestamp = entry.timestamp;
 
-  if (const auto* owned = std::get_if<std::shared_ptr<const std::vector<uint8_t>>>(&entry.payload)) {
-    resolved.data = *owned;
-  } else if (const auto* lazy = std::get_if<std::function<std::vector<uint8_t>()>>(&entry.payload)) {
-    auto bytes = (*lazy)();
-    resolved.data = std::make_shared<const std::vector<uint8_t>>(std::move(bytes));
+  if (const auto* owned = std::get_if<SharedBuffer>(&entry.payload)) {
+    if (*owned) {
+      resolved.payload = sdk::PayloadView{
+          Span<const uint8_t>{(*owned)->data(), (*owned)->size()},
+          sdk::BufferAnchor{*owned},
+      };
+    }
+  } else if (const auto* lazy = std::get_if<LazyCallback>(&entry.payload)) {
+    resolved.payload = (*lazy)();
   }
 
   return resolved;
@@ -322,7 +324,7 @@ void ObjectStore::evictFront(ObjectSeries& series) {
   }
 
   const auto& front = series.entries.front();
-  if (const auto* owned = std::get_if<std::shared_ptr<const std::vector<uint8_t>>>(&front.payload)) {
+  if (const auto* owned = std::get_if<SharedBuffer>(&front.payload); owned != nullptr && *owned) {
     series.memory_bytes -= (*owned)->size();
   }
 

pj_datastore/src/plugin_data_host.cpp

@@ -1392,7 +1392,9 @@ bool sourceObjectPushLazy(
     // the lambda; destructor runs exactly once when ObjectStore drops the
     // entry (retention, evict, removeTopic, clear, or store teardown).
     auto holder = std::make_shared<PluginFetchCtx>(fetch_fn, fetch_ctx, fetch_ctx_destroy);
-    auto closure = [holder]() -> std::vector<uint8_t> { return holder->invoke(); };
+    // Plugins return raw bytes via the C ABI; wrap them as a PayloadView whose
+    // anchor is a shared_ptr<const vector<uint8_t>>, per the pushLazy contract.
+    auto closure = [holder]() -> sdk::PayloadView { return sdk::makePayloadView(holder->invoke()); };
     auto result = impl->store.pushLazy(ObjectTopicId{topic.id}, timestamp_ns, std::move(closure));
     if (!result) {
       impl->setError(result.error());
@@ -1433,10 +1435,10 @@ void sourceObjectSetRetentionBudget(
 // Toolbox object read host trampolines
 // ---------------------------------------------------------------------------
 
-/// Box holding the shared_ptr that keeps ObjectStore bytes alive. One
-/// allocated per successful read_latest_at; freed by release_bytes.
+/// Heap holder for the PayloadView backing PJ_object_bytes_handle_t.
+/// Allocated by read_latest_at; freed by release_bytes.
 struct ObjectBytesBox {
-  std::shared_ptr<const std::vector<uint8_t>> bytes;
+  sdk::PayloadView payload;
 };
 
 PJ_object_topic_handle_t toolboxObjectLookupTopic(void* ctx, PJ_string_view_t topic_name) noexcept {
@@ -1506,12 +1508,12 @@ bool toolboxObjectReadLatestAt(
   *out_handle = nullptr;
   try {
     auto entry = impl->store.latestAt(ObjectTopicId{topic.id}, timestamp_ns);
-    if (!entry.has_value() || entry->data == nullptr) {
+    if (!entry.has_value() || entry->payload.anchor == nullptr) {
       impl->setError("no entry at-or-before timestamp");
       propagateError(out_error, impl->last_error.c_str());
       return false;
     }
-    auto* box = new ObjectBytesBox{std::move(entry->data)};
+    auto* box = new ObjectBytesBox{std::move(entry->payload)};
     *out_handle = reinterpret_cast<PJ_object_bytes_handle_t>(box);
     if (out_timestamp != nullptr) {
       *out_timestamp = entry->timestamp;
@@ -1540,14 +1542,14 @@ void toolboxObjectGetBytes(PJ_object_bytes_handle_t handle, const uint8_t** out_
     return;
   }
   auto* box = reinterpret_cast<ObjectBytesBox*>(handle);
-  if (!box->bytes) {
+  if (box->payload.bytes.empty()) {
     return;
   }
   if (out_data != nullptr) {
-    *out_data = box->bytes->data();
+    *out_data = box->payload.bytes.data();
   }
   if (out_size != nullptr) {
-    *out_size = box->bytes->size();
+    *out_size = box->payload.bytes.size();
   }
 }
 
@@ -1631,7 +1633,7 @@ bool parserObjectPushLazy(
   }
   try {
     auto holder = std::make_shared<PluginFetchCtx>(fetch_fn, fetch_ctx, fetch_ctx_destroy);
-    auto closure = [holder]() -> std::vector<uint8_t> { return holder->invoke(); };
+    auto closure = [holder]() -> sdk::PayloadView { return sdk::makePayloadView(holder->invoke()); };
     auto result = impl->store.pushLazy(impl->bound_topic, timestamp_ns, std::move(closure));
     if (!result) {
       impl->setError(result.error());