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>

Author: facontidavide

pj_base/include/pj_base/buffer_anchor.hpp

@@ -46,17 +46,19 @@ 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)) {}
 };
 
 /// 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.
+/// whose Span points at that vector's contents. Use this when the producer
+/// materializes bytes from a source that has no natural anchor (e.g. a
+/// C-ABI fetch returning a raw byte buffer); when an upstream allocation
+/// already exists (a chunk cache, an mmap region), construct the PayloadView
+/// directly with that anchor to avoid the 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/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,17 +40,30 @@ struct ObjectTopicDescriptor {
   std::string metadata_json;
 };
 
+/// Eager payload alternative: shared ownership of an owned byte buffer.
+/// memoryUsage() counts the buffer's size against the retention budget.
+using SharedBuffer = std::shared_ptr<const std::vector<uint8_t>>;
+
+/// Lazy payload alternative: idempotent, thread-safe callable returning a
+/// PayloadView (Span + BufferAnchor). The anchor may be any shared_ptr-backed
+/// storage (chunk cache, mmap, etc.); type erasure is preserved end-to-end
+/// through the store. Lazy entries are not counted against the retention
+/// budget — bytes are owned upstream, not by the store.
+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;
+  std::variant<SharedBuffer, LazyCallback> payload;
 };
 
+/// Result of resolving an ObjectEntry. Holds the type-erased BufferAnchor
+/// keeping the bytes alive and a Span over the producer-published range.
+/// For eager entries the anchor wraps the SharedBuffer's underlying vector;
+/// for lazy entries it is whatever the producer's PayloadView carries.
 struct ResolvedObjectEntry {
   Timestamp timestamp = 0;
-  std::shared_ptr<const std::vector<uint8_t>> data;
+  sdk::BufferAnchor anchor;
+  Span<const uint8_t> view;
 };
 
 struct RetentionBudget {
@@ -110,7 +124,7 @@ 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
@@ -119,7 +133,7 @@ class ObjectStore {
   // 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);
+  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<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) {
@@ -306,19 +304,15 @@ 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)) {
-    resolved.data = *owned;
-  } else if (const auto* lazy = std::any_cast<std::function<sdk::PayloadView()>>(&entry.payload)) {
-    // Recover the shared_ptr<const vector<uint8_t>> from the PayloadView's
-    // type-erased anchor. Contract: callers of pushLazy must construct the
-    // PayloadView with an anchor that is exactly a
-    // shared_ptr<const std::vector<uint8_t>> — typically the same buffer the
-    // closure's Span points at. The static_pointer_cast reinterprets the
-    // shared_ptr<const void> back to that concrete type, sharing ownership
-    // without copying bytes. Producers that hold their bytes in any other
-    // container should use pushOwned instead.
+  if (const auto* owned = std::get_if<SharedBuffer>(&entry.payload)) {
+    if (*owned) {
+      resolved.anchor = sdk::BufferAnchor{*owned};
+      resolved.view = Span<const uint8_t>{(*owned)->data(), (*owned)->size()};
+    }
+  } else if (const auto* lazy = std::get_if<LazyCallback>(&entry.payload)) {
     auto pv = (*lazy)();
-    resolved.data = std::static_pointer_cast<const std::vector<uint8_t>>(pv.anchor);
+    resolved.anchor = std::move(pv.anchor);
+    resolved.view = pv.bytes;
   }
 
   return resolved;
@@ -330,7 +324,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/src/plugin_data_host.cpp

@@ -1435,10 +1435,12 @@ 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.
+/// Box holding the type-erased anchor that keeps ObjectStore bytes alive,
+/// plus the Span over the producer-published range. One allocated per
+/// successful read_latest_at; freed by release_bytes.
 struct ObjectBytesBox {
-  std::shared_ptr<const std::vector<uint8_t>> bytes;
+  sdk::BufferAnchor anchor;
+  Span<const uint8_t> view;
 };
 
 PJ_object_topic_handle_t toolboxObjectLookupTopic(void* ctx, PJ_string_view_t topic_name) noexcept {
@@ -1508,12 +1510,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->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->anchor), entry->view};
     *out_handle = reinterpret_cast<PJ_object_bytes_handle_t>(box);
     if (out_timestamp != nullptr) {
       *out_timestamp = entry->timestamp;
@@ -1542,14 +1544,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->view.empty()) {
     return;
   }
   if (out_data != nullptr) {
-    *out_data = box->bytes->data();
+    *out_data = box->view.data();
   }
   if (out_size != nullptr) {
-    *out_size = box->bytes->size();
+    *out_size = box->view.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>
@@ -148,7 +150,7 @@ TEST(ObjectStoreTest, LatestAtExact) {
   auto r = store.latestAt(id, 200);
   ASSERT_TRUE(r.has_value());
   EXPECT_EQ(r->timestamp, 200);
-  EXPECT_EQ((*r->data)[0], 0x02);
+  EXPECT_EQ(r->view[0], 0x02);
 }
 
 TEST(ObjectStoreTest, LatestAtBetween) {
@@ -290,8 +292,68 @@ TEST(ObjectStoreTest, PushLazyResolves) {
   auto r = store.latestAt(id, 100);
   ASSERT_TRUE(r.has_value());
   EXPECT_EQ(call_count, 1);
-  EXPECT_EQ(r->data->size(), 2u);
-  EXPECT_EQ((*r->data)[0], 0xDE);
+  EXPECT_EQ(r->view.size(), 2u);
+  EXPECT_EQ(r->view[0], 0xDE);
+}
+
+// Regression: the lazy path must preserve the BufferAnchor's concrete type.
+// Anchor here is a shared_ptr<TestBuffer> — NOT a shared_ptr<vector>. If
+// resolveEntry static_pointer_cast'd to vector (the prior implementation),
+// view would point at garbage and ASAN would flag it. We assert that
+// (a) the bytes the producer published survive resolution unchanged, and
+// (b) the anchor's refcount stays > 0 through the resolve (the store does
+// not silently swap it for a different anchor type).
+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->view.size(), 4u);
+  EXPECT_EQ(r->view[0], 0x11);
+  EXPECT_EQ(r->view[3], 0x44);
+  EXPECT_FALSE(weak_buffer.expired());  // anchor still holds the buffer alive
+}
+
+// Regression: PayloadView::bytes is the *producer-chosen sub-range* of the
+// anchor's storage. resolveEntry must propagate that Span verbatim — not the
+// anchor's full extent. The prior implementation ignored the Span and
+// returned the anchor's whole vector.
+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->view.size(), 10u);
+  EXPECT_EQ(r->view.data(), chunk->data() + 20);
+  EXPECT_EQ(r->view[0], 20);
+  EXPECT_EQ(r->view[9], 29);
 }
 
 // =========================================================================
