| 458 | } |
| 459 | |
| 460 | Status DataEngine::flushTo(DataEngine& dst) { |
| 461 | if (&dst == this) { |
| 462 | return PJ::unexpected("flushTo: source and destination are the same engine"); |
| 463 | } |
| 464 | // Lock both engines for the whole operation; std::lock's deadlock-avoidance |
| 465 | // algorithm makes the acquisition order irrelevant (no ABBA). |
| 466 | std::unique_lock<std::recursive_mutex> lock_this(impl_->mutex_, std::defer_lock); |
| 467 | std::unique_lock<std::recursive_mutex> lock_dst(dst.impl_->mutex_, std::defer_lock); |
| 468 | std::lock(lock_this, lock_dst); |
| 469 | |
| 470 | // Phase 1: validate. Walk every src topic with sealed chunks and look up |
| 471 | // the matching dst topic by descriptor (dataset_id + name). Verify |
| 472 | // monotonicity against dst's current time_max. No mutation yet. |
| 473 | struct Step { |
| 474 | TopicStorage* src; |
| 475 | TopicStorage* dst; |
| 476 | }; |
| 477 | std::vector<Step> plan; |
| 478 | plan.reserve(impl_->topics.size()); |
| 479 | |
| 480 | for (auto it = impl_->topics.begin(); it != impl_->topics.end(); ++it) { |
| 481 | auto& src_storage = *it.value(); |
| 482 | if (src_storage.empty()) { |
| 483 | continue; |
| 484 | } |
| 485 | TopicStorage* dst_storage = nullptr; |
| 486 | for (auto dst_it = dst.impl_->topics.begin(); dst_it != dst.impl_->topics.end(); ++dst_it) { |
| 487 | auto& candidate = *dst_it.value(); |
| 488 | if (candidate.descriptor().dataset_id == src_storage.descriptor().dataset_id && |
| 489 | candidate.descriptor().name == src_storage.descriptor().name) { |
| 490 | dst_storage = &candidate; |
| 491 | break; |
| 492 | } |
| 493 | } |
| 494 | if (dst_storage == nullptr) { |
| 495 | return PJ::unexpected( |
| 496 | "flushTo: destination has no topic '" + src_storage.descriptor().name + "' for dataset " + |
| 497 | std::to_string(src_storage.descriptor().dataset_id)); |
| 498 | } |
| 499 | // The source is a staging engine that is never retention-evicted, so its |
| 500 | // timeMin() is the physical minimum (no retention-floor clamp in play). The |
| 501 | // flushTo contract assumes the source carries no floor; if it ever did, the |
| 502 | // clamped timeMin() could let physically-older source rows past this check. |
| 503 | if (!dst_storage->empty() && src_storage.timeMin() < dst_storage->timeMax()) { |
| 504 | return PJ::unexpected("flushTo: monotonicity violation for topic '" + src_storage.descriptor().name + "'"); |
| 505 | } |
| 506 | plan.push_back({&src_storage, dst_storage}); |
| 507 | } |
| 508 | |
| 509 | // Phase 2: execute. adoptChunksFrom moves sealed_chunks_ directly between |
| 510 | // TopicStorage instances (no column/value copy); the chunks' stats ride along |
| 511 | // by value, so dst's time_min/time_max reflect the new state immediately. The |
| 512 | // topic-id rewrite is a no-op here because flushTo's mirrored dst shares the |
| 513 | // source's TopicId. |
| 514 | for (auto& step : plan) { |
| 515 | adoptChunksFrom(*step.dst, *step.src); |
| 516 | } |
| 517 | |