[dual-timeline 1/N] Add secondary timeline metadata, root offset ring, and lifecycle management

Introduce the foundational infrastructure for dual-timeline support:

Types (category/mpt/detail/timeline.hpp):
- timeline_id enum (primary/secondary) and NUM_TIMELINES constant
- timeline_compaction_state struct with per-timeline compaction boundary

Metadata (category/mpt/detail/db_metadata.hpp):
- secondary_timeline_header_t carved from future_variables_unused:
  version_lower_bound_, next_version_, active_ flag
- Total db_metadata size unchanged (528512 bytes) for backward compat

UpdateAux (trie.hpp, trie.cpp, update_aux.cpp):
- Per-timeline compaction state array (timeline_[NUM_TIMELINES]) with
  tl(timeline_id) accessor, replacing bare member fields
- Secondary root offset ring mapped from cnv chunk 0's unused space
  (65536 entries, 512KB per copy)
- root_offsets_delegator parameterized on timeline_id
- timeline_active(), get_root_offset_at_version(v, tid),
  db_history_{min,max}_version(tid), version_is_valid_ondisk(v, tid)
- Lifecycle: activate_secondary_timeline (initializes compaction from
  primary), deactivate_secondary_timeline, promote_secondary_to_primary

Tests:
- db_metadata_test: layout, field offsets, zero-init semantics,
  read/write round-trip, memcpy survival, no overlap with consensus
- timeline_test: enum values, default construction, trivial copyability
- update_aux_test: secondary ring mapping, activate/deactivate lifecycle,
  ring push/read, promote, per-timeline version queries,
  version_is_valid_ondisk per-timeline

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: maxkozlovsky

category/mpt/db_metadata_context.cpp

@@ -23,6 +23,7 @@
 #include <category/mpt/config.hpp>
 #include <category/mpt/db_metadata_context.hpp>
 #include <category/mpt/detail/db_metadata.hpp>
+#include <category/mpt/detail/timeline.hpp>
 #include <category/mpt/trie.hpp>
 #include <category/mpt/util.hpp>
 
@@ -105,7 +106,50 @@ DbMetadataContext::DbMetadataContext(AsyncIO &io)
         }
     }
 
-    // Version mismatch detection
+    // Migration: MONAD007 -> MONAD008. Old code left future_variables_unused
+    // filled with 0xff; under the new layout those bytes overlap the newly
+    // introduced secondary_timeline header and would spuriously mark it
+    // active_=1. Zero the header and bump the magic. Each copy is migrated
+    // independently: if a crash strands one copy at MONAD007, on restart the
+    // magic-validation step above restores from the MONAD008 copy when that
+    // side is valid, otherwise the per-copy migration loop below picks up
+    // whichever copies still carry MONAD007. Either path converges to both
+    // copies at MONAD008.
+    auto const is_previous_magic = [](detail::db_metadata const *m) {
+        return 0 == memcmp(
+                        m->magic,
+                        detail::db_metadata::PREVIOUS_MAGIC,
+                        detail::db_metadata::MAGIC_STRING_LEN);
+    };
+    if (is_previous_magic(copies_[0].main) ||
+        is_previous_magic(copies_[1].main)) {
+        if (!can_write_to_map_) {
+            MONAD_ABORT_PRINTF(
+                "Detected pre-dual-timeline DB (magic=%s), which requires "
+                "writable mapping to migrate to %s. Open with write access.",
+                detail::db_metadata::PREVIOUS_MAGIC,
+                detail::db_metadata::MAGIC);
+        }
+        for (auto const &copy : copies_) {
+            auto *const m = copy.main;
+            if (!is_previous_magic(m)) {
+                continue;
+            }
+            auto const g = m->hold_dirty();
+            memset(&m->secondary_timeline, 0, sizeof(m->secondary_timeline));
+            std::atomic_signal_fence(std::memory_order_seq_cst);
+            memcpy(
+                m->magic,
+                detail::db_metadata::MAGIC,
+                detail::db_metadata::MAGIC_STRING_LEN);
+        }
+        LOG_INFO(
+            "Migrated DB metadata from {} to {}.",
+            detail::db_metadata::PREVIOUS_MAGIC,
+            detail::db_metadata::MAGIC);
+    }
+
+    // Version mismatch detection (for any other version that we don't migrate)
     constexpr unsigned magic_version_len = 3;
     constexpr unsigned magic_prefix_len =
         detail::db_metadata::MAGIC_STRING_LEN - magic_version_len;
