refactor(object_store): PayloadView end-to-end; cross-store flushTo; atomic target swap

A consolidated refactor of the ObjectStore / DataEngine surface that
makes PayloadView the universal vocabulary for ownership of bytes
across the SDK, adds zero-copy cross-instance transfer of entries
between stores, and exposes an atomic target-swap primitive on the
plugin data host used to redirect writes between two stores at runtime.

ObjectStore + ObjectEntry

  - ObjectEntry::payload is std::any holding either
    std::shared_ptr<const std::vector<uint8_t>> (eager owned bytes,
    counted against the retention budget) or
    std::function<sdk::PayloadView()> (lazy resolver returning Span +
    BufferAnchor; not counted, bytes owned upstream).
    resolveEntry dispatches via std::any_cast; each branch handles its
    own concrete type. No static_pointer_cast on the lazy anchor — its
    type erasure (BufferAnchor = shared_ptr<const void>) is preserved
    end-to-end so producers can anchor on any shared_ptr<T>.

  - ResolvedObjectEntry consolidates on PayloadView (Span + anchor).
    Removes the previous shared_ptr<const vector<uint8_t>> field,
    which locked consumers to a concrete anchor type and required a
    hidden static_pointer_cast in resolveEntry that would fail silently
    for non-vector anchors.

  - ObjectStore::flushTo(ObjectStore& dst): two-phase, atomic, zero-
    copy bulk transfer of entries between two instances. Topics matched
    by descriptor; monotonicity enforced strictly per series; failure
    leaves both sides untouched. Each ObjectEntry is moved by value
    via std::move; the std::any inside transfers its buffer intact.

  - DataEngine::flushTo(DataEngine& dst): symmetric primitive for the
    columnar scalar store. Each TopicStorage's sealed-chunk deque is
    moved through to the destination; no chunk constructor invoked.

  - sdk::makePayloadView(std::vector<uint8_t>) replaces
    sdk::makeOwnedPayloadView. Wraps a vector into a shared_ptr that
    serves as both the bytes backing and the BufferAnchor.

  - ObjectBytesBox removed from plugin_data_host. The C-ABI toolbox
    handle becomes a heap-allocated PayloadView directly.

plugin_data_host atomic target swap

  - setTarget on engine and parser write hosts atomically swap which
    engine/store receives subsequent writes. Safe under concurrent
    ingest: in-flight writes complete on the previous target.

Version bump

  - 0.4.0 -> 0.5.0. Source-incompatible: ResolvedObjectEntry::data
    renamed to payload (PayloadView instead of shared_ptr<vector>).
    Consumers migrate from entry->data->{data,size,empty}() to
    entry->payload.bytes.{data,size,empty}(); from entry->data to
    entry->payload.anchor for ownership retention.

Author: pabloinigoblasco

pj_base/include/pj_base/buffer_anchor.hpp

@@ -15,6 +15,8 @@
 
 #include <cstdint>
 #include <memory>
+#include <utility>
+#include <vector>
 
 #include "pj_base/span.hpp"
 
@@ -49,5 +51,19 @@ 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. 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.
+inline PayloadView makePayloadView(std::vector<uint8_t> bytes) {
+  auto shared = std::make_shared<const std::vector<uint8_t>>(std::move(bytes));
+  return PayloadView{
+      Span<const uint8_t>{shared->data(), shared->size()},
+      BufferAnchor{shared},
+  };
+}
+
 }  // namespace sdk
 }  // namespace PJ

pj_datastore/include/pj_datastore/engine.hpp

