diff --git a/.github/workflows/run_integration_tests.sh b/.github/workflows/run_integration_tests.sh
index 6d9c0a8784..44bf4c993a 100755
--- a/.github/workflows/run_integration_tests.sh
+++ b/.github/workflows/run_integration_tests.sh
@@ -34,6 +34,8 @@ eth_getLogs/test_17,\
 eth_getLogs/test_18,\
 eth_getLogs/test_19,\
 eth_getLogs/test_20,\
+eth_getTransactionByHash/test_02,\
+parity_listStorageKeys/test_07,\
 trace_replayBlockTransactions/test_29,\
 trace_transaction/test_44,\
 trace_transaction/test_47
diff --git a/silkworm/capi/cli/execute.cpp b/silkworm/capi/cli/execute.cpp
index 7100355a4d..639fbf1b0f 100644
--- a/silkworm/capi/cli/execute.cpp
+++ b/silkworm/capi/cli/execute.cpp
@@ -18,6 +18,7 @@
 
 #include <silkworm/capi/silkworm.h>
 #include <silkworm/db/access_layer.hpp>
+#include <silkworm/db/blocks/step_block_num_converter.hpp>
 #include <silkworm/db/datastore/kvdb/mdbx.hpp>
 #include <silkworm/db/datastore/snapshots/snapshot_repository.hpp>
 #include <silkworm/infra/cli/common.hpp>
