refactor(object_store): widen ResolvedObjectEntry::data to PayloadView; drop ObjectBytesBox

Builds on top of !184 to remove the last hidden contract:
resolveEntry's lazy branch used static_pointer_cast<const vector>(pv.anchor),
which forced every lazy producer to anchor on shared_ptr<vector<uint8_t>>
and defeated zero-copy for any producer that already had a different
concrete anchor (C-ABI ctx, arrow::Buffer, mmap pool, etc.).

ResolvedObjectEntry::data (shared_ptr<const vector<uint8_t>>) is now
ResolvedObjectEntry::payload (sdk::PayloadView). resolveEntry returns
PayloadView verbatim in both branches — no concrete-type assumption.

ObjectBytesBox in plugin_data_host.cpp is removed: with PayloadView on
the resolved entry it was a no-value wrapper; toolboxObjectReadLatestAt
allocates a PayloadView directly and the C-ABI handle is its pointer.

All consumers migrated to entry->payload.bytes / entry->payload.anchor.

This is the §14.5 widening from the streaming-manager mejoras report —
precondition for end-to-end zero-copy ingest from the streaming runtime
capturing PJ_payload_t.anchor as shared_ptr<void>.

Author: pabloinigoblasco

pj_datastore/include/pj_datastore/object_store.hpp