@@ -79,9 +79,26 @@ class DataEngine {
   ///   derived.onSourceCommitted(engine.commitChunks(writer.flushAll()));
   std::vector<PJ::TopicId> commitChunks(std::vector<std::pair<PJ::TopicId, TopicChunk>> chunks);
 
-  /// Evict old chunks outside retention window.
+  /// 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()`.
+  ///
+  /// 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);
+
   // Writer/Reader factories
   /// Create a writer bound to this engine.
   [[nodiscard]] DataWriter createWriter();

pj_datastore/include/pj_datastore/object_store.hpp

@@ -2,6 +2,7 @@
 // Copyright 2026 Davide Faconti
 // SPDX-License-Identifier: MPL-2.0
 
+#include <any>
 #include <cstddef>
 #include <cstdint>
 #include <deque>
@@ -13,9 +14,9 @@
 #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/types.hpp"
 
@@ -40,12 +41,24 @@ struct ObjectTopicDescriptor {
 
 struct ObjectEntry {
   Timestamp timestamp = 0;
-  std::variant<std::shared_ptr<const std::vector<uint8_t>>, std::function<std::vector<uint8_t>()>> payload;
+  // 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;
 };
 
 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 {
@@ -108,8 +121,14 @@ class ObjectStore {
 
   Expected<void, std::string> pushOwned(ObjectTopicId id, Timestamp timestamp, std::vector<uint8_t> payload);
 
-  Expected<void, std::string> pushLazy(
-      ObjectTopicId id, Timestamp timestamp, std::function<std::vector<uint8_t>()> fetch);
+  // 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);
 
   // --- Read ---
 
@@ -136,6 +155,29 @@ class ObjectStore {
   void evictBefore(ObjectTopicId id, Timestamp threshold);
   void evictAllBefore(Timestamp threshold);
 
+  // --- 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.
+  //
+  // 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
+  // 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);
+
   // --- Lifecycle ---
 
   void removeTopic(ObjectTopicId id);

pj_datastore/include/pj_datastore/plugin_data_host.hpp

@@ -31,6 +31,13 @@ class DatastoreSourceWriteHost {
   [[nodiscard]] PJ_source_write_host_t raw() noexcept;
   void flushPending();
 
+  // Atomically swap the destination DataEngine. Mirrors the parser write
+  // host's setTarget: the streaming two-engine flow routes source-level
+  // scalar pushes to a secondary DataEngine during pause and back to the
+  // primary on resume. Pending rows are flushed to the current engine
+  // before the switch. Does not take ownership.
+  void setTarget(DataEngine* target);
+
  private:
   std::unique_ptr<DatastoreSourceWriteHostState> state_;
 };
@@ -51,6 +58,13 @@ class DatastoreSourceObjectWriteHost {
 
   [[nodiscard]] PJ_object_write_host_t raw() noexcept;
 
+  // Atomically swap the destination store. Used by the streaming two-store
+  // flow to route pushes to a secondary ObjectStore during pause and back to
+  // the primary on resume. Must point to a live store; the host does not
+  // take ownership and the caller must keep the target alive for as long as
+  // the host can receive pushes against it.
+  void setTarget(ObjectStore* target) noexcept;
+
  private:
   std::unique_ptr<DatastoreSourceObjectWriteHostState> state_;
 };
@@ -68,6 +82,14 @@ class DatastoreParserWriteHost {
   [[nodiscard]] PJ_parser_write_host_t raw() noexcept;
   void flushPending();
 
+  // Atomically swap the destination DataEngine. Mirrors the object write
+  // host's setTarget: the streaming two-store flow routes scalar pushes to a
+  // secondary DataEngine during pause and back to the primary on resume.
+  // Pending rows are flushed to the current engine before the switch; the
+  // bound topic must exist in `target` with the same TopicId (the streaming
+  // manager registers it lockstep on both engines). Does not take ownership.
+  void setTarget(DataEngine* target);
+
  private:
   std::unique_ptr<DatastoreParserWriteHostState> state_;
 };
@@ -111,6 +133,13 @@ class DatastoreParserObjectWriteHost {
 
   [[nodiscard]] PJ_parser_object_write_host_t raw() noexcept;
 
+  // Atomically swap the destination store. Used by the streaming two-store
+  // flow to route pushes to a secondary ObjectStore during pause and back to
+  // the primary on resume. The bound topic id must exist in `target` (the
+  // streaming manager ensures this via lockstep registerTopic on both
+  // stores). The host does not take ownership of the target.
+  void setTarget(ObjectStore* target) noexcept;
+
  private:
   std::unique_ptr<DatastoreParserObjectWriteHostState> state_;
 };

pj_datastore/include/pj_datastore/topic_storage.hpp

@@ -57,6 +57,8 @@ struct TopicMetadata {
   uint32_t truncated_sample_count = 0;
 };
 
+class DataEngine;
+
 /// Storage container for committed chunks of one topic.
 class TopicStorage {
  public:
@@ -123,6 +125,12 @@ class TopicStorage {
   void setArrayExpansionCount(const std::string& field_path, uint32_t count);
 
  private:
+  // DataEngine::flushTo needs to move sealed_chunks_ between TopicStorage
+  // instances of different engines without copying. Friending it lets the
+  // transfer happen entirely inside DataEngine without exposing the move
+  // primitive on the public TopicStorage API.
+  friend class DataEngine;
+
   TopicId topic_id_;
   TopicDescriptor descriptor_;
   std::deque<TopicChunk> sealed_chunks_;

pj_datastore/src/engine.cpp

@@ -182,6 +182,63 @@ void DataEngine::enforceRetention(Timestamp retention_window_ns) {
   }
 }
 
+Expected<void, std::string> DataEngine::flushTo(DataEngine& dst) {
+  if (&dst == this) {
+    return PJ::unexpected("flushTo: source and destination are the same engine");
+  }
+
+  // Phase 1: validate. Walk every src topic with sealed chunks and look up
+  // the matching dst topic by descriptor (dataset_id + name). Verify
+  // monotonicity against dst's current time_max. No mutation yet.
+  struct Step {
+    TopicStorage* src;
+    TopicStorage* dst;
+  };
+  std::vector<Step> plan;
+  plan.reserve(impl_->topics.size());
+
+  for (auto it = impl_->topics.begin(); it != impl_->topics.end(); ++it) {
+    auto& src_storage = it.value();
+    if (src_storage.empty()) {
+      continue;
+    }
+    TopicStorage* dst_storage = nullptr;
+    for (auto dst_it = dst.impl_->topics.begin(); dst_it != dst.impl_->topics.end(); ++dst_it) {
+      auto& candidate = dst_it.value();
+      if (candidate.descriptor().dataset_id == src_storage.descriptor().dataset_id &&
+          candidate.descriptor().name == src_storage.descriptor().name) {
+        dst_storage = &candidate;
+        break;
+      }
+    }
+    if (dst_storage == nullptr) {
+      return PJ::unexpected(
+          "flushTo: destination has no topic '" + src_storage.descriptor().name + "' for dataset " +
+          std::to_string(src_storage.descriptor().dataset_id));
+    }
+    if (!dst_storage->empty() && src_storage.time_min() < dst_storage->time_max()) {
+      return PJ::unexpected("flushTo: monotonicity violation for topic '" + src_storage.descriptor().name + "'");
+    }
+    plan.push_back({&src_storage, dst_storage});
+  }
+
+  // Phase 2: execute. friend access lets us move sealed_chunks_ directly
+  // between TopicStorage instances of different engines — the deque move
+  // transfers chunk ownership without copying any column data or value
+  // buffers. Each chunk's TopicChunkStats (t_min/t_max/row_count) rides
+  // along inside the chunk by value, so dst's time_min/time_max queries
+  // reflect the new state immediately after the move.
+  for (auto& step : plan) {
+    auto drained = std::move(step.src->sealed_chunks_);
+    step.src->sealed_chunks_.clear();  // post-move state: deque is valid but empty.
+    for (auto& chunk : drained) {
+      step.dst->sealed_chunks_.push_back(std::move(chunk));
+    }
+  }
+
+  return {};
+}
+
 // ---------------------------------------------------------------------------
 // Listing helpers
 // ---------------------------------------------------------------------------