@@ -319,7 +320,7 @@ int build_indexes(SilkwormHandle handle, const BuildIndexesSettings& settings, c
         if (!snapshot_path.has_value())
             throw std::runtime_error("Invalid snapshot path");
 
-        segment::SegmentFileReader& segment = segments.emplace_back(*snapshot_path);
+        segment::SegmentFileReader& segment = segments.emplace_back(*snapshot_path, db::blocks::kStepToBlockNumConverter);
 
         auto mmf = new SilkwormMemoryMappedFile{
             .file_path = make_path(*snapshot_path),
diff --git a/silkworm/capi/silkworm_test.cpp b/silkworm/capi/silkworm_test.cpp
index 1f6bcbeaad..d9702d7322 100644
--- a/silkworm/capi/silkworm_test.cpp
+++ b/silkworm/capi/silkworm_test.cpp
@@ -17,6 +17,7 @@
 #include <silkworm/db/datastore/kvdb/mdbx.hpp>
 #include <silkworm/db/datastore/snapshots/schema.hpp>
 #include <silkworm/db/datastore/snapshots/segment/segment_reader.hpp>
+#include <silkworm/db/state/step_txn_id_converter.hpp>
 #include <silkworm/db/test_util/temp_snapshots.hpp>
 #include <silkworm/db/test_util/test_database_context.hpp>
 #include <silkworm/infra/common/directories.hpp>
@@ -783,6 +784,8 @@ TEST_CASE_METHOD(CApiTest, "CAPI silkworm_execute_blocks_perpetual multiple bloc
 }
 
 TEST_CASE_METHOD(CApiTest, "CAPI silkworm_add_blocks_snapshot_bundle", "[capi]") {
+    static constexpr datastore::StepToTimestampConverter kStepConverter = db::blocks::kStepToBlockNumConverter;
+
     snapshot_test::SampleHeaderSnapshotFile header_segment_file{tmp_dir.path()};
     auto& header_segment_path = header_segment_file.path();
     snapshot_test::SampleBodySnapshotFile body_segment_file{tmp_dir.path()};
@@ -793,20 +796,20 @@ TEST_CASE_METHOD(CApiTest, "CAPI silkworm_add_blocks_snapshot_bundle", "[capi]")
     auto header_index_builder = snapshots::HeaderIndex::make(header_segment_path);
     header_index_builder.set_base_data_id(header_segment_file.block_num_range().start);
     REQUIRE_NOTHROW(header_index_builder.build());
-    snapshots::segment::SegmentFileReader header_segment{header_segment_path};
+    snapshots::segment::SegmentFileReader header_segment{header_segment_path, kStepConverter};
     snapshots::rec_split::AccessorIndex idx_header_hash{header_segment_path.related_path_ext(db::blocks::kIdxExtension)};
 
     auto body_index_builder = snapshots::BodyIndex::make(body_segment_path);
     body_index_builder.set_base_data_id(body_segment_file.block_num_range().start);
     REQUIRE_NOTHROW(body_index_builder.build());
-    snapshots::segment::SegmentFileReader body_segment{body_segment_path};
+    snapshots::segment::SegmentFileReader body_segment{body_segment_path, kStepConverter};
     snapshots::rec_split::AccessorIndex idx_body_number{body_segment_path.related_path_ext(db::blocks::kIdxExtension)};
 
     auto tx_index_builder = snapshots::TransactionIndex::make(body_segment_path, txn_segment_path);
     tx_index_builder.build();
     auto tx_index_hash_to_block_builder = snapshots::TransactionToBlockIndex::make(body_segment_path, txn_segment_path, txn_segment_file.block_num_range().start);
     tx_index_hash_to_block_builder.build();
-    snapshots::segment::SegmentFileReader txn_segment{txn_segment_path};
+    snapshots::segment::SegmentFileReader txn_segment{txn_segment_path, kStepConverter};
     snapshots::rec_split::AccessorIndex idx_txn_hash{txn_segment_path.related_path_ext(db::blocks::kIdxExtension)};
     snapshots::rec_split::AccessorIndex idx_txn_hash_2_block{tx_index_hash_to_block_builder.path()};
 
@@ -938,7 +941,7 @@ TEST_CASE_METHOD(CApiTest, "CAPI silkworm_add_state_snapshot", "[capi]") {
     constexpr uint32_t kZeroSalt{0};
 
     const snapshot_test::SampleAccountsDomainSegmentFile kv_segment_file{tmp_dir.path()};
-    segment::KVSegmentFileReader kv_segment{kv_segment_file.path(), seg::CompressionKind::kAll};
+    segment::KVSegmentFileReader kv_segment{kv_segment_file.path(), db::state::kStepToTxnIdConverter, seg::CompressionKind::kAll};
     const auto kv_segment_path_string{kv_segment_file.path().path().string()};
 
     const snapshot_test::SampleAccountsDomainExistenceIndexFile existence_index_file{tmp_dir.path()};
diff --git a/silkworm/db/blocks/bodies/body_txs_amount_query_test.cpp b/silkworm/db/blocks/bodies/body_txs_amount_query_test.cpp
index 3a36f199c8..a4a6f3a9ae 100644
--- a/silkworm/db/blocks/bodies/body_txs_amount_query_test.cpp
+++ b/silkworm/db/blocks/bodies/body_txs_amount_query_test.cpp
@@ -5,6 +5,7 @@
 
 #include <catch2/catch_test_macros.hpp>
 
+#include <silkworm/db/blocks/step_block_num_converter.hpp>
 #include <silkworm/db/test_util/temp_snapshots.hpp>
 #include <silkworm/infra/common/directories.hpp>
 #include <silkworm/infra/common/log.hpp>
@@ -15,7 +16,7 @@ namespace silkworm::snapshots {
 TEST_CASE("BodyTxsAmountSegmentQuery") {
     TemporaryDirectory tmp_dir;
     test_util::SampleBodySnapshotFile snapshot_file{tmp_dir.path()};
-    segment::SegmentFileReader snapshot{snapshot_file.path()};
+    segment::SegmentFileReader snapshot{snapshot_file.path(), db::blocks::kStepToBlockNumConverter};
 
     BodyTxsAmountSegmentQuery query{snapshot};
     auto result = query.exec();
diff --git a/silkworm/db/blocks/headers/header_segment.cpp b/silkworm/db/blocks/headers/header_segment.cpp
index bc59b9f361..e121b5b0ad 100644
--- a/silkworm/db/blocks/headers/header_segment.cpp
+++ b/silkworm/db/blocks/headers/header_segment.cpp
@@ -28,8 +28,11 @@ void decode_word_into_header(ByteView word, BlockHeader& header) {
     success_or_throw(decode_result, "decode_word_into_header: rlp::decode error");
 }
 
-void check_sanity_of_header_with_metadata(const BlockHeader& header, datastore::StepRange step_range) {
-    auto block_num_range = db::blocks::kStepToBlockNumConverter.timestamp_range_from_step_range(step_range);
+void check_sanity_of_header_with_metadata(
+    const BlockHeader& header,
+    datastore::StepRange step_range,
+    const datastore::StepToTimestampConverter& step_converter) {
+    auto block_num_range = step_converter.timestamp_range_from_step_range(step_range);
     BlockNum block_from = block_num_range.start;
     BlockNum block_to = block_num_range.end;
     ensure((header.number >= block_from) && (header.number < block_to), [&]() {
diff --git a/silkworm/db/blocks/headers/header_segment.hpp b/silkworm/db/blocks/headers/header_segment.hpp
index f1da24184c..c4c69e2b60 100644
--- a/silkworm/db/blocks/headers/header_segment.hpp
+++ b/silkworm/db/blocks/headers/header_segment.hpp
@@ -13,7 +13,10 @@ namespace silkworm::snapshots {
 
 void encode_word_from_header(Bytes& word, const BlockHeader& header);
 void decode_word_into_header(ByteView word, BlockHeader& header);
-void check_sanity_of_header_with_metadata(const BlockHeader& header, datastore::StepRange step_range);
+void check_sanity_of_header_with_metadata(
+    const BlockHeader& header,
+    datastore::StepRange step_range,
+    const datastore::StepToTimestampConverter& step_converter);
 
 struct HeaderSegmentWordEncoder : public Encoder {
     BlockHeader value;
@@ -39,8 +42,8 @@ struct HeaderSegmentWordDecoder : public Decoder {
         decode_word_into_header(word, value);
     }
 
-    void check_sanity_with_metadata(const SnapshotPath& path) override {
-        check_sanity_of_header_with_metadata(value, path.step_range());
+    void check_sanity_with_metadata(const SnapshotPath& path, const datastore::StepToTimestampConverter& step_converter) override {
+        check_sanity_of_header_with_metadata(value, path.step_range(), step_converter);
     }
 };
 
diff --git a/silkworm/db/blocks/schema_config.cpp b/silkworm/db/blocks/schema_config.cpp
index 909ff8545a..5470263703 100644
--- a/silkworm/db/blocks/schema_config.cpp
+++ b/silkworm/db/blocks/schema_config.cpp
@@ -11,6 +11,7 @@ namespace silkworm::db::blocks {
 snapshots::Schema::RepositoryDef make_blocks_repository_schema() {
     snapshots::Schema::RepositoryDef repository_schema;
     repository_schema.index_salt_file_name("salt-blocks.txt");
+    repository_schema.step_size(kStepSizeForBlockSnapshots);
     snapshots::Schema::EntityDef& schema = repository_schema.default_entity();
 
     schema.segment(kHeaderSegmentName)
@@ -48,12 +49,12 @@ snapshots::SnapshotRepository make_blocks_repository(
     std::filesystem::path dir_path,
     bool open,
     std::optional<uint32_t> index_salt) {
+    auto schema = make_blocks_repository_schema();
     return snapshots::SnapshotRepository{
         kBlocksRepositoryName,
         std::move(dir_path),
         open,
-        make_blocks_repository_schema(),
-        kStepToBlockNumConverter,
+        schema,
         index_salt,
         make_blocks_index_builders_factory(),
         std::nullopt,  // no domain caches
diff --git a/silkworm/db/blocks/transactions/txn_index.cpp b/silkworm/db/blocks/transactions/txn_index.cpp
index 78bc0192f6..587b50aa20 100644
--- a/silkworm/db/blocks/transactions/txn_index.cpp
+++ b/silkworm/db/blocks/transactions/txn_index.cpp
@@ -6,6 +6,7 @@
 #include <silkworm/db/blocks/bodies/body_txs_amount_query.hpp>
 #include <silkworm/db/datastore/snapshots/segment/segment_reader.hpp>
 
+#include "silkworm/db/blocks/step_block_num_converter.hpp"
 #include "txn_segment_word_codec.hpp"
 
 namespace silkworm::snapshots {
@@ -17,7 +18,7 @@ Bytes TransactionKeyFactory::make(ByteView key_data, uint64_t i) {
 std::pair<uint64_t, uint64_t> TransactionIndex::compute_txs_amount(
     SnapshotPath bodies_segment_path,
     std::optional<MemoryMappedRegion> bodies_segment_region) {
-    segment::SegmentFileReader body_segment{std::move(bodies_segment_path), bodies_segment_region};
+    segment::SegmentFileReader body_segment{std::move(bodies_segment_path), db::blocks::kStepToBlockNumConverter, bodies_segment_region};
     auto result = BodyTxsAmountSegmentQuery{body_segment}.exec();
     return {result.first_tx_id, result.count};
 }
diff --git a/silkworm/db/capi/db.cpp b/silkworm/db/capi/db.cpp
index d7b2f190e8..69f30c10f6 100644
--- a/silkworm/db/capi/db.cpp
+++ b/silkworm/db/capi/db.cpp
@@ -44,11 +44,13 @@ static SnapshotPath parse_snapshot_path(const char* file_path) {
 static void build_inverted_index_bundle_data(
     const SilkwormInvertedIndexSnapshot& snapshot,
     const Schema::EntityDef& entity_def,
+    const datastore::StepToTimestampConverter& step_converter,
     SnapshotBundleEntityData& data) {
     data.kv_segments.emplace(
         Schema::kInvIdxKVSegmentName,
         segment::KVSegmentFileReader{
             parse_snapshot_path(snapshot.segment.file_path),
+            step_converter,
             entity_def.kv_segment(Schema::kInvIdxKVSegmentName).compression_kind(),
             make_region(snapshot.segment),
         });
@@ -62,21 +64,24 @@ static void build_inverted_index_bundle_data(
 
 static snapshots::SnapshotBundleEntityData build_inverted_index_bundle_data(
     const SilkwormInvertedIndexSnapshot& snapshot,
-    const Schema::EntityDef& entity_def) {
+    const Schema::EntityDef& entity_def,
+    const datastore::StepToTimestampConverter& step_converter) {
     SnapshotBundleEntityData data;
-    build_inverted_index_bundle_data(snapshot, entity_def, data);
+    build_inverted_index_bundle_data(snapshot, entity_def, step_converter, data);
     return data;
 }
 
 static snapshots::SnapshotBundleEntityData build_domain_bundle_data(
     const SilkwormDomainSnapshot& snapshot,
     const Schema::EntityDef& entity_def,
+    const datastore::StepToTimestampConverter& step_converter,
     uint32_t index_salt) {
     SnapshotBundleEntityData data;
     data.kv_segments.emplace(
         Schema::kDomainKVSegmentName,
         segment::KVSegmentFileReader{
             parse_snapshot_path(snapshot.segment.file_path),
+            step_converter,
             entity_def.kv_segment(Schema::kDomainKVSegmentName).compression_kind(),
             make_region(snapshot.segment),
         });
@@ -105,12 +110,14 @@ static snapshots::SnapshotBundleEntityData build_domain_bundle_data(
 
 static snapshots::SnapshotBundleEntityData build_history_bundle_data(
     const SilkwormHistorySnapshot& snapshot,
-    const Schema::EntityDef& entity_def) {
+    const Schema::EntityDef& entity_def,
+    const datastore::StepToTimestampConverter& step_converter) {
     SnapshotBundleEntityData data;
     data.segments.emplace(
         Schema::kHistorySegmentName,
         segment::SegmentFileReader{
             parse_snapshot_path(snapshot.segment.file_path),
+            step_converter,
             make_region(snapshot.segment),
         });
     data.accessor_indexes.emplace(
@@ -120,7 +127,7 @@ static snapshots::SnapshotBundleEntityData build_history_bundle_data(
             make_region(snapshot.accessor_index),
         });
 
-    build_inverted_index_bundle_data(snapshot.inverted_index, entity_def, data);
+    build_inverted_index_bundle_data(snapshot.inverted_index, entity_def, step_converter, data);
     return data;
 }
 
@@ -129,22 +136,24 @@ static snapshots::SnapshotBundle build_state_snapshot_bundle_latest(
     const Schema::RepositoryDef& schema,
     uint32_t salt) {
     SnapshotBundleData bundle_data;
+    datastore::StepToTimestampConverter step_converter = schema.make_step_converter();
+
     bundle_data.entities.emplace(
         db::state::kDomainNameAccounts,
-        build_domain_bundle_data(bundle->accounts, schema.domain(db::state::kDomainNameAccounts), salt));
+        build_domain_bundle_data(bundle->accounts, schema.domain(db::state::kDomainNameAccounts), step_converter, salt));
     bundle_data.entities.emplace(
         db::state::kDomainNameStorage,
-        build_domain_bundle_data(bundle->storage, schema.domain(db::state::kDomainNameStorage), salt));
+        build_domain_bundle_data(bundle->storage, schema.domain(db::state::kDomainNameStorage), step_converter, salt));
     bundle_data.entities.emplace(
         db::state::kDomainNameCode,
-        build_domain_bundle_data(bundle->code, schema.domain(db::state::kDomainNameCode), salt));
+        build_domain_bundle_data(bundle->code, schema.domain(db::state::kDomainNameCode), step_converter, salt));
     // TODO(canepat): enable after fixing .kvi configuration with IndexList-like implementation
     // bundle_data.entities.emplace(
     //     db::state::kDomainNameCommitment,
-    //     build_domain_bundle_data(bundle->commitment, schema.domain(db::state::kDomainNameCommitment), salt));
+    //     build_domain_bundle_data(bundle->commitment, schema.domain(db::state::kDomainNameCommitment), step_converter, salt));
     bundle_data.entities.emplace(
         db::state::kDomainNameReceipts,
-        build_domain_bundle_data(bundle->receipts, schema.domain(db::state::kDomainNameReceipts), salt));
+        build_domain_bundle_data(bundle->receipts, schema.domain(db::state::kDomainNameReceipts), step_converter, salt));
 
     return SnapshotBundle{
         parse_snapshot_path(bundle->accounts.segment.file_path).step_range(),
@@ -156,32 +165,33 @@ static snapshots::SnapshotBundle build_state_snapshot_bundle_historical(
     const SilkwormStateSnapshotBundleHistorical* bundle,
     const Schema::RepositoryDef& schema) {
     SnapshotBundleData bundle_data;
+    datastore::StepToTimestampConverter step_converter = schema.make_step_converter();
 
     bundle_data.entities.emplace(
         db::state::kDomainNameAccounts,
-        build_history_bundle_data(bundle->accounts, schema.history(db::state::kDomainNameAccounts)));
+        build_history_bundle_data(bundle->accounts, schema.history(db::state::kDomainNameAccounts), step_converter));
     bundle_data.entities.emplace(
         db::state::kDomainNameStorage,
-        build_history_bundle_data(bundle->storage, schema.history(db::state::kDomainNameStorage)));
+        build_history_bundle_data(bundle->storage, schema.history(db::state::kDomainNameStorage), step_converter));
     bundle_data.entities.emplace(
         db::state::kDomainNameCode,
-        build_history_bundle_data(bundle->code, schema.history(db::state::kDomainNameCode)));
+        build_history_bundle_data(bundle->code, schema.history(db::state::kDomainNameCode), step_converter));
     bundle_data.entities.emplace(
         db::state::kDomainNameReceipts,
-        build_history_bundle_data(bundle->receipts, schema.history(db::state::kDomainNameReceipts)));
+        build_history_bundle_data(bundle->receipts, schema.history(db::state::kDomainNameReceipts), step_converter));
 
     bundle_data.entities.emplace(
         db::state::kInvIdxNameLogAddress,
-        build_inverted_index_bundle_data(bundle->log_addresses, schema.inverted_index(db::state::kInvIdxNameLogAddress)));
+        build_inverted_index_bundle_data(bundle->log_addresses, schema.inverted_index(db::state::kInvIdxNameLogAddress), step_converter));
     bundle_data.entities.emplace(
         db::state::kInvIdxNameLogTopics,
-        build_inverted_index_bundle_data(bundle->log_topics, schema.inverted_index(db::state::kInvIdxNameLogTopics)));
+        build_inverted_index_bundle_data(bundle->log_topics, schema.inverted_index(db::state::kInvIdxNameLogTopics), step_converter));
     bundle_data.entities.emplace(
         db::state::kInvIdxNameTracesFrom,
-        build_inverted_index_bundle_data(bundle->traces_from, schema.inverted_index(db::state::kInvIdxNameTracesFrom)));
+        build_inverted_index_bundle_data(bundle->traces_from, schema.inverted_index(db::state::kInvIdxNameTracesFrom), step_converter));
     bundle_data.entities.emplace(
         db::state::kInvIdxNameTracesTo,
-        build_inverted_index_bundle_data(bundle->traces_to, schema.inverted_index(db::state::kInvIdxNameTracesTo)));
+        build_inverted_index_bundle_data(bundle->traces_to, schema.inverted_index(db::state::kInvIdxNameTracesTo), step_converter));
 
     return SnapshotBundle{
         parse_snapshot_path(bundle->accounts.segment.file_path).step_range(),
@@ -281,13 +291,17 @@ SILKWORM_EXPORT int silkworm_build_recsplit_indexes(SilkwormHandle handle, struc
     return SILKWORM_OK;
 }
 
-static snapshots::SnapshotBundle build_blocks_snapshot_bundle(const SilkwormBlocksSnapshotBundle* bundle) {
+static snapshots::SnapshotBundle build_blocks_snapshot_bundle(
+    const SilkwormBlocksSnapshotBundle* bundle,
+    const Schema::RepositoryDef& schema) {
     snapshots::SnapshotBundleEntityData data;
+    datastore::StepToTimestampConverter step_converter = schema.make_step_converter();
 
     data.segments.emplace(
         db::blocks::kHeaderSegmentName,
         snapshots::segment::SegmentFileReader{
             parse_snapshot_path(bundle->headers.segment.file_path),
+            step_converter,
             make_region(bundle->headers.segment),
         });
     data.accessor_indexes.emplace(
@@ -301,6 +315,7 @@ static snapshots::SnapshotBundle build_blocks_snapshot_bundle(const SilkwormBloc
         db::blocks::kBodySegmentName,
         snapshots::segment::SegmentFileReader{
             parse_snapshot_path(bundle->bodies.segment.file_path),
+            step_converter,
             make_region(bundle->bodies.segment),
         });
     data.accessor_indexes.emplace(
@@ -314,6 +329,7 @@ static snapshots::SnapshotBundle build_blocks_snapshot_bundle(const SilkwormBloc
         db::blocks::kTxnSegmentName,
         snapshots::segment::SegmentFileReader{
             parse_snapshot_path(bundle->transactions.segment.file_path),
+            step_converter,
             make_region(bundle->transactions.segment),
         });
     data.accessor_indexes.emplace(
@@ -351,7 +367,7 @@ SILKWORM_EXPORT int silkworm_add_blocks_snapshot_bundle(
 
         auto& repository = handle->db->blocks_repository;
 
-        repository.add_snapshot_bundle(build_blocks_snapshot_bundle(bundle));
+        repository.add_snapshot_bundle(build_blocks_snapshot_bundle(bundle, repository.schema()));
         return SILKWORM_OK;
     } catch (const InvalidSnapshotPathException&) {
         return SILKWORM_INVALID_PATH;
diff --git a/silkworm/db/cli/snapshots.cpp b/silkworm/db/cli/snapshots.cpp
index 7779697a45..72b32951a1 100644
--- a/silkworm/db/cli/snapshots.cpp
+++ b/silkworm/db/cli/snapshots.cpp
@@ -39,6 +39,7 @@
 #include <silkworm/db/datastore/snapshots/snapshot_repository.hpp>
 #include <silkworm/db/snapshot_recompress.hpp>
 #include <silkworm/db/snapshot_sync.hpp>
+#include <silkworm/db/state/step_txn_id_converter.hpp>
 #include <silkworm/db/tables.hpp>
 #include <silkworm/infra/cli/common.hpp>
 #include <silkworm/infra/cli/shutdown_signal.hpp>
@@ -274,7 +275,7 @@ void decode_segment(const SnapshotSubcommandSettings& settings, int repetitions)
     SILK_INFO << "Decode snapshot: " << snapshot_path->path();
     std::chrono::time_point start{std::chrono::steady_clock::now()};
     for (int i = 0; i < repetitions; ++i) {
-        SegmentFileReader snapshot{*snapshot_path};
+        SegmentFileReader snapshot{*snapshot_path, db::blocks::kStepToBlockNumConverter};
     }
     std::chrono::duration elapsed{std::chrono::steady_clock::now() - start};
     SILK_INFO << "Decode snapshot elapsed: " << as_milliseconds(elapsed) << " msec";
@@ -329,7 +330,7 @@ void count_bodies(const SnapshotSubcommandSettings& settings, int repetitions) {
         if (settings.segment_file_name) {
             const auto snapshot_path{SnapshotPath::parse(std::filesystem::path{*settings.segment_file_name})};
             ensure(snapshot_path.has_value(), "count_bodies: invalid snapshot_file path format");
-            SegmentFileReader body_segment{*snapshot_path};
+            SegmentFileReader body_segment{*snapshot_path, db::blocks::kStepToBlockNumConverter};
             std::tie(num_bodies, num_txns) = count_bodies_in_one(settings, body_segment);
         } else {
             std::tie(num_bodies, num_txns) = count_bodies_in_all(settings);
@@ -373,7 +374,7 @@ void count_headers(const SnapshotSubcommandSettings& settings, int repetitions)
         if (settings.segment_file_name) {
             const auto snapshot_path{SnapshotPath::parse(std::filesystem::path{*settings.segment_file_name})};
             ensure(snapshot_path.has_value(), "count_headers: invalid snapshot_file path format");
-            SegmentFileReader header_segment{*snapshot_path};
+            SegmentFileReader header_segment{*snapshot_path, db::blocks::kStepToBlockNumConverter};
             num_headers = count_headers_in_one(settings, header_segment);
         } else {
             num_headers = count_headers_in_all(settings);
@@ -452,7 +453,7 @@ void open_btree_index(const SnapshotSubcommandSettings& settings) {
 
     std::chrono::time_point start{std::chrono::steady_clock::now()};
 
-    segment::KVSegmentFileReader kv_segment{*kv_segment_path, seg::CompressionKind::kAll};
+    segment::KVSegmentFileReader kv_segment{*kv_segment_path, db::state::kStepToTxnIdConverter, seg::CompressionKind::kAll};
 
     btree::BTreeIndex bt_index{bt_index_path.path()};
     SILK_INFO << "Starting KV scan and BTreeIndex check, total keys: " << bt_index.key_count();
@@ -733,7 +734,7 @@ void lookup_body_in_one(const SnapshotSubcommandSettings& settings, BlockNum blo
     ensure(snapshot_path.has_value(), "lookup_body: --snapshot_file is invalid snapshot file");
 
     std::chrono::time_point start{std::chrono::steady_clock::now()};
-    SegmentFileReader body_segment{*snapshot_path};
+    SegmentFileReader body_segment{*snapshot_path, db::blocks::kStepToBlockNumConverter};
 
     rec_split::AccessorIndex idx_body_number{snapshot_path->related_path_ext(db::blocks::kIdxExtension)};
 
@@ -851,7 +852,7 @@ void lookup_txn_by_hash_in_one(const SnapshotSubcommandSettings& settings, const
     ensure(snapshot_path.has_value(), "lookup_tx_by_hash_in_one: --snapshot_file is invalid snapshot file");
 
     std::chrono::time_point start{std::chrono::steady_clock::now()};
-    SegmentFileReader txn_segment{*snapshot_path};
+    SegmentFileReader txn_segment{*snapshot_path, db::blocks::kStepToBlockNumConverter};
 
     {
         rec_split::AccessorIndex idx_txn_hash{snapshot_path->related_path_ext(db::blocks::kIdxExtension)};
@@ -913,7 +914,7 @@ void lookup_txn_by_id_in_one(const SnapshotSubcommandSettings& settings, uint64_
     ensure(snapshot_path.has_value(), "lookup_txn_by_id_in_one: --snapshot_file is invalid snapshot file");
 
     std::chrono::time_point start{std::chrono::steady_clock::now()};
-    SegmentFileReader txn_segment{*snapshot_path};
+    SegmentFileReader txn_segment{*snapshot_path, db::blocks::kStepToBlockNumConverter};
 
     {
         rec_split::AccessorIndex idx_txn_hash{snapshot_path->related_path_ext(db::blocks::kIdxExtension)};
diff --git a/silkworm/db/datastore/snapshots/btree/btree_index_test.cpp b/silkworm/db/datastore/snapshots/btree/btree_index_test.cpp
index 9e9d230f10..2d6d222ac6 100644
--- a/silkworm/db/datastore/snapshots/btree/btree_index_test.cpp
+++ b/silkworm/db/datastore/snapshots/btree/btree_index_test.cpp
@@ -101,7 +101,7 @@ TEST_CASE("BTreeIndex", "[snapshots][btree]") {
     const auto [kv_file_path, bt_file_path] = sample_3_keys_kv_and_bt_files(tmp_dir);
 
     // Open the KV and BT index files
-    segment::KVSegmentFileReader kv_segment{kv_file_path, seg::CompressionKind::kNone};
+    segment::KVSegmentFileReader kv_segment{kv_file_path, {}, seg::CompressionKind::kNone};
     BTreeIndex bt_index{bt_file_path};
     bt_index.warmup_if_empty_or_check(kv_segment);
     REQUIRE(bt_index.key_count() == 3);
diff --git a/silkworm/db/datastore/snapshots/common/codec.hpp b/silkworm/db/datastore/snapshots/common/codec.hpp
index 6af7cfc1d1..8b78d05c07 100644
--- a/silkworm/db/datastore/snapshots/common/codec.hpp
+++ b/silkworm/db/datastore/snapshots/common/codec.hpp
@@ -6,6 +6,10 @@
 #include <silkworm/core/common/base.hpp>
 #include <silkworm/core/common/bytes.hpp>
 
+namespace silkworm::datastore {
+struct StepToTimestampConverter;
+}  // namespace silkworm::datastore
+
 namespace silkworm::snapshots {
 
 class SnapshotPath;
@@ -24,7 +28,8 @@ struct Decoder {
     virtual ~Decoder() = default;
     using Word = BytesOrByteView;
     virtual void decode_word(Word& word) = 0;  // this allows word to be moved after decoding
-    virtual void check_sanity_with_metadata(const SnapshotPath& /*path*/) {}
+    virtual void decode_word_with_metadata(const SnapshotPath& /*path*/, const datastore::StepToTimestampConverter& /*step_converter*/) {}
+    virtual void check_sanity_with_metadata(const SnapshotPath& /*path*/, const datastore::StepToTimestampConverter& /*step_converter*/) {}
 };
 
 template <class TDecoder>
diff --git a/silkworm/db/datastore/snapshots/elias_fano/elias_fano_decoder.hpp b/silkworm/db/datastore/snapshots/elias_fano/elias_fano_decoder.hpp
deleted file mode 100644
index ff985ee226..0000000000
--- a/silkworm/db/datastore/snapshots/elias_fano/elias_fano_decoder.hpp
+++ /dev/null
@@ -1,27 +0,0 @@
-// Copyright 2025 The Silkworm Authors
-// SPDX-License-Identifier: Apache-2.0
-
-#pragma once
-
-#include "../common/codec.hpp"
-#include "elias_fano_list.hpp"
-
-namespace silkworm::snapshots::elias_fano {
-
-struct EliasFanoDecoder : public snapshots::Decoder {
-    EliasFanoList32 value{EliasFanoList32::empty_list()};
-
-    ~EliasFanoDecoder() override = default;
-
-    void decode_word(Word& word) override {
-        if (word.holds_bytes()) {
-            value = EliasFanoList32::from_encoded_data(std::get<Bytes>(std::move(word)));
-        } else {
-            value = EliasFanoList32::from_encoded_data(std::get<ByteView>(word));
-        }
-    }
-};
-
-static_assert(snapshots::DecoderConcept<EliasFanoDecoder>);
-
-}  // namespace silkworm::snapshots::elias_fano
diff --git a/silkworm/db/datastore/snapshots/elias_fano/elias_fano_decoder_test.cpp b/silkworm/db/datastore/snapshots/elias_fano/elias_fano_decoder_test.cpp
deleted file mode 100644
index 358e318518..0000000000
--- a/silkworm/db/datastore/snapshots/elias_fano/elias_fano_decoder_test.cpp
+++ /dev/null
@@ -1,30 +0,0 @@
-// Copyright 2025 The Silkworm Authors
-// SPDX-License-Identifier: Apache-2.0
-
-#include "elias_fano_decoder.hpp"
-
-#include <sstream>
-
-#include <catch2/catch_test_macros.hpp>
-
-#include <silkworm/core/common/bytes_to_string.hpp>
-
-namespace silkworm::snapshots::elias_fano {
-
-TEST_CASE("EliasFanoDecoder") {
-    EliasFanoList32Builder expected_list{3, 3};
-    expected_list.add_offset(1);
-    expected_list.add_offset(2);
-    expected_list.add_offset(3);
-    expected_list.build();
-    std::stringstream expected_list_stream;
-    expected_list_stream << expected_list;
-    const auto expected_list_str = expected_list_stream.str();
-
-    EliasFanoDecoder decoder;
-    auto expected_list_bytes = BytesOrByteView{string_to_bytes(expected_list_str)};
-    decoder.decode_word(expected_list_bytes);
-    CHECK(decoder.value == expected_list.as_view());
-}
-
-}  // namespace silkworm::snapshots::elias_fano
diff --git a/silkworm/db/datastore/snapshots/elias_fano/elias_fano_list.cpp b/silkworm/db/datastore/snapshots/elias_fano/elias_fano_list.cpp
index 18545a668a..0b6bc9aa56 100644
--- a/silkworm/db/datastore/snapshots/elias_fano/elias_fano_list.cpp
+++ b/silkworm/db/datastore/snapshots/elias_fano/elias_fano_list.cpp
@@ -45,29 +45,38 @@
 
 namespace silkworm::snapshots::elias_fano {
 
-EliasFanoList32 EliasFanoList32::from_encoded_data(std::span<const uint8_t> encoded_data) {
-    ensure(encoded_data.size() >= kCountLength + kULength, "EliasFanoList32::from_encoded_data data too short");
+EliasFanoList32 EliasFanoList32::from_encoded_data(BytesOrByteView encoded_data_holder) {
+    std::span<const uint8_t> encoded_data{ByteView{encoded_data_holder}};
+    const uint64_t data_offset = kCountLength + kULength;
+    ensure(encoded_data.size() >= data_offset, "EliasFanoList32::from_encoded_data data too short");
     const uint64_t last = endian::load_big_u64(encoded_data.data());
     const uint64_t u = endian::load_big_u64(encoded_data.subspan(kCountLength).data());
-    const auto remaining_data = encoded_data.subspan(kCountLength + kULength);
-    return EliasFanoList32{last + 1, u - 1, remaining_data};
+    return EliasFanoList32{last + 1, u - 1, data_offset, std::move(encoded_data_holder)};
+}
+
+EliasFanoList32 EliasFanoList32::from_encoded_data(std::span<const uint8_t> encoded_data) {
+    return from_encoded_data(BytesOrByteView{ByteView{encoded_data}});
 }
 
 EliasFanoList32 EliasFanoList32::from_encoded_data(ByteView encoded_data) {
-    return from_encoded_data(std::span<const uint8_t>{encoded_data});
+    return from_encoded_data(BytesOrByteView{encoded_data});
 }
 
 EliasFanoList32 EliasFanoList32::from_encoded_data(Bytes encoded_data) {
-    auto elias_fano_list = from_encoded_data(std::span<const uint8_t>{encoded_data});
-    elias_fano_list.data_holder_ = std::move(encoded_data);
-    return elias_fano_list;
+    return from_encoded_data(BytesOrByteView{std::move(encoded_data)});
 }
 
-EliasFanoList32::EliasFanoList32(uint64_t count, uint64_t max_value, std::span<const uint8_t> encoded_data)
+EliasFanoList32::EliasFanoList32(
+    uint64_t count,
+    uint64_t max_value,
+    uint64_t data_offset,
+    BytesOrByteView data_holder)
     : count_{count},
       u_{max_value + 1},
-      data_{reinterpret_cast<const uint64_t*>(encoded_data.data()), EliasFanoList32::total_words(count, max_value)} {
-    SILKWORM_ASSERT(EliasFanoList32::total_words(count, max_value) * sizeof(uint64_t) <= encoded_data.size());
+      data_offset_{data_offset},
+      total_words_{EliasFanoList32::total_words(count, max_value)},
+      data_holder_{std::move(data_holder)} {
+    SILKWORM_ASSERT(EliasFanoList32::total_words(count, max_value) * sizeof(uint64_t) + data_offset <= ByteView{data_holder_}.size());
     derive_fields();
 }
 
@@ -164,7 +173,8 @@ std::ostream& operator<<(std::ostream& os, const EliasFanoList32& ef) {
     os.write(reinterpret_cast<const char*>(uint64_buffer.data()), sizeof(uint64_t));
     SILK_DEBUG << "[index] written EF upper: " << ef.u_;
 
-    os.write(reinterpret_cast<const char*>(ef.data_.data()), static_cast<std::streamsize>(ef.data_.size() * sizeof(uint64_t)));
+    auto data = ef.data();
+    os.write(reinterpret_cast<const char*>(data.data()), static_cast<std::streamsize>(data.size() * sizeof(uint64_t)));
     return os;
 }
 
@@ -197,9 +207,11 @@ uint64_t EliasFanoList32::derive_fields() {
     uint64_t words_upper_bits = (count_ + (u_ >> l_) + 63) / 64;
     uint64_t jump_words = jump_size_words(count_);
     uint64_t total_words = words_lower_bits + words_upper_bits + jump_words;
-    lower_bits_ = data_.subspan(0, words_lower_bits);
-    upper_bits_ = data_.subspan(words_lower_bits, words_upper_bits);
-    jump_ = data_.subspan(words_lower_bits + words_upper_bits, jump_words);
+
+    auto data = this->data();
+    lower_bits_ = data.subspan(0, words_lower_bits);
+    upper_bits_ = data.subspan(words_lower_bits, words_upper_bits);
+    jump_ = data.subspan(words_lower_bits + words_upper_bits, jump_words);
 
     return total_words;
 }
diff --git a/silkworm/db/datastore/snapshots/elias_fano/elias_fano_list.hpp b/silkworm/db/datastore/snapshots/elias_fano/elias_fano_list.hpp
index 4b39f16895..6ef26cd183 100644
--- a/silkworm/db/datastore/snapshots/elias_fano/elias_fano_list.hpp
+++ b/silkworm/db/datastore/snapshots/elias_fano/elias_fano_list.hpp
@@ -37,6 +37,8 @@
 #include <span>
 #include <utility>
 
+#include <silkworm/core/common/bytes.hpp>
+
 #include "../common/encoding/sequence.hpp"
 #include "../common/util/iterator/list_iterator.hpp"
 
@@ -47,6 +49,9 @@ class EliasFanoList32 {
   public:
     using value_type = uint64_t;
 
+    //! Create a new 32-bit EF list from the given encoded data (i.e. data plus data header)
+    static EliasFanoList32 from_encoded_data(BytesOrByteView encoded_data_holder);
+
     //! Create a new 32-bit EF list from the given encoded data (i.e. data plus data header)
     static EliasFanoList32 from_encoded_data(std::span<const uint8_t> encoded_data);
 
@@ -60,8 +65,13 @@ class EliasFanoList32 {
     //! Create a new 32-bit EF list from an existing data sequence
     //! \param count
     //! \param max_value
-    //! \param encoded_data the existing data sequence (portion exceeding the total words will be ignored)
-    EliasFanoList32(uint64_t count, uint64_t max_value, std::span<const uint8_t> encoded_data);
+    //! \param data_offset offset in the data_holder to the data sequence
+    //! \param data_holder the data (portion exceeding the total words will be ignored)
+    EliasFanoList32(
+        uint64_t count,
+        uint64_t max_value,
+        uint64_t data_offset,
+        BytesOrByteView data_holder);
 
     size_t size() const { return count_; }
 
@@ -69,9 +79,12 @@ class EliasFanoList32 {
 
     uint64_t min() const { return at(0); }
 
-    std::span<const uint64_t> data() const { return data_; }
+    std::span<const uint64_t> data() const {
+        const uint64_t* data = reinterpret_cast<const uint64_t*>(ByteView{data_holder_}.data() + data_offset_);
+        return {data, total_words_};
+    }
 
-    size_t encoded_data_size() const { return kCountLength + kULength + data_.size() * sizeof(uint64_t); }
+    size_t encoded_data_size() const { return kCountLength + kULength + total_words_ * sizeof(uint64_t); }
 
     uint64_t at(size_t i) const;
     uint64_t operator[](size_t i) const { return at(i); }
@@ -84,23 +97,17 @@ class EliasFanoList32 {
     friend std::ostream& operator<<(std::ostream& os, const EliasFanoList32& ef);
 
     bool operator==(const EliasFanoList32& other) const {
-        return (count_ == other.count_) && (u_ == other.u_) && std::ranges::equal(data_, other.data_);
+        return (count_ == other.count_) && (u_ == other.u_) && std::ranges::equal(data(), other.data());
     }
 
     static uint64_t total_words(uint64_t count, uint64_t max_value);
     static uint64_t jump_size_words(uint64_t count);
 
-    static EliasFanoList32 empty_list() {
-        return EliasFanoList32{};
-    }
-
     using Iterator = ListIterator<EliasFanoList32, value_type>;
     Iterator begin() const { return Iterator{*this, 0}; }
     Iterator end() const { return Iterator{*this, size()}; }
 
   private:
-    EliasFanoList32() = default;
-
     uint64_t upper(uint64_t i) const;
     uint64_t derive_fields();
 
@@ -115,10 +122,10 @@ class EliasFanoList32 {
     //! The strict upper bound on the EF data points, i.e. max + 1
     uint64_t u_{0};
     uint64_t l_{0};
-    //! Lightweight view over the EF encoded data sequence.
-    std::span<const uint64_t> data_;
-    //! Copy of the EF encoded data sequence when it must be kept for lifetime reasons
-    std::optional<Bytes> data_holder_{};
+    uint64_t data_offset_{0};
+    uint64_t total_words_{0};
+    //! The EF encoded data sequence
+    BytesOrByteView data_holder_{};
 };
 
 //! 32-bit Elias-Fano (EF) list writer that can be used to encode one monotone non-decreasing sequence
@@ -139,9 +146,12 @@ class EliasFanoList32Builder {
 
     EliasFanoList32 as_view() const {
         const auto max_value = u_ - 1;
-        return EliasFanoList32{count_,
-                               max_value,
-                               {reinterpret_cast<const uint8_t*>(data_.data()), EliasFanoList32::total_words(count_, max_value) * sizeof(uint64_t)}};
+        return EliasFanoList32{
+            count_,
+            max_value,
+            0,
+            BytesOrByteView{ByteView{reinterpret_cast<const uint8_t*>(data_.data()), EliasFanoList32::total_words(count_, max_value) * sizeof(uint64_t)}},
+        };
     };
 
     size_t size() const { return count_; }
diff --git a/silkworm/db/datastore/snapshots/elias_fano/elias_fano_list_test.cpp b/silkworm/db/datastore/snapshots/elias_fano/elias_fano_list_test.cpp
index 803596ccd4..0f13f3aaa2 100644
--- a/silkworm/db/datastore/snapshots/elias_fano/elias_fano_list_test.cpp
+++ b/silkworm/db/datastore/snapshots/elias_fano/elias_fano_list_test.cpp
@@ -10,6 +10,7 @@
 
 #include <catch2/catch_test_macros.hpp>
 
+#include <silkworm/core/common/bytes_to_string.hpp>
 #include <silkworm/core/common/endian.hpp>
 #include <silkworm/core/common/util.hpp>
 
@@ -114,9 +115,7 @@ TEST_CASE("EliasFanoList32", "[silkworm][recsplit][elias_fano]") {
         CHECK(to_hex(ef_bytes) == to_expected_hex(ef_test.offsets.size() - 1, ef_test.expected_u, ef_test.expected_data));
 
         // Decode monotone ascending integer sequence from Elias-Fano representation and compare with original
-        constexpr size_t kParamsSize{2 * sizeof(uint64_t)};  // count + u length in bytes
-        std::span<uint8_t> data{ef_bytes.data() + kParamsSize, ef_bytes.size() - kParamsSize};
-        EliasFanoList32 ef_list_copy{ef_test.offsets.size(), ef_test.expected_u - 1, data};
+        EliasFanoList32 ef_list_copy = EliasFanoList32::from_encoded_data(ef_bytes);
         for (uint64_t i{0}; i < ef_test.offsets.size(); ++i) {
             const uint64_t x = ef_list_copy.at(i);
             CHECK(x == ef_test.offsets[i]);
@@ -152,4 +151,17 @@ TEST_CASE("EliasFanoList32::seek", "[silkworm][recsplit][elias_fano]") {
     CHECK(ef_list.seek(70, true) == SeekResult{offsets.size() - 1, 62});
 }
 
+TEST_CASE("EliasFanoList32::from_encoded_data") {
+    EliasFanoList32Builder expected_list{3, 3};
+    expected_list.add_offset(1);
+    expected_list.add_offset(2);
+    expected_list.add_offset(3);
+    expected_list.build();
+    std::stringstream expected_list_stream;
+    expected_list_stream << expected_list;
+    const Bytes expected_list_bytes = string_to_bytes(expected_list_stream.str());
+
+    CHECK(EliasFanoList32::from_encoded_data(expected_list_bytes) == expected_list.as_view());
+}
+
 }  // namespace silkworm::snapshots::elias_fano
diff --git a/silkworm/db/datastore/snapshots/history_range_by_keys_query.hpp b/silkworm/db/datastore/snapshots/history_range_by_keys_query.hpp
index 7b473282d4..a96abda048 100644
--- a/silkworm/db/datastore/snapshots/history_range_by_keys_query.hpp
+++ b/silkworm/db/datastore/snapshots/history_range_by_keys_query.hpp
@@ -40,7 +40,7 @@ struct HistoryRangeByKeysSegmentQuery {
         datastore::Timestamp timestamp,
         bool ascending,
         Bytes key_data,
-        const elias_fano::EliasFanoList32& key_timestamps) const {
+        const InvertedIndexTimestampList& key_timestamps) const {
         SILKWORM_ASSERT(ascending);  // descending is not implemented
 
         // find the first key timestamp within the ts_range
@@ -66,7 +66,7 @@ struct HistoryRangeByKeysSegmentQuery {
         auto ii_reader = entity_.inverted_index.kv_segment_reader<RawDecoder<Bytes>>();
         auto begin_it = offset ? ii_reader.seek(*offset) : ii_reader.end();
 
-        auto lookup_kv_pair_func = [query = *this, timestamp, ascending](std::pair<Bytes&, elias_fano::EliasFanoList32&> ii_entry) {
+        auto lookup_kv_pair_func = [query = *this, timestamp, ascending](std::pair<Bytes&, InvertedIndexTimestampList&>&& ii_entry) {
             return query.lookup_kv_pair(timestamp, ascending, std::move(ii_entry.first), ii_entry.second);
         };
 
diff --git a/silkworm/db/datastore/snapshots/history_range_in_period_query.hpp b/silkworm/db/datastore/snapshots/history_range_in_period_query.hpp
index 496109a3da..210ec47b3b 100644
--- a/silkworm/db/datastore/snapshots/history_range_in_period_query.hpp
+++ b/silkworm/db/datastore/snapshots/history_range_in_period_query.hpp
@@ -33,7 +33,7 @@ struct HistoryRangeInPeriodSegmentQuery {
         datastore::TimestampRange ts_range,
         bool ascending,
         Bytes key_data,
-        const elias_fano::EliasFanoList32& key_timestamps) const {
+        const InvertedIndexTimestampList& key_timestamps) const {
         SILKWORM_ASSERT(ascending);  // descending is not implemented
 
         // find the first key timestamp within the ts_range
@@ -58,7 +58,7 @@ struct HistoryRangeInPeriodSegmentQuery {
     auto exec(datastore::TimestampRange ts_range, bool ascending) {
         SILKWORM_ASSERT(ascending);  // descending is not implemented
 
-        auto lookup_kv_pair_func = [query = *this, ts_range, ascending](std::pair<Bytes&, elias_fano::EliasFanoList32&> ii_entry) {
+        auto lookup_kv_pair_func = [query = *this, ts_range, ascending](std::pair<Bytes&, InvertedIndexTimestampList&>&& ii_entry) {
             return query.lookup_kv_pair(ts_range, ascending, std::move(ii_entry.first), ii_entry.second);
         };
 
diff --git a/silkworm/db/datastore/snapshots/inverted_index.hpp b/silkworm/db/datastore/snapshots/inverted_index.hpp
index d9df236989..7d594aee11 100644
--- a/silkworm/db/datastore/snapshots/inverted_index.hpp
+++ b/silkworm/db/datastore/snapshots/inverted_index.hpp
@@ -3,7 +3,7 @@
 
 #pragma once
 
-#include "elias_fano/elias_fano_decoder.hpp"
+#include "inverted_index_ts_list_codec.hpp"
 #include "rec_split/accessor_index.hpp"
 #include "segment/kv_segment_reader.hpp"
 
@@ -14,8 +14,8 @@ struct InvertedIndex {
     const rec_split::AccessorIndex& accessor_index;
 
     template <DecoderConcept TKeyDecoder>
-    segment::KVSegmentReader<TKeyDecoder, elias_fano::EliasFanoDecoder> kv_segment_reader() {
-        return segment::KVSegmentReader<TKeyDecoder, elias_fano::EliasFanoDecoder>{kv_segment};
+    segment::KVSegmentReader<TKeyDecoder, InvertedIndexTimestampListDecoder> kv_segment_reader() {
+        return segment::KVSegmentReader<TKeyDecoder, InvertedIndexTimestampListDecoder>{kv_segment};
     }
 };
 
diff --git a/silkworm/db/datastore/snapshots/inverted_index_find_by_key_segment_query.hpp b/silkworm/db/datastore/snapshots/inverted_index_find_by_key_segment_query.hpp
index b9089fb84d..e8b2d94fa8 100644
--- a/silkworm/db/datastore/snapshots/inverted_index_find_by_key_segment_query.hpp
+++ b/silkworm/db/datastore/snapshots/inverted_index_find_by_key_segment_query.hpp
@@ -16,8 +16,8 @@
 
 namespace silkworm::snapshots {
 
-inline auto timestamp_range_filter(elias_fano::EliasFanoList32 list, datastore::TimestampRange ts_range, bool ascending) {
-    using Iterator = elias_fano::EliasFanoList32::Iterator;
+inline auto timestamp_range_filter(InvertedIndexTimestampList list, datastore::TimestampRange ts_range, bool ascending) {
+    using Iterator = InvertedIndexTimestampList::Iterator;
 
     size_t start = 0;
     size_t end = list.size();
@@ -31,7 +31,7 @@ inline auto timestamp_range_filter(elias_fano::EliasFanoList32 list, datastore::
         start = end;
     }
 
-    auto range_from_list = [ts_range, ascending, start, end](const elias_fano::EliasFanoList32& list1) {
+    auto range_from_list = [ts_range, ascending, start, end](const InvertedIndexTimestampList& list1) {
         auto list_range = [&list1, start, end]() { return std::ranges::subrange{Iterator{list1, start}, Iterator{list1, end}}; };
         return silkworm::views::if_view(
             ascending,
@@ -56,21 +56,21 @@ struct InvertedIndexFindByKeySegmentQuery {
 
     using Key = decltype(TKeyEncoder::value);
 
-    std::optional<elias_fano::EliasFanoList32> exec(Key key) {
+    std::optional<InvertedIndexTimestampList> exec(Key key) {
         TKeyEncoder key_encoder;
         key_encoder.value = std::move(key);
         ByteView key_data = key_encoder.encode_word();
         return exec_raw(key_data);
     }
 
-    std::optional<elias_fano::EliasFanoList32> exec_raw(ByteView key_data) {
+    std::optional<InvertedIndexTimestampList> exec_raw(ByteView key_data) {
         auto offset = entity_.accessor_index.lookup_by_key(key_data);
         if (!offset) {
             return std::nullopt;
         }
 
         auto reader = entity_.kv_segment_reader<RawDecoder<Bytes>>();
-        std::optional<std::pair<Bytes, elias_fano::EliasFanoList32>> result = reader.seek_one(*offset);
+        std::optional<std::pair<Bytes, InvertedIndexTimestampList>> result = reader.seek_one(*offset);
 
         // ensure that the found key matches to avoid lookup_by_key false positives
         if (result && (result->first == key_data)) {
@@ -81,7 +81,7 @@ struct InvertedIndexFindByKeySegmentQuery {
     }
 
     auto exec_filter(Key key, datastore::TimestampRange ts_range, bool ascending) {
-        return timestamp_range_filter(exec(std::move(key)).value_or(elias_fano::EliasFanoList32::empty_list()), ts_range, ascending);
+        return timestamp_range_filter(exec(std::move(key)).value_or(InvertedIndexTimestampList{}), ts_range, ascending);
     }
 
   private:
diff --git a/silkworm/db/datastore/snapshots/inverted_index_seek_query.hpp b/silkworm/db/datastore/snapshots/inverted_index_seek_query.hpp
index 67bc7ab9d9..e4ef080fe0 100644
--- a/silkworm/db/datastore/snapshots/inverted_index_seek_query.hpp
+++ b/silkworm/db/datastore/snapshots/inverted_index_seek_query.hpp
@@ -35,7 +35,7 @@ struct InvertedIndexSeekSegmentQuery {
             return std::nullopt;
         }
 
-        elias_fano::EliasFanoList32& list = *list_opt;
+        InvertedIndexTimestampList& list = *list_opt;
         const auto seek_result = list.seek(timestamp);
         if (!seek_result) {
             return std::nullopt;
diff --git a/silkworm/db/datastore/snapshots/inverted_index_ts_list.cpp b/silkworm/db/datastore/snapshots/inverted_index_ts_list.cpp
new file mode 100644
index 0000000000..79304ab4ce
--- /dev/null
+++ b/silkworm/db/datastore/snapshots/inverted_index_ts_list.cpp
@@ -0,0 +1,75 @@
+// Copyright 2025 The Silkworm Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "inverted_index_ts_list.hpp"
+
+namespace silkworm::snapshots {
+
+size_t InvertedIndexTimestampList::size() const {
+    switch (static_cast<Alternative>(list_.index())) {
+        case Alternative::kEmpty:
+            return 0;
+        case Alternative::kEliasFano:
+            return ef_list().size();
+        case Alternative::kSimple:
+            return simple_list().size;
+        default:
+            SILKWORM_ASSERT(false);
+    }
+}
+
+InvertedIndexTimestampList::value_type InvertedIndexTimestampList::SimpleList::at(size_t i) const {
+    auto values = reinterpret_cast<const uint32_t*>(ByteView{data}.data() + offset);
+    return base_timestamp + values[i];
+}
+
+InvertedIndexTimestampList::value_type InvertedIndexTimestampList::at(size_t i) const {
+    switch (static_cast<Alternative>(list_.index())) {
+        case Alternative::kEmpty:
+            SILKWORM_ASSERT(false);
+            return 0;
+        case Alternative::kEliasFano:
+            return ef_list().at(i);
+        case Alternative::kSimple:
+            return simple_list().at(i);
+        default:
+            SILKWORM_ASSERT(false);
+            return 0;
+    }
+}
+
+std::optional<InvertedIndexTimestampList::SeekResult> InvertedIndexTimestampList::SimpleList::seek(value_type value, bool reverse) const {
+    const auto& list = *this;
+    if (!reverse) {
+        for (size_t i = 0; i < list.size; ++i) {
+            value_type current_value = at(i);
+            if (current_value >= value)
+                return SeekResult{i, current_value};
+        }
+        return std::nullopt;
+    } else {  // NOLINT(readability-else-after-return)
+        for (size_t j = 0; j < list.size; ++j) {
+            size_t i = list.size - 1 - j;
+            value_type current_value = at(i);
+            if (current_value <= value)
+                return SeekResult{i, current_value};
+        }
+        return std::nullopt;
+    }
+}
+
+std::optional<InvertedIndexTimestampList::SeekResult> InvertedIndexTimestampList::seek(value_type value, bool reverse) const {
+    switch (static_cast<Alternative>(list_.index())) {
+        case Alternative::kEmpty:
+            return std::nullopt;
+        case Alternative::kEliasFano:
+            return ef_list().seek(value, reverse);
+        case Alternative::kSimple:
+            return simple_list().seek(value, reverse);
+        default:
+            SILKWORM_ASSERT(false);
+            return std::nullopt;
+    }
+}
+
+}  // namespace silkworm::snapshots
diff --git a/silkworm/db/datastore/snapshots/inverted_index_ts_list.hpp b/silkworm/db/datastore/snapshots/inverted_index_ts_list.hpp
new file mode 100644
index 0000000000..786f8adbb2
--- /dev/null
+++ b/silkworm/db/datastore/snapshots/inverted_index_ts_list.hpp
@@ -0,0 +1,83 @@
+// Copyright 2025 The Silkworm Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include <optional>
+#include <span>
+#include <utility>
+#include <variant>
+
+#include <silkworm/core/common/assert.hpp>
+#include <silkworm/core/common/bytes.hpp>
+
+#include "../common/timestamp.hpp"
+#include "elias_fano/elias_fano_list.hpp"
+
+namespace silkworm::snapshots {
+
+class InvertedIndexTimestampList {
+  public:
+    using value_type = datastore::Timestamp;
+
+    InvertedIndexTimestampList() : list_{std::monostate{}} {}
+    explicit InvertedIndexTimestampList(elias_fano::EliasFanoList32 list) : list_{std::move(list)} {}
+
+    using SeekResult = std::pair<size_t, value_type>;
+
+    struct SimpleList {
+        BytesOrByteView data;
+        value_type base_timestamp;
+        size_t offset;
+        size_t size;
+
+        value_type at(size_t i) const;
+        value_type operator[](size_t i) const { return at(i); }
+        std::optional<SeekResult> seek(value_type value, bool reverse) const;
+    };
+
+    InvertedIndexTimestampList(BytesOrByteView data, value_type base_timestamp, size_t offset, size_t size)
+        : list_{
+              SimpleList{
+                  std::move(data),
+                  base_timestamp,
+                  offset,
+                  size,
+              },
+          } {
+        SILKWORM_ASSERT(ByteView{data}.size() >= offset + size * sizeof(uint32_t));
+    }
+
+    size_t size() const;
+
+    value_type at(size_t i) const;
+    value_type operator[](size_t i) const { return at(i); }
+
+    //! Find the first index where at(i) >= value if reverse = false.
+    //! Find the last index where at(i) <= value if reverse = true.
+    //! \return (i, value) or nullopt if not found
+    std::optional<SeekResult> seek(value_type value, bool reverse = false) const;
+
+    using Iterator = ListIterator<InvertedIndexTimestampList>;
+    Iterator begin() const { return Iterator{*this, 0}; }
+    Iterator end() const { return Iterator{*this, size()}; }
+
+  private:
+    enum class Alternative : size_t {
+        kEmpty,
+        kEliasFano,
+        kSimple,
+    };
+
+    const elias_fano::EliasFanoList32& ef_list() const {
+        return std::get<elias_fano::EliasFanoList32>(list_);
+    }
+
+    const SimpleList& simple_list() const {
+        return std::get<SimpleList>(list_);
+    }
+
+    std::variant<std::monostate, elias_fano::EliasFanoList32, SimpleList> list_;
+};
+
+}  // namespace silkworm::snapshots
diff --git a/silkworm/db/datastore/snapshots/inverted_index_ts_list_codec.cpp b/silkworm/db/datastore/snapshots/inverted_index_ts_list_codec.cpp
new file mode 100644
index 0000000000..bf48fe0b60
--- /dev/null
+++ b/silkworm/db/datastore/snapshots/inverted_index_ts_list_codec.cpp
@@ -0,0 +1,33 @@
+// Copyright 2025 The Silkworm Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "inverted_index_ts_list_codec.hpp"
+
+namespace silkworm::snapshots {
+
+static_assert(snapshots::DecoderConcept<InvertedIndexTimestampListDecoder>);
+
+static constexpr uint8_t kNonV1EncodingMask = 0b10000000;
+static constexpr uint8_t kSimpleEncodingSizeMask = 0b00001111;
+static constexpr uint8_t kSimpleEncodingMaxByte = 0b10001111;
+
+void InvertedIndexTimestampListDecoder::decode_word(Word& word) {
+    ByteView data{word};
+    if (data.empty()) {
+        value = {};
+    } else if ((data[0] & kNonV1EncodingMask) == 0) {
+        auto list = elias_fano::EliasFanoList32::from_encoded_data(std::move(word));
+        value = InvertedIndexTimestampList{std::move(list)};
+    } else if ((data[0] | kSimpleEncodingSizeMask) == kSimpleEncodingMaxByte) {
+        value = InvertedIndexTimestampList{
+            std::move(word),
+            base_timestamp,
+            1,
+            static_cast<size_t>(data[0] & kSimpleEncodingSizeMask) + 1,
+        };
+    } else {
+        throw std::runtime_error{"InvertedIndexTimestampListDecoder: unsupported encoding"};
+    }
+}
+
+}  // namespace silkworm::snapshots
diff --git a/silkworm/db/datastore/snapshots/inverted_index_ts_list_codec.hpp b/silkworm/db/datastore/snapshots/inverted_index_ts_list_codec.hpp
new file mode 100644
index 0000000000..e5f894054f
--- /dev/null
+++ b/silkworm/db/datastore/snapshots/inverted_index_ts_list_codec.hpp
@@ -0,0 +1,25 @@
+// Copyright 2025 The Silkworm Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include "../common/step_timestamp_converter.hpp"
+#include "common/codec.hpp"
+#include "common/snapshot_path.hpp"
+#include "inverted_index_ts_list.hpp"
+
+namespace silkworm::snapshots {
+
+struct InvertedIndexTimestampListDecoder : public snapshots::Decoder {
+    InvertedIndexTimestampList value;
+    datastore::Timestamp base_timestamp{0};
+
+    ~InvertedIndexTimestampListDecoder() override = default;
+    void decode_word(Word& word) override;
+
+    void decode_word_with_metadata(const SnapshotPath& path, const datastore::StepToTimestampConverter& step_converter) override {
+        base_timestamp = step_converter.timestamp_from_step(path.step_range().start);
+    }
+};
+
+}  // namespace silkworm::snapshots
diff --git a/silkworm/db/datastore/snapshots/schema.hpp b/silkworm/db/datastore/snapshots/schema.hpp
index db75517884..ad8ce12e01 100644
--- a/silkworm/db/datastore/snapshots/schema.hpp
+++ b/silkworm/db/datastore/snapshots/schema.hpp
@@ -7,6 +7,7 @@
 #include <vector>
 
 #include "../common/entity_name.hpp"
+#include "../common/step_timestamp_converter.hpp"
 #include "common/snapshot_path.hpp"
 #include "segment/seg/compression_kind.hpp"
 
@@ -187,10 +188,17 @@ class Schema {
             return *this;
         }
 
+        RepositoryDef& step_size(size_t value) {
+            step_size_ = value;
+            return *this;
+        }
+
         const datastore::EntityMap<std::shared_ptr<EntityDef>>& entities() const { return entity_defs_; }
         std::vector<std::string> file_extensions() const;
         std::optional<std::pair<datastore::EntityName, datastore::EntityName>> entity_name_by_path(const SnapshotPath& path) const;
         const std::string& index_salt_file_name() const { return index_salt_file_name_.value(); }
+        size_t step_size() const { return step_size_.value(); }
+        datastore::StepToTimestampConverter make_step_converter() const { return datastore::StepToTimestampConverter{step_size()}; }
 
       private:
         friend DomainDef;
@@ -204,6 +212,7 @@ class Schema {
 
         datastore::EntityMap<std::shared_ptr<EntityDef>> entity_defs_;
         std::optional<std::string> index_salt_file_name_;
+        std::optional<size_t> step_size_;
     };
 
     RepositoryDef& repository(datastore::EntityName name) {
diff --git a/silkworm/db/datastore/snapshots/segment/kv_segment_reader.cpp b/silkworm/db/datastore/snapshots/segment/kv_segment_reader.cpp
index 6b574435bd..6b04ff74c7 100644
--- a/silkworm/db/datastore/snapshots/segment/kv_segment_reader.cpp
+++ b/silkworm/db/datastore/snapshots/segment/kv_segment_reader.cpp
@@ -11,9 +11,11 @@ namespace silkworm::snapshots::segment {
 
 KVSegmentFileReader::KVSegmentFileReader(
     SnapshotPath path,
+    datastore::StepToTimestampConverter step_converter,
     seg::CompressionKind compression_kind,
     std::optional<MemoryMappedRegion> segment_region)
     : path_(std::move(path)),
+      step_converter_{std::move(step_converter)},
       decompressor_{path_.path(), segment_region, compression_kind} {
 }
 
@@ -33,9 +35,12 @@ KVSegmentFileReader::Iterator& KVSegmentFileReader::Iterator::operator++() {
 
         if (has_next) {
             if (decoder) {
+                if (path_) {
+                    decoder->decode_word_with_metadata(*path_, step_converter_);
+                }
                 decoder->decode_word(*it_);
                 if (path_) {
-                    decoder->check_sanity_with_metadata(*path_);
+                    decoder->check_sanity_with_metadata(*path_, step_converter_);
                 }
             }
         } else {
@@ -75,23 +80,25 @@ KVSegmentFileReader::Iterator KVSegmentFileReader::begin(std::shared_ptr<Decoder
     }
 
     if (key_decoder) {
+        key_decoder->decode_word_with_metadata(path_, step_converter_);
         key_decoder->decode_word(*it);
-        key_decoder->check_sanity_with_metadata(path_);
+        key_decoder->check_sanity_with_metadata(path_, step_converter_);
     }
 
     if (value_decoder) {
         ++it;
+        value_decoder->decode_word_with_metadata(path_, step_converter_);
         value_decoder->decode_word(*it);
-        value_decoder->check_sanity_with_metadata(path_);
+        value_decoder->check_sanity_with_metadata(path_, step_converter_);
     } else {
         it.skip();
     }
 
-    return KVSegmentFileReader::Iterator{std::move(it), std::move(key_decoder), std::move(value_decoder), path()};
+    return KVSegmentFileReader::Iterator{std::move(it), std::move(key_decoder), std::move(value_decoder), path(), step_converter_};
 }
 
 KVSegmentFileReader::Iterator KVSegmentFileReader::end() const {
-    return KVSegmentFileReader::Iterator{decompressor_.end(), {}, {}, path()};
+    return KVSegmentFileReader::Iterator{decompressor_.end(), {}, {}, path(), step_converter_};
 }
 
 KVSegmentFileReader::Iterator KVSegmentFileReader::seek(
@@ -109,23 +116,25 @@ KVSegmentFileReader::Iterator KVSegmentFileReader::seek(
     }
 
     if (key_decoder) {
+        key_decoder->decode_word_with_metadata(path_, step_converter_);
         try {
             key_decoder->decode_word(*it);
         } catch (...) {
             return end();
         }
-        key_decoder->check_sanity_with_metadata(path_);
+        key_decoder->check_sanity_with_metadata(path_, step_converter_);
     }
 
     if (value_decoder) {
         ++it;
+        value_decoder->decode_word_with_metadata(path_, step_converter_);
         value_decoder->decode_word(*it);
-        value_decoder->check_sanity_with_metadata(path_);
+        value_decoder->check_sanity_with_metadata(path_, step_converter_);
     } else {
         it.skip();
     }
 
-    return KVSegmentFileReader::Iterator{std::move(it), std::move(key_decoder), std::move(value_decoder), path()};
+    return KVSegmentFileReader::Iterator{std::move(it), std::move(key_decoder), std::move(value_decoder), path(), step_converter_};
 }
 
 }  // namespace silkworm::snapshots::segment
diff --git a/silkworm/db/datastore/snapshots/segment/kv_segment_reader.hpp b/silkworm/db/datastore/snapshots/segment/kv_segment_reader.hpp
index 0aa123a1d2..b70c1a22c9 100644
--- a/silkworm/db/datastore/snapshots/segment/kv_segment_reader.hpp
+++ b/silkworm/db/datastore/snapshots/segment/kv_segment_reader.hpp
@@ -18,6 +18,7 @@
 #include <silkworm/infra/common/memory_mapped_file.hpp>
 #include <silkworm/infra/common/os.hpp>
 
+#include "../../common/step_timestamp_converter.hpp"
 #include "../common/codec.hpp"
 #include "../common/snapshot_path.hpp"
 #include "seg/decompressor.hpp"
@@ -38,15 +39,18 @@ class KVSegmentFileReader {
             seg::Decompressor::Iterator it,
             std::shared_ptr<Decoder> key_decoder,
             std::shared_ptr<Decoder> value_decoder,
-            SnapshotPath path)
+            SnapshotPath path,
+            datastore::StepToTimestampConverter step_converter)
             : it_(std::move(it)),
               decoders_(std::move(key_decoder), std::move(value_decoder)),
-              path_(std::move(path)) {}
+              path_(std::move(path)),
+              step_converter_(std::move(step_converter)) {}
 
         Iterator()
             : it_{seg::Decompressor::Iterator::make_end()},
               decoders_{{}, {}},
-              path_{std::nullopt} {}
+              path_{std::nullopt},
+              step_converter_{} {}
 
         value_type operator*() const { return decoders_; }
         const value_type* operator->() const { return &decoders_; }
@@ -63,6 +67,7 @@ class KVSegmentFileReader {
         seg::Decompressor::Iterator it_;
         value_type decoders_;
         std::optional<SnapshotPath> path_;
+        datastore::StepToTimestampConverter step_converter_;
     };
 
     static_assert(std::input_iterator<Iterator>);
@@ -72,6 +77,7 @@ class KVSegmentFileReader {
 
     explicit KVSegmentFileReader(
         SnapshotPath path,
+        datastore::StepToTimestampConverter step_converter,
         seg::CompressionKind compression_kind,
         std::optional<MemoryMappedRegion> segment_region = std::nullopt);
 
@@ -100,6 +106,7 @@ class KVSegmentFileReader {
   private:
     //! The path of the segment file for this snapshot
     SnapshotPath path_;
+    datastore::StepToTimestampConverter step_converter_;
 
     seg::Decompressor decompressor_;
 };
diff --git a/silkworm/db/datastore/snapshots/segment/kv_segment_test.cpp b/silkworm/db/datastore/snapshots/segment/kv_segment_test.cpp
index ac7be0ab9e..fe08228b34 100644
--- a/silkworm/db/datastore/snapshots/segment/kv_segment_test.cpp
+++ b/silkworm/db/datastore/snapshots/segment/kv_segment_test.cpp
@@ -64,7 +64,7 @@ TEST_CASE("KVSegmentFile") {
     }
     KVSegmentFileWriter::flush(std::move(file_writer));
 
-    KVSegmentFileReader file_reader{path, kCompressionKind};
+    KVSegmentFileReader file_reader{path, {}, kCompressionKind};
     KVSegmentReader<StringCodec, CharCodec> reader{file_reader};
     for (std::pair<std::string&, char&> entry : reader) {
         CHECK(entry.first == entries[0].first);
diff --git a/silkworm/db/datastore/snapshots/segment/segment_reader.cpp b/silkworm/db/datastore/snapshots/segment/segment_reader.cpp
index de196bb88a..a56352c04c 100644
--- a/silkworm/db/datastore/snapshots/segment/segment_reader.cpp
+++ b/silkworm/db/datastore/snapshots/segment/segment_reader.cpp
@@ -12,9 +12,11 @@ namespace silkworm::snapshots::segment {
 
 SegmentFileReader::SegmentFileReader(
     SnapshotPath path,
+    datastore::StepToTimestampConverter step_converter,
     std::optional<MemoryMappedRegion> segment_region,
     bool is_compressed)
     : path_(std::move(path)),
+      step_converter_{std::move(step_converter)},
       decompressor_{
           path_.path(),
           segment_region,
@@ -31,9 +33,12 @@ SegmentFileReader::Iterator& SegmentFileReader::Iterator::operator++() {
     ++it_;
 
     if (has_next) {
+        if (path_) {
+            decoder_->decode_word_with_metadata(*path_, step_converter_);
+        }
         decoder_->decode_word(*it_);
         if (path_) {
-            decoder_->check_sanity_with_metadata(*path_);
+            decoder_->check_sanity_with_metadata(*path_, step_converter_);
         }
     } else {
         decoder_.reset();
@@ -62,13 +67,14 @@ SegmentFileReader::Iterator SegmentFileReader::begin(std::shared_ptr<Decoder> de
     if (it == decompressor_.end()) {
         return end();
     }
+    decoder->decode_word_with_metadata(path_, step_converter_);
     decoder->decode_word(*it);
-    decoder->check_sanity_with_metadata(path_);
-    return SegmentFileReader::Iterator{std::move(it), std::move(decoder), path()};
+    decoder->check_sanity_with_metadata(path_, step_converter_);
+    return SegmentFileReader::Iterator{std::move(it), std::move(decoder), path(), step_converter_};
 }
 
 SegmentFileReader::Iterator SegmentFileReader::end() const {
-    return SegmentFileReader::Iterator{decompressor_.end(), {}, path()};
+    return SegmentFileReader::Iterator{decompressor_.end(), {}, path(), step_converter_};
 }
 
 SegmentFileReader::Iterator SegmentFileReader::seek(
@@ -79,13 +85,14 @@ SegmentFileReader::Iterator SegmentFileReader::seek(
     if (it == decompressor_.end()) {
         return end();
     }
+    decoder->decode_word_with_metadata(path_, step_converter_);
     try {
         decoder->decode_word(*it);
     } catch (...) {
         return end();
     }
-    decoder->check_sanity_with_metadata(path_);
-    return SegmentFileReader::Iterator{std::move(it), std::move(decoder), path()};
+    decoder->check_sanity_with_metadata(path_, step_converter_);
+    return SegmentFileReader::Iterator{std::move(it), std::move(decoder), path(), step_converter_};
 }
 
 }  // namespace silkworm::snapshots::segment
diff --git a/silkworm/db/datastore/snapshots/segment/segment_reader.hpp b/silkworm/db/datastore/snapshots/segment/segment_reader.hpp
index 60157d5646..3f0aa3efe1 100644
--- a/silkworm/db/datastore/snapshots/segment/segment_reader.hpp
+++ b/silkworm/db/datastore/snapshots/segment/segment_reader.hpp
@@ -18,6 +18,7 @@
 #include <silkworm/infra/common/memory_mapped_file.hpp>
 #include <silkworm/infra/common/os.hpp>
 
+#include "../../common/step_timestamp_converter.hpp"
 #include "../common/codec.hpp"
 #include "../common/snapshot_path.hpp"
 #include "../common/util/iterator/iterator_read_into_vector.hpp"
@@ -48,13 +49,18 @@ class SegmentFileReader {
         Iterator(
             seg::Decompressor::Iterator it,
             std::shared_ptr<Decoder> decoder,
-            SnapshotPath path)
-            : it_(std::move(it)), decoder_(std::move(decoder)), path_(std::move(path)) {}
+            SnapshotPath path,
+            datastore::StepToTimestampConverter step_converter)
+            : it_(std::move(it)),
+              decoder_(std::move(decoder)),
+              path_(std::move(path)),
+              step_converter_(std::move(step_converter)) {}
 
         Iterator()
             : it_{seg::Decompressor::Iterator::make_end()},
               decoder_{},
-              path_{std::nullopt} {}
+              path_{std::nullopt},
+              step_converter_{} {}
 
         value_type operator*() const { return decoder_; }
 
@@ -70,6 +76,7 @@ class SegmentFileReader {
         seg::Decompressor::Iterator it_;
         std::shared_ptr<Decoder> decoder_;
         std::optional<SnapshotPath> path_;
+        datastore::StepToTimestampConverter step_converter_;
     };
 
     static_assert(std::input_iterator<Iterator>);
@@ -79,6 +86,7 @@ class SegmentFileReader {
 
     explicit SegmentFileReader(
         SnapshotPath path,
+        datastore::StepToTimestampConverter step_converter,
         std::optional<MemoryMappedRegion> segment_region = std::nullopt,
         bool is_compressed = true);
 
@@ -101,6 +109,7 @@ class SegmentFileReader {
   private:
     //! The path of the segment file for this snapshot
     SnapshotPath path_;
+    datastore::StepToTimestampConverter step_converter_;
 
     seg::Decompressor decompressor_;
 };
diff --git a/silkworm/db/datastore/snapshots/segment/segment_test.cpp b/silkworm/db/datastore/snapshots/segment/segment_test.cpp
index c8e6e50890..85f0c30122 100644
--- a/silkworm/db/datastore/snapshots/segment/segment_test.cpp
+++ b/silkworm/db/datastore/snapshots/segment/segment_test.cpp
@@ -36,7 +36,7 @@ TEST_CASE("SegmentFile") {
     }
     SegmentFileWriter::flush(std::move(file_writer));
 
-    SegmentFileReader file_reader{path};
+    SegmentFileReader file_reader{path, {}};
     SegmentReader<StringCodec> reader{file_reader};
     for (std::string& item : reader) {
         CHECK(item == items[0]);
diff --git a/silkworm/db/datastore/snapshots/snapshot_bundle.cpp b/silkworm/db/datastore/snapshots/snapshot_bundle.cpp
index 412cafec92..26e5b4a98d 100644
--- a/silkworm/db/datastore/snapshots/snapshot_bundle.cpp
+++ b/silkworm/db/datastore/snapshots/snapshot_bundle.cpp
@@ -29,13 +29,14 @@ static datastore::EntityMap<SnapshotPath> make_snapshot_paths(
 static datastore::EntityMap<SegmentFileReader> open_segments(
     const Schema::EntityDef& entity,
     const std::filesystem::path& dir_path,
-    StepRange range) {
+    StepRange range,
+    datastore::StepToTimestampConverter step_converter) {
     datastore::EntityMap<SegmentFileReader> results;
     for (auto& [name, anyDef] : entity.files()) {
         if (anyDef->format() != Schema::SnapshotFileDef::Format::kSegment) continue;
         auto& def = dynamic_cast<const Schema::SegmentDef&>(*anyDef);
         auto path = def.make_path(dir_path, range);
-        results.emplace(name, SegmentFileReader{std::move(path), std::nullopt, def.compression_enabled()});
+        results.emplace(name, SegmentFileReader{std::move(path), step_converter, std::nullopt, def.compression_enabled()});
     }
     return results;
 }
@@ -43,13 +44,14 @@ static datastore::EntityMap<SegmentFileReader> open_segments(
 static datastore::EntityMap<KVSegmentFileReader> open_kv_segments(
     const Schema::EntityDef& entity,
     const std::filesystem::path& dir_path,
-    StepRange range) {
+    StepRange range,
+    datastore::StepToTimestampConverter step_converter) {
     datastore::EntityMap<KVSegmentFileReader> results;
     for (auto& [name, anyDef] : entity.files()) {
         if (anyDef->format() != Schema::SnapshotFileDef::Format::kKVSegment) continue;
         auto& def = dynamic_cast<const Schema::KVSegmentDef&>(*anyDef);
         auto path = def.make_path(dir_path, range);
-        results.emplace(name, KVSegmentFileReader{std::move(path), def.compression_kind()});
+        results.emplace(name, KVSegmentFileReader{std::move(path), step_converter, def.compression_kind()});
     }
     return results;
 }
@@ -97,13 +99,14 @@ SnapshotBundleData open_bundle_data(
     StepRange step_range,
     std::optional<uint32_t> index_salt) {
     SnapshotBundleData data;
+    StepToTimestampConverter step_converter = schema.make_step_converter();
     for (auto& [name, entity_schema_ptr] : schema.entities()) {
         auto& entity_schema = *entity_schema_ptr;
         data.entities.emplace(
             name,
             SnapshotBundleEntityData{
-                open_segments(entity_schema, dir_path, step_range),
-                open_kv_segments(entity_schema, dir_path, step_range),
+                open_segments(entity_schema, dir_path, step_range, step_converter),
+                open_kv_segments(entity_schema, dir_path, step_range, step_converter),
                 open_accessor_indexes(entity_schema, dir_path, step_range),
                 open_existence_indexes(entity_schema, dir_path, step_range, index_salt),
                 open_btree_indexes(entity_schema, dir_path, step_range),
diff --git a/silkworm/db/datastore/snapshots/snapshot_repository.cpp b/silkworm/db/datastore/snapshots/snapshot_repository.cpp
index ecb4f8dea2..8866c6c5d3 100644
--- a/silkworm/db/datastore/snapshots/snapshot_repository.cpp
+++ b/silkworm/db/datastore/snapshots/snapshot_repository.cpp
@@ -25,7 +25,6 @@ SnapshotRepository::SnapshotRepository(
     std::filesystem::path dir_path,
     bool open,
     Schema::RepositoryDef schema,
-    StepToTimestampConverter step_converter,
     std::optional<uint32_t> index_salt,
     std::unique_ptr<IndexBuildersFactory> index_builders_factory,
     std::optional<DomainGetLatestCaches> domain_caches,
@@ -33,7 +32,7 @@ SnapshotRepository::SnapshotRepository(
     : name_(std::move(name)),
       dir_path_(std::move(dir_path)),
       schema_(std::move(schema)),
-      step_converter_(std::move(step_converter)),
+      step_converter_{schema_.make_step_converter()},
       index_salt_(index_salt),
       index_builders_factory_(std::move(index_builders_factory)),
       bundles_(std::make_shared<Bundles>()),
diff --git a/silkworm/db/datastore/snapshots/snapshot_repository.hpp b/silkworm/db/datastore/snapshots/snapshot_repository.hpp
index feeef1fa2f..3053c1793d 100644
--- a/silkworm/db/datastore/snapshots/snapshot_repository.hpp
+++ b/silkworm/db/datastore/snapshots/snapshot_repository.hpp
@@ -43,7 +43,6 @@ class SnapshotRepository : public SnapshotRepositoryROAccess {
         std::filesystem::path dir_path,
         bool open,
         Schema::RepositoryDef schema,
-        datastore::StepToTimestampConverter step_converter,
         std::optional<uint32_t> index_salt,
         std::unique_ptr<IndexBuildersFactory> index_builders_factory,
         std::optional<DomainGetLatestCaches> domain_caches,
diff --git a/silkworm/db/snapshot_recompress.cpp b/silkworm/db/snapshot_recompress.cpp
index 9e06e6f34f..bfc51e8156 100644
--- a/silkworm/db/snapshot_recompress.cpp
+++ b/silkworm/db/snapshot_recompress.cpp
@@ -9,6 +9,7 @@
 #include "blocks/bodies/body_segment.hpp"
 #include "blocks/headers/header_segment.hpp"
 #include "blocks/schema_config.hpp"
+#include "blocks/step_block_num_converter.hpp"
 #include "blocks/transactions/txn_segment.hpp"
 #include "datastore/snapshots/common/snapshot_path.hpp"
 
@@ -27,7 +28,7 @@ void snapshot_file_recompress(const std::filesystem::path& path) {
     auto path_opt = SnapshotPath::parse(path);
     if (!path_opt) throw std::runtime_error{"bad snapshot path"};
 
-    SegmentFileReader file_reader{*path_opt};
+    SegmentFileReader file_reader{*path_opt, db::blocks::kStepToBlockNumConverter};
 
     auto out_path = path;
     out_path.replace_extension("seg2");
diff --git a/silkworm/db/snapshot_test.cpp b/silkworm/db/snapshot_test.cpp
index bccf528b9f..7dfd81a34d 100644
--- a/silkworm/db/snapshot_test.cpp
+++ b/silkworm/db/snapshot_test.cpp
@@ -12,6 +12,7 @@
 #include "blocks/bodies/body_queries.hpp"
 #include "blocks/headers/header_index.hpp"
 #include "blocks/headers/header_queries.hpp"
+#include "blocks/step_block_num_converter.hpp"
 #include "blocks/transactions/txn_index.hpp"
 #include "blocks/transactions/txn_queries.hpp"
 #include "blocks/transactions/txn_segment_word_codec.hpp"
@@ -27,11 +28,12 @@ using namespace rec_split;
 using namespace segment;
 
 static const SnapshotPath kValidHeadersSegmentPath{*SnapshotPath::parse("v1-014500-015000-headers.seg")};
+static constexpr datastore::StepToTimestampConverter kStepConverter = db::blocks::kStepToBlockNumConverter;
 
 TEST_CASE("Snapshot::open", "[silkworm][node][snapshot][snapshot]") {
     TemporaryDirectory tmp_dir;
     test::TemporarySnapshotFile tmp_snapshot_file{tmp_dir.path(), kValidHeadersSegmentPath.filename(), test::SnapshotHeader{}};
-    SegmentFileReader snapshot{tmp_snapshot_file.path()};
+    SegmentFileReader snapshot{tmp_snapshot_file.path(), kStepConverter};
     CHECK(snapshot.path() == tmp_snapshot_file.path());
     CHECK(snapshot.fs_path() == tmp_snapshot_file.path().path());
 }
@@ -39,7 +41,7 @@ TEST_CASE("Snapshot::open", "[silkworm][node][snapshot][snapshot]") {
 TEST_CASE("Snapshot::for_each_item", "[silkworm][node][snapshot][snapshot]") {
     TemporaryDirectory tmp_dir;
     test::HelloWorldSnapshotFile hello_world_snapshot_file{tmp_dir.path(), kValidHeadersSegmentPath.filename()};
-    SegmentFileReader tmp_snapshot{hello_world_snapshot_file.path()};
+    SegmentFileReader tmp_snapshot{hello_world_snapshot_file.path(), kStepConverter};
     CHECK(!tmp_snapshot.empty());
     CHECK(tmp_snapshot.item_count() == 1);
 
@@ -64,7 +66,7 @@ TEST_CASE("HeaderSnapshot::header_by_number OK", "[silkworm][node][snapshot][ind
     header_index.set_base_data_id(header_segment_file.block_num_range().start);
     REQUIRE_NOTHROW(header_index.build());
 
-    SegmentFileReader header_segment{header_segment_path};
+    SegmentFileReader header_segment{header_segment_path, kStepConverter};
 
     AccessorIndex idx_header_hash{header_segment_path.related_path_ext(db::blocks::kIdxExtension)};
     HeaderFindByBlockNumSegmentQuery header_by_number{{header_segment, idx_header_hash}};
@@ -104,7 +106,7 @@ TEST_CASE("BodySnapshot::body_by_number OK", "[silkworm][node][snapshot][index]"
     body_index.set_base_data_id(body_segment_file.block_num_range().start);
     REQUIRE_NOTHROW(body_index.build());
 
-    SegmentFileReader body_segment{body_segment_path};
+    SegmentFileReader body_segment{body_segment_path, kStepConverter};
 
     AccessorIndex idx_body_number{body_segment_path.related_path_ext(db::blocks::kIdxExtension)};
     BodyFindByBlockNumSegmentQuery body_by_number{{body_segment, idx_body_number}};
@@ -130,7 +132,7 @@ TEST_CASE("TransactionSnapshot::txn_by_id OK", "[silkworm][node][snapshot][index
     auto tx_index = TransactionIndex::make(body_segment_path, txn_segment_path);
     CHECK_NOTHROW(tx_index.build());
 
-    SegmentFileReader txn_segment{txn_segment_path};
+    SegmentFileReader txn_segment{txn_segment_path, kStepConverter};
 
     AccessorIndex idx_txn_hash{txn_segment_path.related_path_ext(db::blocks::kIdxExtension)};
     TransactionFindByIdSegmentQuery txn_by_id{{txn_segment, idx_txn_hash}};
@@ -156,7 +158,7 @@ TEST_CASE("TransactionSnapshot::block_num_by_txn_hash OK", "[silkworm][node][sna
     auto tx_index_hash_to_block = TransactionToBlockIndex::make(body_segment_path, txn_segment_path, txn_segment_file.block_num_range().start);
     REQUIRE_NOTHROW(tx_index_hash_to_block.build());
 
-    SegmentFileReader txn_segment{txn_segment_path};
+    SegmentFileReader txn_segment{txn_segment_path, kStepConverter};
 
     AccessorIndex idx_txn_hash{txn_segment_path.related_path_ext(db::blocks::kIdxExtension)};
     TransactionFindByIdSegmentQuery txn_by_id{{txn_segment, idx_txn_hash}};
@@ -194,7 +196,7 @@ TEST_CASE("TransactionSnapshot::txn_range OK", "[silkworm][node][snapshot][index
     auto tx_index = TransactionIndex::make(body_segment_path, txn_segment_path);
     REQUIRE_NOTHROW(tx_index.build());
 
-    SegmentFileReader txn_segment{txn_segment_path};
+    SegmentFileReader txn_segment{txn_segment_path, kStepConverter};
 
     AccessorIndex idx_txn_hash{txn_segment_path.related_path_ext(db::blocks::kIdxExtension)};
     TransactionRangeFromIdSegmentQuery query{{txn_segment, idx_txn_hash}};
@@ -223,7 +225,7 @@ TEST_CASE("TransactionSnapshot::txn_rlp_range OK", "[silkworm][node][snapshot][i
     auto tx_index = TransactionIndex::make(body_segment_path, txn_segment_path);
     REQUIRE_NOTHROW(tx_index.build());
 
-    SegmentFileReader txn_segment{txn_segment_path};
+    SegmentFileReader txn_segment{txn_segment_path, kStepConverter};
 
     AccessorIndex idx_txn_hash{txn_segment_path.related_path_ext(db::blocks::kIdxExtension)};
     TransactionPayloadRlpRangeFromIdSegmentQuery query{{txn_segment, idx_txn_hash}};
diff --git a/silkworm/db/state/schema_config.cpp b/silkworm/db/state/schema_config.cpp
index 8a482ce5a5..33554e92b1 100644
--- a/silkworm/db/state/schema_config.cpp
+++ b/silkworm/db/state/schema_config.cpp
@@ -15,6 +15,7 @@ namespace silkworm::db::state {
 snapshots::Schema::RepositoryDef make_state_repository_schema_latest() {
     snapshots::Schema::RepositoryDef schema;
     schema.index_salt_file_name("salt-state.txt");
+    schema.step_size(kStepSizeForTemporalSnapshots);
 
     schema.domain(kDomainNameAccounts)
         .tag_override(kDomainAccountsTag);
@@ -33,6 +34,7 @@ snapshots::Schema::RepositoryDef make_state_repository_schema_latest() {
 snapshots::Schema::RepositoryDef make_state_repository_schema_historical() {
     snapshots::Schema::RepositoryDef schema;
     schema.index_salt_file_name("salt-state.txt");
+    schema.step_size(kStepSizeForTemporalSnapshots);
 
     schema.history(kDomainNameAccounts)
         .tag_override(kDomainAccountsTag);
@@ -119,7 +121,6 @@ static snapshots::SnapshotRepository make_state_repository(
         std::move(dir_path),
         open,
         schema,
-        kStepToTxnIdConverter,
         index_salt,
         std::make_unique<StateIndexBuildersFactory>(schema),
         make_domain_caches(index_salt),