[vm] Record txn read sets for hot state promotion

Hot state promotions are currently derived from BlockSTM's captured
reads, which exist for validation and differ between the parallel and
sequential paths (e.g. aggregator v1 reads served through delta
resolution are excluded in parallel but recorded in sequential), and
the accumulation is coupled to the `conflict_penalty_window` config.
The write-side exclusion also misses in-place delayed field rewrites,
aggregator v1 writes/deltas and module writes, so keys written in the
block can still be promoted by the epilogue.

Record the read set at the VM boundary instead, where it is a function
of the transaction and the pre-state only:

- `StorageAdapter` records resource, resource group, table item,
  aggregator v1 and config reads.
- A new `ReadRecordingCodeStorage` wrapper records module fetches,
  including those served by the global module cache (which bypass the
  state view today and require cache priming).

The merged set rides on `AptosTransactionOutput` next to the
`VMOutput`; the committed incarnation's reads are fed to the hot state
accumulator in commit order via `accumulate_hot_state_rw()`, gated
only on the accumulator being enabled. Keys treated as written are
enumerated completely from the change set in `hot_state_write_keys()`.

The derived promotion set intentionally differs from the old one
(e.g. `exists<T>` loads the resource through the resolver and now
counts as a read), so enabling this in production requires its own
gating. Read kind/hotness tagging, an on-chain promotion cap, and
removal of the old summary-based derivation are left for follow-ups.

Also adds a per-block promotions histogram, a `StorageAdapter`
recording unit test, and an e2e test asserting that promotions are
identical across sequential and parallel execution, include module
keys, and exclude every key the block writes.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: wqfish

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

@@ -38,3 +38,6 @@ aptos-gas-schedule = { workspace = true, features = ["testing"] }
 aptos-transaction-simulation = { workspace = true }
 aptos-vm = { workspace = true }
 test-case = { workspace = true }
+
+[lints.rust]
+unexpected_cfgs = { level = "warn", check-cfg = ["cfg(fuzzing)"] }

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

@@ -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/read_recording.rs

