chore: PayloadView constructors + docs + extra tests + bump 0.5.0

Adopt the additions Davide made on top of the consolidated refactor
upstream (PR #105):

- PayloadView gains explicit default + Span/anchor + shared_ptr<vector>
  constructors in buffer_anchor.hpp.
- New section in OBJECT_STORE_DESIGN.md covering the PayloadView contract
  and the eager/lazy dispatch in resolveEntry.
- Additional ObjectStore tests (+56 lines) covering edge cases not in the
  original PR.
- service_registry_builder.hpp tweak.
- conanfile.py + CMakeLists.txt bumped to 0.5.0 (matches the release tag
  cut upstream).

Author: pabloinigoblasco

CMakeLists.txt

@@ -196,7 +196,7 @@ endif()
 if(PJ_INSTALL_SDK)
   include(CMakePackageConfigHelpers)
 
-  set(PJ_PACKAGE_VERSION "0.4.0")
+  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.0` 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.0"
+    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

@@ -41,14 +41,20 @@ using BufferAnchor = std::shared_ptr<const void>;
 struct PayloadView {
   Span<const uint8_t> bytes;
   BufferAnchor anchor;
+
+  PayloadView() = default;
+
+  PayloadView(Span<const uint8_t> bytes_, BufferAnchor anchor_) : bytes(bytes_), anchor(std::move(anchor_)) {}
+
+  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)) {}
 };
 
-/// Convenience helper for producers whose only payload is a plain
-/// std::vector<uint8_t>. Wraps it in a shared_ptr (which becomes both the
-/// owner of the bytes and the type-erased anchor), and returns a PayloadView
-/// whose Span points at that vector's contents. Satisfies the contract
-/// expected by ObjectStore::resolveEntry on the lazy branch: the anchor is a
-/// shared_ptr<const std::vector<uint8_t>> recoverable via static_pointer_cast.
+/// 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{

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

@@ -2,7 +2,6 @@
 // Copyright 2026 Davide Faconti
 // SPDX-License-Identifier: MPL-2.0
 
-#include <any>
 #include <cstddef>
 #include <cstdint>
 #include <deque>
@@ -14,10 +13,12 @@
 #include <string>
 #include <string_view>
 #include <utility>
+#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 {
@@ -39,12 +40,20 @@ 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;
-  // Holds either a shared_ptr<const std::vector<uint8_t>> (eager owned payload)
-  // or a std::function<sdk::PayloadView()> (lazy resolver). resolveEntry
-  // discriminates via std::any_cast.
-  std::any payload;
+  // Holds either a SharedBuffer (eager owned payload, counted against the
+  // retention budget) or a LazyCallback (lazy resolver). resolveEntry
+  // discriminates via std::get_if; the variant is exhaustive over the two
+  // (and only two) payload kinds.
+  std::variant<SharedBuffer, LazyCallback> payload;
 };
 
 struct ResolvedObjectEntry {
@@ -119,16 +128,14 @@ 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);
 
-  // The fetch callable is invoked on every read. It returns a PayloadView
-  // (Span + anchor). When the producer already holds the bytes in memory
-  // behind a shared_ptr (e.g. a streaming buffer being handed off between
-  // stores), the closure can capture that shared_ptr and return a view
-  // backed by it — no copy. For producers that materialize bytes from disk
-  // or other sources, the closure allocates a fresh buffer and uses it as
-  // the anchor.
-  Expected<void, std::string> pushLazy(ObjectTopicId id, Timestamp timestamp, std::function<sdk::PayloadView()> 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. When the producer already holds
+  // the bytes behind a shared_ptr (e.g. a streaming buffer handed off between
+  // stores), the closure captures it and returns a view backed by it.
+  Status pushLazy(ObjectTopicId id, Timestamp timestamp, LazyCallback fetch);
 
   // --- Read ---
 
@@ -167,10 +174,10 @@ class ObjectStore {
   // neither store is mutated.
   //
   // Zero-copy on the payload bytes. Each ObjectEntry is moved into the
-  // destination's series by value; the std::variant inside (or, post-#184,
-  // the std::any) holds either a shared_ptr or a std::function, and moving
-  // it is a pointer/buffer move — bytes captured by the closure or owned by
-  // the shared_ptr are never copied or materialized during the flush. Lazy
+  // destination's series by value; the std::variant inside holds either a
+  // shared_ptr or a std::function, and moving it is a pointer/buffer move —
+  // bytes captured by the closure or owned by the shared_ptr are never
+  // copied or materialized during the flush. Lazy
   // entries preserve their semantics in the destination: their closure is
   // re-invoked only when the destination is 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) {
@@ -95,8 +94,7 @@ Expected<void, std::string> ObjectStore::pushOwned(
   return {};
 }
 
-Expected<void, std::string> ObjectStore::pushLazy(
-    ObjectTopicId id, Timestamp timestamp, std::function<sdk::PayloadView()> fetch) {
+Status ObjectStore::pushLazy(ObjectTopicId id, Timestamp timestamp, LazyCallback fetch) {
   std::shared_lock store_lock(store_mutex_);
   auto* series = findSeries(id);
   if (series == nullptr) {
@@ -375,14 +373,17 @@ ResolvedObjectEntry ObjectStore::resolveEntry(const ObjectEntry& entry) {
   ResolvedObjectEntry resolved;
   resolved.timestamp = entry.timestamp;
 
-  if (const auto* owned = std::any_cast<std::shared_ptr<const std::vector<uint8_t>>>(&entry.payload)) {
+  if (const auto* owned = std::get_if<SharedBuffer>(&entry.payload)) {
     // Owned branch: build a PayloadView whose Span covers the vector and
-    // whose anchor IS the same shared_ptr — refcount bump, no copy.
-    resolved.payload = sdk::PayloadView{
-        Span<const uint8_t>{(*owned)->data(), (*owned)->size()},
-        sdk::BufferAnchor{*owned},
-    };
-  } else if (const auto* lazy = std::any_cast<std::function<sdk::PayloadView()>>(&entry.payload)) {
+    // whose anchor IS the same shared_ptr — refcount bump, no copy. A
+    // default-constructed entry holds a null SharedBuffer, so guard it.
+    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)) {
     // Lazy branch: forward whatever the closure returns. The anchor stays
     // opaque (shared_ptr<const void>); consumers retain it without needing
     // to know the concrete type. No static_pointer_cast here — that was the
@@ -401,7 +402,7 @@ void ObjectStore::evictFront(ObjectSeries& series) {
   }
 
   const auto& front = series.entries.front();
-  if (const auto* owned = std::any_cast<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/tests/object_store_test.cpp

@@ -5,8 +5,10 @@
 
 #include <gtest/gtest.h>
 
+#include <array>
 #include <cstdint>
 #include <functional>
+#include <memory>
 #include <string>
 #include <thread>
 #include <vector>
@@ -294,6 +296,60 @@ TEST(ObjectStoreTest, PushLazyResolves) {
   EXPECT_EQ(r->payload.bytes[0], 0xDE);
 }
 
+// Regression: anchor type-erasure must survive resolveEntry. The anchor here
+// is a shared_ptr<TestBuffer> (not vector); a prior static_pointer_cast to
+// vector would UB.
+TEST(ObjectStoreTest, PushLazyPreservesAnchorType) {
+  struct TestBuffer {
+    std::array<uint8_t, 4> bytes{0x11, 0x22, 0x33, 0x44};
+  };
+  ObjectStore store;
+  auto id = registerTestTopic(store);
+
+  auto buffer = std::make_shared<TestBuffer>();
+  std::weak_ptr<TestBuffer> weak_buffer = buffer;
+
+  store.pushLazy(id, 100, [buffer]() -> sdk::PayloadView {
+    return sdk::PayloadView{
+        Span<const uint8_t>{buffer->bytes.data(), buffer->bytes.size()},
+        sdk::BufferAnchor{buffer},
+    };
+  });
+
+  auto r = store.latestAt(id, 100);
+  ASSERT_TRUE(r.has_value());
+  ASSERT_EQ(r->payload.bytes.size(), 4u);
+  EXPECT_EQ(r->payload.bytes[0], 0x11);
+  EXPECT_EQ(r->payload.bytes[3], 0x44);
+  EXPECT_FALSE(weak_buffer.expired());  // anchor still holds the buffer alive
+}
+
+// Regression: the producer's Span is a sub-range of the anchor's storage.
+// resolveEntry must propagate it verbatim — not the anchor's full extent.
+TEST(ObjectStoreTest, PushLazyHonorsSpanSubview) {
+  ObjectStore store;
+  auto id = registerTestTopic(store);
+
+  auto chunk = std::make_shared<std::vector<uint8_t>>(100);
+  for (size_t i = 0; i < chunk->size(); ++i) {
+    (*chunk)[i] = static_cast<uint8_t>(i);
+  }
+
+  store.pushLazy(id, 100, [chunk]() -> sdk::PayloadView {
+    return sdk::PayloadView{
+        Span<const uint8_t>{chunk->data() + 20, 10},  // bytes [20, 30)
+        sdk::BufferAnchor{chunk},
+    };
+  });
+
+  auto r = store.latestAt(id, 100);
+  ASSERT_TRUE(r.has_value());
+  EXPECT_EQ(r->payload.bytes.size(), 10u);
+  EXPECT_EQ(r->payload.bytes.data(), chunk->data() + 20);
+  EXPECT_EQ(r->payload.bytes[0], 20);
+  EXPECT_EQ(r->payload.bytes[9], 29);
+}
+
 // =========================================================================
 // Timestamp monotonicity
 // =========================================================================

pj_datastore/tests/plugin_data_host_object_test.cpp

@@ -3,6 +3,7 @@
 
 #include <gtest/gtest.h>
 
+#include <algorithm>
 #include <atomic>
 #include <cstdint>
 #include <memory>
@@ -65,7 +66,8 @@ TEST(PluginDataHostObjectTest, PushOwnedStoresBytes) {
   ASSERT_TRUE(resolved.has_value());
   ASSERT_NE(resolved->payload.anchor, nullptr);
   EXPECT_EQ(resolved->payload.bytes.size(), payload.size());
-  EXPECT_EQ(std::vector<uint8_t>(resolved->payload.bytes.begin(), resolved->payload.bytes.end()), payload);
+  EXPECT_TRUE(
+      std::equal(resolved->payload.bytes.begin(), resolved->payload.bytes.end(), payload.begin(), payload.end()));
 }
 
 TEST(PluginDataHostObjectTest, PushLazyRetainsClosureUntilEviction) {
@@ -94,7 +96,9 @@ TEST(PluginDataHostObjectTest, PushLazyRetainsClosureUntilEviction) {
   auto first = f.store.latestAt(ObjectTopicId{topic.id}, 42);
   ASSERT_TRUE(first.has_value());
   ASSERT_NE(first->payload.anchor, nullptr);
-  EXPECT_EQ(std::vector<uint8_t>(first->payload.bytes.begin(), first->payload.bytes.end()), shared->payload);
+  EXPECT_TRUE(
+      std::equal(
+          first->payload.bytes.begin(), first->payload.bytes.end(), shared->payload.begin(), shared->payload.end()));
   EXPECT_GE(shared->fetch_calls.load(), 1);
 
   auto second = f.store.latestAt(ObjectTopicId{topic.id}, 42);
@@ -149,7 +153,9 @@ TEST(PluginDataHostObjectTest, PushLazyDestroyCallbackRunsExactlyOnceOnEviction)
   // Fetch once — the callback runs but the ctx stays alive.
   auto resolved = f.store.latestAt(ObjectTopicId{topic.id}, 100);
   ASSERT_TRUE(resolved.has_value());
-  EXPECT_EQ(std::vector<uint8_t>(resolved->payload.bytes.begin(), resolved->payload.bytes.end()), ctx->payload);
+  EXPECT_TRUE(
+      std::equal(
+          resolved->payload.bytes.begin(), resolved->payload.bytes.end(), ctx->payload.begin(), ctx->payload.end()));
   EXPECT_EQ(ctx->destroy_count.load(), 0);
 
   // Evict — destroy_fn runs exactly once.

pj_datastore/tests/plugin_parser_object_write_test.cpp

@@ -9,6 +9,7 @@
 
 #include <gtest/gtest.h>
 
+#include <algorithm>
 #include <cstdint>
 #include <memory>
 #include <string>
@@ -142,7 +143,8 @@ TEST(ParserObjectWriteHostTest, ParserWritesToBothHostsFromOneParse) {
   ASSERT_TRUE(resolved.has_value());
   ASSERT_NE(resolved->payload.anchor, nullptr);
   const std::vector<uint8_t> expected{0xAA, 0xBB, 0xCC};
-  EXPECT_EQ(std::vector<uint8_t>(resolved->payload.bytes.begin(), resolved->payload.bytes.end()), expected);
+  EXPECT_TRUE(
+      std::equal(resolved->payload.bytes.begin(), resolved->payload.bytes.end(), expected.begin(), expected.end()));
 
   // (Scalar side requires flushing + a read path; Phase-3 scope is proving
   // both hosts were resolved and invoked. Scalar writes go into DataEngine
@@ -202,9 +204,9 @@ TEST(ParserObjectWriteHostTest, ObjectHostViewPushLazyThroughSdk) {
 
   auto resolved = store.latestAt(ObjectTopicId{topic.id}, 10);
   ASSERT_TRUE(resolved.has_value());
-  EXPECT_EQ(
-      std::vector<uint8_t>(resolved->payload.bytes.begin(), resolved->payload.bytes.end()),
-      (std::vector<uint8_t>{0xAA, 0xBB}));
+  const std::vector<uint8_t> expected{0xAA, 0xBB};
+  EXPECT_TRUE(
+      std::equal(resolved->payload.bytes.begin(), resolved->payload.bytes.end(), expected.begin(), expected.end()));
   EXPECT_GE(fetch_calls, 1);
 }
 

pj_plugins/include/pj_plugins/host/service_registry_builder.hpp

@@ -43,7 +43,7 @@ class ServiceRegistryBuilder {
   ///                          not take ownership of ctx/vtable.
   /// @return `Expected<void>`: ok on success, error string on duplicate or
   ///         null fat-pointer field.
-  [[nodiscard]] ::PJ::Expected<void, std::string> tryRegisterService(
+  [[nodiscard]] ::PJ::Status tryRegisterService(
       std::string_view name, uint32_t protocol_version, PJ_service_t service) {
     const std::string service_name(name);
     if (service.ctx == nullptr || service.vtable == nullptr) {