diff --git a/category/execution/ethereum/db/db_snapshot.cpp b/category/execution/ethereum/db/db_snapshot.cpp index 7ee8d12824..42f34cb240 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 2c4e3e3118..f56f06060b 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); }