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>

Author: facontidavide

pj_base/include/pj_base/buffer_anchor.hpp

@@ -51,14 +51,10 @@ struct PayloadView {
         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. 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.
+/// 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/include/pj_datastore/object_store.hpp

@@ -40,30 +40,21 @@ 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.
+/// Eager payload: store-owned bytes, counted 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.
+/// 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<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;
-  sdk::BufferAnchor anchor;
-  Span<const uint8_t> view;
+  sdk::PayloadView payload;
 };
 
 struct RetentionBudget {
@@ -126,13 +117,9 @@ class ObjectStore {
 
   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.
+  // 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

@@ -306,13 +306,13 @@ ResolvedObjectEntry ObjectStore::resolveEntry(const ObjectEntry& entry) {
 
   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()};
+      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)) {
-    auto pv = (*lazy)();
-    resolved.anchor = std::move(pv.anchor);
-    resolved.view = pv.bytes;
+    resolved.payload = (*lazy)();
   }
 
   return resolved;

pj_datastore/src/plugin_data_host.cpp

@@ -1435,12 +1435,10 @@ void sourceObjectSetRetentionBudget(
 // Toolbox object read host trampolines
 // ---------------------------------------------------------------------------
 
-/// 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.
+/// Heap holder for the PayloadView backing PJ_object_bytes_handle_t.
+/// Allocated by read_latest_at; freed by release_bytes.
 struct ObjectBytesBox {
-  sdk::BufferAnchor anchor;
-  Span<const uint8_t> view;
+  sdk::PayloadView payload;
 };
 
 PJ_object_topic_handle_t toolboxObjectLookupTopic(void* ctx, PJ_string_view_t topic_name) noexcept {
@@ -1510,12 +1508,12 @@ bool toolboxObjectReadLatestAt(
   *out_handle = nullptr;
   try {
     auto entry = impl->store.latestAt(ObjectTopicId{topic.id}, timestamp_ns);
-    if (!entry.has_value() || entry->anchor == 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->anchor), entry->view};
+    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;
@@ -1544,14 +1542,14 @@ void toolboxObjectGetBytes(PJ_object_bytes_handle_t handle, const uint8_t** out_
     return;
   }
   auto* box = reinterpret_cast<ObjectBytesBox*>(handle);
-  if (box->view.empty()) {
+  if (box->payload.bytes.empty()) {
     return;
   }
   if (out_data != nullptr) {
-    *out_data = box->view.data();
+    *out_data = box->payload.bytes.data();
   }
   if (out_size != nullptr) {
-    *out_size = box->view.size();
+    *out_size = box->payload.bytes.size();
   }
 }
 

pj_datastore/tests/object_store_test.cpp

@@ -150,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->view[0], 0x02);
+  EXPECT_EQ(r->payload.bytes[0], 0x02);
 }
 
 TEST(ObjectStoreTest, LatestAtBetween) {
@@ -292,17 +292,13 @@ TEST(ObjectStoreTest, PushLazyResolves) {
   auto r = store.latestAt(id, 100);
   ASSERT_TRUE(r.has_value());
   EXPECT_EQ(call_count, 1);
-  EXPECT_EQ(r->view.size(), 2u);
-  EXPECT_EQ(r->view[0], 0xDE);
+  EXPECT_EQ(r->payload.bytes.size(), 2u);
+  EXPECT_EQ(r->payload.bytes[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).
+// 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};
@@ -322,16 +318,14 @@ TEST(ObjectStoreTest, PushLazyPreservesAnchorType) {
 
   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);
+  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: 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.
+// 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);
@@ -350,10 +344,10 @@ TEST(ObjectStoreTest, PushLazyHonorsSpanSubview) {
 
   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);
+  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);
 }
 
 // =========================================================================
@@ -390,13 +384,13 @@ TEST(ObjectStoreTest, HandleSurvivesEviction) {
 
   auto handle = store.latestAt(id, 100);
   ASSERT_TRUE(handle.has_value());
-  EXPECT_EQ(handle->view[0], 0xAA);
+  EXPECT_EQ(handle->payload.bytes[0], 0xAA);
 
   store.evictBefore(id, 150);
   EXPECT_EQ(store.entryCount(id), 1u);
 
-  EXPECT_EQ(handle->view.size(), 4u);
-  EXPECT_EQ(handle->view[0], 0xAA);
+  EXPECT_EQ(handle->payload.bytes.size(), 4u);
+  EXPECT_EQ(handle->payload.bytes[0], 0xAA);
 }
 
 // =========================================================================

pj_datastore/tests/plugin_data_host_object_test.cpp

@@ -64,9 +64,10 @@ 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->anchor, nullptr);
-  EXPECT_EQ(resolved->view.size(), payload.size());
-  EXPECT_TRUE(std::equal(resolved->view.begin(), resolved->view.end(), payload.begin(), payload.end()));
+  ASSERT_NE(resolved->payload.anchor, nullptr);
+  EXPECT_EQ(resolved->payload.bytes.size(), payload.size());
+  EXPECT_TRUE(
+      std::equal(resolved->payload.bytes.begin(), resolved->payload.bytes.end(), payload.begin(), payload.end()));
 }
 
 TEST(PluginDataHostObjectTest, PushLazyRetainsClosureUntilEviction) {
@@ -94,8 +95,10 @@ 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->anchor, nullptr);
-  EXPECT_TRUE(std::equal(first->view.begin(), first->view.end(), shared->payload.begin(), shared->payload.end()));
+  ASSERT_NE(first->payload.anchor, nullptr);
+  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);
@@ -150,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_TRUE(std::equal(resolved->view.begin(), resolved->view.end(), ctx->payload.begin(), ctx->payload.end()));
+  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

@@ -141,9 +141,10 @@ 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->anchor, nullptr);
+  ASSERT_NE(resolved->payload.anchor, nullptr);
   const std::vector<uint8_t> expected{0xAA, 0xBB, 0xCC};
-  EXPECT_TRUE(std::equal(resolved->view.begin(), resolved->view.end(), expected.begin(), expected.end()));
+  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
@@ -204,7 +205,8 @@ TEST(ParserObjectWriteHostTest, ObjectHostViewPushLazyThroughSdk) {
   auto resolved = store.latestAt(ObjectTopicId{topic.id}, 10);
   ASSERT_TRUE(resolved.has_value());
   const std::vector<uint8_t> expected{0xAA, 0xBB};
-  EXPECT_TRUE(std::equal(resolved->view.begin(), resolved->view.end(), expected.begin(), expected.end()));
+  EXPECT_TRUE(
+      std::equal(resolved->payload.bytes.begin(), resolved->payload.bytes.end(), expected.begin(), expected.end()));
   EXPECT_GE(fetch_calls, 1);
 }