[vm] Record txn read sets for hot state promotion

Hot state promotions (the `to_make_hot` set in the block epilogue) are
part of consensus, so their inputs must be deterministic. This reworks how
that set is derived.

## Why the old derivation was problematic

Promotions were fed from BlockSTM's read/write summary, which exists for
conflict detection, not for this:

- **Path-dependent.** The summary differs between parallel and sequential
  execution (e.g. aggregator v1 reads served by delta resolution are
  dropped in parallel but kept in sequential), so the same block could
  promote different keys depending on how it happened to execute — a
  divergence in a consensus-agreed artifact.
- **Coupled to gas config.** It was only populated inside the
  `conflict_penalty_window` branch, so promotions silently depended on an
  unrelated block-gas knob.
- **Incomplete write exclusion.** Its write side misses in-place
  delayed-field rewrites, aggregator v1 writes/deltas and module writes, so
  a key the block writes could still be promoted by the epilogue even
  though the write already makes it hot.

## Approach

Record the read set at the VM boundary, where it is a pure function of the
transaction and the pre-state — identical across parallel and sequential
execution and independent of gas config:

- `StorageAdapter` records resource, resource-group, table-item, aggregator
  v1 and config reads. Respawned sessions share one recorder so all of a
  transaction's sessions accumulate into a single set.
- `ReadRecordingCodeStorage` wraps the code storage and records module
  fetches. It sits above the global module cache, so a module is recorded
  whether served from that cache, the per-block cache or storage.
- A transaction's output carries the data and module keys as two
  collections; they are disjoint by construction, so consumers iterate a
  chained view and nothing pays to merge them.
- Written keys are enumerated directly from the change set, covering every
  write kind the conflict summary missed.

## Worth calling out

The new set intentionally differs from the old one — e.g. `exists<T>` now
loads the resource and counts as a read — so it ships behind the existing
hotness feature flag, which the mixed-version forge suites keep off (see
the preceding forge commit) to avoid old/new nodes disagreeing on
transaction output. Adds a per-block promotions histogram and tests
covering sequential/parallel parity, each read and write kind, and discard
handling. Follow-ups: read-kind/hotness tagging and an on-chain,
byte-based promotion cap.

## Performance

The recording runs unconditionally, so its per-transaction cost has to
stay small. Even a trivial transaction fetches 10+ framework modules and
re-reads the same resources across its sessions, so the hot paths are
engineered around interning, allocation and clone traffic:

- A module's `StateKey` is a pure function of `(address, name)`, yet
  interning one goes through the global, lock-guarded key registry. A
  bounded per-thread memo serves repeats, and module reads are recorded
  directly as memoized `StateKey`s: a transaction allocates no owned
  module ids, and extraction moves the set out instead of interning.
- The interpreter fetches the same module many times in a row. A
  last-recorded fast path (a reused string buffer, so module switches
  allocate nothing) skips the bookkeeping for such runs, and the
  recording wrapper's delegators are `#[inline]` so wrapping does not
  deoptimize the module fetch path.
- All recording sets are keyed by `StateKey` — which carries a
  precomputed 32-byte hash — under FxHash (via the maintained
  `rustc-hash`), pre-sized past the typical framework-module count, and
  keys are cloned only on first sighting; repeats, e.g. framework modules
  touched by every transaction, are the common case.

Measured on the single-node benchmark with interleaved paired runs,
together with the preceding `to_make_hot` ordering commit: ~2-2.5% net
mean TPS cost across representative workloads, and ~0.3-0.4us per
transaction on the sequential move-e2e micros. An ablation build with
recording no-oped attributes the entire cost to the recording machinery
itself (diffuse per-transaction set and clone work rather than any single
hot symbol) and confirms the accumulator commit offsets part of it.
Candidate follow-ups if more needs to come back: pointer-hashed recorder
sets (`StateKey` equality is already pointer identity), pooling the
per-transaction sets, and an indexed key registry to cut refcount traffic
on hot shared keys.

Author: wqfish

Cargo.lock

@@ -29,6 +29,7 @@ move-core-types = { workspace = true }
 move-vm-runtime = { workspace = true }
 move-vm-types = { workspace = true }
 rand = { workspace = true }
+rustc-hash = { workspace = true }
 serde = { workspace = true }
 triomphe = { workspace = true }
 

aptos-move/aptos-vm-types/Cargo.toml

@@ -3,6 +3,7 @@
 
 pub mod code_storage;
 pub mod module_storage;