@@ -235,6 +279,7 @@ DbMetadataContext::DbMetadataContext(AsyncIO &io)
     else {
         // Existing pool: map root offsets immediately
         map_root_offsets();
+        map_secondary_root_offsets();
     }
 }
 #if defined(__GNUC__) && !defined(__clang__)
@@ -306,6 +351,39 @@ void DbMetadataContext::map_root_offsets()
         copies_[0].root_offsets.size());
 }
 
+void DbMetadataContext::map_secondary_root_offsets()
+{
+    // Map the secondary timeline's root offset ring from cnv chunk 0's
+    // unused space, immediately after db_metadata (one copy per half-chunk).
+    // The assertion below verifies the chosen ring size fits.
+    static constexpr size_t SECONDARY_RING_CAPACITY = 65536; // 2^16
+    static constexpr size_t secondary_ring_bytes =
+        SECONDARY_RING_CAPACITY * sizeof(chunk_offset_t);
+
+    auto &cnv_chunk = io_->storage_pool().chunk(storage_pool::cnv, 0);
+    auto const fdr = cnv_chunk.read_fd();
+    auto const fdw = cnv_chunk.write_fd(0);
+    auto const secondary_offset = round_up_align<CPU_PAGE_BITS>(db_map_size_);
+    MONAD_ASSERT(
+        secondary_offset + secondary_ring_bytes <= cnv_chunk.capacity() / 2);
+
+    for (unsigned i = 0; i < 2; i++) {
+        auto const file_offset =
+            fdr.second + i * (cnv_chunk.capacity() / 2) + secondary_offset;
+        auto *ptr = ::mmap(
+            nullptr,
+            secondary_ring_bytes,
+            prot_,
+            mapflags_,
+            (can_write_to_map_ ? fdw : fdr).first,
+            off_t(file_offset));
+        MONAD_ASSERT(ptr != MAP_FAILED);
+        copies_[i].secondary_root_offsets = {
+            start_lifetime_as<chunk_offset_t>((chunk_offset_t *)ptr),
+            SECONDARY_RING_CAPACITY};
+    }
+}
+
 // Version metadata getters
 
 uint64_t DbMetadataContext::get_latest_finalized_version() const noexcept
@@ -571,15 +649,158 @@ chunk_offset_t DbMetadataContext::get_root_offset_at_version(
     return INVALID_OFFSET;
 }
 
