From 85ca1e5a377fec024b87fab0de2e1601bec998c2 Mon Sep 17 00:00:00 2001 From: Maxim Kozlovsky <10603239+maxkozlovsky@users.noreply.github.com> Date: Mon, 27 Jul 2026 17:30:22 -0400 Subject: [PATCH] [db] Frame snapshot storage streams into per-leaf groups A page-encoded target had to buffer a whole shard before writing any of it: the slots of one page arrive spread over an out-of-order stream, and upsert is set-not-merge, so a page written twice loses the slots of the earlier write. The dump already emitted each source leaf's slots contiguously, but nothing in the artifact said so, and the loader could not rely on it. Frame the storage stream as groups, one per source leaf, and spend the stream header's reserved byte on the source's page-key shift. When a group covers a whole target page, the loader assembles the page as the group ends and may flush at any group boundary, so peak memory follows the flush threshold rather than the shard size. That threshold moves from 10 GiB to 1 GiB and gains a setter: with mid-shard flushing gated away it had become unreachable within a shard. Because a storage record grows a length, the format version goes to 2. Both older layouts still load: a version 1 stream, dumped once headers existed but before grouping did, and a stream with no header at all each hold one slot per record, which the loader reads as groups of a single slot and takes the whole-shard path for. Co-Authored-By: Claude Opus 5 (1M context) --- .../execution/ethereum/db/db_snapshot.cpp | 375 ++++++++--- category/execution/ethereum/db/db_snapshot.h | 66 +- .../ethereum/db/db_snapshot_filesystem.cpp | 3 - .../ethereum/test/test_db_snapshot.cpp | 600 +++++++++++++++--- 4 files changed, 825 insertions(+), 219 deletions(-) diff --git a/category/execution/ethereum/db/db_snapshot.cpp b/category/execution/ethereum/db/db_snapshot.cpp index 47b4c00ea1..b6eae99b41 100644 --- a/category/execution/ethereum/db/db_snapshot.cpp +++ b/category/execution/ethereum/db/db_snapshot.cpp @@ -49,8 +49,10 @@ struct monad_db_snapshot_loader ankerl::unordered_dense::segmented_map, MONAD_SNAPSHOT_SHARDS> account_offset_to_update; - // Per-shard page accumulator used only when page_encoded. Maps - // account_offset -> (page_key -> assembled storage_page_t). + // Per-shard page accumulator, used only for a page-encoded target whose + // input stream groups slots more finely than a target page, so that pages + // cannot be closed as they are read. Maps account_offset -> (page_key -> + // assembled storage_page_t). std::array< ankerl::unordered_dense::segmented_map< uint64_t, ankerl::unordered_dense::map< @@ -60,6 +62,7 @@ struct monad_db_snapshot_loader monad::mpt::UpdateList state_updates; monad::mpt::UpdateList code_updates; uint64_t bytes_read; + uint64_t flush_bytes; monad_db_snapshot_loader( uint64_t const block, char const *const *const dbname_paths, @@ -69,6 +72,7 @@ struct monad_db_snapshot_loader , db{open_target_db( dbname_paths, len, sq_thread_cpu, load_to_secondary)} , bytes_read{0} + , flush_bytes{MONAD_SNAPSHOT_DEFAULT_FLUSH_BYTES} { } @@ -119,8 +123,9 @@ uint64_t get_shard(monad::mpt::NibblesView const path) return ret; } -// When the target is page-encoded, drain the accumulator into per-account -// `next` lists. Each page becomes one Update keyed by keccak256(page_key) +// Drain whatever the accumulator holds into per-account `next` lists; it is +// populated only when pages cannot be closed as the stream is read, and empty +// otherwise. Each page becomes one Update keyed by keccak256(page_key) // with value encode_storage_page_db(page_key, page) (or std::nullopt if the // page is empty so the entry is a deletion). The encoded byte_strings are // kept alive in loader->bytes_alloc until the upsert completes; the Update @@ -243,19 +248,30 @@ uint64_t monad_db_snapshot_loader_read_account( return bytes_consumed; } +struct StreamFormat +{ + // Whether a storage stream's records carry a group prefix. False for a + // version 1 stream and for one written before headers existed, both of + // which hold one slot entry per record. + bool grouped; + // Zero unless the stream groups storage slots more coarsely than one slot + // at a time. + uint8_t group_key_shift; +}; + // Consume the stream header if there is one, leaving `stream` at its first // record. -void read_stream_header( +StreamFormat read_stream_header( monad::byte_string_view &stream, monad_snapshot_type const kind) { using namespace monad; if (stream.size() < sizeof(monad_snapshot_stream_header)) { - return; + return {.grouped = false, .group_key_shift = 0}; } auto const header = unaligned_load(stream.data()); if (header.magic != MONAD_SNAPSHOT_STREAM_MAGIC) { - return; + return {.grouped = false, .group_key_shift = 0}; } // A legacy stream can hold the guard byte by chance, but not the magic, so // past the magic a bad guard is corruption rather than an older layout and @@ -267,9 +283,11 @@ void read_stream_header( header.guard, MONAD_SNAPSHOT_STREAM_GUARD); MONAD_ASSERT_PRINTF( - header.version == MONAD_SNAPSHOT_STREAM_VERSION, - "snapshot stream version %u is not supported (expected %u)", + header.version == MONAD_SNAPSHOT_STREAM_VERSION || + header.version == MONAD_SNAPSHOT_STREAM_VERSION_UNGROUPED, + "snapshot stream version %u is not supported (expected %u or %u)", header.version, + MONAD_SNAPSHOT_STREAM_VERSION_UNGROUPED, MONAD_SNAPSHOT_STREAM_VERSION); MONAD_ASSERT_PRINTF( header.kind == kind, @@ -277,6 +295,115 @@ void read_stream_header( header.kind, static_cast(kind)); stream.remove_prefix(sizeof(header)); + if (header.version == MONAD_SNAPSHOT_STREAM_VERSION_UNGROUPED) { + return {.grouped = false, .group_key_shift = 0}; + } + return {.grouped = true, .group_key_shift = header.group_key_shift}; +} + +// Byte length of the encode_storage_db entry at the head of `stream`, which is +// the whole group of an ungrouped stream. +size_t storage_entry_length(monad::byte_string_view const stream) +{ + monad::byte_string_view rest{stream}; + auto const res = monad::decode_storage_db_raw(rest); + MONAD_ASSERT(res.has_value()); + return stream.size() - rest.size(); +} + +void emit_slot_updates( + monad_db_snapshot_loader *const loader, uint64_t const shard, + uint64_t const account_offset, monad::byte_string_view payload) +{ + using namespace monad; + using namespace monad::mpt; + auto &account_update = + loader->account_offset_to_update.at(shard).at(account_offset); + while (!payload.empty()) { + byte_string_view const before{payload}; + auto const res = decode_storage_db_raw(payload); + MONAD_ASSERT(res.has_value()); + account_update.next.push_front(loader->update_alloc.emplace_back(Update{ + .key = loader->hash_alloc.emplace_back( + keccak256(to_bytes(res.value().first))), + .value = before.substr(0, before.size() - payload.size()), + .incarnation = false, + .next = UpdateList{}, + .version = static_cast(loader->block)})); + } +} + +// Assemble the target pages held by one closed group and push them onto the +// owning account. A leaf's slots arrive in ascending key order, so a page is +// complete as soon as the page key changes. Plural because a group_key_shift +// coarser than the target's page key is legal and would put several pages in +// one group; no dumper emits one, as source and target shift are the same +// constant. +void emit_page_updates( + monad_db_snapshot_loader *const loader, uint64_t const shard, + uint64_t const account_offset, monad::byte_string_view payload) +{ + using namespace monad; + using namespace monad::mpt; + auto &account_update = + loader->account_offset_to_update.at(shard).at(account_offset); + storage_page_t page; + bytes32_t page_key{}; + bool have_page = false; + auto const emit = [&] { + // Dropped rather than written as a deletion the way the accumulator + // path does, which is equivalent only because the target starts empty. + if (page.is_empty()) { + return; + } + account_update.next.push_front(loader->update_alloc.emplace_back(Update{ + .key = loader->hash_alloc.emplace_back( + keccak256({page_key.bytes, sizeof(page_key.bytes)})), + .value = byte_string_view{loader->bytes_alloc.emplace_back( + encode_storage_page_db(page_key, page))}, + .incarnation = false, + .next = UpdateList{}, + .version = static_cast(loader->block)})); + }; + while (!payload.empty()) { + auto const res = decode_storage_db_raw(payload); + MONAD_ASSERT(res.has_value()); + bytes32_t const slot_key = to_bytes(res.value().first); + bytes32_t const key = compute_page_key(slot_key); + if (!have_page) { + page_key = key; + have_page = true; + } + else if (key != page_key) { + // Ascending order is what makes a page complete here. Were the + // group unordered, one page key could recur after being emitted and + // the second Update would overwrite the first (upsert is + // set-not-merge), silently dropping slots. + MONAD_ASSERT(key > page_key); + emit(); + page = storage_page_t{}; + page_key = key; + } + page.set(compute_slot_offset(slot_key), to_bytes(res.value().second)); + } + emit(); +} + +// Merge the group's slots into the whole-shard accumulator, for a stream whose +// groups do not cover a target page. +void accumulate_page_slots( + monad_db_snapshot_loader *const loader, uint64_t const shard, + uint64_t const account_offset, monad::byte_string_view payload) +{ + using namespace monad; + auto &account_pages = loader->page_accumulator.at(shard)[account_offset]; + while (!payload.empty()) { + auto const res = decode_storage_db_raw(payload); + MONAD_ASSERT(res.has_value()); + bytes32_t const slot_key = to_bytes(res.value().first); + account_pages[compute_page_key(slot_key)].set( + compute_slot_offset(slot_key), to_bytes(res.value().second)); + } } class NibblePath @@ -332,7 +459,10 @@ using SnapshotWriteFn = uint64_t (*)( void *user); // Writes the records of every stream of one dump, and is shared by every clone -// of the traverse machine as well as by the eth-header writes outside it. +// of the traverse machine as well as by the eth-header writes outside it. The +// parallel traverse is single threaded (see Db::traverse), so one scratch +// buffer serves all clones: a storage group is built and written within a +// single down() call. // // Every record goes through here so that no stream can be opened without its // header: a stream missing one is indistinguishable from a stream written @@ -341,12 +471,16 @@ class SnapshotStreamWriter { SnapshotWriteFn const write_; void *const user_; + // Group key shift stamped on a storage stream's header; zero for a + // slot-encoded source, whose groups hold one slot each. + uint8_t const storage_group_key_shift_; std::array< std::array, MONAD_SNAPSHOT_SHARDS> header_written_{}; // Length of each shard's account stream counted from its first record, so // that the offsets it hands out do not shift when a header is present. std::array account_bytes_written_{}; + monad::byte_string group_buffer_; // Written lazily so that a kind a shard has no records for leaves a // zero-length stream rather than a header-only one. @@ -361,7 +495,9 @@ class SnapshotStreamWriter .magic = MONAD_SNAPSHOT_STREAM_MAGIC, .version = MONAD_SNAPSHOT_STREAM_VERSION, .kind = static_cast(kind), - .reserved = 0, + .group_key_shift = kind == MONAD_SNAPSHOT_STORAGE + ? storage_group_key_shift_ + : uint8_t{0}, .guard = MONAD_SNAPSHOT_STREAM_GUARD}; std::array bytes; monad::unaligned_store(bytes.data(), header); @@ -372,9 +508,12 @@ class SnapshotStreamWriter } public: - SnapshotStreamWriter(SnapshotWriteFn const write, void *const user) + SnapshotStreamWriter( + SnapshotWriteFn const write, void *const user, + uint8_t const storage_group_key_shift) : write_{write} , user_{user} + , storage_group_key_shift_{storage_group_key_shift} { } @@ -389,7 +528,7 @@ class SnapshotStreamWriter } // Appends one account record, returning the offset it occupies in the - // shard's account stream, which is how a storage record names its account. + // shard's account stream, which is how a storage group names its account. uint64_t write_account_record( uint64_t const shard, unsigned char const *const bytes, size_t const len) @@ -399,6 +538,37 @@ class SnapshotStreamWriter write_record(shard, MONAD_SNAPSHOT_ACCOUNT, bytes, len); return offset; } + + // Appends one storage group, prefixed with `account_offset` and its payload + // length. `append_payload` appends the group's slot-encoded entries; a + // group whose payload stays empty writes nothing. + template + void write_storage_group( + uint64_t const shard, uint64_t const account_offset, + AppendPayload const &append_payload) + { + constexpr size_t prefix = MONAD_SNAPSHOT_STORAGE_GROUP_HEADER_SIZE; + group_buffer_.clear(); + group_buffer_.resize(prefix); + append_payload(group_buffer_); + + size_t const payload_len = group_buffer_.size() - prefix; + if (payload_len == 0) { + return; + } + MONAD_ASSERT(payload_len <= std::numeric_limits::max()); + + unsigned char *const cursor = group_buffer_.data(); + monad::unaligned_store(cursor, account_offset); + monad::unaligned_store( + cursor + sizeof(account_offset), + static_cast(payload_len)); + write_record( + shard, + MONAD_SNAPSHOT_STORAGE, + group_buffer_.data(), + group_buffer_.size()); + } }; struct MonadSnapshotTraverseMachine : public monad::mpt::TraverseMachine @@ -483,42 +653,39 @@ struct MonadSnapshotTraverseMachine : public monad::mpt::TraverseMachine } else { MONAD_ASSERT(path.length() == (HASH_SIZE * 2)); - // Emit one slot-format storage entry, prefixed with the owning - // account's offset so the loader can re-link it. - auto const emit_slot = [&](byte_string_view const entry) { - writer.write_record( - shard, - MONAD_SNAPSHOT_STORAGE, - reinterpret_cast( - &account_offset), - sizeof(account_offset)); - writer.write_record( - shard, - MONAD_SNAPSHOT_STORAGE, - entry.data(), - entry.size()); - }; - if (page_encoded) { - // Source db is page-encoded: expand the storage leaf into - // one slot-encoded entry per non-zero slot so the dumped - // snapshot stays slot-granular and loads unchanged. - auto const decoded = - decode_storage_page_leaf(byte_string_view{val}); - MONAD_ASSERT(decoded.has_value()); - for (auto const [slot_key, slot_val] : - decoded.value().slots()) { - emit_slot(encode_storage_db(slot_key, slot_val)); - } - } - else { - emit_slot(val); - } + write_storage_group(shard, val); } } return true; } + // Turn one storage leaf into one group of the shard's storage stream. The + // slots of a page-encoded leaf are expanded to slot-encoded entries so the + // format stays slot-granular, but they stay together in one group, which is + // what lets the loader rebuild a page without holding the whole shard. + void + write_storage_group(uint64_t const shard, monad::byte_string_view const val) + { + using namespace monad; + + MONAD_ASSERT(account_offset != std::numeric_limits::max()); + + writer.write_storage_group( + shard, account_offset, [&](byte_string &payload) { + if (!page_encoded) { + payload += val; + return; + } + auto const decoded = decode_storage_page_leaf(val); + MONAD_ASSERT(decoded.has_value()); + for (auto const [slot_key, slot_val] : + decoded.value().slots()) { + payload += encode_storage_db(slot_key, slot_val); + } + }); + } + virtual void up(unsigned char const, monad::mpt::Node const &node) override { if (path.length() == 0) { @@ -592,7 +759,13 @@ bool monad_db_dump_snapshot( io_context, dump_from_secondary ? timeline_id::secondary : timeline_id::primary}; - SnapshotStreamWriter writer{write, user}; + bool const page_encoded = + db.state_machine_type() == state_machine_kind::monad; + SnapshotStreamWriter writer{ + write, + user, + page_encoded ? static_cast(storage_page_t::PAGE_KEY_SHIFT) + : uint8_t{0}}; for (uint64_t b = block < 256 ? 0 : block - 255; b <= block; ++b) { uint64_t const header_shard = block - b; if (header_shard % total_shards != shard_number) { @@ -635,10 +808,7 @@ bool monad_db_dump_snapshot( } MonadSnapshotTraverseMachine machine{ - writer, - total_shards, - shard_number, - db.state_machine_type() == state_machine_kind::monad}; + writer, total_shards, shard_number, page_encoded}; bool const success = db.traverse(finalized_root, machine, block, dump_concurrency_limit); if (!success) { @@ -647,13 +817,13 @@ bool monad_db_dump_snapshot( return success; } -// Loads the standard slot-encoded snapshot (the format produced by -// monad_db_dump_snapshot against a slot db) into one timeline: +// Loads a snapshot (the slot-granular format produced by +// monad_db_dump_snapshot) into one timeline: // * load_to_secondary == false: the primary timeline. // * load_to_secondary == true: an already-activated secondary timeline. // The target's storage encoding is derived from its persisted -// state_machine_kind; a page-encoded target converts slot leaves to page -// leaves on the fly. The target's kind must already be stamped on disk. +// state_machine_kind; a page-encoded target assembles page leaves from the slot +// entries on the fly. The target's kind must already be stamped on disk. monad_db_snapshot_loader *monad_db_snapshot_loader_create( uint64_t const block, char const *const *const dbname_paths, size_t const len, unsigned const sq_thread_cpu, @@ -671,6 +841,13 @@ monad_db_snapshot_loader *monad_db_snapshot_loader_create( return loader; } +void monad_db_snapshot_loader_set_flush_bytes( + monad_db_snapshot_loader *const loader, uint64_t const bytes) +{ + MONAD_ASSERT(loader); + loader->flush_bytes = bytes; +} + void monad_db_snapshot_loader_load( monad_db_snapshot_loader *const loader, uint64_t const shard, unsigned char const *const eth_header, size_t const eth_header_len, @@ -680,7 +857,6 @@ void monad_db_snapshot_loader_load( { using namespace monad; using namespace monad::mpt; - constexpr size_t BYTES_READ_BEFORE_FLUSH = 10ull * 1024 * 1024 * 1024; MONAD_ASSERT(loader); // Account offsets index from the first account record, so the storage loop // below must resolve them against this header-stripped view rather than the @@ -692,7 +868,7 @@ void monad_db_snapshot_loader_load( for (uint64_t account_offset = 0; account_offset != accounts.size();) { account_offset += monad_db_snapshot_loader_read_account( loader, shard, account_offset, accounts); - if (loader->bytes_read >= BYTES_READ_BEFORE_FLUSH) { + if (loader->bytes_read >= loader->flush_bytes) { monad_db_snapshot_loader_flush(loader); } MONAD_ASSERT(account_offset <= accounts.size()); @@ -702,63 +878,60 @@ void monad_db_snapshot_loader_load( if (storage) { MONAD_ASSERT(account); byte_string_view storage_view{storage, storage_len}; - read_stream_header(storage_view, MONAD_SNAPSHOT_STORAGE); - auto &account_offset_to_update = + auto const [grouped, group_key_shift] = + read_stream_header(storage_view, MONAD_SNAPSHOT_STORAGE); + // A group closes a target page only if its key is at least as coarse; + // otherwise a page's slots span groups that arrive in any order, so + // nothing can be written until the whole shard has been read. + bool const close_pages_per_group = + group_key_shift >= storage_page_t::PAGE_KEY_SHIFT; + bool const can_flush = !loader->page_encoded() || close_pages_per_group; + auto const &account_offset_to_update = loader->account_offset_to_update.at(shard); while (!storage_view.empty()) { - MONAD_ASSERT(storage_view.size() >= sizeof(uint64_t)); + MONAD_ASSERT( + storage_view.size() > + (grouped ? MONAD_SNAPSHOT_STORAGE_GROUP_HEADER_SIZE + : sizeof(uint64_t))); uint64_t const account_offset = unaligned_load(storage_view.data()); + size_t payload_len; + if (grouped) { + payload_len = unaligned_load( + storage_view.data() + sizeof(account_offset)); + storage_view.remove_prefix( + MONAD_SNAPSHOT_STORAGE_GROUP_HEADER_SIZE); + } + else { + storage_view.remove_prefix(sizeof(account_offset)); + payload_len = storage_entry_length(storage_view); + } + // A dump never writes an empty group, so one here would silently + // drop the slots the reader expected to find in it. + MONAD_ASSERT(payload_len != 0); + MONAD_ASSERT(payload_len <= storage_view.size()); + byte_string_view const payload{storage_view.substr(0, payload_len)}; + storage_view.remove_prefix(payload_len); + if (!account_offset_to_update.contains(account_offset)) { monad_db_snapshot_loader_read_account( loader, shard, account_offset, accounts); } - storage_view.remove_prefix(sizeof(account_offset)); - byte_string_view const before{storage_view}; - uint64_t consumed; - if (loader->page_encoded()) { - // The storage byte stream concatenates multiple - // [account_offset, leaf.value()] entries, so we use - // decode_storage_db_raw which advances the view in place - // and tolerates trailing bytes. Convert the raw views to - // bytes32_t (right-aligned) for the page accumulator. - auto const res = decode_storage_db_raw(storage_view); - MONAD_ASSERT(res.has_value()); - bytes32_t const slot_key = to_bytes(res.value().first); - bytes32_t const slot_val = to_bytes(res.value().second); - consumed = before.size() - storage_view.size(); - bytes32_t const pg_key = compute_page_key(slot_key); - uint8_t const slot_off = compute_slot_offset(slot_key); - auto &shard_pages = loader->page_accumulator.at(shard); - shard_pages[account_offset][pg_key].set(slot_off, slot_val); + if (!loader->page_encoded()) { + emit_slot_updates(loader, shard, account_offset, payload); + } + else if (close_pages_per_group) { + emit_page_updates(loader, shard, account_offset, payload); } else { - auto const res = decode_storage_db_raw(storage_view); - MONAD_ASSERT(res.has_value()); - auto &update = account_offset_to_update.at(account_offset); - consumed = before.size() - storage_view.size(); - update.next.push_front(loader->update_alloc.emplace_back(Update{ - .key = loader->hash_alloc.emplace_back( - keccak256(to_bytes(res.value().first))), - .value = before.substr(0, consumed), - .next = UpdateList{}, - .version = static_cast(loader->block)})); + accumulate_page_slots(loader, shard, account_offset, payload); } - loader->bytes_read += consumed; - // When page-encoded, all slots that share a page_key must be in - // the same flush. A mid-loop flush would emit a page Update for - // the slots seen so far; later slots in the same page would start - // a fresh accumulator entry and the next flush would emit another - // Update for the same keccak256(page_key), causing the mpt - // upsert to overwrite the earlier page (set-not-merge). Defer - // flushing until the unconditional final flush at end of load(). - // - // Consequence: the page accumulator holds a whole shard's storage - // in RAM before that final flush. With the current state size this - // is not a problem. There will be a follow up to bound the memory - // usage. - if (!loader->page_encoded() && - loader->bytes_read >= BYTES_READ_BEFORE_FLUSH) { + + loader->bytes_read += + (grouped ? MONAD_SNAPSHOT_STORAGE_GROUP_HEADER_SIZE + : sizeof(account_offset)) + + payload_len; + if (can_flush && loader->bytes_read >= loader->flush_bytes) { monad_db_snapshot_loader_flush(loader); } } @@ -782,7 +955,7 @@ void monad_db_snapshot_loader_load( .version = static_cast(loader->block)})); code_view.remove_prefix(size); loader->bytes_read += sizeof(uint64_t) + size; - if (loader->bytes_read >= BYTES_READ_BEFORE_FLUSH) { + if (loader->bytes_read >= loader->flush_bytes) { monad_db_snapshot_loader_flush(loader); } } diff --git a/category/execution/ethereum/db/db_snapshot.h b/category/execution/ethereum/db/db_snapshot.h index 9472b0337d..ded0fc3cc5 100644 --- a/category/execution/ethereum/db/db_snapshot.h +++ b/category/execution/ethereum/db/db_snapshot.h @@ -39,23 +39,41 @@ inline constexpr unsigned MONAD_SNAPSHOT_FILES_PER_SHARD = 4; // // eth_header := rlp(header) // account := encode_account_db(address, account) ... -// storage := [account_offset: uint64][encode_storage_db(key, value)] ... +// storage := group* // code := [size: uint64][code] ... // -// A storage record is prefixed by the offset of the owning account within the -// shard's account stream, which is how the loader relinks it whatever order the -// records arrive in. Those offsets count from the first account record rather -// than from the stream header, so they do not depend on whether the header is -// present. +// group := [account_offset: uint64][payload_len: uint32][payload] // -// A dump always writes the header, but a reader must also accept a stream that -// lacks one and parse its records from byte 0: that is how a snapshot written -// before the header existed is recognised. +// Each group is prefixed by the offset of the owning account within the shard's +// account stream, and holds the encode_storage_db(slot_key, slot_value) entries +// of one storage leaf of the source db, in ascending slot-key order. +// payload_len is never zero. Account offsets count from the first account +// record, not from the stream header, so they do not depend on whether the +// header is present. +// +// A group is closed: it holds every non-zero slot of its source leaf, and no +// later group in the stream repeats any of them. When group_key_shift is also +// at least as coarse as the target's page key, that is what lets the loader +// emit a finished page as its group ends instead of buffering a whole shard. +// +// Two older storage layouts must also be read. A version 1 header, and no +// header at all, both denote ungrouped storage records: +// +// storage := [account_offset: uint64][one encode_storage_db entry] ... +// +// which the loader reads as groups of a single slot. No header at all is how a +// snapshot written before headers existed is recognised; nothing outside the +// storage stream differs between version 1 and version 2. // // Scalars are native-endian, which the format takes to be little-endian. inline constexpr uint32_t MONAD_SNAPSHOT_STREAM_MAGIC = 0x5347534d; // "MSGS" -inline constexpr uint8_t MONAD_SNAPSHOT_STREAM_VERSION = 1; +inline constexpr uint8_t MONAD_SNAPSHOT_STREAM_VERSION = 2; +// Version 1 framed every stream the same way but held one slot entry per +// storage record, and left the group_key_shift byte reserved. +inline constexpr uint8_t MONAD_SNAPSHOT_STREAM_VERSION_UNGROUPED = 1; inline constexpr uint8_t MONAD_SNAPSHOT_STREAM_GUARD = 0xff; +inline constexpr size_t MONAD_SNAPSHOT_STORAGE_GROUP_HEADER_SIZE = + sizeof(uint64_t) + sizeof(uint32_t); struct monad_snapshot_stream_header { @@ -65,9 +83,13 @@ struct monad_snapshot_stream_header // kind is rejected rather than misparsed. Nothing here identifies the // shard, so files swapped between shards still load. uint8_t kind; - // Zero. Readers ignore it, so a later revision may give it a meaning - // without a version bump only if an unaware reader can correctly skip it. - uint8_t reserved; + // Meaning depends on kind; zero for every kind but MONAD_SNAPSHOT_STORAGE, + // where it is the number of low slot-key bits that do not participate in + // grouping: a group holds every non-zero slot sharing the remaining high + // bits. Zero there too means one slot per group, which groups nothing. + // Reserved in version 1, whose readers were told to accept any value, so it + // carries no meaning for a version 1 stream. + uint8_t group_key_shift; // MONAD_SNAPSHOT_STREAM_GUARD. The magic's first byte on disk is below // 0xc0, so a header can never be mistaken for the RLP list that opens an // eth_header or account stream; the guard is the most significant byte when @@ -85,6 +107,14 @@ static_assert(sizeof(struct monad_snapshot_stream_header) == 8); static_assert(std::endian::native == std::endian::little); static_assert((MONAD_SNAPSHOT_STREAM_MAGIC & 0xff) < 0xc0); +// Snapshot bytes the loader buffers before an intermediate upsert. This bounds +// its peak memory, except for storage the loader cannot close page by page (see +// monad_db_snapshot_loader_set_flush_bytes). Each flush costs one extra +// incremental merklizing upsert, so it is set well above a typical shard and +// only bites on outsized ones. A shard is flushed when its load ends whatever +// the threshold, which is what keeps its buffered storage from outliving it. +inline constexpr uint64_t MONAD_SNAPSHOT_DEFAULT_FLUSH_BYTES = 1ull << 30; + extern "C" { #endif @@ -112,6 +142,16 @@ struct monad_db_snapshot_loader *monad_db_snapshot_loader_create( uint64_t block, char const *const *dbname_paths, size_t len, unsigned sq_thread_cpu, bool load_to_secondary); +// Override MONAD_SNAPSHOT_DEFAULT_FLUSH_BYTES. Storage honours it only when the +// loader can close pages as it reads, that is when the target is slot-encoded +// or the stream's group_key_shift covers a whole target page. Otherwise — a +// page-encoded target reading a stream whose groups hold one slot each, which +// is every stream dumped from a slot-encoded db and every stream with no header +// at all — a shard's storage has to be assembled in full before any of it can +// be written. +void monad_db_snapshot_loader_set_flush_bytes( + struct monad_db_snapshot_loader *loader, uint64_t bytes); + void monad_db_snapshot_loader_load( struct monad_db_snapshot_loader *loader, uint64_t shard, unsigned char const *eth_header, size_t, unsigned char const *account, diff --git a/category/execution/ethereum/db/db_snapshot_filesystem.cpp b/category/execution/ethereum/db/db_snapshot_filesystem.cpp index 37770d898b..50f1103e4b 100644 --- a/category/execution/ethereum/db/db_snapshot_filesystem.cpp +++ b/category/execution/ethereum/db/db_snapshot_filesystem.cpp @@ -170,9 +170,6 @@ void monad_db_snapshot_load_filesystem( { std::filesystem::path const root{std::format("{}/{}", snapshot_dir, block)}; MONAD_ASSERT(std::filesystem::is_directory(root)); - // The input snapshot is always slot-encoded (the standard format produced - // by monad_db_dump_snapshot from a slot db). If the target timeline is - // page-encoded, the loader converts slot leaves to page leaves on the fly. monad_db_snapshot_loader *const loader = monad_db_snapshot_loader_create( block, dbname_paths, len, sq_thread_cpu, load_to_secondary); diff --git a/category/execution/ethereum/test/test_db_snapshot.cpp b/category/execution/ethereum/test/test_db_snapshot.cpp index 1a1ae5a533..b4994de73a 100644 --- a/category/execution/ethereum/test/test_db_snapshot.cpp +++ b/category/execution/ethereum/test/test_db_snapshot.cpp @@ -46,9 +46,11 @@ #include #include #include +#include #include #include #include +#include namespace monad::mpt::test { @@ -133,6 +135,50 @@ namespace bytes.size()}; } + // Load a snapshot directory the way monad_db_snapshot_load_filesystem does, + // but with a caller-chosen flush threshold: the smaller it is, the more + // often the loader writes what it holds mid-shard. A threshold this low is + // safe to ask for whatever the stream, because the loader ignores it for + // storage it cannot close page by page. Skips the blake3 verification the + // filesystem loader does. + void load_snapshot( + std::string const &dbname, std::filesystem::path const &root, + uint64_t const block, uint64_t const flush_bytes, + bool const load_to_secondary) + { + char const *dbname_paths[] = {dbname.c_str()}; + auto *const loader = monad_db_snapshot_loader_create( + block, + dbname_paths, + 1, + static_cast(-1), + load_to_secondary); + monad_db_snapshot_loader_set_flush_bytes(loader, flush_bytes); + for (auto const &dir : std::filesystem::directory_iterator{ + root / std::to_string(block)}) { + uint64_t const shard = std::stoull(dir.path().stem()); + auto const eth_header = read_file(dir.path() / "eth_header"); + auto const account = read_file(dir.path() / "account"); + auto const storage = read_file(dir.path() / "storage"); + auto const code = read_file(dir.path() / "code"); + auto const ptr = [](monad::byte_string const &b) { + return b.empty() ? nullptr : b.data(); + }; + monad_db_snapshot_loader_load( + loader, + shard, + ptr(eth_header), + eth_header.size(), + ptr(account), + account.size(), + ptr(storage), + storage.size(), + ptr(code), + code.size()); + } + monad_db_snapshot_loader_destroy(loader); + } + void write_file( std::filesystem::path const &path, monad::byte_string_view const bytes) { @@ -158,64 +204,158 @@ namespace MONAD_ASSERT(header.magic == MONAD_SNAPSHOT_STREAM_MAGIC); MONAD_ASSERT(header.version == MONAD_SNAPSHOT_STREAM_VERSION); MONAD_ASSERT(header.kind == kind); - MONAD_ASSERT(header.reserved == 0); MONAD_ASSERT(header.guard == MONAD_SNAPSHOT_STREAM_GUARD); view.remove_prefix(sizeof(header)); return view; } - // Rewrite a snapshot in the layout that predates stream headers. + // Rewrite a grouped storage stream, header already removed, as the + // ungrouped records of version 1: one account offset per slot entry. + monad::byte_string ungroup_storage_stream(monad::byte_string_view view) + { + using namespace monad; + byte_string ungrouped; + while (!view.empty()) { + MONAD_ASSERT( + view.size() >= MONAD_SNAPSHOT_STORAGE_GROUP_HEADER_SIZE); + uint64_t const account_offset = + unaligned_load(view.data()); + uint32_t const payload_len = + unaligned_load(view.data() + sizeof(account_offset)); + view.remove_prefix(MONAD_SNAPSHOT_STORAGE_GROUP_HEADER_SIZE); + byte_string_view payload{view.substr(0, payload_len)}; + view.remove_prefix(payload_len); + std::array offset_bytes; + unaligned_store(offset_bytes.data(), account_offset); + while (!payload.empty()) { + byte_string_view const before{payload}; + auto const entry = decode_storage_db_raw(payload); + MONAD_ASSERT(entry.has_value()); + ungrouped.append(offset_bytes.data(), offset_bytes.size()); + ungrouped += before.substr(0, before.size() - payload.size()); + } + } + return ungrouped; + } + + // Rewrite a snapshot in the layout that predates stream headers: no headers + // anywhere, and one account offset per storage slot entry rather than per + // group. void strip_stream_headers(std::filesystem::path const &version_dir) { + using namespace monad; for (auto const &dir : std::filesystem::directory_iterator{version_dir}) { for (auto const &[name, kind] : STREAM_FILES) { auto const path = dir.path() / name; - auto const framed = read_file(path); - if (framed.empty()) { + auto const stream = read_file(path); + if (stream.empty()) { continue; } - write_file(path, strip_stream_header(framed, kind)); + byte_string_view const stripped{ + strip_stream_header(stream, kind)}; + write_file( + path, + kind == MONAD_SNAPSHOT_STORAGE + ? byte_string_view{ungroup_storage_stream(stripped)} + : stripped); } } } - // Load a snapshot directory the way monad_db_snapshot_load_filesystem does, - // but without verifying checksums, so a test can rewrite the streams first. - void load_snapshot( - std::string const &dbname, std::filesystem::path const &root, - uint64_t const block, bool const load_to_secondary) + monad::byte_string version_1_stream_header(monad_snapshot_type const kind) { - char const *dbname_paths[] = {dbname.c_str()}; - auto *const loader = monad_db_snapshot_loader_create( - block, - dbname_paths, - 1, - static_cast(-1), - load_to_secondary); - for (auto const &dir : std::filesystem::directory_iterator{ - root / std::to_string(block)}) { - uint64_t const shard = std::stoull(dir.path().stem()); - auto const eth_header = read_file(dir.path() / "eth_header"); - auto const account = read_file(dir.path() / "account"); - auto const storage = read_file(dir.path() / "storage"); - auto const code = read_file(dir.path() / "code"); - auto const ptr = [](monad::byte_string const &b) { - return b.empty() ? nullptr : b.data(); - }; - monad_db_snapshot_loader_load( - loader, - shard, - ptr(eth_header), - eth_header.size(), - ptr(account), - account.size(), - ptr(storage), - storage.size(), - ptr(code), - code.size()); + monad_snapshot_stream_header const header{ + .magic = MONAD_SNAPSHOT_STREAM_MAGIC, + .version = MONAD_SNAPSHOT_STREAM_VERSION_UNGROUPED, + .kind = static_cast(kind), + .group_key_shift = 0, // reserved in version 1 + .guard = MONAD_SNAPSHOT_STREAM_GUARD}; + monad::byte_string bytes; + bytes.resize(sizeof(header)); + monad::unaligned_store(bytes.data(), header); + return bytes; + } + + // Rewrite a snapshot as version 1: same headers but for the version byte, + // and a storage stream whose records hold one slot entry each. + void downgrade_to_version_1(std::filesystem::path const &version_dir) + { + using namespace monad; + for (auto const &dir : + std::filesystem::directory_iterator{version_dir}) { + for (auto const &[name, kind] : STREAM_FILES) { + auto const path = dir.path() / name; + auto const stream = read_file(path); + if (stream.empty()) { + continue; + } + byte_string_view const stripped{ + strip_stream_header(stream, kind)}; + byte_string rewritten{version_1_stream_header(kind)}; + rewritten += kind == MONAD_SNAPSHOT_STORAGE + ? ungroup_storage_stream(stripped) + : byte_string{stripped}; + write_file(path, rewritten); + } } - monad_db_snapshot_loader_destroy(loader); + } + + struct StorageGroup + { + uint64_t account_offset; + monad::bytes32_t page_key; + size_t slots; + }; + + // Split a storage stream into its groups, asserting that each holds the + // slots of exactly one page in ascending key order. + std::vector parse_storage_stream( + monad::byte_string_view view, uint8_t const expected_group_key_shift) + { + using namespace monad; + MONAD_ASSERT(view.size() >= sizeof(monad_snapshot_stream_header)); + auto const header = + unaligned_load(view.data()); + EXPECT_EQ(header.magic, MONAD_SNAPSHOT_STREAM_MAGIC); + EXPECT_EQ(header.version, MONAD_SNAPSHOT_STREAM_VERSION); + EXPECT_EQ(header.kind, MONAD_SNAPSHOT_STORAGE); + EXPECT_EQ(header.group_key_shift, expected_group_key_shift); + view.remove_prefix(sizeof(header)); + + std::vector groups; + while (!view.empty()) { + MONAD_ASSERT( + view.size() >= MONAD_SNAPSHOT_STORAGE_GROUP_HEADER_SIZE); + uint64_t const account_offset = + unaligned_load(view.data()); + uint32_t const payload_len = + unaligned_load(view.data() + sizeof(account_offset)); + view.remove_prefix(MONAD_SNAPSHOT_STORAGE_GROUP_HEADER_SIZE); + EXPECT_LE(payload_len, view.size()); + byte_string_view payload{view.substr(0, payload_len)}; + view.remove_prefix(payload_len); + + StorageGroup group{account_offset, bytes32_t{}, 0}; + bytes32_t last_slot_key{}; + while (!payload.empty()) { + auto const entry = decode_storage_db_raw(payload); + EXPECT_TRUE(entry.has_value()); + bytes32_t const slot_key = to_bytes(entry.value().first); + if (group.slots == 0) { + group.page_key = compute_page_key(slot_key); + } + else { + EXPECT_GT(slot_key, last_slot_key); + EXPECT_EQ(compute_page_key(slot_key), group.page_key); + } + last_slot_key = slot_key; + ++group.slots; + } + EXPECT_GT(group.slots, 0u); + groups.push_back(group); + } + return groups; } } @@ -740,11 +880,18 @@ namespace constexpr std::array PAGE_SLOTS{ 0x0000, 0x0001, 0x0002, 0x007f, 0x0080, 0x0081, 0x0100, 0x01ff}; constexpr size_t PAGES_PER_ACCOUNT = 4; - std::array const PAGE_ADDRS{ + constexpr size_t MULTI_SLOT_PAGES_PER_ACCOUNT = 2; + // Address{5} and Address{15} hash into the same shard, so one of them + // writes its storage groups at a non-zero account offset: with one account + // per shard every offset is zero and a loader that ignored the field would + // pass. + std::array const PAGE_ADDRS{ monad::Address{1}, monad::Address{2}, monad::Address{3}, - monad::Address{4}}; + monad::Address{4}, + monad::Address{5}, + monad::Address{15}}; monad::bytes32_t page_slot_key(uint16_t const raw) { @@ -826,8 +973,54 @@ namespace return tdb2.state_root(); } - void dump_page_source( - std::string const &dbname, std::filesystem::path const &root) + // Populate the slot-encoded primary timeline of `dbname` with the same + // state build_page_source puts on a page-encoded secondary, so a restore of + // either dump into a page-encoded target must reach the same root. + void build_slot_source(std::string const &dbname) + { + using namespace monad; + using namespace monad::mpt; + + mpt::Db db{ + std::make_unique(), + OnDiskDbConfig{.dbname_paths = {dbname}}}; + load_header({}, db, BlockHeader{.number = 0}); + db.update_finalized_version(0); + + StateDeltas deltas; + Code code_delta; + for (auto const &addr : PAGE_ADDRS) { + StorageDeltas storage; + for (auto const raw : PAGE_SLOTS) { + storage.emplace( + page_slot_key(raw), + StorageDelta{bytes32_t{}, page_slot_value(addr, raw)}); + } + auto const code = page_code(addr); + bytes32_t const code_hash = to_bytes(keccak256(code)); + code_delta.emplace(code_hash, vm::make_shared_intercode(code)); + deltas.emplace( + addr, + StateDelta{ + .account = + {std::nullopt, + Account{.balance = 1, .code_hash = code_hash}}, + .storage = storage}); + } + TrieDb tdb{db}; + MONAD_ASSERT(!tdb.is_page_encoded()); + monad::test::commit_simple( + tdb, + deltas, + code_delta, + bytes32_t{PAGE_BLOCK}, + BlockHeader{.number = PAGE_BLOCK}); + tdb.finalize(PAGE_BLOCK, bytes32_t{PAGE_BLOCK}); + } + + void dump_source( + std::string const &dbname, std::filesystem::path const &root, + bool const from_secondary) { auto *const context = monad_db_snapshot_filesystem_write_user_context_create( @@ -843,10 +1036,22 @@ namespace 2048, 1, 0, - /*dump_from_secondary=*/true)); + from_secondary)); monad_db_snapshot_filesystem_write_user_context_destroy(context); } + void dump_page_source( + std::string const &dbname, std::filesystem::path const &root) + { + dump_source(dbname, root, /*from_secondary=*/true); + } + + void dump_slot_source( + std::string const &dbname, std::filesystem::path const &root) + { + dump_source(dbname, root, /*from_secondary=*/false); + } + void activate_page_secondary(std::string const &dbname) { using namespace monad; @@ -900,6 +1105,49 @@ namespace db.traverse_blocking(state_cursor.value(), counter, PAGE_BLOCK)); EXPECT_EQ(counter.count, PAGE_ADDRS.size() * PAGES_PER_ACCOUNT); } + + // Restore `root` into a fresh slot-encoded primary at the smallest + // threshold there is, and check every slot of every account round-trips. + void restore_and_verify_slot_target( + std::string const &dbname, std::filesystem::path const &root) + { + using namespace monad; + using namespace monad::mpt; + + { + mpt::Db dest_init{ + std::make_unique(), + OnDiskDbConfig{.dbname_paths = {dbname}}}; + monad::mpt::test::DbAccessor::aux(dest_init) + .metadata_ctx() + .set_state_machine_kind( + timeline_id::primary, state_machine_kind::ethereum); + } + load_snapshot( + dbname, + root, + PAGE_BLOCK, + /*flush_bytes=*/1, + /*load_to_secondary=*/false); + + AsyncIOContext io_context{ + ReadOnlyOnDiskDbConfig{.dbname_paths = {dbname}}}; + mpt::Db db{io_context}; + TrieDb tdb{db}; + ASSERT_FALSE(tdb.is_page_encoded()); + tdb.set_block_and_prefix(PAGE_BLOCK); + Incarnation const inc{0, 0}; + for (auto const &addr : PAGE_ADDRS) { + ASSERT_TRUE(tdb.read_account(addr).has_value()); + for (auto const raw : PAGE_SLOTS) { + EXPECT_EQ( + tdb.read_storage(addr, inc, page_slot_key(raw)), + page_slot_value(addr, raw)) + << "addr=" << static_cast(addr.bytes[19]) << " slot=0x" + << std::hex << raw; + } + } + } } // Every stream a shard writes opens with a header naming its version and kind. @@ -929,8 +1177,95 @@ TEST(DbBinarySnapshot, SnapshotStreamHeaders) } } +// A page-encoded source dumps the slots of each page as one closed group, which +// is what lets a page-encoded target write pages out as it reads. +TEST(DbBinarySnapshot, PageGroupedStorageStream) +{ + using namespace monad; + + TempDb const src_db; + TempDir const snapshot_dir; + + build_page_source(src_db.path); + dump_page_source(src_db.path, snapshot_dir.path); + + size_t total_groups = 0; + size_t total_slots = 0; + size_t multi_slot_groups = 0; + size_t groups_past_the_first_account = 0; + for (auto const &dir : std::filesystem::directory_iterator{ + snapshot_dir.path / std::to_string(PAGE_BLOCK)}) { + auto const storage = read_file(dir.path() / "storage"); + if (storage.empty()) { + continue; + } + auto const groups = + parse_storage_stream(storage, storage_page_t::PAGE_KEY_SHIFT); + std::set> seen; + for (auto const &group : groups) { + EXPECT_TRUE( + seen.emplace(group.account_offset, group.page_key).second) + << "page spread over more than one group"; + total_slots += group.slots; + if (group.slots > 1) { + ++multi_slot_groups; + } + if (group.account_offset != 0) { + ++groups_past_the_first_account; + } + } + total_groups += groups.size(); + } + EXPECT_EQ(total_groups, PAGE_ADDRS.size() * PAGES_PER_ACCOUNT); + EXPECT_EQ(total_slots, PAGE_ADDRS.size() * PAGE_SLOTS.size()); + EXPECT_EQ( + multi_slot_groups, PAGE_ADDRS.size() * MULTI_SLOT_PAGES_PER_ACCOUNT); + // Two of PAGE_ADDRS share a shard, so the offset field is exercised rather + // than being zero everywhere. + EXPECT_GT(groups_past_the_first_account, 0u); +} + +// Restore a page-encoded snapshot into a page-encoded target twice: once +// flushing at every group boundary, once with the default threshold, which a +// snapshot this small never reaches, so it flushes only when the load ends. +// Both must reproduce the source root, which is what makes the threshold a pure +// memory knob. +TEST(DbBinarySnapshot, PageToPageRestoreIndependentOfFlushThreshold) +{ + using namespace monad; + + TempDb const src_db; + TempDb const per_page_db; + TempDb const per_shard_db; + TempDir const snapshot_dir; + + bytes32_t const source_root = build_page_source(src_db.path); + dump_page_source(src_db.path, snapshot_dir.path); + + activate_page_secondary(per_page_db.path); + load_snapshot( + per_page_db.path, + snapshot_dir.path, + PAGE_BLOCK, + /*flush_bytes=*/1, + /*load_to_secondary=*/true); + verify_page_restore(per_page_db.path, source_root); + + activate_page_secondary(per_shard_db.path); + char const *dest_paths[] = {per_shard_db.path.c_str()}; + monad_db_snapshot_load_filesystem( + dest_paths, + 1, + static_cast(-1), + snapshot_dir.path.c_str(), + PAGE_BLOCK, + /*load_to_secondary=*/true); + verify_page_restore(per_shard_db.path, source_root); +} + // A snapshot with no stream headers at all — the layout dumped before they -// existed — still restores into either encoding. +// existed — still restores into either encoding, buffering the shard for a +// page-encoded target as it always did. TEST(DbBinarySnapshot, HeaderlessSnapshotRestores) { using namespace monad; @@ -945,47 +1280,91 @@ TEST(DbBinarySnapshot, HeaderlessSnapshotRestores) dump_page_source(src_db.path, snapshot_dir.path); strip_stream_headers(snapshot_dir.path / std::to_string(PAGE_BLOCK)); + // A shift of zero tells the loader nothing about where pages end, so the + // one byte threshold must not tempt it into writing a page before the shard + // is read out. activate_page_secondary(page_db.path); load_snapshot( page_db.path, snapshot_dir.path, PAGE_BLOCK, + /*flush_bytes=*/1, /*load_to_secondary=*/true); verify_page_restore(page_db.path, source_root); - { - mpt::Db dest_init{ - std::make_unique(), - OnDiskDbConfig{.dbname_paths = {slot_db.path}}}; - monad::mpt::test::DbAccessor::aux(dest_init) - .metadata_ctx() - .set_state_machine_kind( - timeline_id::primary, state_machine_kind::ethereum); - } + restore_and_verify_slot_target(slot_db.path, snapshot_dir.path); +} + +// A version 1 snapshot — headers throughout, but a storage stream whose records +// hold one slot each — restores into either encoding. This is the layout of a +// snapshot dumped after stream headers landed and before grouping did. +TEST(DbBinarySnapshot, Version1SnapshotRestores) +{ + using namespace monad; + + TempDb const src_db; + TempDb const page_db; + TempDb const slot_db; + TempDir const snapshot_dir; + + bytes32_t const source_root = build_page_source(src_db.path); + dump_page_source(src_db.path, snapshot_dir.path); + downgrade_to_version_1(snapshot_dir.path / std::to_string(PAGE_BLOCK)); + + // Version 1 leaves the shift byte reserved, so the loader must not read a + // page-closing promise out of it however small the threshold. + activate_page_secondary(page_db.path); load_snapshot( - slot_db.path, + page_db.path, snapshot_dir.path, PAGE_BLOCK, - /*load_to_secondary=*/false); - { - AsyncIOContext io_context{ - ReadOnlyOnDiskDbConfig{.dbname_paths = {slot_db.path}}}; - mpt::Db db{io_context}; - TrieDb tdb{db}; - ASSERT_FALSE(tdb.is_page_encoded()); - tdb.set_block_and_prefix(PAGE_BLOCK); - Incarnation const inc{0, 0}; - for (auto const &addr : PAGE_ADDRS) { - ASSERT_TRUE(tdb.read_account(addr).has_value()); - for (auto const raw : PAGE_SLOTS) { - EXPECT_EQ( - tdb.read_storage(addr, inc, page_slot_key(raw)), - page_slot_value(addr, raw)) - << "addr=" << static_cast(addr.bytes[19]) << " slot=0x" - << std::hex << raw; - } + /*flush_bytes=*/1, + /*load_to_secondary=*/true); + verify_page_restore(page_db.path, source_root); + + restore_and_verify_slot_target(slot_db.path, snapshot_dir.path); +} + +// A snapshot dumped from a slot-encoded db is grouped, but every group holds +// one slot, so the loader cannot close a target page as it reads and must hold +// the shard however small the threshold. Restoring it into a page-encoded +// target has to reach the same root as restoring a page-encoded dump of the +// same state. +TEST(DbBinarySnapshot, SlotSourceToPageTargetIgnoresFlushThreshold) +{ + using namespace monad; + + TempDb const page_src_db; + TempDb const slot_src_db; + TempDb const dest_db; + TempDir const snapshot_dir; + + bytes32_t const page_root = build_page_source(page_src_db.path); + build_slot_source(slot_src_db.path); + dump_slot_source(slot_src_db.path, snapshot_dir.path); + + size_t storage_streams = 0; + for (auto const &dir : std::filesystem::directory_iterator{ + snapshot_dir.path / std::to_string(PAGE_BLOCK)}) { + auto const storage = read_file(dir.path() / "storage"); + if (storage.empty()) { + continue; + } + for (auto const &group : parse_storage_stream(storage, 0)) { + EXPECT_EQ(group.slots, 1u) << "a slot leaf holds one slot"; } + ++storage_streams; } + EXPECT_GT(storage_streams, 0u); + + activate_page_secondary(dest_db.path); + load_snapshot( + dest_db.path, + snapshot_dir.path, + PAGE_BLOCK, + /*flush_bytes=*/1, + /*load_to_secondary=*/true); + verify_page_restore(dest_db.path, page_root); } // Dump from a page-encoded secondary timeline, then load into a fresh @@ -1019,8 +1398,40 @@ TEST(DbBinarySnapshot, DumpFromSecondaryPageDb) TempDb const src_db; TempDb const dest_db; + TempDb const flushing_dest_db; TempDir const snapshot_dir; + auto const stamp_slot_target = [](std::string const &dbname) { + mpt::Db dest_init{ + std::make_unique(), + OnDiskDbConfig{.dbname_paths = {dbname}}}; + monad::mpt::test::DbAccessor::aux(dest_init) + .metadata_ctx() + .set_state_machine_kind( + timeline_id::primary, state_machine_kind::ethereum); + }; + + // Verify the target is slot-encoded and every slot round-trips. + auto const verify_slots = [&](std::string const &dbname) { + AsyncIOContext io_context{ + ReadOnlyOnDiskDbConfig{.dbname_paths = {dbname}}}; + mpt::Db db{io_context}; + TrieDb tdb{db}; + ASSERT_FALSE(tdb.is_page_encoded()); + tdb.set_block_and_prefix(BLOCK); + Incarnation const inc{0, 0}; + for (auto const &addr : ADDRS) { + ASSERT_TRUE(tdb.read_account(addr).has_value()); + for (auto const b : SLOT_BYTES) { + EXPECT_EQ( + tdb.read_storage(addr, inc, make_slot(b)), + make_val(addr, b)) + << "addr=" << static_cast(addr.bytes[19]) << " slot=0x" + << std::hex << static_cast(b); + } + } + }; + // Build a slot-encoded primary with a page-encoded secondary, and populate // the secondary (MonadOnDiskMachine stamps its kind = monad). { @@ -1083,15 +1494,7 @@ TEST(DbBinarySnapshot, DumpFromSecondaryPageDb) /*dump_from_secondary=*/true)); monad_db_snapshot_filesystem_write_user_context_destroy(context); - { - mpt::Db dest_init{ - std::make_unique(), - OnDiskDbConfig{.dbname_paths = {dest_db.path}}}; - monad::mpt::test::DbAccessor::aux(dest_init) - .metadata_ctx() - .set_state_machine_kind( - timeline_id::primary, state_machine_kind::ethereum); - } + stamp_slot_target(dest_db.path); char const *dest_path[] = {dest_db.path.c_str()}; monad_db_snapshot_load_filesystem( dest_path, @@ -1102,24 +1505,17 @@ TEST(DbBinarySnapshot, DumpFromSecondaryPageDb) /*load_to_secondary=*/false); } - // Verify the target is slot-encoded and every slot round-trips. - { - AsyncIOContext io_context{ - ReadOnlyOnDiskDbConfig{.dbname_paths = {dest_db.path}}}; - mpt::Db db{io_context}; - TrieDb tdb{db}; - ASSERT_FALSE(tdb.is_page_encoded()); - tdb.set_block_and_prefix(BLOCK); - Incarnation const inc{0, 0}; - for (auto const &addr : ADDRS) { - ASSERT_TRUE(tdb.read_account(addr).has_value()); - for (auto const b : SLOT_BYTES) { - EXPECT_EQ( - tdb.read_storage(addr, inc, make_slot(b)), - make_val(addr, b)) - << "addr=" << static_cast(addr.bytes[19]) << " slot=0x" - << std::hex << static_cast(b); - } - } - } + verify_slots(dest_db.path); + + // The same multi-slot groups into a slot-encoded target, flushing after + // every group: a slot leaf stands alone, so nothing depends on the group + // surviving until the end of the shard. + stamp_slot_target(flushing_dest_db.path); + load_snapshot( + flushing_dest_db.path, + snapshot_dir.path, + BLOCK, + /*flush_bytes=*/1, + /*load_to_secondary=*/false); + verify_slots(flushing_dest_db.path); }