+pub mod read_recording;
 
 mod state_view_adapter;
 pub use state_view_adapter::{AptosCodeStorageAdapter, AsAptosCodeStorage};

aptos-move/aptos-vm-types/src/module_and_script_storage/mod.rs

@@ -0,0 +1,257 @@
+// Copyright (c) Aptos Foundation
+// Licensed pursuant to the Innovation-Enabling Source Code License, available at https://github.com/aptos-labs/aptos-core/blob/main/LICENSE
+
+#![allow(clippy::duplicated_attributes)]
+
+use crate::{
+    module_and_script_storage::module_storage::AptosModuleStorage,
+    resolver::{ambassador_impl_BlockSynchronizationKillSwitch, BlockSynchronizationKillSwitch},
+};
+use ambassador::delegate_to_methods;
+use aptos_types::state_store::{state_key::StateKey, state_value::StateValueMetadata};
+use bytes::Bytes;
+use move_binary_format::{
+    errors::{PartialVMResult, VMResult},
+    file_format::CompiledScript,
+    CompiledModule,
+};
+use move_core_types::{
+    account_address::AccountAddress,
+    identifier::{IdentStr, Identifier},
+    language_storage::ModuleId,
+};
+use move_vm_runtime::{
+    ambassador_impl_LayoutCache, ambassador_impl_WithRuntimeEnvironment, LayoutCache,
+    LayoutCacheEntry, Module, ModuleStorage, RuntimeEnvironment, Script, StructKey,
+    WithRuntimeEnvironment,
+};
+use move_vm_types::code::{ambassador_impl_ScriptCache, Code, ScriptCache};
+use rustc_hash::{FxHashMap, FxHashSet};
+use std::{cell::RefCell, sync::Arc};
+
+thread_local! {
+    /// A module's `StateKey` is a pure function of its `(address, name)` and never changes, yet
+    /// interning one goes through the global, lock-guarded `StateKey` registry. Worker threads
+    /// re-read the same modules on every transaction, so memoize the interned keys per thread to
+    /// keep that lock off the extraction path. Bounded so an unbounded module working set can't
+    /// grow it without limit; the hot (framework) modules are seen first and stay cached.
+    static MODULE_STATE_KEYS: RefCell<FxHashMap<AccountAddress, FxHashMap<Identifier, StateKey>>> =
+        RefCell::new(FxHashMap::default());
+}
+
+/// Cap on distinct addresses memoized per thread, bounding the memo under a very large or
+/// adversarial module working set.
+const MODULE_STATE_KEY_CACHE_MAX_ADDRESSES: usize = 1 << 13;
+
+/// Interns a module `StateKey`, serving repeats from the per-thread memo so that the common case
+/// (a module already seen by this thread) avoids the global registry lock entirely.
+fn interned_module_state_key(address: &AccountAddress, name: &IdentStr) -> StateKey {
+    MODULE_STATE_KEYS.with(|cache| {
+        let mut cache = cache.borrow_mut();
+        if let Some(key) = cache.get(address).and_then(|names| names.get(name)) {
+            return key.clone();
+        }
+        let key = StateKey::module(address, name);
+        // Only new addresses are gated by the cap; an already-cached address keeps accumulating
+        // its modules so it is never left half-populated.
+        if cache.len() < MODULE_STATE_KEY_CACHE_MAX_ADDRESSES || cache.contains_key(address) {
+            cache
+                .entry(*address)
+                .or_default()
+                .insert(name.to_owned(), key.clone());
+        }
+        key
+    })
+}
+
+/// Wraps a code storage and records every module the VM fetches through it, so that module
+/// accesses become part of the transaction's observed read set (the basis for hot state
+/// promotion).
+///
+/// Recorded reads are kept directly as interned `StateKey`s served from the per-thread memo,
+/// so steady state a transaction's recording performs no allocation beyond growing its key
+/// set: no owned module ids, and no interning pass at extraction.
+///
+/// Scripts are not state items, so script cache accesses are not recorded.
+pub struct ReadRecordingCodeStorage<'a, C> {
+    code_storage: &'a C,
+    module_reads: RefCell<FxHashSet<StateKey>>,
+    /// The previously recorded `(address, name)`. The interpreter fetches the same module many
+    /// times in a row, so this lets a burst of accesses skip the memo and set lookups. A
+    /// `String` buffer rather than an `Identifier` so updating it on a module switch reuses
+    /// the allocation instead of making a fresh one.
+    last_recorded: RefCell<(AccountAddress, String)>,
+}
+
+impl<'a, C> ReadRecordingCodeStorage<'a, C> {
+    pub fn new(code_storage: &'a C) -> Self {
+        Self {
+            code_storage,
+            // Even a trivial transaction touches 10+ framework modules through its prologue
+            // and epilogue, so start with room for the typical count and skip the rehashes.
+            module_reads: RefCell::new(FxHashSet::with_capacity_and_hasher(24, Default::default())),
+            // Module names are never empty, so an empty name means "nothing recorded yet".
+            last_recorded: RefCell::new((AccountAddress::ZERO, String::new())),
+        }
+    }
+
+    /// Returns the state keys of modules fetched so far, deduplicated by key.
+    pub fn into_recorded_reads(self) -> FxHashSet<StateKey> {
+        self.module_reads.into_inner()
+    }
+
+    #[inline]
+    fn record(&self, address: &AccountAddress, module_name: &IdentStr) {
+        {
+            // Fast path: a run of accesses to the same module needs no further work. Only an
+            // exact (address, name) match is skipped, so the recorded set is identical either
+            // way.
+            let last = self.last_recorded.borrow();
+            if last.0 == *address && last.1.as_str() == module_name.as_str() {
+                return;
+            }
+        }
+        let key = interned_module_state_key(address, module_name);
+        self.module_reads.borrow_mut().insert(key);
+        let mut last = self.last_recorded.borrow_mut();
+        last.0 = *address;
+        last.1.clear();
+        last.1.push_str(module_name.as_str());
+    }
+}
+
+#[delegate_to_methods]
+#[delegate(
+    WithRuntimeEnvironment,
+    target_ref = "inner",
+    where = "C: WithRuntimeEnvironment"
+)]
+#[delegate(LayoutCache, target_ref = "inner", where = "C: LayoutCache")]
+#[delegate(
+    BlockSynchronizationKillSwitch,
+    target_ref = "inner",
+    where = "C: BlockSynchronizationKillSwitch"
+)]
+impl<C> ReadRecordingCodeStorage<'_, C> {
+    /// Returns the wrapped code storage.
+    fn inner(&self) -> &C {
+        self.code_storage
+    }
+}
+
+impl<C: ModuleStorage> ModuleStorage for ReadRecordingCodeStorage<'_, C> {
+    #[inline]
+    fn unmetered_check_module_exists(
+        &self,
+        address: &AccountAddress,
+        module_name: &IdentStr,
+    ) -> VMResult<bool> {
+        self.record(address, module_name);
+        self.code_storage
+            .unmetered_check_module_exists(address, module_name)
+    }
+
+    #[inline]
+    fn unmetered_get_module_bytes(
+        &self,
+        address: &AccountAddress,
+        module_name: &IdentStr,
+    ) -> VMResult<Option<Bytes>> {
+        self.record(address, module_name);
+        self.code_storage
+            .unmetered_get_module_bytes(address, module_name)
+    }
+
+    #[inline]
+    fn unmetered_get_module_hash_and_size(
+        &self,
+        address: &AccountAddress,
+        module_name: &IdentStr,
+    ) -> VMResult<Option<([u8; 32], usize)>> {
+        self.record(address, module_name);
+        self.code_storage
+            .unmetered_get_module_hash_and_size(address, module_name)
+    }
+
+    #[inline]
+    fn unmetered_get_module_size(
+        &self,
+        address: &AccountAddress,
+        module_name: &IdentStr,
+    ) -> VMResult<Option<usize>> {
+        self.record(address, module_name);
+        self.code_storage
+            .unmetered_get_module_size(address, module_name)
+    }
+
+    #[inline]
+    fn unmetered_get_deserialized_module(
+        &self,
+        address: &AccountAddress,
+        module_name: &IdentStr,
+    ) -> VMResult<Option<Arc<CompiledModule>>> {
+        self.record(address, module_name);
+        self.code_storage
+            .unmetered_get_deserialized_module(address, module_name)
+    }
+
+    #[inline]
+    fn unmetered_get_eagerly_verified_module(
+        &self,
+        address: &AccountAddress,
+        module_name: &IdentStr,
+    ) -> VMResult<Option<Arc<Module>>> {
+        self.record(address, module_name);
+        self.code_storage
+            .unmetered_get_eagerly_verified_module(address, module_name)
+    }
+
+    #[inline]
+    fn unmetered_get_lazily_verified_module(
+        &self,
+        module_id: &ModuleId,
+    ) -> VMResult<Option<Arc<Module>>> {
+        self.record(module_id.address(), module_id.name());
+        self.code_storage
+            .unmetered_get_lazily_verified_module(module_id)
+    }
+
+    #[cfg(fuzzing)]
+    #[inline]
+    fn unmetered_get_module_skip_verification(
+        &self,
+        address: &AccountAddress,
+        module_name: &IdentStr,
+    ) -> VMResult<Option<Arc<Module>>> {
+        self.record(address, module_name);
+        self.code_storage
+            .unmetered_get_module_skip_verification(address, module_name)
+    }
+}
+
+impl<C: AptosModuleStorage> AptosModuleStorage for ReadRecordingCodeStorage<'_, C> {
+    #[inline]
+    fn unmetered_get_module_state_value_metadata(
+        &self,
+        address: &AccountAddress,
+        module_name: &IdentStr,
+    ) -> PartialVMResult<Option<StateValueMetadata>> {
+        self.record(address, module_name);
+        self.code_storage
+            .unmetered_get_module_state_value_metadata(address, module_name)
+    }
+}
+
+#[delegate_to_methods]
+#[delegate(ScriptCache, target_ref = "as_script_cache")]
+impl<C> ReadRecordingCodeStorage<'_, C>
+where
+    C: ScriptCache<Key = [u8; 32], Deserialized = CompiledScript, Verified = Script>,
+{
+    /// Returns the wrapped script cache.
+    fn as_script_cache(
+        &self,
+    ) -> &dyn ScriptCache<Key = [u8; 32], Deserialized = CompiledScript, Verified = Script> {
+        self.code_storage
+    }
+}