@@ -0,0 +1,212 @@
+// 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 crate::{
+    module_and_script_storage::module_storage::AptosModuleStorage,
+    resolver::BlockSynchronizationKillSwitch,
+};
+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, language_storage::ModuleId,
+};
+use move_vm_runtime::{
+    LayoutCache, LayoutCacheEntry, Module, ModuleStorage, RuntimeEnvironment, Script, StructKey,
+    WithRuntimeEnvironment,
+};
+use move_vm_types::code::{Code, ScriptCache};
+use std::{cell::RefCell, collections::HashSet, sync::Arc};
+
+/// Wraps a code storage and records the state key of 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). Recording happens here, at the VM boundary, so the derived read set
+/// is the same no matter which executor (sequential, BlockSTM, ...) or cache tier (global
+/// cross-block cache vs per-block cache vs storage) served the module.
+///
+/// Scripts are not state items, so script cache accesses are not recorded.
+pub struct ReadRecordingCodeStorage<'a, C> {
+    code_storage: &'a C,
+    module_reads: RefCell<HashSet<StateKey>>,
+}
+
+impl<'a, C> ReadRecordingCodeStorage<'a, C> {
+    pub fn new(code_storage: &'a C) -> Self {
+        Self {
+            code_storage,
+            module_reads: RefCell::new(HashSet::new()),
+        }
+    }
+
+    /// Returns the state keys of modules fetched so far, clearing the recorded set.
+    pub fn take_recorded_reads(&self) -> HashSet<StateKey> {
+        self.module_reads.take()
+    }
+
+    fn record(&self, address: &AccountAddress, module_name: &IdentStr) {
+        self.module_reads
+            .borrow_mut()
+            .insert(StateKey::module(address, module_name));
+    }
+}
+
+impl<C: WithRuntimeEnvironment> WithRuntimeEnvironment for ReadRecordingCodeStorage<'_, C> {
+    fn runtime_environment(&self) -> &RuntimeEnvironment {
+        self.code_storage.runtime_environment()
+    }
+}
+
+impl<C: LayoutCache> LayoutCache for ReadRecordingCodeStorage<'_, C> {
+    fn get_struct_layout(&self, key: &StructKey) -> Option<LayoutCacheEntry> {
+        self.code_storage.get_struct_layout(key)
+    }
+
+    fn store_struct_layout(&self, key: &StructKey, entry: LayoutCacheEntry) -> PartialVMResult<()> {
+        self.code_storage.store_struct_layout(key, entry)
+    }
+
+    fn remove_struct_layout(&self, key: &StructKey) {
+        self.code_storage.remove_struct_layout(key)
+    }
+}
+
+impl<C: ModuleStorage> ModuleStorage for ReadRecordingCodeStorage<'_, C> {
+    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)
+    }
+
+    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)
+    }
+
+    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)
+    }
+
+    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)
+    }
+
+    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)
+    }
+
+    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)
+    }
+
+    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)]
+    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> {
+    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)
+    }
+}
+
+impl<C> ScriptCache for ReadRecordingCodeStorage<'_, C>
+where
+    C: ScriptCache<Key = [u8; 32], Deserialized = CompiledScript, Verified = Script>,
+{
+    type Deserialized = CompiledScript;
+    type Key = [u8; 32];
+    type Verified = Script;
+
+    fn insert_deserialized_script(
+        &self,
+        key: Self::Key,
+        deserialized_script: Self::Deserialized,
+    ) -> Arc<Self::Deserialized> {
+        self.code_storage
+            .insert_deserialized_script(key, deserialized_script)
+    }
+
+    fn insert_verified_script(
+        &self,
+        key: Self::Key,
+        verified_script: Self::Verified,
+    ) -> Arc<Self::Verified> {
+        self.code_storage
+            .insert_verified_script(key, verified_script)
+    }
+
+    fn get_script(&self, key: &Self::Key) -> Option<Code<Self::Deserialized, Self::Verified>> {
+        self.code_storage.get_script(key)
+    }
+
+    fn num_scripts(&self) -> usize {
+        self.code_storage.num_scripts()
+    }
+}
+
+impl<C: BlockSynchronizationKillSwitch> BlockSynchronizationKillSwitch
+    for ReadRecordingCodeStorage<'_, C>
+{
+    fn interrupt_requested(&self) -> bool {
+        self.code_storage.interrupt_requested()
+    }
+}

aptos-move/aptos-vm/src/block_executor/mod.rs

@@ -59,13 +59,25 @@ use vm_wrapper::AptosExecutorTask;
 pub struct AptosTransactionOutput {
     vm_output: Option<VMOutput>,
     committed_output: OnceCell<TransactionOutput>,
+    /// State keys read by the VM during the execution (incarnation) that produced this
+    /// output. Recorded at the resolver / code storage boundary, so the content does not
+    /// depend on the executor mode (sequential vs BlockSTM) serving the reads. Consumed by
+    /// the block executor to derive hot state promotions; never part of the committed
+    /// output. TODO(HotState): also record the read kind (exists/metadata/value) and the
+    /// observed slot hotness, so the promotion policy can filter on them.
+    read_set: HashSet<StateKey>,
 }
 
 impl AptosTransactionOutput {
     pub fn new(output: VMOutput) -> Self {
+        Self::new_with_read_set(output, HashSet::new())
+    }
+
+    pub fn new_with_read_set(output: VMOutput, read_set: HashSet<StateKey>) -> Self {
         Self {
             vm_output: Some(output),
             committed_output: OnceCell::new(),
+            read_set,
         }
     }
 
@@ -107,6 +119,7 @@ impl<'a> AfterMaterializationOutput<SignatureVerifiedTransaction>
 /// Before materialization guard wrapper that holds a read lock.
 pub struct BeforeMaterializationGuard<'a> {
     guard: &'a VMOutput,
+    read_set: &'a HashSet<StateKey>,
 }
 
 impl BeforeMaterializationOutput<SignatureVerifiedTransaction> for BeforeMaterializationGuard<'_> {
