Method rollback

pj_runtime/src/SessionManager.cpp:383–412

}

void RefillGuard::rollback() {
  // (a) Adapters drop any partial-refill chunk pointers cached via progress notifies.
  session_->notifyDatasetAboutToBeReplaced(dataset_id_);
  // (b) Scalar: drop partial refill chunks + retire topics it added, then move the
  //     prior chunks back into the stable ids.
  session_->dataEngine().reattachDatasetChunks(dataset_id_, std::move(scalar_snapshot_));
  // (c) Object: evict topics the refill ADDED (drops their store series + parser
  //     slots) BEFORE reattach, so reattach's own "current - prior" removal is a
  //     no-op. evictObjectTopics re-takes store_mutex_, so it must run OUTSIDE it —
  //     it does here (no ObjectStore lock held on this thread).
  std::unordered_set<uint32_t> prior;
  prior.reserve(prior_object_topic_ids_.size());
  for (const ObjectTopicId id : prior_object_topic_ids_) {
    prior.insert(id.id);
  }
  std::vector<ObjectTopicId> added;
  for (const ObjectTopicId id : session_->objectStore().listTopics(dataset_id_)) {
    if (prior.find(id.id) == prior.end()) {
      added.push_back(id);
    }
  }
  if (!added.empty()) {
    session_->evictObjectTopics(added);
  }
  // (d) Object: move the prior entries back into the stable ids.
  session_->objectStore().reattachDataset(dataset_id_, std::move(object_snapshot_));
  // (e) UI: reflect the restored topic set (non-live).
  const std::vector<TopicId> current = session_->dataEngine().listTopics(dataset_id_);
  session_->notifyIngest(QVector<TopicId>(current.begin(), current.end()), /*live=*/false);
}

std::size_t RefillGuard::snapshotBytes() const noexcept {
  std::size_t bytes = 0;