@@ -328,13 +390,13 @@ TEST(ObjectStoreTest, HandleSurvivesEviction) {
 
   auto handle = store.latestAt(id, 100);
   ASSERT_TRUE(handle.has_value());
-  EXPECT_EQ((*handle->data)[0], 0xAA);
+  EXPECT_EQ(handle->view[0], 0xAA);
 
   store.evictBefore(id, 150);
   EXPECT_EQ(store.entryCount(id), 1u);
 
-  EXPECT_EQ(handle->data->size(), 4u);
-  EXPECT_EQ((*handle->data)[0], 0xAA);
+  EXPECT_EQ(handle->view.size(), 4u);
+  EXPECT_EQ(handle->view[0], 0xAA);
 }
 
 // =========================================================================

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>
@@ -63,9 +64,9 @@ TEST(PluginDataHostObjectTest, PushOwnedStoresBytes) {
   EXPECT_EQ(f.store.entryCount(store_id), 2U);
   auto resolved = f.store.latestAt(store_id, 2000);
   ASSERT_TRUE(resolved.has_value());
-  ASSERT_NE(resolved->data, nullptr);
-  EXPECT_EQ(resolved->data->size(), payload.size());
-  EXPECT_EQ(*resolved->data, payload);
+  ASSERT_NE(resolved->anchor, nullptr);
+  EXPECT_EQ(resolved->view.size(), payload.size());
+  EXPECT_TRUE(std::equal(resolved->view.begin(), resolved->view.end(), payload.begin(), payload.end()));
 }
 
 TEST(PluginDataHostObjectTest, PushLazyRetainsClosureUntilEviction) {
@@ -93,8 +94,8 @@ TEST(PluginDataHostObjectTest, PushLazyRetainsClosureUntilEviction) {
   // Each read invokes the fetch closure.
   auto first = f.store.latestAt(ObjectTopicId{topic.id}, 42);
   ASSERT_TRUE(first.has_value());
-  ASSERT_NE(first->data, nullptr);
-  EXPECT_EQ(*first->data, shared->payload);
+  ASSERT_NE(first->anchor, nullptr);
+  EXPECT_TRUE(std::equal(first->view.begin(), first->view.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 +150,7 @@ 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(*resolved->data, ctx->payload);
+  EXPECT_TRUE(std::equal(resolved->view.begin(), resolved->view.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>
@@ -140,9 +141,9 @@ TEST(ParserObjectWriteHostTest, ParserWritesToBothHostsFromOneParse) {
   // Object-store side: bytes landed.
   auto resolved = store.latestAt(ObjectTopicId{obj_topic.id}, 100);
   ASSERT_TRUE(resolved.has_value());
-  ASSERT_NE(resolved->data, nullptr);
+  ASSERT_NE(resolved->anchor, nullptr);
   const std::vector<uint8_t> expected{0xAA, 0xBB, 0xCC};
-  EXPECT_EQ(*resolved->data, expected);
+  EXPECT_TRUE(std::equal(resolved->view.begin(), resolved->view.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,7 +203,8 @@ TEST(ParserObjectWriteHostTest, ObjectHostViewPushLazyThroughSdk) {
 
   auto resolved = store.latestAt(ObjectTopicId{topic.id}, 10);
   ASSERT_TRUE(resolved.has_value());
-  EXPECT_EQ(*resolved->data, (std::vector<uint8_t>{0xAA, 0xBB}));
+  const std::vector<uint8_t> expected{0xAA, 0xBB};
+  EXPECT_TRUE(std::equal(resolved->view.begin(), resolved->view.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) {