@@ -156,6 +169,22 @@ impl BeforeMaterializationOutput<SignatureVerifiedTransaction> for BeforeMateria
         writes
     }
 
+    fn hot_state_read_keys(&self) -> HashSet<StateKey> {
+        self.read_set.clone()
+    }
+
+    fn hot_state_write_keys(&self) -> HashSet<StateKey> {
+        // Every key receiving a value write becomes hot via the write itself, so the hot
+        // state accumulator must treat it as written and not promote it separately. Unlike
+        // get_write_summary (conflict detection), this includes in-place delayed field
+        // rewrites, aggregator v1 writes/deltas and module writes.
+        let mut keys: HashSet<StateKey> = self.guard.resource_write_set().keys().cloned().collect();
+        keys.extend(self.guard.module_write_set().keys().cloned());
+        keys.extend(self.guard.aggregator_v1_write_set().keys().cloned());
+        keys.extend(self.guard.aggregator_v1_delta_set().keys().cloned());
+        keys
+    }
+
     // TODO: get rid of the cloning data-structures in the following APIs.
     fn resource_group_write_set(
         &self,
@@ -381,6 +410,7 @@ impl BlockExecutorTransactionOutput for AptosTransactionOutput {
                 .vm_output
                 .as_ref()
                 .ok_or_else(|| code_invariant_error("Output must be set but not materialized"))?,
+            read_set: &self.read_set,
         })
     }
 

aptos-move/aptos-vm/src/block_executor/vm_wrapper.rs

@@ -15,7 +15,9 @@ use aptos_types::{
 use aptos_vm_environment::environment::AptosEnvironment;
 use aptos_vm_logging::{log_schema::AdapterLogSchema, prelude::*};
 use aptos_vm_types::{
-    module_and_script_storage::code_storage::AptosCodeStorage,
+    module_and_script_storage::{
+        code_storage::AptosCodeStorage, read_recording::ReadRecordingCodeStorage,
+    },
     resolver::{BlockSynchronizationKillSwitch, ExecutorView, ResourceGroupView},
 };
 use fail::fail_point;
@@ -61,11 +63,20 @@ impl ExecutorTask for AptosExecutorTask {
 
         let log_context = AdapterLogSchema::new(self.id, txn_idx as usize);
         let resolver = self.vm.as_move_resolver_with_group_view(view);
-        match self
-            .vm
-            .execute_single_transaction(txn, &resolver, view, &log_context, auxiliary_info)
-        {
+        let code_storage = ReadRecordingCodeStorage::new(view);
+        match self.vm.execute_single_transaction(
+            txn,
+            &resolver,
+            &code_storage,
+            &log_context,
+            auxiliary_info,
+        ) {
             Ok((vm_status, vm_output)) => {
+                // The state keys the VM read during this execution (incarnation). For the
+                // committed incarnation this is the transaction's logical read set, used to
+                // derive hot state promotions.
+                let mut read_set = resolver.take_recorded_reads();
+                read_set.extend(code_storage.take_recorded_reads());
                 if vm_output.status().is_discarded() {
                     speculative_trace!(
                         &log_context,
@@ -87,13 +98,17 @@ impl ExecutorTask for AptosExecutorTask {
                         &log_context,
                         "Reconfiguration occurred: restart required".into()
                     );
-                    ExecutionStatus::SkipRest(AptosTransactionOutput::new(vm_output))
+                    ExecutionStatus::SkipRest(AptosTransactionOutput::new_with_read_set(
+                        vm_output, read_set,
+                    ))
                 } else {
                     assert!(
                         Self::is_transaction_dynamic_change_set_capable(txn),
                         "DirectWriteSet should always create SkipRest transaction, validate_waypoint_change_set provides this guarantee"
                     );
-                    ExecutionStatus::Success(AptosTransactionOutput::new(vm_output))
+                    ExecutionStatus::Success(AptosTransactionOutput::new_with_read_set(
+                        vm_output, read_set,
+                    ))
                 }
             },
             // execute_single_transaction only returns an error when transactions that should never fail