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_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

@@ -49,11 +49,24 @@ 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.
   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.
   sdk::PayloadView payload;
 };
 
@@ -119,7 +132,9 @@ class ObjectStore {
 
   // 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.
+  // 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.
   Status pushLazy(ObjectTopicId id, Timestamp timestamp, LazyCallback fetch);
 
   // --- Read ---
@@ -147,6 +162,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 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
 // ---------------------------------------------------------------------------

pj_datastore/src/object_store.cpp

@@ -79,7 +79,8 @@ Status ObjectStore::pushOwned(ObjectTopicId id, Timestamp timestamp, std::vector
     return unexpected("timestamp not monotonically non-decreasing");
   }
 
-  size_t payload_size = payload.size();
+  const size_t payload_size = payload.size();
+
   auto shared_data = std::make_shared<const std::vector<uint8_t>>(std::move(payload));
 
   ObjectEntry entry;
@@ -264,6 +265,74 @@ void ObjectStore::evictAllBefore(Timestamp threshold) {
   }
 }
 
+// --- Cross-store flush ---
+
+Expected<void, std::string> ObjectStore::flushTo(ObjectStore& dst) {
+  if (&dst == this) {
+    return unexpected("flushTo: source and destination are the same store");
+  }
+
+  // Deterministic lock order by address to avoid deadlock with concurrent flushTo calls.
+  ObjectStore* first = this < &dst ? this : &dst;
+  ObjectStore* second = first == this ? &dst : this;
+  std::unique_lock first_lock(first->store_mutex_);
+  std::unique_lock second_lock(second->store_mutex_);
+
+  // Phase 1: validate every source series can be matched to a destination
+  // topic by descriptor and that the move respects monotonicity. No mutation.
+  struct Step {
+    ObjectSeries* src;
+    ObjectSeries* dst;
+  };
+  std::vector<Step> plan;
+  plan.reserve(topics_.size());
+
+  for (auto& [src_id, src_series] : topics_) {
+    if (src_series->entry_timestamps.empty()) {
+      continue;
+    }
+    ObjectSeries* dst_series = nullptr;
+    for (auto& [dst_id, dst_series_ptr] : dst.topics_) {
+      if (dst_series_ptr->descriptor.dataset_id == src_series->descriptor.dataset_id &&
+          dst_series_ptr->descriptor.topic_name == src_series->descriptor.topic_name) {
+        dst_series = dst_series_ptr.get();
+        break;
+      }
+    }
+    if (dst_series == nullptr) {
+      return unexpected(
+          "flushTo: destination has no topic '" + src_series->descriptor.topic_name + "' for dataset " +
+          std::to_string(src_series->descriptor.dataset_id));
+    }
+    if (!dst_series->entry_timestamps.empty() &&
+        src_series->entry_timestamps.front() < dst_series->entry_timestamps.back()) {
+      return unexpected("flushTo: monotonicity violation for topic '" + src_series->descriptor.topic_name + "'");
+    }
+    plan.push_back({src_series.get(), dst_series});
+  }
+
+  // Phase 2: execute the moves. Holding both store_mutex_ unique means no
+  // other reader or writer can observe an intermediate state; per-series
+  // mutexes are not needed because no concurrent access can occur.
+  for (auto& step : plan) {
+    for (auto& entry : step.src->entries) {
+      step.dst->entries.push_back(std::move(entry));
+    }
+    step.dst->entry_timestamps.insert(
+        step.dst->entry_timestamps.end(), step.src->entry_timestamps.begin(), step.src->entry_timestamps.end());
+    step.dst->memory_bytes += step.src->memory_bytes;
+
+    step.src->entries.clear();
+    step.src->entry_timestamps.clear();
+    step.src->memory_bytes = 0;
+
+    const Timestamp newest = step.dst->entry_timestamps.empty() ? 0 : step.dst->entry_timestamps.back();
+    applyRetention(*step.dst, newest);
+  }
+
+  return {};
+}
+
 // --- Lifecycle ---
 
 void ObjectStore::removeTopic(ObjectTopicId id) {
@@ -305,13 +374,22 @@ 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.
     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.
     resolved.payload = (*lazy)();
   }