+bool DbMetadataContext::timeline_active(timeline_id const tid) const noexcept
+{
+    if (tid == timeline_id::primary) {
+        return true;
+    }
+    return start_lifetime_as<std::atomic<uint8_t> const>(
+               &copies_[0].main->secondary_timeline.active_)
+               ->load(std::memory_order_acquire) != 0;
+}
+
+chunk_offset_t DbMetadataContext::get_root_offset_at_version(
+    uint64_t const version, timeline_id const tid) const noexcept
+{
+    if (!timeline_active(tid)) {
+        return INVALID_OFFSET;
+    }
+    if (tid == timeline_id::primary) {
+        return get_root_offset_at_version(version);
+    }
+    auto const ro = root_offsets(tid);
+    if (ro.empty()) {
+        return INVALID_OFFSET;
+    }
+    auto const max_version = ro.max_version();
+    if (max_version == INVALID_BLOCK_NUM || version > max_version) {
+        return INVALID_OFFSET;
+    }
+    // Secondary's valid range is bounded by ring capacity (wrap) and by the
+    // header's version_lower_bound_.
+    auto const capacity_min_version =
+        max_version >= ro.capacity() ? max_version - ro.capacity() + 1 : 0;
+    auto const header_lower_bound =
+        start_lifetime_as<std::atomic_uint64_t const>(
+            &copies_[0].main->secondary_timeline.version_lower_bound_)
+            ->load(std::memory_order_acquire);
+    if (version < std::max(capacity_min_version, header_lower_bound)) {
+        return INVALID_OFFSET;
+    }
+    return ro[version];
+}
+
+uint64_t
+DbMetadataContext::db_history_max_version(timeline_id const tid) const noexcept
+{
+    if (!timeline_active(tid)) {
+        return INVALID_BLOCK_NUM;
+    }
+    if (tid == timeline_id::primary) {
+        return db_history_max_version();
+    }
+    return root_offsets(tid).max_version();
+}
+
+uint64_t DbMetadataContext::db_history_min_valid_version(
+    timeline_id const tid) const noexcept
+{
+    if (!timeline_active(tid)) {
+        return INVALID_BLOCK_NUM;
+    }
+    if (tid == timeline_id::primary) {
+        return db_history_min_valid_version();
+    }
+    auto const ro = root_offsets(tid);
+    if (ro.empty() || ro.max_version() == INVALID_BLOCK_NUM) {
+        return INVALID_BLOCK_NUM;
+    }
+    return start_lifetime_as<std::atomic_uint64_t const>(
+               &copies_[0].main->secondary_timeline.version_lower_bound_)
+        ->load(std::memory_order_acquire);
+}
+
+void DbMetadataContext::activate_secondary_header(uint64_t const fork_version)
+{
+    MONAD_ASSERT(!timeline_active(timeline_id::secondary));
+    MONAD_ASSERT(!copies_[0].secondary_root_offsets.empty());
+
+    for (auto const &copy : copies_) {
+        std::fill(
+            copy.secondary_root_offsets.begin(),
+            copy.secondary_root_offsets.end(),
+            INVALID_OFFSET);
+    }
+    // Release-order stores with active_ last: a reader that observes
+    // active_=1 is guaranteed to see the populated version fields.
+    for (auto const &copy : copies_) {
+        auto *const m = copy.main;
+        auto const g = m->hold_dirty();
+        start_lifetime_as<std::atomic_uint64_t>(
+            &m->secondary_timeline.version_lower_bound_)
+            ->store(fork_version, std::memory_order_release);
+        start_lifetime_as<std::atomic_uint64_t>(
+            &m->secondary_timeline.next_version_)
+            ->store(fork_version, std::memory_order_release);
+        start_lifetime_as<std::atomic<uint8_t>>(&m->secondary_timeline.active_)
+            ->store(1, std::memory_order_release);
+    }
+    LOG_INFO("Activated secondary timeline at fork version {}", fork_version);
+}
+
+void DbMetadataContext::deactivate_secondary_header()
+{
+    MONAD_ASSERT(timeline_active(timeline_id::secondary));
+    // Clear active_ only. Version fields are defined only when active_!=0,
+    // so a reader that observes active_=0 ignores them — leaving them
+    // unchanged avoids the torn-read hazard where a racing reader sees
+    // stale active_=1 alongside newly-zeroed version fields.
+    for (auto const &copy : copies_) {
+        auto *const m = copy.main;
+        auto const g = m->hold_dirty();
+        start_lifetime_as<std::atomic<uint8_t>>(&m->secondary_timeline.active_)
+            ->store(0, std::memory_order_release);
+    }
+    LOG_INFO("Deactivated secondary timeline");
+}
+
+void DbMetadataContext::unsafe_swap_primary_secondary_headers() noexcept
+{
+    MONAD_ASSERT(timeline_active(timeline_id::secondary));
+    auto const atomic_swap = [](uint64_t &a, uint64_t &b) {
+        auto *const pa = start_lifetime_as<std::atomic_uint64_t>(&a);
+        auto *const pb = start_lifetime_as<std::atomic_uint64_t>(&b);
+        auto const va = pa->load(std::memory_order_acquire);
+        auto const vb = pb->load(std::memory_order_acquire);
+        pa->store(vb, std::memory_order_release);
+        pb->store(va, std::memory_order_release);
+    };
+    for (auto const &copy : copies_) {
+        auto *const m = copy.main;
+        auto const g = m->hold_dirty();
+        atomic_swap(
+            m->root_offsets.version_lower_bound_,
+            m->secondary_timeline.version_lower_bound_);
+        atomic_swap(
+            m->root_offsets.next_version_, m->secondary_timeline.next_version_);
+    }
+    swap_root_offsets_spans();
+}
+
 DbMetadataContext::~DbMetadataContext()
 {
-    // munmap root_offsets
     for (auto &copy : copies_) {
         if (copy.root_offsets.data() != nullptr) {
             (void)::munmap(
                 copy.root_offsets.data(), copy.root_offsets.size_bytes());
             copy.root_offsets = {};
         }
+        if (copy.secondary_root_offsets.data() != nullptr) {
+            (void)::munmap(
+                copy.secondary_root_offsets.data(),
+                copy.secondary_root_offsets.size_bytes());
+            copy.secondary_root_offsets = {};
+        }
     }
     // munmap db_metadata
     if (copies_[0].main != nullptr) {
@@ -711,6 +932,13 @@ void DbMetadataContext::init_new_pool(
         detail::db_metadata::MAGIC_STRING_LEN);
 
     map_root_offsets();
+    map_secondary_root_offsets();
+    for (auto const &copy : copies_) {
+        std::fill(
+            copy.secondary_root_offsets.begin(),
+            copy.secondary_root_offsets.end(),
+            INVALID_OFFSET);
+    }
     // Set history length, MUST be after root offsets are mapped
     if (history_len.has_value()) {
         update_history_length_metadata(*history_len);