db: snapshots freezer (#2144)

silkworm/db/access_layer.cpp

@@ -22,7 +22,6 @@
 #include <silkworm/core/common/assert.hpp>
 #include <silkworm/core/common/empty_hashes.hpp>
 #include <silkworm/core/common/endian.hpp>
-#include <silkworm/core/types/block_body_for_storage.hpp>
 #include <silkworm/core/types/evmc_bytes32.hpp>
 #include <silkworm/db/bodies/body_queries.hpp>
 #include <silkworm/db/headers/header_queries.hpp>
@@ -456,14 +455,23 @@ bool read_body(ROTxn& txn, BlockNum block_number, const uint8_t (&hash)[kHashLen
     return read_body(txn, key, read_senders, out);
 }
 
-bool read_body(ROTxn& txn, const Bytes& key, bool read_senders, BlockBody& out) {
+std::optional<BlockBodyForStorage> read_body_for_storage(ROTxn& txn, const Bytes& key) {
     auto cursor = txn.ro_cursor(table::kBlockBodies);
     auto data{cursor->find(to_slice(key), false)};
     if (!data) {
-        return false;
+        return std::nullopt;
     }
     ByteView data_view{from_slice(data.value)};
     auto body{unwrap_or_throw(decode_stored_block_body(data_view))};
+    return body;
+}
+
+bool read_body(ROTxn& txn, const Bytes& key, bool read_senders, BlockBody& out) {
+    auto body_opt = read_body_for_storage(txn, key);
+    if (!body_opt) {
+        return false;
+    }
+    BlockBodyForStorage& body = *body_opt;
 
     std::swap(out.ommers, body.ommers);
     std::swap(out.withdrawals, body.withdrawals);
@@ -477,12 +485,10 @@ bool read_body(ROTxn& txn, const Bytes& key, bool read_senders, BlockBody& out)
 
 bool read_rlp_transactions(ROTxn& txn, BlockNum height, const evmc::bytes32& hash, std::vector<Bytes>& rlp_txs) {
     const auto key{block_key(height, hash.bytes)};
-    auto cursor = txn.ro_cursor(table::kBlockBodies);
-    const auto data{cursor->find(to_slice(key), false)};
-    if (!data) return false;
+    auto body_opt = read_body_for_storage(txn, key);
+    if (!body_opt) return false;
+    auto& body = *body_opt;
 
-    ByteView data_view{from_slice(data.value)};
-    const auto body{unwrap_or_throw(decode_stored_block_body(data_view))};
     ensure(body.txn_count > 1, [&]() { return "unexpected txn_count=" + std::to_string(body.txn_count) + " for key=" + std::to_string(height); });
     read_rlp_transactions(txn, body.base_txn_id + 1, body.txn_count - 2, rlp_txs);
 
@@ -497,6 +503,18 @@ bool read_body(ROTxn& txn, const evmc::bytes32& h, BlockBody& body) {
     return db::read_body(txn, *block_num, h.bytes, /*read_senders=*/false, body);
 }
 
+bool read_canonical_body(ROTxn& txn, BlockNum block_number, bool read_senders, BlockBody& body) {
+    auto hash = read_canonical_hash(txn, block_number);
+    if (!hash) return false;
+    return read_body(txn, block_number, hash->bytes, read_senders, body);
+}
+
+std::optional<BlockBodyForStorage> read_canonical_body_for_storage(ROTxn& txn, BlockNum height) {
+    auto hash = read_canonical_hash(txn, height);
+    if (!hash) return std::nullopt;
+    return read_body_for_storage(txn, block_key(height, hash->bytes));
+}
+
 bool read_canonical_block(ROTxn& txn, BlockNum height, Block& block) {
     std::optional<evmc::bytes32> h = read_canonical_hash(txn, height);
     if (!h) return false;

silkworm/db/access_layer.hpp

@@ -30,6 +30,7 @@
 #include <silkworm/core/chain/config.hpp>
 #include <silkworm/core/types/account.hpp>
 #include <silkworm/core/types/block.hpp>
+#include <silkworm/core/types/block_body_for_storage.hpp>
 #include <silkworm/core/types/hash.hpp>
 #include <silkworm/core/types/receipt.hpp>
 #include <silkworm/db/mdbx/mdbx.hpp>
@@ -109,6 +110,10 @@ void write_canonical_header_hash(RWTxn& txn, const uint8_t (&hash)[kHashLength],
                              bool read_senders, BlockBody& out);
 [[nodiscard]] bool read_body(ROTxn& txn, const evmc::bytes32& hash, BlockNum bn, BlockBody& body);
 [[nodiscard]] bool read_body(ROTxn& txn, const evmc::bytes32& hash, BlockBody& body);
+[[nodiscard]] bool read_canonical_body(ROTxn& txn, BlockNum block_number, bool read_senders, BlockBody& body);
+
+[[nodiscard]] std::optional<BlockBodyForStorage> read_body_for_storage(ROTxn& txn, const Bytes& key);
+[[nodiscard]] std::optional<BlockBodyForStorage> read_canonical_body_for_storage(ROTxn& txn, BlockNum height);
 
 //! \brief Read the canonical block at specified height
 [[nodiscard]] bool read_canonical_block(ROTxn& txn, BlockNum height, Block& block);

silkworm/db/bodies/body_snapshot_freezer.cpp

@@ -0,0 +1,37 @@
+/*
+   Copyright 2024 The Silkworm Authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+#include "body_snapshot_freezer.hpp"
+
+#include <stdexcept>
+
+#include <silkworm/db/access_layer.hpp>
+
+#include "body_snapshot.hpp"
+
+namespace silkworm::db {
+
+void BodySnapshotFreezer::copy(ROTxn& txn, BlockNumRange range, snapshots::SnapshotFileWriter& file_writer) const {
+    snapshots::BodySnapshotWriter writer{file_writer};
+    auto out = writer.out();
+    for (BlockNum i = range.first; i < range.second; i++) {
+        auto value_opt = read_canonical_body_for_storage(txn, i);
+        if (!value_opt) throw std::runtime_error{"BodySnapshotFreezer::copy missing body for block " + std::to_string(i)};
+        *out++ = *value_opt;
+    }
+}
+
+}  // namespace silkworm::db

silkworm/db/bodies/body_snapshot_freezer.hpp

@@ -0,0 +1,29 @@
+/*
+   Copyright 2024 The Silkworm Authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+#pragma once
+
+#include <silkworm/db/snapshot_freezer.hpp>
+
+namespace silkworm::db {
+
+class BodySnapshotFreezer : public SnapshotFreezer {
+  public:
+    ~BodySnapshotFreezer() override = default;
+    void copy(ROTxn& txn, BlockNumRange range, snapshots::SnapshotFileWriter& file_writer) const override;
+};
+
+}  // namespace silkworm::db

silkworm/db/data_migration.cpp

@@ -0,0 +1,31 @@
+/*
+   Copyright 2024 The Silkworm Authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+#include "data_migration.hpp"
+
+namespace silkworm::db {
+
+void DataMigration::run() {
+    cleanup();
+    auto command = next_command();
+    if (!command) return;
+    auto result = migrate(std::move(command));
+    index(result);
+    commit(result);
+    cleanup();
+}
+
+}  // namespace silkworm::db

silkworm/db/data_migration.hpp

@@ -0,0 +1,44 @@
+/*
+   Copyright 2024 The Silkworm Authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+#pragma once
+
+#include <memory>
+
+namespace silkworm::db {
+
+struct DataMigrationCommand {
+    virtual ~DataMigrationCommand() = default;
+};
+
+struct DataMigrationResult {
+    virtual ~DataMigrationResult() = default;
+};
+
+struct DataMigration {
+    virtual ~DataMigration() = default;
+
+    void run();
+
+  protected:
+    virtual std::unique_ptr<DataMigrationCommand> next_command();
+    virtual std::shared_ptr<DataMigrationResult> migrate(std::unique_ptr<DataMigrationCommand> command) = 0;
+    virtual void index(std::shared_ptr<DataMigrationResult> result) = 0;
+    virtual void commit(std::shared_ptr<DataMigrationResult> result) = 0;
+    virtual void cleanup() = 0;
+};
+
+}  // namespace silkworm::db

silkworm/db/freezer.cpp

@@ -0,0 +1,150 @@
+/*
+   Copyright 2024 The Silkworm Authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+#include "freezer.hpp"
+
+#include <cassert>
+#include <filesystem>
+#include <stdexcept>
+#include <vector>
+
+#include <silkworm/core/common/base.hpp>
+
+#include "access_layer.hpp"
+#include "bodies/body_snapshot_freezer.hpp"
+#include "headers/header_snapshot_freezer.hpp"
+#include "prune_mode.hpp"
+#include "snapshot_freezer.hpp"
+#include "snapshots/path.hpp"
+#include "snapshots/snapshot_bundle.hpp"
+#include "snapshots/snapshot_writer.hpp"
+#include "transactions/txn_snapshot_freezer.hpp"
+
+namespace silkworm::db {
+
+using namespace silkworm::snapshots;
+
+struct FreezerCommand : public DataMigrationCommand {
+    BlockNumRange range;
+
+    FreezerCommand(BlockNumRange range1)
+        : range(std::move(range1)) {}
+    ~FreezerCommand() override = default;
+};
+
+struct FreezerResult : public DataMigrationResult {
+    SnapshotBundle bundle;
+
+    FreezerResult(SnapshotBundle bundle1)
+        : bundle(std::move(bundle1)) {}
+    ~FreezerResult() override = default;
+};
+
+static BlockNum get_tip_num(ROTxn& txn) {
+    auto [num, _] = db::read_canonical_head(txn);
+    return num;
+}
+
+std::unique_ptr<DataMigrationCommand> Freezer::next_command() {
+    BlockNum last_frozen = snapshots_.max_block_available();
+    BlockNum start = (last_frozen > 0) ? last_frozen + 1 : 0;
+    BlockNum end = start + kChunkSize;
+
+    BlockNum tip = [this] {
+        auto db_tx = db_access_.start_ro_tx();
+        return get_tip_num(db_tx);
+    }();
+
+    if (end + kFullImmutabilityThreshold <= tip) {
+        return std::make_unique<FreezerCommand>(FreezerCommand{{start, end}});
+    }
+    return {};
+}
+
+static const SnapshotFreezer& get_snapshot_freezer(SnapshotType type) {
+    static HeaderSnapshotFreezer header_snapshot_freezer;
+    static BodySnapshotFreezer body_snapshot_freezer;
+    static TransactionSnapshotFreezer txn_snapshot_freezer;
+
+    switch (type) {
+        case snapshots::headers:
+            return header_snapshot_freezer;
+        case snapshots::bodies:
+            return body_snapshot_freezer;
+        case snapshots::transactions:
+            return txn_snapshot_freezer;
+        default:
+            assert(false);
+            throw std::runtime_error("invalid type");
+    }
+}
+
+std::shared_ptr<DataMigrationResult> Freezer::migrate(std::unique_ptr<DataMigrationCommand> command) {
+    auto& freezer_command = dynamic_cast<FreezerCommand&>(*command);
+    auto range = freezer_command.range;
+
+    auto bundle = snapshots_.bundle_factory().make(tmp_dir_path_, range);
+    for (auto& snapshot_ref : bundle.snapshots()) {
+        auto path = snapshot_ref.get().path();
+        SnapshotFileWriter file_writer{path, tmp_dir_path_};
+        {
+            auto db_tx = db_access_.start_ro_tx();
+            auto& freezer = get_snapshot_freezer(path.type());
+            freezer.copy(db_tx, range, file_writer);
+        }
+        SnapshotFileWriter::flush(std::move(file_writer));
+    }
+
+    return std::make_shared<FreezerResult>(std::move(bundle));
+}
+
+void Freezer::index(std::shared_ptr<DataMigrationResult> result) {
+    auto& freezer_result = dynamic_cast<FreezerResult&>(*result);
+    auto& bundle = freezer_result.bundle;
+
+    for (auto& snapshot_ref : bundle.snapshots()) {
+        SnapshotPath snapshot_path = snapshot_ref.get().path();
+        auto index_builders = snapshots_.bundle_factory().index_builders(snapshot_path);
+        for (auto& index_builder : index_builders) {
+            index_builder->build();
+        }
+    }
+}
+
+static void move_file(const std::filesystem::path& path, const std::filesystem::path& target_dir_path) {
+    std::filesystem::rename(path, target_dir_path / path.filename());
+}
+
+void Freezer::commit(std::shared_ptr<DataMigrationResult> result) {
+    auto& freezer_result = dynamic_cast<FreezerResult&>(*result);
+    auto& bundle = freezer_result.bundle;
+
+    for (auto& index_ref : bundle.indexes()) {
+        move_file(index_ref.get().path().path(), snapshots_.path());
+    }
+    for (auto& snapshot_ref : bundle.snapshots()) {
+        move_file(snapshot_ref.get().path().path(), snapshots_.path());
+    }
+
+    auto final_bundle = snapshots_.bundle_factory().make(snapshots_.path(), bundle.block_range());
+    snapshots_.add_snapshot_bundle(std::move(final_bundle));
+}
+
+void Freezer::cleanup() {
+    // TODO
+}
+
+}  // namespace silkworm::db