aptos-move/aptos-vm-types/src/module_and_script_storage/read_recording.rs

@@ -24,6 +24,7 @@ use move_core_types::{
 };
 use move_vm_runtime::execution_tracing::Trace;
 use move_vm_types::delayed_values::delayed_field_id::DelayedFieldID;
+use rustc_hash::FxHashSet;
 use std::{collections::BTreeMap, mem};
 
 /// Output produced by the VM after executing a transaction.
@@ -261,3 +262,28 @@ impl VMOutput {
         self.into_transaction_output()
     }
 }
+
+/// A transaction's read set, used for hot-state promotion. Unordered at the
+/// per-transaction level; ordering is imposed later when aggregating per-block.
+///
+/// Data and module keys are kept as recorded. Both sides are already deduplicated and
+/// they can never contain the same key (module and data state keys are disjoint), so
+/// merging them into one set would only re-hash every module key.
+#[derive(Clone, Debug, Default)]
+pub struct UnorderedReadSet {
+    data_keys: FxHashSet<StateKey>,
+    module_keys: FxHashSet<StateKey>,
+}
+
+impl UnorderedReadSet {
+    pub fn new(data_keys: FxHashSet<StateKey>, module_keys: FxHashSet<StateKey>) -> Self {
+        Self {
+            data_keys,
+            module_keys,
+        }
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = &StateKey> {
+        self.data_keys.iter().chain(self.module_keys.iter())
+    }
+}

aptos-move/aptos-vm-types/src/output.rs

@@ -1,6 +1,7 @@
 // Copyright (c) Aptos Foundation
 // Licensed pursuant to the Innovation-Enabling Source Code License, available at https://github.com/aptos-labs/aptos-core/blob/main/LICENSE
 
+use ambassador::delegatable_trait;
 use aptos_aggregator::resolver::{TAggregatorV1View, TDelayedFieldView};
 use aptos_types::{
     serde_helper::bcs_utils::size_u32_as_uleb128,
@@ -22,6 +23,7 @@ use std::collections::{BTreeMap, HashMap};
 /// Allows requesting an immediate interrupt to ongoing transaction execution. For example, this
 /// allows an early return from a useless speculative execution when block execution has already
 /// halted (e.g. due to gas limit, committing only a block prefix).
+#[delegatable_trait]
 pub trait BlockSynchronizationKillSwitch {
     fn interrupt_requested(&self) -> bool;
 }

aptos-move/aptos-vm-types/src/resolver.rs

@@ -73,6 +73,7 @@ once_cell = { workspace = true }
 ouroboros = { workspace = true }
 rand = { workspace = true }
 rayon = { workspace = true }
+rustc-hash = { workspace = true }
 serde = { workspace = true }
 triomphe = { workspace = true }