chore: minor changes

Adopt the follow-up tweaks from upstream PR #105: small reshuffles in
engine.hpp / object_store.hpp / object_store.cpp / engine.cpp /
plugin_data_host.cpp (net -32 lines across the five files). No semantic
or API impact; tightens declarations and trims a few unused includes.

Author: facontidavide

pj_datastore/include/pj_datastore/engine.hpp

@@ -82,22 +82,18 @@ class DataEngine {
   /// Evict old chunks outside the retention window.
   void enforceRetention(PJ::Timestamp retention_window_ns);
 
-  /// Move every committed chunk from this engine into `dst`, leaving this
-  /// engine's topic storages empty (datasets, topics, schemas, and time
-  /// domains remain registered). Topics are matched by descriptor
-  /// (`dataset_id` + `name`); both engines must have the matching topics
-  /// registered or the call fails without partial mutation. Monotonicity is
-  /// enforced strictly per topic: the source's earliest chunk timestamp must
-  /// be greater than or equal to the destination's current `time_max()`.
+  /// Move every committed chunk into `dst`, leaving this engine's storages
+  /// empty (datasets, topics, schemas, time domains stay registered). Topics
+  /// are matched by descriptor (`dataset_id` + `name`); both engines must have
+  /// them registered. Monotonicity is enforced per topic: the source's
+  /// earliest chunk timestamp must be >= the destination's `time_max()`. Any
+  /// failure mutates neither engine.
   ///
-  /// Zero-copy on the chunk data: the destination's `std::deque<TopicChunk>`
-  /// receives the source's chunks via `std::move`, never via copy. The chunk
-  /// internals (column buffers, value arrays) are pointer/buffer moves.
-  ///
-  /// Schema compatibility is the caller's responsibility — typically the
-  /// destination is kept in lockstep with the source through parallel
-  /// `createDataset` / `createTopic` registration at startup.
-  PJ::Expected<void, std::string> flushTo(DataEngine& dst);
+  /// Zero-copy: dst's `std::deque<TopicChunk>` receives the chunks via
+  /// `std::move` (column buffers/value arrays are pointer moves). Schema
+  /// compatibility is the caller's responsibility — typically dst is kept in
+  /// lockstep with the source via parallel registration at startup.
+  PJ::Status flushTo(DataEngine& dst);
 
   // Writer/Reader factories
   /// Create a writer bound to this engine.

pj_datastore/include/pj_datastore/object_store.hpp

@@ -49,24 +49,15 @@ using LazyCallback = std::function<sdk::PayloadView()>;
 
 struct ObjectEntry {
   Timestamp timestamp = 0;
-  // 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.
+  // Eager owned bytes or a lazy resolver; resolveEntry discriminates via std::get_if.
   std::variant<SharedBuffer, LazyCallback> payload;
 };
 
 struct ResolvedObjectEntry {
   Timestamp timestamp = 0;
-  // 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.
+  // Non-owning Span over the bytes plus an opaque anchor (any shared_ptr<T>).
+  // Consumers read `payload.bytes`; retain `payload.anchor` to keep the bytes
+  // alive past the resolve call. resolveEntry never casts the anchor.
   sdk::PayloadView payload;
 };
 
@@ -130,11 +121,9 @@ class ObjectStore {
 
   Status pushOwned(ObjectTopicId id, Timestamp timestamp, std::vector<uint8_t> payload);
 
-  // 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.
+  // Fetcher runs on every read; the store retains the anchor via PayloadView
+  // and never copies. The closure can return a view over bytes the producer
+  // already owns (chunk cache, mmap, hand-off between stores).
   Status pushLazy(ObjectTopicId id, Timestamp timestamp, LazyCallback fetch);
 
   // --- Read ---
@@ -164,26 +153,17 @@ class ObjectStore {
 
   // --- Cross-store flush ---
 
-  // Move every entry from this store into `dst`, leaving this store empty
-  // (topic registrations are preserved). Topics are matched by descriptor
-  // (dataset_id + topic_name); both stores must have registered the same
-  // descriptors or the call fails without partial mutation. For each series,
-  // monotonicity is enforced strictly: the earliest timestamp being moved
-  // must be greater than or equal to the destination's current last
-  // timestamp. On any validation failure the call returns an error and
-  // neither store is mutated.
+  // Move every entry into `dst`, leaving this store empty (registrations kept).
+  // Topics are matched by descriptor (dataset_id + topic_name); both stores
+  // must share descriptors. Monotonicity is enforced per series: the earliest
+  // moved timestamp must be >= the destination's last. Any validation failure
+  // returns an error and mutates neither store.
   //
-  // Zero-copy on the payload bytes. Each ObjectEntry is moved into the
-  // 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.
-  //
-  // After the move, the destination's retention budget is applied to each
-  // touched series in normal order.
-  Expected<void, std::string> flushTo(ObjectStore& dst);
+  // Zero-copy: each ObjectEntry is moved by value, so the variant's shared_ptr
+  // or closure transfers as a pointer move — bytes are never copied. Lazy
+  // entries keep their semantics; their closure re-runs only on a dst read.
+  // Afterward, dst's retention budget is applied to each touched series.
+  Status flushTo(ObjectStore& dst);
 
   // --- Lifecycle ---
 

pj_datastore/src/engine.cpp

@@ -182,7 +182,7 @@ void DataEngine::enforceRetention(Timestamp retention_window_ns) {
   }
 }
 
-Expected<void, std::string> DataEngine::flushTo(DataEngine& dst) {
+Status DataEngine::flushTo(DataEngine& dst) {
   if (&dst == this) {
     return PJ::unexpected("flushTo: source and destination are the same engine");
   }

pj_datastore/src/object_store.cpp

@@ -267,7 +267,7 @@ void ObjectStore::evictAllBefore(Timestamp threshold) {
 
 // --- Cross-store flush ---
 
-Expected<void, std::string> ObjectStore::flushTo(ObjectStore& dst) {
+Status ObjectStore::flushTo(ObjectStore& dst) {
   if (&dst == this) {
     return unexpected("flushTo: source and destination are the same store");
   }
@@ -374,22 +374,17 @@ ResolvedObjectEntry ObjectStore::resolveEntry(const ObjectEntry& entry) {
   resolved.timestamp = entry.timestamp;
 
   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. A
-    // default-constructed entry holds a null SharedBuffer, so guard it.
+    // Span the vector, anchor on 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
-    // 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.
+    // Forward the closure's PayloadView verbatim. The anchor stays opaque (no
+    // cast), so producers can back it with arrow::Buffer, mmap, or a C-ABI anchor.
     resolved.payload = (*lazy)();
   }
 

pj_datastore/src/plugin_data_host.cpp

@@ -1456,12 +1456,9 @@ void sourceObjectSetRetentionBudget(
 // Toolbox object read host trampolines
 // ---------------------------------------------------------------------------
 
-// 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_bytes_handle_t is a heap-allocated sdk::PayloadView: its anchor
+// keeps the buffer alive until the plugin calls release_bytes. No wrapper
+// struct needed — PayloadView already carries the Span + anchor.
 PJ_object_topic_handle_t toolboxObjectLookupTopic(void* ctx, PJ_string_view_t topic_name) noexcept {
   auto* impl = static_cast<DatastoreToolboxObjectReadHostState*>(ctx);
   try {