@@ -49,7 +49,16 @@ struct ObjectEntry {
 
 struct ResolvedObjectEntry {
   Timestamp timestamp = 0;
-  std::shared_ptr<const std::vector<uint8_t>> data;
+  // PayloadView lets the entry hold an opaque anchor (any shared_ptr<T>) plus
+  // a non-owning Span over the bytes. Consumers read `payload.bytes` for the
+  // data; they retain `payload.anchor` if they need the bytes to outlive the
+  // resolve call. Both `pushOwned` and `pushLazy` paths land here:
+  //   - owned:  payload.bytes spans the shared_ptr<vector<uint8_t>>; anchor IS that shared_ptr.
+  //   - lazy:   payload is whatever the closure returns; anchor can be any shared_ptr<T>
+  //             (e.g. a C-ABI payload anchor wrapped as shared_ptr<void>).
+  // resolveEntry never casts the anchor to a concrete type; the type erasure
+  // stays opaque all the way to the consumer.
+  sdk::PayloadView payload;
 };
 
 struct RetentionBudget {

pj_datastore/src/object_store.cpp

@@ -376,18 +376,20 @@ ResolvedObjectEntry ObjectStore::resolveEntry(const ObjectEntry& entry) {
   resolved.timestamp = entry.timestamp;
 
   if (const auto* owned = std::any_cast<std::shared_ptr<const std::vector<uint8_t>>>(&entry.payload)) {
-    resolved.data = *owned;
+    // 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)) {
-    // 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.
-    auto pv = (*lazy)();
-    resolved.data = std::static_pointer_cast<const std::vector<uint8_t>>(pv.anchor);
+    // 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
+    // hidden contract that locked the slot to shared_ptr<vector<uint8_t>>;
+    // dropping it lets producers anchor on arrow::Buffer, mmap pools, or
+    // C-ABI payload anchors equally.
+    resolved.payload = (*lazy)();
   }
 
   return resolved;

pj_datastore/src/plugin_data_host.cpp

@@ -1456,12 +1456,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.
-struct ObjectBytesBox {
-  std::shared_ptr<const std::vector<uint8_t>> bytes;
-};
-
+// The C-ABI exposes the bytes via a void-handle (PJ_object_bytes_handle_t)
+// that the plugin must later release. We allocate a sdk::PayloadView on the
+// heap and reinterpret_cast its pointer to the handle: the PayloadView's
+// anchor is what keeps the underlying buffer alive until the plugin calls
+// release_bytes. No wrapper struct needed — PayloadView already carries
+// bytes (Span) + anchor (BufferAnchor) in one value.
 PJ_object_topic_handle_t toolboxObjectLookupTopic(void* ctx, PJ_string_view_t topic_name) noexcept {
   auto* impl = static_cast<DatastoreToolboxObjectReadHostState*>(ctx);
   try {
@@ -1529,13 +1529,13 @@ 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)};
-    *out_handle = reinterpret_cast<PJ_object_bytes_handle_t>(box);
+    auto* payload_handle = new sdk::PayloadView(std::move(entry->payload));
+    *out_handle = reinterpret_cast<PJ_object_bytes_handle_t>(payload_handle);
     if (out_timestamp != nullptr) {
       *out_timestamp = entry->timestamp;
     }
@@ -1562,23 +1562,23 @@ void toolboxObjectGetBytes(PJ_object_bytes_handle_t handle, const uint8_t** out_
   if (handle == nullptr) {
     return;
   }
-  auto* box = reinterpret_cast<ObjectBytesBox*>(handle);
-  if (!box->bytes) {
+  const auto* payload = reinterpret_cast<const sdk::PayloadView*>(handle);
+  if (payload->anchor == nullptr) {
     return;
   }
   if (out_data != nullptr) {
-    *out_data = box->bytes->data();
+    *out_data = payload->bytes.data();
   }
   if (out_size != nullptr) {
-    *out_size = box->bytes->size();
+    *out_size = payload->bytes.size();
   }
 }
 
 void toolboxObjectReleaseBytes(PJ_object_bytes_handle_t handle) noexcept {
   if (handle == nullptr) {
     return;
   }
-  delete reinterpret_cast<ObjectBytesBox*>(handle);
+  delete reinterpret_cast<sdk::PayloadView*>(handle);
 }
 
 std::size_t toolboxObjectEntryCount(void* ctx, PJ_object_topic_handle_t topic) noexcept {

pj_datastore/tests/object_store_test.cpp

@@ -148,7 +148,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->payload.bytes[0], 0x02);
 }
 
 TEST(ObjectStoreTest, LatestAtBetween) {
@@ -290,8 +290,8 @@ 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->payload.bytes.size(), 2u);
+  EXPECT_EQ(r->payload.bytes[0], 0xDE);
 }
 
 // =========================================================================
@@ -328,13 +328,13 @@ TEST(ObjectStoreTest, HandleSurvivesEviction) {
 
   auto handle = store.latestAt(id, 100);
   ASSERT_TRUE(handle.has_value());
-  EXPECT_EQ((*handle->data)[0], 0xAA);
+  EXPECT_EQ(handle->payload.bytes[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->payload.bytes.size(), 4u);
+  EXPECT_EQ(handle->payload.bytes[0], 0xAA);
 }
 
 // =========================================================================
@@ -490,13 +490,13 @@ TEST(ObjectStoreFlushTest, BasicMoveOwnedEntries) {
   auto e0 = dst.at(dst_id, 0);
   ASSERT_TRUE(e0.has_value());
   EXPECT_EQ(e0->timestamp, 100);
-  EXPECT_EQ(e0->data->size(), 8u);
-  EXPECT_EQ((*e0->data)[0], 0xAA);
+  EXPECT_EQ(e0->payload.bytes.size(), 8u);
+  EXPECT_EQ(e0->payload.bytes[0], 0xAA);
 
   auto e2 = dst.at(dst_id, 2);
   ASSERT_TRUE(e2.has_value());
   EXPECT_EQ(e2->timestamp, 300);
-  EXPECT_EQ((*e2->data)[0], 0xCC);
+  EXPECT_EQ(e2->payload.bytes[0], 0xCC);
 }
 
 TEST(ObjectStoreFlushTest, PreservesTopicRegistrationOnSrc) {
@@ -605,8 +605,8 @@ TEST(ObjectStoreFlushTest, MultipleTopicsFlushed) {
 
   EXPECT_EQ(dst.entryCount(dst_a), 2u);
   EXPECT_EQ(dst.entryCount(dst_b), 2u);
-  EXPECT_EQ((*dst.at(dst_a, 0)->data)[0], 0xAA);
-  EXPECT_EQ((*dst.at(dst_b, 1)->data)[0], 0xDD);
+  EXPECT_EQ(dst.at(dst_a, 0)->payload.bytes[0], 0xAA);
+  EXPECT_EQ(dst.at(dst_b, 1)->payload.bytes[0], 0xDD);
 }
 
 TEST(ObjectStoreFlushTest, LazyEntriesPreserveSemantics) {
@@ -615,9 +615,9 @@ TEST(ObjectStoreFlushTest, LazyEntriesPreserveSemantics) {
   auto dst_id = registerSameDescriptor(dst);
 
   int src_invocations = 0;
-  src.pushLazy(src_id, 100, [&src_invocations]() -> std::vector<uint8_t> {
+  src.pushLazy(src_id, 100, [&src_invocations]() -> sdk::PayloadView {
     ++src_invocations;
-    return {0xDE, 0xAD, 0xBE, 0xEF};
+    return sdk::makePayloadView({0xDE, 0xAD, 0xBE, 0xEF});
   });
 
   // Flush itself must NOT invoke the closure — it just moves the std::function.
@@ -632,8 +632,8 @@ TEST(ObjectStoreFlushTest, LazyEntriesPreserveSemantics) {
   auto resolved = dst.latestAt(dst_id, 100);
   ASSERT_TRUE(resolved.has_value());
   EXPECT_EQ(src_invocations, 1);
-  ASSERT_EQ(resolved->data->size(), 4u);
-  EXPECT_EQ((*resolved->data)[0], 0xDE);
+  ASSERT_EQ(resolved->payload.bytes.size(), 4u);
+  EXPECT_EQ(resolved->payload.bytes[0], 0xDE);
 }
 
 TEST(ObjectStoreFlushTest, RetentionBudgetAppliedAfterFlush) {
@@ -668,13 +668,13 @@ TEST(ObjectStoreFlushTest, ZeroCopyOwnershipChainSurvives) {
   src.pushOwned(src_id, 100, makePayload(16, 0x77));
   auto pre_handle = src.at(src_id, 0);
   ASSERT_TRUE(pre_handle.has_value());
-  const auto* pre_ptr = pre_handle->data.get();
+  const auto* pre_ptr = pre_handle->payload.anchor.get();
 
   ASSERT_TRUE(src.flushTo(dst).has_value());
 
   auto post_handle = dst.at(dst_id, 0);
   ASSERT_TRUE(post_handle.has_value());
-  EXPECT_EQ(post_handle->data.get(), pre_ptr) << "shared_ptr identity must survive the flush";
+  EXPECT_EQ(post_handle->payload.anchor.get(), pre_ptr) << "shared_ptr identity must survive the flush";
 }
 
 }  // namespace

pj_datastore/tests/plugin_data_host_object_test.cpp

@@ -63,9 +63,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->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);
 }
 
 TEST(PluginDataHostObjectTest, PushLazyRetainsClosureUntilEviction) {
@@ -93,8 +93,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->payload.anchor, nullptr);
+  EXPECT_EQ(std::vector<uint8_t>(first->payload.bytes.begin(), first->payload.bytes.end()), shared->payload);
   EXPECT_GE(shared->fetch_calls.load(), 1);
 
   auto second = f.store.latestAt(ObjectTopicId{topic.id}, 42);
@@ -149,7 +149,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_EQ(std::vector<uint8_t>(resolved->payload.bytes.begin(), resolved->payload.bytes.end()), ctx->payload);
   EXPECT_EQ(ctx->destroy_count.load(), 0);
 
   // Evict — destroy_fn runs exactly once.

pj_datastore/tests/plugin_parser_object_write_test.cpp

@@ -140,9 +140,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->payload.anchor, nullptr);
   const std::vector<uint8_t> expected{0xAA, 0xBB, 0xCC};
-  EXPECT_EQ(*resolved->data, expected);
+  EXPECT_EQ(std::vector<uint8_t>(resolved->payload.bytes.begin(), resolved->payload.bytes.end()), expected);
 
   // (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 +202,9 @@ 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}));
+  EXPECT_EQ(
+      std::vector<uint8_t>(resolved->payload.bytes.begin(), resolved->payload.bytes.end()),
+      (std::vector<uint8_t>{0xAA, 0xBB}));
   EXPECT_GE(fetch_calls, 1);
 }