| 349 | } |
| 350 | |
| 351 | void RefillGuard::pruneVanishedTopics() { |
| 352 | if (session_ == nullptr) { |
| 353 | return; |
| 354 | } |
| 355 | // Scalar: a prior topic still empty after the refill vanished from the new file. |
| 356 | // Collect under one engine lock (getTopicStorage needs it held), then retire — |
| 357 | // retireTopic re-locks (recursive mutex) and clears its already-empty deque. |
| 358 | std::vector<TopicId> vanished_scalar; |
| 359 | { |
| 360 | auto lock = session_->dataEngine().lockEngine(); |
| 361 | for (const TopicId topic_id : scalar_snapshot_.prior_topic_ids) { |
| 362 | const TopicStorage* storage = session_->dataEngine().getTopicStorage(topic_id); |
| 363 | if (storage != nullptr && storage->empty()) { |
| 364 | vanished_scalar.push_back(topic_id); |
| 365 | } |
| 366 | } |
| 367 | } |
| 368 | for (const TopicId topic_id : vanished_scalar) { |
| 369 | session_->dataEngine().retireTopic(topic_id); |
| 370 | } |
| 371 | // Object: a prior object topic with no entries after the refill vanished too. |
| 372 | std::vector<ObjectTopicId> vanished_object; |
| 373 | for (const ObjectTopicId object_topic_id : prior_object_topic_ids_) { |
| 374 | if (session_->objectStore().entryCount(object_topic_id) == 0) { |
| 375 | vanished_object.push_back(object_topic_id); |
| 376 | } |
| 377 | } |
| 378 | if (!vanished_object.empty()) { |
| 379 | session_->evictObjectTopics(vanished_object); // drops the empty store series + its parser slot |
| 380 | } |
| 381 | } |
| 382 | |
| 383 | void RefillGuard::rollback() { |
| 384 | // (a) Adapters drop any partial-refill chunk pointers cached via progress notifies. |