Repository: stellar/rs-soroban-env
Pull request #1359 (2024-02-29 10:10:19)

Decompose VM instantiation costs and add module cache
This is a sketch of the first step in the plan for addressing #1292 and eventually #827 and #1313, namely:

  - add a bunch of cost types that decompose the current "worst case" VM instantiation cost type
  - continue to charge the worst case on initial contract upload
  - _store_ the decomposed cost-type inputs in the ledger, since we can observe them after the initial upload parse
  - _use_ those decomposed cost-type inputs when doing runtime instantiation
  - add a module cache
  - populate the module cache with all modules on host startup
  - use cached modules during instantiation

This PR has accompanying changes in XDR and wasmi:

  - https://github.com/stellar/stellar-xdr/pull/177
  - https://github.com/stellar/rs-stellar-xdr/pull/346

Remaining to do:

  - [x] determine what the correct set of decomposed cost types even is
  - [ ] ~add more code to wasmi to enable observing more inputs~
  - [x] add cost-type runners / calibrations for all the decomposed cost-types
  - [x] protocol-gate this new behaviour
  - [x] make the linker-loop do less work to match the tighter cost model (i.e. complete #1292)
  - [x] possibly _duplicate_ the set of cost types added here, so we have a cached and uncached version of each, and implement a solution to #827 
  - [ ] possibly decompose the set sufficiently to model what will happen when we take wasmi 0.32 and enable lazy translation (i.e. complete #1313)

graydon: Updated significantly:

  - Does intra-tx caching
  - Does minimal linking
  - Splits parsing from instantiation to, I think, be _more_ ready for wasmi 0.9
  - Drives parser manually to get refined costs rather than delegating to wasmi
  - Added calibration and expanded refined costs a bit more
  - Added checks for complexity of const exprs
  - Protocol gated
  - feature="next" gated

This is getting close to where I want to merge it, I think it's a fairly good set of perf wins.
graydon: Updated to address all comments (and all revisions from CAP review process today)

Patch:
diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
index 3c934a38..5977dffb 100644
--- a/.github/workflows/rust.yml
+++ b/.github/workflows/rust.yml
@@ -44,7 +44,7 @@ jobs:
     continue-on-error: ${{ matrix.checks == 'advisories' }}
     steps:
     - uses: actions/checkout@v3
-    - uses: EmbarkStudios/cargo-deny-action@e0a440755b184aa50374330fa75cca0f84fcb59a
+    - uses: EmbarkStudios/cargo-deny-action@b01e7a8cfb1f496c52d77361e84c1840d8246393
       with:
         command: check ${{ matrix.checks }}
 
diff --git a/Cargo.lock b/Cargo.lock
index ef16fdd4..4c7bc59a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1366,6 +1366,7 @@ dependencies = [
  "static_assertions",
  "stellar-xdr",
  "tracy-client",
+ "wasmparser",
 ]
 
 [[package]]
@@ -1420,6 +1421,7 @@ dependencies = [
  "thousands",
  "tracy-client",
  "wasm-encoder",
+ "wasmparser",
  "wasmprinter",
 ]
 
@@ -1515,7 +1517,7 @@ dependencies = [
 [[package]]
 name = "stellar-xdr"
 version = "20.1.0"
-source = "git+https://github.com/stellar/rs-stellar-xdr?rev=2e0f7f7d42fcd6c3c42eb0d65570fba9f5193d7e#2e0f7f7d42fcd6c3c42eb0d65570fba9f5193d7e"
+source = "git+https://github.com/stellar/rs-stellar-xdr?rev=44b7e2d4cdf27a3611663e82828de56c5274cba0#44b7e2d4cdf27a3611663e82828de56c5274cba0"
 dependencies = [
  "arbitrary",
  "base64 0.13.1",
diff --git a/Cargo.toml b/Cargo.toml
index 1e74df1e..fc3ae86f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -29,12 +29,14 @@ soroban-env-guest = { version = "=20.3.0", path = "soroban-env-guest" }
 soroban-env-host = { version = "=20.3.0", path = "soroban-env-host" }
 soroban-env-macros = { version = "=20.3.0", path = "soroban-env-macros" }
 soroban-builtin-sdk-macros = { version = "=20.3.0", path = "soroban-builtin-sdk-macros" }
+# NB: this must match the wasmparser version wasmi is using
+wasmparser = "=0.116.1"
 
 # NB: When updating, also update the version in rs-soroban-env dev-dependencies
 [workspace.dependencies.stellar-xdr]
 version = "=20.1.0"
 git = "https://github.com/stellar/rs-stellar-xdr"
-rev = "2e0f7f7d42fcd6c3c42eb0d65570fba9f5193d7e"
+rev = "44b7e2d4cdf27a3611663e82828de56c5274cba0"
 default-features = false
 
 [workspace.dependencies.wasmi]
diff --git a/soroban-bench-utils/src/tracker.rs b/soroban-bench-utils/src/tracker.rs
index c77c88b0..0875c39f 100644
--- a/soroban-bench-utils/src/tracker.rs
+++ b/soroban-bench-utils/src/tracker.rs
@@ -111,10 +111,10 @@ mod cpu {
 
 #[cfg(not(any(target_os = "linux", target_os = "macos")))]
 mod cpu {
-    pub struct InstructionCounter(u64);
+    pub struct InstructionCounter;
     impl InstructionCounter {
         pub fn new() -> Self {
-            InstructionCounter(0)
+            InstructionCounter
         }
         pub fn begin(&mut self) {}
         pub fn end_and_count(&mut self) -> u64 {
diff --git a/soroban-env-common/Cargo.toml b/soroban-env-common/Cargo.toml
index 7d058e7f..57428f37 100644
--- a/soroban-env-common/Cargo.toml
+++ b/soroban-env-common/Cargo.toml
@@ -17,6 +17,7 @@ crate-git-revision = "=0.0.6"
 soroban-env-macros = { workspace = true }
 stellar-xdr = { workspace = true, default-features = false, features = [ "curr" ] }
 wasmi = { workspace = true, optional = true }
+wasmparser = { workspace = true, optional = true}
 serde = { version = "=1.0.192", features = ["derive"], optional = true }
 static_assertions = "=1.1.0"
 ethnum = "=1.5.0"
@@ -34,7 +35,7 @@ num-traits = "=0.2.17"
 [features]
 std = ["stellar-xdr/std", "stellar-xdr/base64"]
 serde = ["dep:serde", "stellar-xdr/serde"]
-wasmi = ["dep:wasmi"]
+wasmi = ["dep:wasmi", "dep:wasmparser"]
 testutils = ["dep:arbitrary", "stellar-xdr/arbitrary"]
 next = ["stellar-xdr/next", "soroban-env-macros/next"]
 tracy = ["dep:tracy-client"]
diff --git a/soroban-env-common/src/error.rs b/soroban-env-common/src/error.rs
index 6e83b35c..e967d461 100644
--- a/soroban-env-common/src/error.rs
+++ b/soroban-env-common/src/error.rs
@@ -277,6 +277,13 @@ impl From<wasmi::Error> for Error {
     }
 }
 
+#[cfg(feature = "wasmi")]
+impl From<wasmparser::BinaryReaderError> for Error {
+    fn from(_: wasmparser::BinaryReaderError) -> Self {
+        Error::from_type_and_code(ScErrorType::WasmVm, ScErrorCode::InvalidInput)
+    }
+}
+
 impl Error {
     // NB: we don't provide a "get_type" to avoid casting a bad bit-pattern into
     // an ScErrorType. Instead we provide an "is_type" to check any specific
diff --git a/soroban-env-host/Cargo.toml b/soroban-env-host/Cargo.toml
index 2276dc27..e0f1f952 100644
--- a/soroban-env-host/Cargo.toml
+++ b/soroban-env-host/Cargo.toml
@@ -8,7 +8,7 @@ license = "Apache-2.0"
 version.workspace = true
 readme = "../README.md"
 edition = "2021"
-rust-version.workspace = true
+rust-version = "1.74"
 build = "build.rs"
 
 exclude = ["observations/"]
@@ -17,6 +17,7 @@ exclude = ["observations/"]
 soroban-builtin-sdk-macros = { workspace = true }
 soroban-env-common = { workspace = true, features = ["std", "wasmi", "shallow-val-hash"] }
 wasmi = { workspace = true }
+wasmparser = { workspace = true }
 stellar-strkey = "=0.0.8"
 static_assertions = "=1.1.0"
 sha2 = "=0.10.8"
@@ -86,7 +87,7 @@ rustversion = "1.0"
 [dev-dependencies.stellar-xdr]
 version = "=20.1.0"
 git = "https://github.com/stellar/rs-stellar-xdr"
-rev = "2e0f7f7d42fcd6c3c42eb0d65570fba9f5193d7e"
+rev = "44b7e2d4cdf27a3611663e82828de56c5274cba0"
 default-features = false
 features = ["arbitrary"]
 
diff --git a/soroban-env-host/benches/common/cost_types/vm_ops.rs b/soroban-env-host/benches/common/cost_types/vm_ops.rs
index d011ace8..a71c9db8 100644
--- a/soroban-env-host/benches/common/cost_types/vm_ops.rs
+++ b/soroban-env-host/benches/common/cost_types/vm_ops.rs
@@ -1,45 +1,267 @@
-#[allow(unused)]
-use super::wasm_insn_exec::{wasm_module_with_4n_insns, wasm_module_with_n_internal_funcs};
-use crate::common::{util, HostCostMeasurement};
-use rand::{rngs::StdRng, Rng};
+use super::wasm_insn_exec::wasm_module_with_n_internal_funcs;
+use crate::common::HostCostMeasurement;
+use rand::rngs::StdRng;
 use soroban_env_host::{
     cost_runner::{VmInstantiationRun, VmInstantiationSample},
+    vm::{ParsedModule, VersionedContractCodeCostInputs},
     xdr, Host,
 };
+use std::rc::Rc;
 
+// Protocol 20 coarse cost model.
 pub(crate) struct VmInstantiationMeasure;
 
-// This measures the cost of instantiating a host::Vm on a variety of possible
-// wasm modules, of different sizes. The input value should be the size of the
-// module, though for now we're just selecting modules from the fixed example
-// repertoire. Costs should be linear.
-impl HostCostMeasurement for VmInstantiationMeasure {
-    type Runner = VmInstantiationRun;
-
-    fn new_best_case(_host: &Host, _rng: &mut StdRng) -> VmInstantiationSample {
-        let id: xdr::Hash = [0; 32].into();
-        let wasm: Vec<u8> = soroban_test_wasms::ADD_I32.into();
-        VmInstantiationSample { id: Some(id), wasm }
-    }
-
-    fn new_worst_case(_host: &Host, _rng: &mut StdRng, input: u64) -> VmInstantiationSample {
-        let id: xdr::Hash = [0; 32].into();
-        // generate a test wasm contract with many trivial internal functions,
-        // which represents the worst case in terms of work needed for WASM parsing.
-        let n = (Self::INPUT_BASE_SIZE + input * 30) as usize;
-        let wasm = wasm_module_with_n_internal_funcs(n);
-        // replace the above two lines with these below to test with wasm contracts
-        // with a single function of many instructions. In both tests the cpu grows
-        // linearly with the contract size however the slopes are very different.
-        // let n = (input * 50) as usize;
-        // let wasm = wasm_module_with_4n_insns(n);
-        VmInstantiationSample { id: Some(id), wasm }
-    }
-
-    fn new_random_case(_host: &Host, rng: &mut StdRng, _input: u64) -> VmInstantiationSample {
-        let id: xdr::Hash = [0; 32].into();
-        let idx = rng.gen_range(0..=10) % util::TEST_WASMS.len();
-        let wasm = util::TEST_WASMS[idx].into();
-        VmInstantiationSample { id: Some(id), wasm }
-    }
+// This measures the cost of parsing Wasm and/or instantiating a host::Vm on a
+// variety of possible Wasm modules, of different sizes.
+macro_rules! impl_measurement_for_instantiation_cost_type {
+    ($RUNNER:ty, $MEASURE:ty, $BUILD:ident, $USE_REFINED_INPUTS:expr, $MAGNITUDE:expr) => {
+        impl HostCostMeasurement for $MEASURE {
+            type Runner = $RUNNER;
+
+            fn new_random_case(
+                _host: &Host,
+                _rng: &mut StdRng,
+                input: u64,
+            ) -> VmInstantiationSample {
+                let id: xdr::Hash = [0; 32].into();
+                let n = (Self::INPUT_BASE_SIZE + input * $MAGNITUDE) as usize;
+                let wasm = $BUILD(n);
+                #[allow(unused_mut)]
+                let mut cost_inputs = VersionedContractCodeCostInputs::V0 {
+                    wasm_bytes: wasm.len(),
+                };
+                #[cfg(feature = "next")]
+                if $USE_REFINED_INPUTS {
+                    cost_inputs = VersionedContractCodeCostInputs::V1(
+                        soroban_env_host::vm::ParsedModule::extract_refined_contract_cost_inputs(
+                            _host,
+                            &wasm[..],
+                        )
+                        .unwrap(),
+                    )
+                }
+                let module = Rc::new(
+                    ParsedModule::new_with_isolated_engine(_host, &wasm, cost_inputs.clone())
+                        .unwrap(),
+                );
+                VmInstantiationSample {
+                    id: Some(id),
+                    wasm,
+                    module,
+                }
+            }
+        }
+    };
+}
+
+// Protocol 20 coarse unified, or protocol 21 coarse parse-phase cost model
+impl_measurement_for_instantiation_cost_type!(
+    VmInstantiationRun,
+    VmInstantiationMeasure,
+    wasm_module_with_n_internal_funcs,
+    false,
+    30
+);
+
+// Protocol 21 cost models.
+#[cfg(feature = "next")]
+pub(crate) use v21::*;
+#[cfg(feature = "next")]
+mod v21 {
+    use super::super::wasm_insn_exec::{
+        wasm_module_with_n_data_segment_bytes, wasm_module_with_n_data_segments,
+        wasm_module_with_n_elem_segments, wasm_module_with_n_exports, wasm_module_with_n_globals,
+        wasm_module_with_n_imports, wasm_module_with_n_insns, wasm_module_with_n_internal_funcs,
+        wasm_module_with_n_table_entries, wasm_module_with_n_types,
+    };
+    use super::*;
+    use soroban_env_host::{
+        cost_runner::{
+            InstantiateWasmDataSegmentBytesRun, InstantiateWasmDataSegmentsRun,
+            InstantiateWasmElemSegmentsRun, InstantiateWasmExportsRun, InstantiateWasmFunctionsRun,
+            InstantiateWasmGlobalsRun, InstantiateWasmImportsRun, InstantiateWasmInstructionsRun,
+            InstantiateWasmTableEntriesRun, InstantiateWasmTypesRun, ParseWasmDataSegmentBytesRun,
+            ParseWasmDataSegmentsRun, ParseWasmElemSegmentsRun, ParseWasmExportsRun,
+            ParseWasmFunctionsRun, ParseWasmGlobalsRun, ParseWasmImportsRun,
+            ParseWasmInstructionsRun, ParseWasmTableEntriesRun, ParseWasmTypesRun,
+            VmCachedInstantiationRun, VmInstantiationSample,
+        },
+        xdr, Host,
+    };
+
+    pub(crate) struct VmCachedInstantiationMeasure;
+
+    pub(crate) struct ParseWasmInstructionsMeasure;
+    pub(crate) struct ParseWasmFunctionsMeasure;
+    pub(crate) struct ParseWasmGlobalsMeasure;
+    pub(crate) struct ParseWasmTableEntriesMeasure;
+    pub(crate) struct ParseWasmTypesMeasure;
+    pub(crate) struct ParseWasmDataSegmentsMeasure;
+    pub(crate) struct ParseWasmElemSegmentsMeasure;
+    pub(crate) struct ParseWasmImportsMeasure;
+    pub(crate) struct ParseWasmExportsMeasure;
+    pub(crate) struct ParseWasmDataSegmentBytesMeasure;
+
+    pub(crate) struct InstantiateWasmInstructionsMeasure;
+    pub(crate) struct InstantiateWasmFunctionsMeasure;
+    pub(crate) struct InstantiateWasmGlobalsMeasure;
+    pub(crate) struct InstantiateWasmTableEntriesMeasure;
+    pub(crate) struct InstantiateWasmTypesMeasure;
+    pub(crate) struct InstantiateWasmDataSegmentsMeasure;
+    pub(crate) struct InstantiateWasmElemSegmentsMeasure;
+    pub(crate) struct InstantiateWasmImportsMeasure;
+    pub(crate) struct InstantiateWasmExportsMeasure;
+    pub(crate) struct InstantiateWasmDataSegmentBytesMeasure;
+
+    // Protocol 21 coarse instantiation-phase cost model
+    impl_measurement_for_instantiation_cost_type!(
+        VmCachedInstantiationRun,
+        VmCachedInstantiationMeasure,
+        wasm_module_with_n_internal_funcs,
+        false,
+        30
+    );
+
+    // Protocol 21 refined cost model
+    impl_measurement_for_instantiation_cost_type!(
+        ParseWasmInstructionsRun,
+        ParseWasmInstructionsMeasure,
+        wasm_module_with_n_insns,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        ParseWasmFunctionsRun,
+        ParseWasmFunctionsMeasure,
+        wasm_module_with_n_internal_funcs,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        ParseWasmGlobalsRun,
+        ParseWasmGlobalsMeasure,
+        wasm_module_with_n_globals,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        ParseWasmTableEntriesRun,
+        ParseWasmTableEntriesMeasure,
+        wasm_module_with_n_table_entries,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        ParseWasmTypesRun,
+        ParseWasmTypesMeasure,
+        wasm_module_with_n_types,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        ParseWasmDataSegmentsRun,
+        ParseWasmDataSegmentsMeasure,
+        wasm_module_with_n_data_segments,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        ParseWasmElemSegmentsRun,
+        ParseWasmElemSegmentsMeasure,
+        wasm_module_with_n_elem_segments,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        ParseWasmImportsRun,
+        ParseWasmImportsMeasure,
+        wasm_module_with_n_imports,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        ParseWasmExportsRun,
+        ParseWasmExportsMeasure,
+        wasm_module_with_n_exports,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        ParseWasmDataSegmentBytesRun,
+        ParseWasmDataSegmentBytesMeasure,
+        wasm_module_with_n_data_segment_bytes,
+        true,
+        200000
+    );
+
+    impl_measurement_for_instantiation_cost_type!(
+        InstantiateWasmInstructionsRun,
+        InstantiateWasmInstructionsMeasure,
+        wasm_module_with_n_insns,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        InstantiateWasmFunctionsRun,
+        InstantiateWasmFunctionsMeasure,
+        wasm_module_with_n_internal_funcs,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        InstantiateWasmGlobalsRun,
+        InstantiateWasmGlobalsMeasure,
+        wasm_module_with_n_globals,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        InstantiateWasmTableEntriesRun,
+        InstantiateWasmTableEntriesMeasure,
+        wasm_module_with_n_table_entries,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        InstantiateWasmTypesRun,
+        InstantiateWasmTypesMeasure,
+        wasm_module_with_n_types,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        InstantiateWasmDataSegmentsRun,
+        InstantiateWasmDataSegmentsMeasure,
+        wasm_module_with_n_data_segments,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        InstantiateWasmElemSegmentsRun,
+        InstantiateWasmElemSegmentsMeasure,
+        wasm_module_with_n_elem_segments,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        InstantiateWasmImportsRun,
+        InstantiateWasmImportsMeasure,
+        wasm_module_with_n_imports,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        InstantiateWasmExportsRun,
+        InstantiateWasmExportsMeasure,
+        wasm_module_with_n_exports,
+        true,
+        30
+    );
+    impl_measurement_for_instantiation_cost_type!(
+        InstantiateWasmDataSegmentBytesRun,
+        InstantiateWasmDataSegmentBytesMeasure,
+        wasm_module_with_n_data_segment_bytes,
+        true,
+        200000
+    );
 }
diff --git a/soroban-env-host/benches/common/cost_types/wasm_insn_exec.rs b/soroban-env-host/benches/common/cost_types/wasm_insn_exec.rs
index 4453462b..67cf4f34 100644
--- a/soroban-env-host/benches/common/cost_types/wasm_insn_exec.rs
+++ b/soroban-env-host/benches/common/cost_types/wasm_insn_exec.rs
@@ -2,6 +2,7 @@ use crate::common::HostCostMeasurement;
 use rand::{rngs::StdRng, RngCore};
 use soroban_env_host::{cost_runner::*, xdr::Hash, Host, Symbol, Vm};
 use soroban_synth_wasm::{Arity, GlobalRef, ModEmitter, Operand};
+use wasm_encoder::{ConstExpr, ExportKind, ValType};
 
 // These are fp numbers to minimize rounding during overhead calculation.
 // The fact they both turned out to be "whole" numbers is pure luck.
@@ -13,20 +14,56 @@ struct WasmModule {
     overhead: u64,
 }
 
+// ModEmitter's default constructors are a little too spartan for our needs, we
+// want our benchmarks to all have at least one imported function and at least
+// one defined and exported function, so we're in the right performance tier.
+// But we also don't want to go changing those constructors since it'll perturb
+// a lot of non-benchmark users.
+trait ModEmitterExt {
+    fn bench_default() -> Self;
+    fn bench_from_configs(mem_pages: u32, elem_count: u32) -> Self;
+    fn add_bench_import(self) -> Self;
+    fn add_bench_export(self) -> Self;
+    fn add_bench_baseline_material(self) -> Self;
+}
+
+impl ModEmitterExt for ModEmitter {
+    fn add_bench_import(mut self) -> Self {
+        self.import_func("t", "_", Arity(0));
+        self
+    }
+    fn add_bench_export(self) -> Self {
+        let mut fe = self.func(Arity(0), 0);
+        fe.push(Symbol::try_from_small_str("pass").unwrap());
+        fe.finish_and_export("default")
+    }
+    fn add_bench_baseline_material(self) -> Self {
+        self.add_bench_import().add_bench_export()
+    }
+
+    fn bench_default() -> Self {
+        Self::add_bench_baseline_material(ModEmitter::default())
+    }
+
+    fn bench_from_configs(mem_pages: u32, elem_count: u32) -> Self {
+        Self::add_bench_baseline_material(ModEmitter::from_configs(mem_pages, elem_count))
+    }
+}
+
 pub fn wasm_module_with_n_internal_funcs(n: usize) -> Vec<u8> {
-    let mut me = ModEmitter::default();
+    let mut me = ModEmitter::bench_default();
     for _ in 0..n {
         let mut fe = me.func(Arity(0), 0);
         fe.push(Symbol::try_from_small_str("pass").unwrap());
         (me, _) = fe.finish();
     }
-    let mut fe = me.func(Arity(0), 0);
-    fe.push(Symbol::try_from_small_str("pass").unwrap());
-    fe.finish_and_export("test").finish()
+    me.finish()
 }
 
-pub fn wasm_module_with_4n_insns(n: usize) -> Vec<u8> {
-    let mut fe = ModEmitter::default().func(Arity(1), 0);
+pub fn wasm_module_with_n_insns(n: usize) -> Vec<u8> {
+    // We actually emit 4 instructions per loop iteration, so we need to divide by 4.
+    let n = 1 + (n / 4);
+    let mut fe = ModEmitter::bench_default().func(Arity(1), 0);
     let arg = fe.args[0];
     fe.push(Operand::Const64(1));
     for i in 0..n {
@@ -39,6 +76,139 @@ pub fn wasm_module_with_4n_insns(n: usize) -> Vec<u8> {
     fe.push(Symbol::try_from_small_str("pass").unwrap());
     fe.finish_and_export("test").finish()
 }
+pub fn wasm_module_with_n_globals(n: usize) -> Vec<u8> {
+    let mut me = ModEmitter::bench_default();
+    for i in 0..n {
+        me.global(ValType::I64, true, &ConstExpr::i64_const(i as i64));
+    }
+    me.finish()
+}
+
+pub fn wasm_module_with_n_imports(n: usize) -> Vec<u8> {
+    let mut me = ModEmitter::default().add_bench_import();
+    let names = Vm::get_all_host_functions();
+    for (module, name, arity) in names.iter().take(n) {
+        if *module == "t" {
+            continue;
+        }
+        me.import_func(module, name, Arity(*arity));
+    }
+    me.add_bench_export().finish()
+}
+
+pub fn wasm_module_with_n_exports(n: usize) -> Vec<u8> {
+    let me = ModEmitter::bench_default();
+    let mut fe = me.func(Arity(0), 0);
+    fe.push(Symbol::try_from_small_str("pass").unwrap());
+    let (mut me, fid) = fe.finish();
+    for i in 0..n {
+        me.export(format!("_{i}").as_str(), ExportKind::Func, fid.0);
+    }
+    me.finish()
+}
+
+pub fn wasm_module_with_n_table_entries(n: usize) -> Vec<u8> {
+    let me = ModEmitter::bench_from_configs(1, n as u32);
+    let mut fe = me.func(Arity(0), 0);
+    fe.push(Symbol::try_from_small_str("pass").unwrap());
+    let (mut me, f) = fe.finish();
+    let funcs = vec![f; n];
+    me.define_elem_funcs(&funcs);
+    me.finish()
+}
+
+pub fn wasm_module_with_n_types(mut n: usize) -> Vec<u8> {
+    let mut me = ModEmitter::bench_default();
+    // There's a max of 1,000,000 types, so we just make a loop
+    // that covers more than that many combinations, and break when we've got
+    // to the requested number.
+    let val_types = &[ValType::I32, ValType::I64];
+
+    'top: for a in val_types {
+        for b in val_types {
+            for c in val_types {
+                for d in val_types {
+                    for e in val_types {
+                        for f in val_types {
+                            for g in val_types {
+                                for h in val_types {
+                                    for i in val_types {
+                                        for j in val_types {
+                                            for aa in val_types {
+                                                for bb in val_types {
+                                                    for cc in val_types {
+                                                        for dd in val_types {
+                                                            for ee in val_types {
+                                                                for ff in val_types {
+                                                                    for gg in val_types {
+                                                                        for hh in val_types {
+                                                                            for ii in val_types {
+                                                                                for jj in val_types
+                                                                                {
+                                                                                    if n == 0 {
+                                                                                        break 'top;
+                                                                                    }
+                                                                                    n -= 1;
+                                                                                    let params = &[
+                                                                                        *a, *b, *c,
+                                                                                        *d, *e, *f,
+                                                                                        *g, *h, *i,
+                                                                                        *j, *aa,
+                                                                                        *bb, *cc,
+                                                                                        *dd, *ee,
+                                                                                        *ff, *gg,
+                                                                                        *hh, *ii,
+                                                                                        *jj,
+                                                                                    ];
+                                                                                    me.add_raw_fn_type(params, &[]);
+                                                                                }
+                                                                            }
+                                                                        }
+                                                                    }
+                                                                }
+                                                            }
+                                                        }
+                                                    }
+                                                }
+                                            }
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+    me.finish()
+}
+
+pub fn wasm_module_with_n_elem_segments(n: usize) -> Vec<u8> {
+    let me = ModEmitter::bench_from_configs(1, n as u32);
+    let mut fe = me.func(Arity(0), 0);
+    fe.push(Symbol::try_from_small_str("pass").unwrap());
+    let (mut me, f) = fe.finish();
+    for _ in 0..n {
+        me.define_elem_funcs(&[f]);
+    }
+    me.finish()
+}
+
+pub fn wasm_module_with_n_data_segments(n: usize) -> Vec<u8> {
+    let mem_offset = n as u32 * 1024;
+    let mut me = ModEmitter::bench_from_configs(1 + mem_offset / 65536, 0);
+    for _ in 0..n {
+        me.define_data_segment(n as u32 * 1024, vec![1, 2, 3, 4]);
+    }
+    me.finish()
+}
+
+pub fn wasm_module_with_n_data_segment_bytes(n: usize) -> Vec<u8> {
+    let mut me = ModEmitter::bench_from_configs(1 + (n / 0x10000) as u32, 0);
+    me.define_data_segment(0, vec![0xff; n]);
+    me.finish()
+}
 
 fn wasm_module_with_mem_grow(n_pages: usize) -> Vec<u8> {
     let mut fe = ModEmitter::default().func(Arity(0), 0);
@@ -51,7 +221,7 @@ fn wasm_module_with_mem_grow(n_pages: usize) -> Vec<u8> {
     fe.finish_and_export("test").finish()
 }
 
-// A wasm module with a single const to serve as the baseline
+// A Wasm module with a single const to serve as the baseline
 fn wasm_module_baseline_pass() -> WasmModule {
     let mut fe = ModEmitter::default().func(Arity(0), 0);
     fe.push(Symbol::try_from_small_str("pass").unwrap());
@@ -59,7 +229,7 @@ fn wasm_module_baseline_pass() -> WasmModule {
     WasmModule { wasm, overhead: 0 }
 }
 
-// A wasm module with a single trap to serve as the baseline
+// A Wasm module with a single trap to serve as the baseline
 fn wasm_module_baseline_trap() -> WasmModule {
     let mut fe = ModEmitter::default().func(Arity(0), 0);
     fe.trap();
diff --git a/soroban-env-host/benches/common/experimental/vm_ops.rs b/soroban-env-host/benches/common/experimental/vm_ops.rs
index 943a73d6..aaf04290 100644
--- a/soroban-env-host/benches/common/experimental/vm_ops.rs
+++ b/soroban-env-host/benches/common/experimental/vm_ops.rs
@@ -1,5 +1,4 @@
 #[allow(unused)]
-use super::wasm_insn_exec::{wasm_module_with_4n_insns, wasm_module_with_n_internal_funcs};
 use crate::common::{util, HostCostMeasurement};
 use rand::{rngs::StdRng, Rng, RngCore};
 use soroban_env_host::{
diff --git a/soroban-env-host/benches/common/measure.rs b/soroban-env-host/benches/common/measure.rs
index f60c51b4..51f54300 100644
--- a/soroban-env-host/benches/common/measure.rs
+++ b/soroban-env-host/benches/common/measure.rs
@@ -68,10 +68,11 @@ impl Measurements {
             .iter()
             .map(|m| {
                 let mut e = m.clone();
-                e.inputs = e.inputs.map(|i| i / e.iterations);
-                e.cpu_insns = e.cpu_insns.saturating_sub(self.baseline.cpu_insns) / e.iterations;
-                e.mem_bytes = e.mem_bytes.saturating_sub(self.baseline.mem_bytes) / e.iterations;
-                e.time_nsecs = e.time_nsecs.saturating_sub(self.baseline.time_nsecs) / e.iterations;
+                let iterations = e.iterations.max(1);
+                e.inputs = e.inputs.map(|i| i / iterations);
+                e.cpu_insns = e.cpu_insns.saturating_sub(self.baseline.cpu_insns) / iterations;
+                e.mem_bytes = e.mem_bytes.saturating_sub(self.baseline.mem_bytes) / iterations;
+                e.time_nsecs = e.time_nsecs.saturating_sub(self.baseline.time_nsecs) / iterations;
                 e
             })
             .collect()
diff --git a/soroban-env-host/benches/common/mod.rs b/soroban-env-host/benches/common/mod.rs
index 0d981a38..249d88c9 100644
--- a/soroban-env-host/benches/common/mod.rs
+++ b/soroban-env-host/benches/common/mod.rs
@@ -82,6 +82,33 @@ pub(crate) fn for_each_host_cost_measurement<B: Benchmark>(
     call_bench::<B, Int256PowMeasure>(&mut params)?;
     call_bench::<B, Int256ShiftMeasure>(&mut params)?;
     call_bench::<B, ChaCha20DrawBytesMeasure>(&mut params)?;
+
+    #[cfg(feature = "next")]
+    {
+        call_bench::<B, VmCachedInstantiationMeasure>(&mut params)?;
+
+        call_bench::<B, ParseWasmInstructionsMeasure>(&mut params)?;
+        call_bench::<B, ParseWasmFunctionsMeasure>(&mut params)?;
+        call_bench::<B, ParseWasmGlobalsMeasure>(&mut params)?;
+        call_bench::<B, ParseWasmTableEntriesMeasure>(&mut params)?;
+        call_bench::<B, ParseWasmTypesMeasure>(&mut params)?;
+        call_bench::<B, ParseWasmDataSegmentsMeasure>(&mut params)?;
+        call_bench::<B, ParseWasmElemSegmentsMeasure>(&mut params)?;
+        call_bench::<B, ParseWasmImportsMeasure>(&mut params)?;
+        call_bench::<B, ParseWasmExportsMeasure>(&mut params)?;
+        call_bench::<B, ParseWasmDataSegmentBytesMeasure>(&mut params)?;
+
+        call_bench::<B, InstantiateWasmInstructionsMeasure>(&mut params)?;
+        call_bench::<B, InstantiateWasmFunctionsMeasure>(&mut params)?;
+        call_bench::<B, InstantiateWasmGlobalsMeasure>(&mut params)?;
+        call_bench::<B, InstantiateWasmTableEntriesMeasure>(&mut params)?;
+        call_bench::<B, InstantiateWasmTypesMeasure>(&mut params)?;
+        call_bench::<B, InstantiateWasmDataSegmentsMeasure>(&mut params)?;
+        call_bench::<B, InstantiateWasmElemSegmentsMeasure>(&mut params)?;
+        call_bench::<B, InstantiateWasmImportsMeasure>(&mut params)?;
+        call_bench::<B, InstantiateWasmExportsMeasure>(&mut params)?;
+        call_bench::<B, InstantiateWasmDataSegmentBytesMeasure>(&mut params)?;
+    }
     // These three mem ones are derived analytically, we do not calibrate them typically
     if std::env::var("INCLUDE_ANALYTICAL_COSTTYPES").is_ok() {
         call_bench::<B, MemAllocMeasure>(&mut params)?;
diff --git a/soroban-env-host/benches/common/util.rs b/soroban-env-host/benches/common/util.rs
index a41c222f..e3e70a78 100644
--- a/soroban-env-host/benches/common/util.rs
+++ b/soroban-env-host/benches/common/util.rs
@@ -10,6 +10,9 @@ pub(crate) fn test_host() -> Host {
     .unwrap();
     host.as_budget().reset_unlimited().unwrap();
     host.as_budget().reset_fuel_config().unwrap();
+    if std::env::var("DEBUG_BENCH_HOST").is_ok() {
+        host.enable_debug().unwrap();
+    }
     host
 }
 
diff --git a/soroban-env-host/benches/worst_case_linear_models.rs b/soroban-env-host/benches/worst_case_linear_models.rs
index 00c06337..87a5f188 100644
--- a/soroban-env-host/benches/worst_case_linear_models.rs
+++ b/soroban-env-host/benches/worst_case_linear_models.rs
@@ -68,6 +68,87 @@ fn write_cost_params_table<T: Display>(
     tw.flush()
 }
 
+fn correct_multi_variable_models(
+    params: &mut BTreeMap<CostType, (MeteredCostComponent, MeteredCostComponent)>,
+) {
+    // Several cost types actually represent additional terms a cost model that
+    // we're decomposing into multiple variables, such as the cost of VM
+    // instantiation. When we charge these costs, we charge each variable
+    // separately, i.e. to charge a 5-variable cost we'll make 5 calls to the
+    // budget. Only the first of these 5 calls should have a constant factor,
+    // the rest should have zero as their constant (since they only contribute a
+    // new linear term), but the calibration code will have put the same (or
+    // nearly-the-same) nonzero constant term in each `CostComponent`. We
+    // correct this here by zeroing out the constant term in all but the first
+    // `CostComponent` of each set, (and attempting to confirm that they all
+    // have roughly-the-same constant term).
+    use ContractCostType::*;
+    const MULTI_VARIABLE_COST_GROUPS: &[&[ContractCostType]] = &[
+        &[
+            ParseWasmInstructions,
+            ParseWasmFunctions,
+            ParseWasmGlobals,
+            ParseWasmTableEntries,
+            ParseWasmTypes,
+            ParseWasmDataSegments,
+            ParseWasmElemSegments,
+            ParseWasmImports,
+            ParseWasmExports,
+            ParseWasmDataSegmentBytes,
+        ],
+        &[
+            InstantiateWasmInstructions,
+            InstantiateWasmFunctions,
+            InstantiateWasmGlobals,
+            InstantiateWasmTableEntries,
+            InstantiateWasmTypes,
+            InstantiateWasmDataSegments,
+            InstantiateWasmElemSegments,
+            InstantiateWasmImports,
+            InstantiateWasmExports,
+            InstantiateWasmDataSegmentBytes,
+        ],
+    ];
+    for group in MULTI_VARIABLE_COST_GROUPS {
+        let mut iter = group.iter();
+        if let Some(first) = iter.next() {
+            let Some((first_cpu, first_mem)) = params.get(&CostType::Contract(*first)).cloned()
+            else {
+                continue;
+            };
+            for ty in iter {
+                let Some((cpu, mem)) = params.get_mut(&CostType::Contract(*ty)) else {
+                    continue;
+                };
+                let cpu_const_diff_ratio = (cpu.const_term as f64 - first_cpu.const_term as f64)
+                    / first_cpu.const_term as f64;
+                let mem_const_diff_ratio = (mem.const_term as f64 - first_mem.const_term as f64)
+                    / first_mem.const_term as f64;
+                assert!(
+                    cpu_const_diff_ratio < 0.25,
+                    "cost type {:?} has too large a constant CPU term over {:?}: {:?} vs. {:?} ({:?} diff)",
+                    ty,
+                    first,
+                    cpu.const_term,
+                    first_cpu.const_term,
+                    cpu_const_diff_ratio
+                );
+                assert!(
+                    mem_const_diff_ratio < 0.25,
+                    "cost type {:?} has too large a constant memory term over {:?}: {:?} vs. {:?} ({:?} diff)",
+                    ty,
+                    first,
+                    mem.const_term,
+                    first_mem.const_term,
+                    mem_const_diff_ratio
+                );
+                cpu.const_term = 0;
+                mem.const_term = 0;
+            }
+        }
+    }
+}
+
 fn write_budget_params_code(
     params: &BTreeMap<CostType, (MeteredCostComponent, MeteredCostComponent)>,
     wasm_tier_cost: &BTreeMap<WasmInsnTier, u64>,
@@ -295,22 +376,24 @@ fn process_tier(
     params_wasm: &BTreeMap<CostType, (MeteredCostComponent, MeteredCostComponent)>,
     insn_tier: &[WasmInsnType],
 ) -> u64 {
-    println!("\n");
-    println!("\n{:=<100}", "");
-    println!("\"{:?}\" tier", tier);
-
     let (params_tier, ave_cpu_per_fuel) = extract_tier(params_wasm, insn_tier);
 
-    let mut tw = TabWriter::new(vec![])
-        .padding(5)
-        .alignment(Alignment::Right);
-    write_cost_params_table::<WasmInsnType>(&mut tw, &params_tier).unwrap();
-    eprintln!("{}", String::from_utf8(tw.into_inner().unwrap()).unwrap());
-    println!(
-        "average cpu insns per fuel for \"{:?}\" tier: {}",
-        tier, ave_cpu_per_fuel
-    );
-    println!("{:=<100}\n", "");
+    if !params_tier.is_empty() {
+        println!("\n");
+        println!("\n{:=<100}", "");
+        println!("\"{:?}\" tier", tier);
+
+        let mut tw = TabWriter::new(vec![])
+            .padding(5)
+            .alignment(Alignment::Right);
+        write_cost_params_table::<WasmInsnType>(&mut tw, &params_tier).unwrap();
+        eprintln!("{}", String::from_utf8(tw.into_inner().unwrap()).unwrap());
+        println!(
+            "average cpu insns per fuel for \"{:?}\" tier: {}",
+            tier, ave_cpu_per_fuel
+        );
+        println!("{:=<100}\n", "");
+    }
     ave_cpu_per_fuel
 }
 
@@ -333,12 +416,18 @@ fn extract_wasmi_fuel_costs(
 
 #[cfg(all(test, any(target_os = "linux", target_os = "macos")))]
 fn main() -> std::io::Result<()> {
-    let params = if std::env::var("RUN_EXPERIMENT").is_err() {
+    let mut params = if std::env::var("RUN_EXPERIMENT").is_err() {
         for_each_host_cost_measurement::<WorstCaseLinearModels>()?
     } else {
         for_each_experimental_cost_measurement::<WorstCaseLinearModels>()?
     };
-    let params_wasm = for_each_wasm_insn_measurement::<WorstCaseLinearModels>()?;
+    let params_wasm = if std::env::var("SKIP_WASM_INSNS").is_err() {
+        for_each_wasm_insn_measurement::<WorstCaseLinearModels>()?
+    } else {
+        BTreeMap::new()
+    };
+
+    correct_multi_variable_models(&mut params);
 
     let mut tw = TabWriter::new(vec![])
         .padding(5)
diff --git a/soroban-env-host/src/budget.rs b/soroban-env-host/src/budget.rs
index 87ad7cd4..8b3e1ddb 100644
--- a/soroban-env-host/src/budget.rs
+++ b/soroban-env-host/src/budget.rs
@@ -23,7 +23,7 @@ use crate::{
 
 use dimension::{BudgetDimension, IsCpu, IsShadowMode};
 
-#[derive(Debug, Clone, Default, PartialEq, Eq, PartialOrd, Ord)]
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord)]
 pub struct CostTracker {
     pub iterations: u64,
     pub inputs: Option<u64>,
@@ -46,11 +46,11 @@ struct BudgetTracker {
 impl Default for BudgetTracker {
     fn default() -> Self {
         let mut mt = Self {
-            cost_tracker: Default::default(),
+            cost_tracker: [CostTracker::default(); ContractCostType::variants().len()],
             meter_count: Default::default(),
             #[cfg(any(test, feature = "testutils", feature = "bench"))]
             wasm_memory: Default::default(),
-            time_tracker: Default::default(),
+            time_tracker: [0_u64; ContractCostType::variants().len()],
         };
         for (ct, tracker) in ContractCostType::variants()
             .iter()
@@ -92,6 +92,47 @@ impl Default for BudgetTracker {
                 ContractCostType::Int256Pow => (),
                 ContractCostType::Int256Shift => (),
                 ContractCostType::ChaCha20DrawBytes => init_input(), // number of random bytes to draw
+
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmInstructions => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmFunctions => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmGlobals => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmTableEntries => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmTypes => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmDataSegments => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmElemSegments => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmImports => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmExports => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmDataSegmentBytes => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmInstructions => (),
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmFunctions => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmGlobals => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmTableEntries => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmTypes => (),
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmDataSegments => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmElemSegments => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmImports => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmExports => init_input(),
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmDataSegmentBytes => init_input(),
             }
         }
         mt
@@ -368,6 +409,107 @@ impl Default for BudgetImpl {
                     cpu.const_term = 1058;
                     cpu.lin_term = ScaledU64(501);
                 }
+
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmInstructions => {
+                    cpu.const_term = 72736;
+                    cpu.lin_term = ScaledU64(25420);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmFunctions => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(536688);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmGlobals => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(176902);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmTableEntries => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(29639);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmTypes => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(1048891);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmDataSegments => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(236970);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmElemSegments => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(317249);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmImports => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(694667);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmExports => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(427037);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmDataSegmentBytes => {
+                    cpu.const_term = 66075;
+                    cpu.lin_term = ScaledU64(28);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmInstructions => {
+                    cpu.const_term = 25059;
+                    cpu.lin_term = ScaledU64(0);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmFunctions => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(7503);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmGlobals => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(10761);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmTableEntries => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(3211);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmTypes => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(0);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmDataSegments => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(16370);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmElemSegments => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(28309);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmImports => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(683461);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmExports => {
+                    cpu.const_term = 0;
+                    cpu.lin_term = ScaledU64(297065);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmDataSegmentBytes => {
+                    cpu.const_term = 25191;
+                    cpu.lin_term = ScaledU64(14);
+                }
             }
 
             // define the memory cost model parameters
@@ -471,6 +613,108 @@ impl Default for BudgetImpl {
                     mem.const_term = 0;
                     mem.lin_term = ScaledU64(0);
                 }
+
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmInstructions => {
+                    mem.const_term = 17564;
+                    mem.lin_term = ScaledU64(6457);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmFunctions => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(47464);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmGlobals => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(13420);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmTableEntries => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(6285);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmTypes => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(64670);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmDataSegments => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(29074);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmElemSegments => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(48095);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmImports => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(102890);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmExports => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(36394);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::ParseWasmDataSegmentBytes => {
+                    mem.const_term = 17580;
+                    mem.lin_term = ScaledU64(257);
+                }
+
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmInstructions => {
+                    mem.const_term = 70192;
+                    mem.lin_term = ScaledU64(0);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmFunctions => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(14613);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmGlobals => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(6833);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmTableEntries => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(1025);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmTypes => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(0);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmDataSegments => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(129632);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmElemSegments => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(13665);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmImports => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(77273);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmExports => {
+                    mem.const_term = 0;
+                    mem.lin_term = ScaledU64(9176);
+                }
+                #[cfg(feature = "next")]
+                ContractCostType::InstantiateWasmDataSegmentBytes => {
+                    mem.const_term = 69256;
+                    mem.lin_term = ScaledU64(126);
+                }
             }
         }
 
@@ -790,7 +1034,7 @@ impl Budget {
             .tracker
             .cost_tracker
             .get(ty as usize)
-            .map(|x| x.clone())
+            .map(|x| *x)
             .ok_or_else(|| (ScErrorType::Budget, ScErrorCode::InternalError).into())
     }
 
diff --git a/soroban-env-host/src/budget/dimension.rs b/soroban-env-host/src/budget/dimension.rs
index 326b9d66..75a62a08 100644
--- a/soroban-env-host/src/budget/dimension.rs
+++ b/soroban-env-host/src/budget/dimension.rs
@@ -8,7 +8,7 @@ use core::fmt::Debug;
 pub(crate) struct IsCpu(pub(crate) bool);
 pub(crate) struct IsShadowMode(pub(crate) bool);
 
-#[derive(Clone, Default)]
+#[derive(Clone)]
 pub(crate) struct BudgetDimension {
     /// A set of cost models that map input values (eg. event counts, object
     /// sizes) from some CostType to whatever concrete resource type is being
@@ -36,6 +36,18 @@ pub(crate) struct BudgetDimension {
     pub(crate) shadow_total_count: u64,
 }
 
+impl Default for BudgetDimension {
+    fn default() -> Self {
+        Self {
+            cost_models: [MeteredCostComponent::default(); ContractCostType::variants().len()],
+            limit: Default::default(),
+            total_count: Default::default(),
+            shadow_limit: Default::default(),
+            shadow_total_count: Default::default(),
+        }
+    }
+}
+
 impl Debug for BudgetDimension {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         writeln!(
diff --git a/soroban-env-host/src/budget/model.rs b/soroban-env-host/src/budget/model.rs
index 39b9c20d..4cda325f 100644
--- a/soroban-env-host/src/budget/model.rs
+++ b/soroban-env-host/src/budget/model.rs
@@ -38,7 +38,7 @@ pub trait HostCostModel {
 const COST_MODEL_LIN_TERM_SCALE_BITS: u32 = 7;
 
 /// A helper type that wraps an u64 to signify the wrapped value have been scaled.
-#[derive(Clone, Default, Debug)]
+#[derive(Clone, Copy, Default, Debug)]
 pub struct ScaledU64(pub(crate) u64);
 
 impl ScaledU64 {
@@ -81,7 +81,7 @@ impl From<f64> for ScaledU64 {
     }
 }
 
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Copy, Debug, Default)]
 pub struct MeteredCostComponent {
     pub const_term: u64,
     pub lin_term: ScaledU64,
diff --git a/soroban-env-host/src/budget/wasmi_helper.rs b/soroban-env-host/src/budget/wasmi_helper.rs
index b0c94e37..b976d4c0 100644
--- a/soroban-env-host/src/budget/wasmi_helper.rs
+++ b/soroban-env-host/src/budget/wasmi_helper.rs
@@ -1,5 +1,8 @@
 use crate::{
-    budget::AsBudget, host::error::TryBorrowOrErr, xdr::ContractCostType, Host, HostError,
+    budget::{AsBudget, Budget},
+    host::error::TryBorrowOrErr,
+    xdr::ContractCostType,
+    Host, HostError,
 };
 use wasmi::{errors, FuelConsumptionMode, FuelCosts, ResourceLimiter};
 
@@ -110,9 +113,9 @@ pub(crate) fn load_calibrated_fuel_costs() -> FuelCosts {
     fuel_costs
 }
 
-pub(crate) fn get_wasmi_config(host: &Host) -> Result<wasmi::Config, HostError> {
+pub(crate) fn get_wasmi_config(budget: &Budget) -> Result<wasmi::Config, HostError> {
     let mut config = wasmi::Config::default();
-    let fuel_costs = host.as_budget().0.try_borrow_or_err()?.fuel_costs;
+    let fuel_costs = budget.0.try_borrow_or_err()?.fuel_costs;
 
     // Turn off most optional wasm features, leaving on some post-MVP features
     // commonly enabled by Rust and Clang. Make sure all unused features are
diff --git a/soroban-env-host/src/cost_runner/cost_types/vm_ops.rs b/soroban-env-host/src/cost_runner/cost_types/vm_ops.rs
index b3082940..b21d02c0 100644
--- a/soroban-env-host/src/cost_runner/cost_types/vm_ops.rs
+++ b/soroban-env-host/src/cost_runner/cost_types/vm_ops.rs
@@ -1,19 +1,23 @@
 use crate::{
     cost_runner::{CostRunner, CostType},
-    xdr::ContractCostType::VmInstantiation,
-    xdr::Hash,
+    vm::ParsedModule,
+    xdr::{ContractCostType::VmInstantiation, Hash},
     Vm,
 };
 use std::{hint::black_box, rc::Rc};
 
-pub struct VmInstantiationRun;
-
 #[derive(Clone)]
 pub struct VmInstantiationSample {
     pub id: Option<Hash>,
     pub wasm: Vec<u8>,
+    pub module: Rc<ParsedModule>,
 }
 
+// Protocol 20 coarse and unified cost model
+#[cfg(not(feature = "next"))]
+pub struct VmInstantiationRun;
+
+#[cfg(not(feature = "next"))]
 impl CostRunner for VmInstantiationRun {
     const COST_TYPE: CostType = CostType::Contract(VmInstantiation);
 
@@ -24,7 +28,15 @@ impl CostRunner for VmInstantiationRun {
     type RecycledType = (Option<Rc<Vm>>, Vec<u8>);
 
     fn run_iter(host: &crate::Host, _iter: u64, sample: Self::SampleType) -> Self::RecycledType {
-        let vm = black_box(Vm::new(host, sample.id.unwrap(), &sample.wasm[..]).unwrap());
+        let vm = black_box(
+            Vm::new_with_cost_inputs(
+                host,
+                sample.id.unwrap(),
+                &sample.wasm[..],
+                sample.module.cost_inputs.clone(),
+            )
+            .unwrap(),
+        );
         (Some(vm), sample.wasm)
     }
 
@@ -37,3 +49,194 @@ impl CostRunner for VmInstantiationRun {
         black_box((None, sample.wasm))
     }
 }
+
+// Protocol 21 refined and split/caching cost model.
+#[cfg(feature = "next")]
+pub use v21::*;
+#[cfg(feature = "next")]
+mod v21 {
+    use super::*;
+    use crate::vm::ParsedModule;
+    use crate::xdr::ContractCostType::{
+        InstantiateWasmDataSegmentBytes, InstantiateWasmDataSegments, InstantiateWasmElemSegments,
+        InstantiateWasmExports, InstantiateWasmFunctions, InstantiateWasmGlobals,
+        InstantiateWasmImports, InstantiateWasmInstructions, InstantiateWasmTableEntries,
+        InstantiateWasmTypes, ParseWasmDataSegmentBytes, ParseWasmDataSegments,
+        ParseWasmElemSegments, ParseWasmExports, ParseWasmFunctions, ParseWasmGlobals,
+        ParseWasmImports, ParseWasmInstructions, ParseWasmTableEntries, ParseWasmTypes,
+        VmCachedInstantiation,
+    };
+
+    macro_rules! impl_costrunner_for_parse_cost_type {
+        ($RUNNER:ty, $COST:ident) => {
+            impl CostRunner for $RUNNER {
+                const COST_TYPE: CostType = CostType::Contract($COST);
+
+                const RUN_ITERATIONS: u64 = 10;
+
+                type SampleType = VmInstantiationSample;
+
+                type RecycledType = (Option<Rc<ParsedModule>>, Vec<u8>);
+
+                fn run_iter(
+                    host: &crate::Host,
+                    _iter: u64,
+                    sample: Self::SampleType,
+                ) -> Self::RecycledType {
+                    let module = black_box(Rc::new(
+                        ParsedModule::new(
+                            host,
+                            sample.module.module.engine(),
+                            &sample.wasm[..],
+                            sample.module.cost_inputs.clone(),
+                        )
+                        .unwrap(),
+                    ));
+                    (Some(module), sample.wasm)
+                }
+
+                fn run_baseline_iter(
+                    host: &crate::Host,
+                    _iter: u64,
+                    sample: Self::SampleType,
+                ) -> Self::RecycledType {
+                    black_box(host.charge_budget($COST, Some(0)).unwrap());
+                    black_box((None, sample.wasm))
+                }
+            }
+        };
+    }
+
+    macro_rules! impl_costrunner_for_instantiation_cost_type {
+        ($RUNNER:ty, $COST:ident, $IS_CONST:expr) => {
+            impl CostRunner for $RUNNER {
+                const COST_TYPE: CostType = CostType::Contract($COST);
+
+                const RUN_ITERATIONS: u64 = 10;
+
+                type SampleType = VmInstantiationSample;
+
+                type RecycledType = (Option<Rc<Vm>>, Vec<u8>);
+
+                fn run_iter(
+                    host: &crate::Host,
+                    _iter: u64,
+                    sample: Self::SampleType,
+                ) -> Self::RecycledType {
+                    let vm = black_box(
+                        Vm::from_parsed_module(host, sample.id.unwrap(), sample.module).unwrap(),
+                    );
+                    (Some(vm), sample.wasm)
+                }
+
+                fn run_baseline_iter(
+                    host: &crate::Host,
+                    _iter: u64,
+                    sample: Self::SampleType,
+                ) -> Self::RecycledType {
+                    if $IS_CONST {
+                        black_box(host.charge_budget($COST, None).unwrap());
+                    } else {
+                        black_box(host.charge_budget($COST, Some(0)).unwrap());
+                    }
+                    black_box((None, sample.wasm))
+                }
+            }
+        };
+    }
+
+    // This cost-type is recycled as unrefined-model, parse-only phase.
+    pub struct VmInstantiationRun;
+    // This cost-type is recycled as unrefined-model, instantiate-only phase.
+    pub struct VmCachedInstantiationRun;
+
+    pub struct ParseWasmInstructionsRun;
+    pub struct ParseWasmFunctionsRun;
+    pub struct ParseWasmGlobalsRun;
+    pub struct ParseWasmTableEntriesRun;
+    pub struct ParseWasmTypesRun;
+    pub struct ParseWasmDataSegmentsRun;
+    pub struct ParseWasmElemSegmentsRun;
+    pub struct ParseWasmImportsRun;
+    pub struct ParseWasmExportsRun;
+    pub struct ParseWasmDataSegmentBytesRun;
+
+    pub struct InstantiateWasmInstructionsRun;
+    pub struct InstantiateWasmFunctionsRun;
+    pub struct InstantiateWasmGlobalsRun;
+    pub struct InstantiateWasmTableEntriesRun;
+    pub struct InstantiateWasmTypesRun;
+    pub struct InstantiateWasmDataSegmentsRun;
+    pub struct InstantiateWasmElemSegmentsRun;
+    pub struct InstantiateWasmImportsRun;
+    pub struct InstantiateWasmExportsRun;
+    pub struct InstantiateWasmDataSegmentBytesRun;
+
+    impl_costrunner_for_parse_cost_type!(VmInstantiationRun, VmInstantiation);
+    impl_costrunner_for_parse_cost_type!(ParseWasmInstructionsRun, ParseWasmInstructions);
+    impl_costrunner_for_parse_cost_type!(ParseWasmFunctionsRun, ParseWasmFunctions);
+    impl_costrunner_for_parse_cost_type!(ParseWasmGlobalsRun, ParseWasmGlobals);
+    impl_costrunner_for_parse_cost_type!(ParseWasmTableEntriesRun, ParseWasmTableEntries);
+    impl_costrunner_for_parse_cost_type!(ParseWasmTypesRun, ParseWasmTypes);
+    impl_costrunner_for_parse_cost_type!(ParseWasmDataSegmentsRun, ParseWasmDataSegments);
+    impl_costrunner_for_parse_cost_type!(ParseWasmElemSegmentsRun, ParseWasmElemSegments);
+    impl_costrunner_for_parse_cost_type!(ParseWasmImportsRun, ParseWasmImports);
+    impl_costrunner_for_parse_cost_type!(ParseWasmExportsRun, ParseWasmExports);
+    impl_costrunner_for_parse_cost_type!(ParseWasmDataSegmentBytesRun, ParseWasmDataSegmentBytes);
+
+    impl_costrunner_for_instantiation_cost_type!(
+        VmCachedInstantiationRun,
+        VmCachedInstantiation,
+        false
+    );
+    impl_costrunner_for_instantiation_cost_type!(
+        InstantiateWasmInstructionsRun,
+        InstantiateWasmInstructions,
+        true
+    );
+    impl_costrunner_for_instantiation_cost_type!(
+        InstantiateWasmFunctionsRun,
+        InstantiateWasmFunctions,
+        false
+    );
+    impl_costrunner_for_instantiation_cost_type!(
+        InstantiateWasmGlobalsRun,
+        InstantiateWasmGlobals,
+        false
+    );
+    impl_costrunner_for_instantiation_cost_type!(
+        InstantiateWasmTableEntriesRun,
+        InstantiateWasmTableEntries,
+        false
+    );
+    impl_costrunner_for_instantiation_cost_type!(
+        InstantiateWasmTypesRun,
+        InstantiateWasmTypes,
+        true
+    );
+    impl_costrunner_for_instantiation_cost_type!(
+        InstantiateWasmDataSegmentsRun,
+        InstantiateWasmDataSegments,
+        false
+    );
+    impl_costrunner_for_instantiation_cost_type!(
+        InstantiateWasmElemSegmentsRun,
+        InstantiateWasmElemSegments,
+        false
+    );
+    impl_costrunner_for_instantiation_cost_type!(
+        InstantiateWasmImportsRun,
+        InstantiateWasmImports,
+        false
+    );
+    impl_costrunner_for_instantiation_cost_type!(
+        InstantiateWasmExportsRun,
+        InstantiateWasmExports,
+        false
+    );
+    impl_costrunner_for_instantiation_cost_type!(
+        InstantiateWasmDataSegmentBytesRun,
+        InstantiateWasmDataSegmentBytes,
+        false
+    );
+}
diff --git a/soroban-env-host/src/host.rs b/soroban-env-host/src/host.rs
index ecd9a511..c9d07f8c 100644
--- a/soroban-env-host/src/host.rs
+++ b/soroban-env-host/src/host.rs
@@ -10,6 +10,7 @@ use crate::{
     impl_wrapping_obj_to_num,
     num::*,
     storage::Storage,
+    vm::ModuleCache,
     xdr::{
         int128_helpers, AccountId, Asset, ContractCostType, ContractEventType, ContractExecutable,
         ContractIdPreimage, ContractIdPreimageFromAddress, CreateContractArgs, Duration, Hash,
@@ -79,6 +80,7 @@ pub struct CoverageScoreboard {
 
 #[derive(Clone, Default)]
 struct HostImpl {
+    module_cache: RefCell<Option<ModuleCache>>,
     source_account: RefCell<Option<AccountId>>,
     ledger: RefCell<Option<LedgerInfo>>,
     objects: RefCell<Vec<HostObject>>,
@@ -186,7 +188,12 @@ macro_rules! impl_checked_borrow_helpers {
         }
     };
 }
-
+impl_checked_borrow_helpers!(
+    module_cache,
+    Option<ModuleCache>,
+    try_borrow_module_cache,
+    try_borrow_module_cache_mut
+);
 impl_checked_borrow_helpers!(
     source_account,
     Option<AccountId>,
@@ -315,6 +322,7 @@ impl Host {
         #[cfg(all(not(target_family = "wasm"), feature = "tracy"))]
         let _client = tracy_client::Client::start();
         Self(Rc::new(HostImpl {
+            module_cache: RefCell::new(None),
             source_account: RefCell::new(None),
             ledger: RefCell::new(None),
             objects: Default::default(),
@@ -345,6 +353,15 @@ impl Host {
         }))
     }
 
+    pub fn maybe_add_module_cache(&self) -> Result<(), HostError> {
+        if cfg!(feature = "next")
+            && self.get_ledger_protocol_version()? >= ModuleCache::MIN_LEDGER_VERSION
+        {
+            *self.try_borrow_module_cache_mut()? = Some(ModuleCache::new(self)?);
+        }
+        Ok(())
+    }
+
     pub fn set_source_account(&self, source_account: AccountId) -> Result<(), HostError> {
         *self.try_borrow_source_account_mut()? = Some(source_account);
         Ok(())
diff --git a/soroban-env-host/src/host/data_helper.rs b/soroban-env-host/src/host/data_helper.rs
index 89d6858b..cada88a8 100644
--- a/soroban-env-host/src/host/data_helper.rs
+++ b/soroban-env-host/src/host/data_helper.rs
@@ -6,6 +6,7 @@ use crate::{
     err,
     host::metered_clone::{MeteredAlloc, MeteredClone},
     storage::{InstanceStorageMap, Storage},
+    vm::VersionedContractCodeCostInputs,
     xdr::{
         AccountEntry, AccountId, Asset, BytesM, ContractCodeEntry, ContractDataDurability,
         ContractDataEntry, ContractExecutable, ContractIdPreimage, ExtensionPoint, Hash,
@@ -122,7 +123,10 @@ impl Host {
         )
     }
 
-    pub(crate) fn retrieve_wasm_from_storage(&self, wasm_hash: &Hash) -> Result<BytesM, HostError> {
+    pub(crate) fn retrieve_wasm_from_storage(
+        &self,
+        wasm_hash: &Hash,
+    ) -> Result<(BytesM, VersionedContractCodeCostInputs), HostError> {
         let key = self.contract_code_ledger_key(wasm_hash)?;
         match &self
             .try_borrow_storage_mut()?
@@ -130,7 +134,23 @@ impl Host {
             .map_err(|e| self.decorate_contract_code_storage_error(e, wasm_hash))?
             .data
         {
-            LedgerEntryData::ContractCode(e) => e.code.metered_clone(self),
+            LedgerEntryData::ContractCode(e) => {
+                let code = e.code.metered_clone(self)?;
+                #[allow(unused_mut)]
+                let mut costs = VersionedContractCodeCostInputs::V0 {
+                    wasm_bytes: code.len(),
+                };
+                #[cfg(feature = "next")]
+                match &e.ext {
+                    crate::xdr::ContractCodeEntryExt::V0 => (),
+                    crate::xdr::ContractCodeEntryExt::V1(v1) => {
+                        costs = VersionedContractCodeCostInputs::V1(
+                            v1.cost_inputs.metered_clone(self.as_budget())?,
+                        )
+                    }
+                };
+                Ok((code, costs))
+            }
             _ => Err(err!(
                 self,
                 (ScErrorType::Storage, ScErrorCode::InternalError),
diff --git a/soroban-env-host/src/host/declared_size.rs b/soroban-env-host/src/host/declared_size.rs
index d8ad6f5d..d6ab4a43 100644
--- a/soroban-env-host/src/host/declared_size.rs
+++ b/soroban-env-host/src/host/declared_size.rs
@@ -1,3 +1,5 @@
+#[cfg(feature = "next")]
+use crate::xdr::{ContractCodeCostInputs, ContractCodeEntryV1};
 use crate::{
     auth::{
         AccountAuthorizationTracker, AccountAuthorizationTrackerSnapshot, AuthorizedInvocation,
@@ -157,7 +159,11 @@ impl_declared_size_type!(LedgerKeyContractCode, 36);
 impl_declared_size_type!(LedgerEntryExt, 33);
 impl_declared_size_type!(AccountEntry, 216);
 impl_declared_size_type!(TrustLineEntry, 128);
+#[cfg(feature = "next")]
+impl_declared_size_type!(ContractCodeCostInputs, 40);
 impl_declared_size_type!(ContractCodeEntry, 64);
+#[cfg(feature = "next")]
+impl_declared_size_type!(ContractCodeEntryV1, 40);
 // TtlEntry must be declared as it's used in e2e to build
 // The TtlEntryMap, but is not otherwise cloned anywhere.
 impl_declared_size_type!(TtlEntry, 36);
@@ -172,7 +178,80 @@ impl_declared_size_type!(CreateContractArgs, 98);
 impl_declared_size_type!(InvokeContractArgs, 88);
 impl_declared_size_type!(ContractIdPreimage, 65);
 impl_declared_size_type!(ContractDataDurability, 4);
+
+// NB: ExtensionPoint is a 1-variant enum with no payload, which Rust optimizes
+// to take zero bytes of memroy -- but in XDR it's a 4-byte type like any other
+// union.
+//
+// It exists to help allow the protocol to evolve, as a placeholder type in XDR.
+// ExtensionPoints are sprinkled around the XDR definitions where we expect to
+// need to add stuff in the future. Any field with ExtensionPoint as its type
+// can be _replaced_ in subsequent protocols with some _other_ union type that
+// has a variant with a zero discriminant, and some other non-zero-discriminant
+// variants with new payloads. If old data (written when the field was
+// ExtensionPoint) is read by new code (which knows about the new union type),
+// the new code will just interpret the zero-discriminant old value as the zero
+// case in the new type. All well and good!
+//
+// But none of this works in Rust, because Rust doesn't treat unions (enums) as
+// variable size: it allocates space for the largest variant that can occupy the
+// enum, and it allocates zero bytes for ExtensionPoints. So when you upgrade a
+// union field in XDR, if the new variant is bigger than all existing variants,
+// the Rust size of the field will just change (and since ExtensionPoint is
+// zero-sized, this happens _any_ time you replace an ExtensionPoint with some
+// nonempty type). There's no real getting around this.
+//
+// "Luckily" (from the perspective of deterministic replay) we don't charge the
+// cost model based on Rust's "real sizes" of anything; we charge the cost model
+// based on stable _declared_ sizes we write down explicitly here. So what
+// happens when you upgrade a union field in a way that changes the Rust size of
+// a type is that the Rust size _diverges a bit_ from the declared size: in
+// other words the metering based on declared sizes gets a bit inaccurate, and
+// the test below (`test_declared_size`) that checks real sizes are less than or
+// equal to declared sizes needs to have an exception written into it to handle
+// this divergence.
+//
+// This is not ideal. It means in some cases the metering will be off, at least
+// when reading old data into the new enum (and, say, cloning it) we'll treat it
+// as small when really it is a bit bigger.
+//
+// We can recover a degree of correctness at least in the cases where we're
+// writing new data (occupying the new variants of the extended enum) by
+// essentially pretending that the ext field is a sort of magical smart pointer
+// that occupies zero bytes of its own but points to the new variant as "deep
+// substructure" (the same way we do with real pointer fields).
+//
+// In other words, we can use the following idiom:
+//
+//  - Assume the enclosing struct was `S`, and it previously had an
+//    `ExtensionPoint` `S.ext`, and now it has some other type in `S.ext`. Let's
+//    call the new type we want to add `T` and say we added it by replacing the
+//    `ExtensionPoint` with an enum `E` with variants `V0` (covering the old
+//    `ExtensionPoint`) and `V1(T)`.
+//
+//  - Previously we charged nothing to clone `S.ext`, because it was zero-sized.
+//    So we have to continue to charge nothing when cloning the new field when
+//    it is in its `E::V0` state: that's what the old `ExtensionPoint` data will
+//    be interpreted as when it's deserialized, so we have to treat it as we did
+//    before. This is the case where metering has become incorrect, because `E`
+//    is larger than `ExtensionPoint` but we can't acknowledge that fact.
+//
+//  - To charge nothing for `E::V0`, we actually have to _not_ declare `E` as a
+//    `MeteredClone` / `declared_size` type itself. Rather, inside the body of
+//    `S::charge_for_substructure` we match on `S.ext` and do nothing in the
+//    `E::V0` case, only do nonzero work in the `E::V1` case.
+//
+//  - In the `E::V1` case we can call through to `T::charge_for_substructure`,
+//    treating it almost like it was a separate heap allocation. It still gets
+//    charged for, just "as if" it were out-of-line, rather than inline with `S`
+//    where it happens to be located.
+//
+// This is all a bit weird and awful. It's a mistake that we only noticed after
+// we finalized the metering system of Soroban. We're stuck with it at this
+// point. Hopefully most instances will be fairly benign (and if types get
+// upgraded to their new variants, hopefully also transient!)
 impl_declared_size_type!(ExtensionPoint, 0);
+
 impl_declared_size_type!(ScContractInstance, 64);
 impl_declared_size_type!(SorobanAuthorizationEntry, 240);
 impl_declared_size_type!(SorobanAuthorizedInvocation, 128);
@@ -312,8 +391,21 @@ mod test {
         expect!["8"].assert_eq(size_of::<SymbolSmallIter>().to_string().as_str());
         expect!["32"].assert_eq(size_of::<U256>().to_string().as_str());
         expect!["32"].assert_eq(size_of::<I256>().to_string().as_str());
+
+        #[rustversion::before(1.77)]
         #[cfg(target_arch = "x86_64")]
-        expect!["40"].assert_eq(size_of::<HostObject>().to_string().as_str());
+        fn check_x64_host_object_size_that_changed_at_rust_1_77() {
+            expect!["40"].assert_eq(size_of::<HostObject>().to_string().as_str());
+        }
+        #[rustversion::since(1.77)]
+        #[cfg(target_arch = "x86_64")]
+        fn check_x64_host_object_size_that_changed_at_rust_1_77() {
+            expect!["48"].assert_eq(size_of::<HostObject>().to_string().as_str());
+        }
+
+        #[cfg(target_arch = "x86_64")]
+        check_x64_host_object_size_that_changed_at_rust_1_77();
+
         #[cfg(target_arch = "aarch64")]
         expect!["48"].assert_eq(size_of::<HostObject>().to_string().as_str());
         expect!["16"].assert_eq(size_of::<HostError>().to_string().as_str());
@@ -382,7 +474,17 @@ mod test {
         expect!["33"].assert_eq(size_of::<LedgerEntryExt>().to_string().as_str());
         expect!["216"].assert_eq(size_of::<AccountEntry>().to_string().as_str());
         expect!["128"].assert_eq(size_of::<TrustLineEntry>().to_string().as_str());
+        #[cfg(feature = "next")]
+        expect!["40"].assert_eq(size_of::<ContractCodeCostInputs>().to_string().as_str());
+        #[cfg(not(feature = "next"))]
         expect!["56"].assert_eq(size_of::<ContractCodeEntry>().to_string().as_str());
+        // ContractCodeEntry had an ExtensionPoint added to it and is now 40
+        // bytes larger than its original size (and for some reason its declared
+        // size was 64 bytes, even though its original size wasonly 56 bytes)
+        #[cfg(feature = "next")]
+        expect!["104"].assert_eq(size_of::<ContractCodeEntry>().to_string().as_str());
+        #[cfg(feature = "next")]
+        expect!["40"].assert_eq(size_of::<ContractCodeEntryV1>().to_string().as_str());
         expect!["36"].assert_eq(size_of::<TtlEntry>().to_string().as_str());
 
         // NB: a couple structs shrank between rust 1.75 and 1.76 but this is harmless
@@ -448,6 +550,19 @@ mod test {
                     <$t as DeclaredSizeForMetering>::DECLARED_SIZE
                 );
             };
+            // This variant allows accounting for the case where an
+            // ExtensionPoint has been upgraded to some other type, causing the
+            // enclosing type to exceed its declared size: you can include an
+            // extension type as a second argument and it will be added to the
+            // declared size for the sake of retaining _some_ check here.
+            // Unfortunately this is kinda the best we can do. See the long
+            // comment above around the zero declared_size of ExtensionPoint.
+            ($t:ty, $ext:ty) => {
+                ma::assert_le!(
+                    size_of::<$t>() as u64,
+                    <$t as DeclaredSizeForMetering>::DECLARED_SIZE + (size_of::<$ext>() as u64)
+                );
+            };
         }
         // primitive types
         assert_mem_size_le_declared_size!(bool);
@@ -556,7 +671,14 @@ mod test {
         assert_mem_size_le_declared_size!(LedgerEntryExt);
         assert_mem_size_le_declared_size!(AccountEntry);
         assert_mem_size_le_declared_size!(TrustLineEntry);
+        #[cfg(feature = "next")]
+        assert_mem_size_le_declared_size!(ContractCodeCostInputs);
+        #[cfg(not(feature = "next"))]
         assert_mem_size_le_declared_size!(ContractCodeEntry);
+        #[cfg(feature = "next")]
+        assert_mem_size_le_declared_size!(ContractCodeEntry, ContractCodeEntryV1);
+        #[cfg(feature = "next")]
+        assert_mem_size_le_declared_size!(ContractCodeEntryV1);
         assert_mem_size_le_declared_size!(TtlEntry);
         assert_mem_size_le_declared_size!(LedgerKey);
         assert_mem_size_le_declared_size!(LedgerEntry);
diff --git a/soroban-env-host/src/host/frame.rs b/soroban-env-host/src/host/frame.rs
index e3afa2be..fea20138 100644
--- a/soroban-env-host/src/host/frame.rs
+++ b/soroban-env-host/src/host/frame.rs
@@ -640,8 +640,14 @@ impl Host {
         let args_vec = args.to_vec();
         match &instance.executable {
             ContractExecutable::Wasm(wasm_hash) => {
-                let code_entry = self.retrieve_wasm_from_storage(&wasm_hash)?;
-                let vm = Vm::new(self, id.metered_clone(self)?, code_entry.as_slice())?;
+                let contract_id = id.metered_clone(self)?;
+                let vm = if let Some(cache) = &*self.try_borrow_module_cache()? {
+                    let module = cache.get_module(self, wasm_hash)?;
+                    Vm::from_parsed_module(self, contract_id, module)?
+                } else {
+                    let (code, costs) = self.retrieve_wasm_from_storage(&wasm_hash)?;
+                    Vm::new_with_cost_inputs(self, contract_id, code.as_slice(), costs)?
+                };
                 let relative_objects = Vec::new();
                 self.with_frame(
                     Frame::ContractVM {
diff --git a/soroban-env-host/src/host/lifecycle.rs b/soroban-env-host/src/host/lifecycle.rs
index 7259f75e..b7ec0263 100644
--- a/soroban-env-host/src/host/lifecycle.rs
+++ b/soroban-env-host/src/host/lifecycle.rs
@@ -5,12 +5,13 @@ use crate::{
         metered_clone::{MeteredAlloc, MeteredClone},
         metered_write_xdr, ContractReentryMode, CreateContractArgs,
     },
+    vm::Vm,
     xdr::{
         Asset, ContractCodeEntry, ContractDataDurability, ContractExecutable, ContractIdPreimage,
         ContractIdPreimageFromAddress, ExtensionPoint, Hash, LedgerKey, LedgerKeyContractCode,
         ScAddress, ScErrorCode, ScErrorType,
     },
-    AddressObject, BytesObject, Host, HostError, Symbol, TryFromVal, Vm,
+    AddressObject, BytesObject, Host, HostError, Symbol, TryFromVal,
 };
 use std::rc::Rc;
 
@@ -164,6 +165,16 @@ impl Host {
             )
         })?;
 
+        // We're going to make a code entry with an ext field which, if zero,
+        // can be read as either a protocol v20 ExtensionPoint::V0 value, or a
+        // protocol v21 ContractCodeEntryExt::V0 value, depending on which
+        // protocol we were compiled with.
+        #[cfg(not(feature = "next"))]
+        let ext = ExtensionPoint::V0;
+
+        #[cfg(feature = "next")]
+        let mut ext = crate::xdr::ContractCodeEntryExt::V0;
+
         // Instantiate a temporary / throwaway VM using this wasm. This will do
         // both quick checks like "does this wasm have the right protocol number
         // to run on this network" and also a full parse-and-link pass to check
@@ -182,6 +193,20 @@ impl Host {
                 Hash(hash_bytes.metered_clone(self)?),
                 wasm_bytes_m.as_slice(),
             )?;
+            #[cfg(feature = "next")]
+            if self.get_ledger_protocol_version()? >= super::ModuleCache::MIN_LEDGER_VERSION {
+                // At this point we do a secondary parse on what we've checked to be a valid
+                // module in order to extract a refined cost model, which we'll store in the
+                // code entry's ext field, for future parsing and instantiations.
+                _check_vm.module.cost_inputs.charge_for_parsing(self)?;
+                ext = crate::xdr::ContractCodeEntryExt::V1(crate::xdr::ContractCodeEntryV1 {
+                    ext: ExtensionPoint::V0,
+                    cost_inputs: crate::vm::ParsedModule::extract_refined_contract_cost_inputs(
+                        self,
+                        wasm_bytes_m.as_slice(),
+                    )?,
+                });
+            }
         }
 
         let hash_obj = self.add_host_object(self.scbytes_from_slice(hash_bytes.as_slice())?)?;
@@ -191,24 +216,46 @@ impl Host {
             }),
             self,
         )?;
-        if !self
-            .try_borrow_storage_mut()?
+
+        let mut storage = self.try_borrow_storage_mut()?;
+
+        // We will definitely put the contract in the ledger if it isn't there yet.
+        #[allow(unused_mut)]
+        let mut should_put_contract = !storage
             .has(&code_key, self.as_budget())
-            .map_err(|e| self.decorate_contract_code_storage_error(e, &Hash(hash_bytes)))?
+            .map_err(|e| self.decorate_contract_code_storage_error(e, &Hash(hash_bytes)))?;
+
+        // We may also, in the cache-supporting protocol, overwrite the contract if its ext field changed.
+        #[cfg(feature = "next")]
+        if !should_put_contract
+            && self.get_ledger_protocol_version()? >= super::ModuleCache::MIN_LEDGER_VERSION
         {
-            self.with_mut_storage(|storage| {
-                let data = ContractCodeEntry {
-                    hash: Hash(hash_bytes),
-                    ext: ExtensionPoint::V0,
-                    code: wasm_bytes_m,
-                };
-                storage.put(
+            let entry = storage
+                .get(&code_key, self.as_budget())
+                .map_err(|e| self.decorate_contract_code_storage_error(e, &Hash(hash_bytes)))?;
+            if let crate::xdr::LedgerEntryData::ContractCode(ContractCodeEntry {
+                ext: old_ext,
+                ..
+            }) = &entry.data
+            {
+                should_put_contract = *old_ext != ext;
+            }
+        }
+
+        if should_put_contract {
+            let data = ContractCodeEntry {
+                hash: Hash(hash_bytes),
+                ext,
+                code: wasm_bytes_m,
+            };
+            storage
+                .put(
                     &code_key,
                     &Host::new_contract_code(self, data)?,
                     Some(self.get_min_live_until_ledger(ContractDataDurability::Persistent)?),
                     self.as_budget(),
                 )
-            })?;
+                .map_err(|e| self.decorate_contract_code_storage_error(e, &Hash(hash_bytes)))?;
         }
         Ok(hash_obj)
     }
diff --git a/soroban-env-host/src/host/metered_clone.rs b/soroban-env-host/src/host/metered_clone.rs
index b29be1f5..1b3c24bd 100644
--- a/soroban-env-host/src/host/metered_clone.rs
+++ b/soroban-env-host/src/host/metered_clone.rs
@@ -18,6 +18,8 @@
 
 use std::{cell::RefCell, iter::FromIterator, mem, rc::Rc};
 
+#[cfg(feature = "next")]
+use crate::xdr::{ContractCodeCostInputs, ContractCodeEntryExt, ContractCodeEntryV1};
 use crate::{
     budget::{AsBudget, DepthLimiter},
     builtin_contracts::base_types::Address,
@@ -314,6 +316,10 @@ impl MeteredClone for TimePoint {}
 impl MeteredClone for Duration {}
 impl MeteredClone for Hash {}
 impl MeteredClone for Uint256 {}
+#[cfg(feature = "next")]
+impl MeteredClone for ContractCodeCostInputs {}
+#[cfg(feature = "next")]
+impl MeteredClone for ContractCodeEntryV1 {}
 impl MeteredClone for ContractExecutable {}
 impl MeteredClone for AccountId {}
 impl MeteredClone for ScAddress {}
@@ -557,6 +563,12 @@ impl MeteredClone for ContractCodeEntry {
     const IS_SHALLOW: bool = false;
 
     fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
+        #[cfg(feature = "next")]
+        // self.ext is a former ExtensionEntry; see note on ExtensionEntry in declared_size.rs
+        match &self.ext {
+            ContractCodeEntryExt::V0 => (),
+            ContractCodeEntryExt::V1(v1) => v1.charge_for_substructure(budget.clone())?,
+        }
         self.code.charge_for_substructure(budget)
     }
 }
diff --git a/soroban-env-host/src/vm.rs b/soroban-env-host/src/vm.rs
index a469d1cb..b48208c2 100644
--- a/soroban-env-host/src/vm.rs
+++ b/soroban-env-host/src/vm.rs
@@ -11,6 +11,8 @@
 mod dispatch;
 mod fuel_refillable;
 mod func_info;
+mod module_cache;
+mod parsed_module;
 
 #[cfg(feature = "bench")]
 pub(crate) use dispatch::dummy0;
@@ -19,29 +21,58 @@ pub(crate) use dispatch::protocol_gated_dummy;
 
 use crate::{
     budget::{get_wasmi_config, AsBudget, Budget},
-    err,
     host::{
         error::TryBorrowOrErr,
         metered_clone::MeteredContainer,
         metered_hash::{CountingHasher, MeteredHash},
     },
-    meta::{self, get_ledger_protocol_version},
-    xdr::{ContractCostType, Hash, Limited, ReadXdr, ScEnvMetaEntry, ScErrorCode, ScErrorType},
+    xdr::{ContractCostType, Hash, ScErrorCode, ScErrorType},
     ConversionError, Host, HostError, Symbol, SymbolStr, TryIntoVal, Val, WasmiMarshal,
-    DEFAULT_XDR_RW_LIMITS,
 };
-use std::{cell::RefCell, io::Cursor, rc::Rc, time::Instant};
+use std::{cell::RefCell, collections::BTreeSet, rc::Rc};
 
 use fuel_refillable::FuelRefillable;
 use func_info::HOST_FUNCTIONS;
 
-use wasmi::{Engine, Instance, Linker, Memory, Module, Store, Value};
+pub use module_cache::ModuleCache;
+pub use parsed_module::{ParsedModule, VersionedContractCodeCostInputs};
+
+use wasmi::{Instance, Linker, Memory, Store, Value};
 
 use crate::VmCaller;
 use wasmi::{Caller, StoreContextMut};
+
 impl wasmi::core::HostError for HostError {}
 
 const MAX_VM_ARGS: usize = 32;
+#[cfg(feature = "next")]
+const WASM_STD_MEM_PAGE_SIZE_IN_BYTES: u32 = 0x10000;
+
+struct VmInstantiationTimer {
+    #[cfg(not(target_family = "wasm"))]
+    host: Host,
+    #[cfg(not(target_family = "wasm"))]
+    start: std::time::Instant,
+}
+impl VmInstantiationTimer {
+    fn new(_host: Host) -> Self {
+        VmInstantiationTimer {
+            #[cfg(not(target_family = "wasm"))]
+            host: _host,
+            #[cfg(not(target_family = "wasm"))]
+            start: std::time::Instant::now(),
+        }
+    }
+}
+#[cfg(not(target_family = "wasm"))]
+impl Drop for VmInstantiationTimer {
+    fn drop(&mut self) {
+        let _ = self.host.as_budget().track_time(
+            ContractCostType::VmInstantiation,
+            self.start.elapsed().as_nanos() as u64,
+        );
+    }
+}
 
 /// A [Vm] is a thin wrapper around an instance of [wasmi::Module]. Multiple
 /// [Vm]s may be held in a single [Host], and each contains a single WASM module
@@ -56,10 +87,8 @@ const MAX_VM_ARGS: usize = 32;
 /// will fail.
 pub struct Vm {
     pub(crate) contract_id: Hash,
-    // TODO: consider moving store and possibly module to Host so they can be
-    // recycled across calls. Or possibly beyond, to be recycled across txs.
-    // https://github.com/stellar/rs-soroban-env/issues/827
-    module: Module,
+    #[allow(dead_code)]
+    pub(crate) module: Rc<ParsedModule>,
     store: RefCell<Store<Host>>,
     instance: Instance,
     pub(crate) memory: Option<Memory>,
@@ -72,160 +101,41 @@ impl std::hash::Hash for Vm {
 }
 
 impl Vm {
-    fn check_contract_interface_version(
-        host: &Host,
-        interface_version: u64,
-    ) -> Result<(), HostError> {
-        let want_proto = {
-            let ledger_proto = host.get_ledger_protocol_version()?;
-            let env_proto = get_ledger_protocol_version(meta::INTERFACE_VERSION);
-            if ledger_proto <= env_proto {
-                // ledger proto should be before or equal to env proto
-                ledger_proto
-            } else {
-                return Err(err!(
-                    host,
-                    (ScErrorType::Context, ScErrorCode::InternalError),
-                    "ledger protocol number is ahead of supported env protocol number",
-                    ledger_proto,
-                    env_proto
-                ));
-            }
-        };
-
-        // Not used when "next" is enabled
-        #[cfg(not(feature = "next"))]
-        let got_pre = meta::get_pre_release_version(interface_version);
-
-        let got_proto = get_ledger_protocol_version(interface_version);
-
-        if got_proto < want_proto {
-            // Old protocols are finalized, we only support contracts
-            // with similarly finalized (zero) prerelease numbers.
-            //
-            // Note that we only enable this check if the "next" feature isn't enabled
-            // because a "next" stellar-core can still run a "curr" test using non-finalized
-            // test wasms. The "next" feature isn't safe for production and is meant to
-            // simulate the protocol version after the one currently supported in
-            // stellar-core, so bypassing this check for "next" is safe.
-            #[cfg(not(feature = "next"))]
-            if got_pre != 0 {
-                return Err(err!(
-                    host,
-                    (ScErrorType::WasmVm, ScErrorCode::InvalidInput),
-                    "contract pre-release number for old protocol is nonzero",
-                    got_pre
-                ));
-            }
-        } else if got_proto == want_proto {
-            // Relax this check as well for the "next" feature to allow for flexibility while testing.
-            // stellar-core can pass in an older protocol version, in which case the pre-release version
-            // will not match up with the "next" feature (The "next" pre-release version is always 1).
-            #[cfg(not(feature = "next"))]
-            {
-                // Current protocol might have a nonzero prerelease number; we will
-                // allow it only if it matches the current prerelease exactly.
-                let want_pre = meta::get_pre_release_version(meta::INTERFACE_VERSION);
-                if want_pre != got_pre {
-                    return Err(err!(
-                        host,
-                        (ScErrorType::WasmVm, ScErrorCode::InvalidInput),
-                        "contract pre-release number for current protocol does not match host",
-                        got_pre,
-                        want_pre
-                    ));
-                }
-            }
-        } else {
-            // Future protocols we don't allow. It might be nice (in the sense
-            // of "allowing uploads of a future-protocol contract that will go
-            // live as soon as the network upgrades to it") but there's a risk
-            // that the "future" protocol semantics baked in to a contract
-            // differ from the final semantics chosen by the network, so to be
-            // conservative we avoid even allowing this.
-            return Err(err!(
-                host,
-                (ScErrorType::WasmVm, ScErrorCode::InvalidInput),
-                "contract protocol number is newer than host",
-                got_proto
-            ));
-        }
-        Ok(())
-    }
-
-    fn check_meta_section(host: &Host, m: &Module) -> Result<u64, HostError> {
-        if let Some(env_meta) = Self::module_custom_section(m, meta::ENV_META_V0_SECTION_NAME) {
-            let mut limits = DEFAULT_XDR_RW_LIMITS;
-            limits.len = env_meta.len();
-            let mut cursor = Limited::new(Cursor::new(env_meta), limits);
-            if let Some(env_meta_entry) = ScEnvMetaEntry::read_xdr_iter(&mut cursor).next() {
-                let ScEnvMetaEntry::ScEnvMetaKindInterfaceVersion(v) =
-                    host.map_err(env_meta_entry)?;
-                Vm::check_contract_interface_version(host, v)?;
-                Ok(v)
-            } else {
-                Err(host.err(
-                    ScErrorType::WasmVm,
-                    ScErrorCode::InvalidInput,
-                    "contract missing environment interface version",
-                    &[],
-                ))
-            }
-        } else {
-            Err(host.err(
-                ScErrorType::WasmVm,
-                ScErrorCode::InvalidInput,
-                "contract missing metadata section",
-                &[],
-            ))
-        }
-    }
-
-    fn check_max_args(host: &Host, m: &Module) -> Result<(), HostError> {
-        for e in m.exports() {
-            match e.ty() {
-                wasmi::ExternType::Func(f) => {
-                    if f.params().len() > MAX_VM_ARGS || f.results().len() > MAX_VM_ARGS {
-                        return Err(host.err(
-                            ScErrorType::WasmVm,
-                            ScErrorCode::InvalidInput,
-                            "Too many arguments or results in wasm export",
-                            &[],
-                        ));
-                    }
-                }
-                _ => (),
-            }
-        }
-        Ok(())
+    #[cfg(feature = "testutils")]
+    pub fn get_all_host_functions() -> Vec<(&'static str, &'static str, u32)> {
+        HOST_FUNCTIONS
+            .iter()
+            .map(|hf| (hf.mod_str, hf.fn_str, hf.arity))
+            .collect()
     }
 
-    /// Instantiates a VM given the arguments provided in [`Self::new`]
+    /// Instantiates a VM given the arguments provided in [`Self::new`],
+    /// or [`Self::new_from_module_cache`]
     fn instantiate(
         host: &Host,
         contract_id: Hash,
-        module_wasm_code: &[u8],
+        parsed_module: Rc<ParsedModule>,
     ) -> Result<Rc<Self>, HostError> {
-        host.charge_budget(
-            ContractCostType::VmInstantiation,
-            Some(module_wasm_code.len() as u64),
-        )?;
+        let _span = tracy_span!("Vm::instantiate");
 
-        let config = get_wasmi_config(host)?;
-        let engine = Engine::new(&config);
-        let module = {
-            let _span0 = tracy_span!("parse module");
-            host.map_err(Module::new(&engine, module_wasm_code))?
-        };
+        let engine = parsed_module.module.engine();
+        let mut linker = <Linker<Host>>::new(engine);
+        let mut store = Store::new(engine, host.clone());
 
-        Self::check_max_args(host, &module)?;
-        let interface_version = Self::check_meta_section(host, &module)?;
-        let contract_proto = get_ledger_protocol_version(interface_version);
+        parsed_module.cost_inputs.charge_for_instantiation(host)?;
 
-        let mut store = Store::new(&engine, host.clone());
         store.limiter(|host| host);
 
-        let mut linker = <Linker<Host>>::new(&engine);
+        let module_imports: BTreeSet<(&str, &str)> = parsed_module
+            .module
+            .imports()
+            .filter(|i| i.ty().func().is_some())
+            .map(|i| {
+                let mod_str = i.module();
+                let fn_str = i.name();
+                (mod_str, fn_str)
+            })
+            .collect();
 
         {
             // We perform link-time protocol version gating here.
@@ -248,19 +158,24 @@ impl Vm {
             let ledger_proto = host.with_ledger_info(|li| Ok(li.protocol_version))?;
             for hf in HOST_FUNCTIONS {
                 if let Some(min_proto) = hf.min_proto {
-                    if contract_proto < min_proto || ledger_proto < min_proto {
+                    if parsed_module.proto_version < min_proto || ledger_proto < min_proto {
                         // We skip linking this hf instead of returning an error
                         // because we have to support old contracts during replay.
                         continue;
                     }
                 }
                 if let Some(max_proto) = hf.max_proto {
-                    if contract_proto > max_proto || ledger_proto > max_proto {
+                    if parsed_module.proto_version > max_proto || ledger_proto > max_proto {
                         // We skip linking this hf instead of returning an error
                         // because we have to support old contracts during replay.
                         continue;
                     }
                 }
+                // We only link the functions that are actually used by the
+                // contract. Linking is quite expensive.
+                if !module_imports.contains(&(hf.mod_str, hf.fn_str)) {
+                    continue;
+                }
                 let func = (hf.wrap)(&mut store);
                 host.map_err(
                     linker
@@ -272,7 +187,7 @@ impl Vm {
 
         let not_started_instance = {
             let _span0 = tracy_span!("instantiate module");
-            host.map_err(linker.instantiate(&mut store, &module))?
+            host.map_err(linker.instantiate(&mut store, &parsed_module.module))?
         };
 
         let instance = host.map_err(
@@ -292,13 +207,23 @@ impl Vm {
         // boundary.
         Ok(Rc::new(Self {
             contract_id,
-            module,
+            module: parsed_module,
             store: RefCell::new(store),
             instance,
             memory,
         }))
     }
 
+    pub fn from_parsed_module(
+        host: &Host,
+        contract_id: Hash,
+        parsed_module: Rc<ParsedModule>,
+    ) -> Result<Rc<Self>, HostError> {
+        let _span = tracy_span!("Vm::from_parsed_module");
+        VmInstantiationTimer::new(host.clone());
+        Self::instantiate(host, contract_id, parsed_module)
+    }
+
     /// Constructs a new instance of a [Vm] within the provided [Host],
     /// establishing a new execution context for a contract identified by
     /// `contract_id` with WASM bytecode provided in `module_wasm_code`.
@@ -317,26 +242,26 @@ impl Vm {
     ///
     /// This method is called automatically as part of [Host::invoke_function]
     /// and does not usually need to be called from outside the crate.
-    pub fn new(
+
+    pub fn new(host: &Host, contract_id: Hash, wasm: &[u8]) -> Result<Rc<Self>, HostError> {
+        let cost_inputs = VersionedContractCodeCostInputs::V0 {
+            wasm_bytes: wasm.len(),
+        };
+        Self::new_with_cost_inputs(host, contract_id, wasm, cost_inputs)
+    }
+
+    pub fn new_with_cost_inputs(
         host: &Host,
         contract_id: Hash,
-        module_wasm_code: &[u8],
+        wasm: &[u8],
+        cost_inputs: VersionedContractCodeCostInputs,
     ) -> Result<Rc<Self>, HostError> {
         let _span = tracy_span!("Vm::new");
-
-        if cfg!(not(target_family = "wasm")) {
-            let now = Instant::now();
-            let vm = Self::instantiate(host, contract_id, module_wasm_code)?;
-
-            host.as_budget().track_time(
-                ContractCostType::VmInstantiation,
-                now.elapsed().as_nanos() as u64,
-            )?;
-
-            Ok(vm)
-        } else {
-            Self::instantiate(host, contract_id, module_wasm_code)
-        }
+        VmInstantiationTimer::new(host.clone());
+        let config = get_wasmi_config(host.as_budget())?;
+        let engine = wasmi::Engine::new(&config);
+        let parsed_module = Rc::new(ParsedModule::new(host, &engine, wasm, cost_inputs)?);
+        Self::instantiate(host, contract_id, parsed_module)
     }
 
     pub(crate) fn get_memory(&self, host: &Host) -> Result<Memory, HostError> {
@@ -481,20 +406,10 @@ impl Vm {
         self.metered_func_call(host, func_sym, wasm_args.as_slice())
     }
 
-    fn module_custom_section(m: &Module, name: impl AsRef<str>) -> Option<&[u8]> {
-        m.custom_sections().iter().find_map(|s| {
-            if &*s.name == name.as_ref() {
-                Some(&*s.data)
-            } else {
-                None
-            }
-        })
-    }
-
     /// Returns the raw bytes content of a named custom section from the WASM
     /// module loaded into the [Vm], or `None` if no such custom section exists.
     pub fn custom_section(&self, name: impl AsRef<str>) -> Option<&[u8]> {
-        Self::module_custom_section(&self.module, name)
+        self.module.custom_section(name)
     }
 
     /// Utility function that synthesizes a `VmCaller<Host>` configured to point
diff --git a/soroban-env-host/src/vm/func_info.rs b/soroban-env-host/src/vm/func_info.rs
index 48123d70..7914828b 100644
--- a/soroban-env-host/src/vm/func_info.rs
+++ b/soroban-env-host/src/vm/func_info.rs
@@ -11,6 +11,10 @@ pub(crate) struct HostFuncInfo {
     /// as.
     pub(crate) fn_str: &'static str,
 
+    /// Number of I64-typed wasm arguments the function takes.
+    #[allow(dead_code)]
+    pub(crate) arity: u32,
+
     /// Function that takes a wasmi::Store and _wraps_ a dispatch function
     /// for this host function, with the specific type of the dispatch function,
     /// into a Func in the Store.
@@ -23,38 +27,54 @@ pub(crate) struct HostFuncInfo {
     pub(crate) max_proto: Option<u32>,
 }
 
+macro_rules! fn_arity {
+    (($($args:ident : $tys:ident),*)) => {
+        fn_arity!(@count_args 0, $($args:$tys)*)
+    };
+    (@count_args $n:expr, ) => {
+        $n
+    };
+    (@count_args $n:expr, $arg:ident:$ty:ident $($args:ident:$tys:ident)*) => {
+        fn_arity!(@count_args $n+1, $($args:$tys)*)
+    };
+}
+
 macro_rules! host_function_info_helper {
-    {$mod_str:literal, $fn_id:literal, $func_id:ident, $min_proto:literal, $max_proto:literal} => {
+    {$mod_str:literal, $fn_id:literal, $args:tt, $func_id:ident, $min_proto:literal, $max_proto:literal} => {
         HostFuncInfo {
             mod_str: $mod_str,
             fn_str: $fn_id,
+            arity: fn_arity!($args),
             wrap: |store| Func::wrap(store, dispatch::$func_id),
             min_proto: Some($min_proto),
             max_proto: Some($max_proto),
         }
     };
-    {$mod_str:literal, $fn_id:literal, $func_id:ident, $min_proto:literal, } => {
+    {$mod_str:literal, $fn_id:literal, $args:tt, $func_id:ident, $min_proto:literal, } => {
         HostFuncInfo {
             mod_str: $mod_str,
             fn_str: $fn_id,
+            arity: fn_arity!($args),
             wrap: |store| Func::wrap(store, dispatch::$func_id),
             min_proto: Some($min_proto),
             max_proto: None,
         }
     };
-    {$mod_str:literal, $fn_id:literal, $func_id:ident, , $max_proto:literal} => {
+    {$mod_str:literal, $fn_id:literal, $args:tt, $func_id:ident, , $max_proto:literal} => {
         HostFuncInfo {
             mod_str: $mod_str,
             fn_str: $fn_id,
+            arity: fn_arity!($args),
             wrap: |store| Func::wrap(store, dispatch::$func_id),
             min_proto: None,
             max_proto: Some($max_proto),
         }
     };
-    {$mod_str:literal, $fn_id:literal, $func_id:ident, , } => {
+    {$mod_str:literal, $fn_id:literal, $args:tt, $func_id:ident, , } => {
         HostFuncInfo {
             mod_str: $mod_str,
             fn_str: $fn_id,
+            arity: fn_arity!($args),
             wrap: |store| Func::wrap(store, dispatch::$func_id),
             min_proto: None,
             max_proto: None,
@@ -119,7 +139,7 @@ macro_rules! generate_host_function_infos {
                     // block repetition-level from the outer pattern in the
                     // expansion, flattening all functions from all 'mod' blocks
                     // into the a single array of HostFuncInfo structs.
-                    host_function_info_helper!{$mod_str, $fn_id, $func_id, $($min_proto)?, $($max_proto)?},
+                    host_function_info_helper!{$mod_str, $fn_id, $args, $func_id, $($min_proto)?, $($max_proto)?},
                 )*
             )*
         ];
diff --git a/soroban-env-host/src/vm/module_cache.rs b/soroban-env-host/src/vm/module_cache.rs
new file mode 100644
index 00000000..81facb2c
--- /dev/null
+++ b/soroban-env-host/src/vm/module_cache.rs
@@ -0,0 +1,100 @@
+use super::parsed_module::{ParsedModule, VersionedContractCodeCostInputs};
+use crate::{
+    budget::{get_wasmi_config, AsBudget},
+    host::metered_clone::MeteredClone,
+    xdr::{Hash, ScErrorCode, ScErrorType},
+    Host, HostError, MeteredOrdMap,
+};
+use std::rc::Rc;
+use wasmi::Engine;
+
+/// A [ModuleCache] is a cache of a set of Wasm modules that have been parsed
+/// but not yet instantiated, along with a shared and reusable [Engine] storing
+/// their code. The cache must be populated eagerly with all the contracts in a
+/// single [Host]'s lifecycle (at least) added all at once, since each wasmi
+/// [Engine] is locked during execution and no new modules can be added to it.
+#[derive(Clone, Default)]
+pub struct ModuleCache {
+    pub(crate) engine: Engine,
+    modules: MeteredOrdMap<Hash, Rc<ParsedModule>, Host>,
+}
+
+impl ModuleCache {
+    // ModuleCache should not be active until protocol version 21.
+    pub const MIN_LEDGER_VERSION: u32 = 21;
+
+    pub fn new(host: &Host) -> Result<Self, HostError> {
+        let config = get_wasmi_config(host.as_budget())?;
+        let engine = Engine::new(&config);
+        let modules = MeteredOrdMap::new();
+        #[allow(unused_mut)]
+        let mut cache = Self { engine, modules };
+        #[cfg(feature = "next")]
+        cache.add_stored_contracts(host)?;
+        Ok(cache)
+    }
+
+    #[cfg(feature = "next")]
+    pub fn add_stored_contracts(&mut self, host: &Host) -> Result<(), HostError> {
+        use crate::xdr::{ContractCodeEntry, ContractCodeEntryExt, LedgerEntryData, LedgerKey};
+        for (k, v) in host.try_borrow_storage()?.map.iter(host.as_budget())? {
+            if let LedgerKey::ContractCode(_) = &**k {
+                if let Some((e, _)) = v {
+                    if let LedgerEntryData::ContractCode(ContractCodeEntry { code, hash, ext }) =
+                        &e.data
+                    {
+                        let code_cost_inputs = match ext {
+                            ContractCodeEntryExt::V0 => VersionedContractCodeCostInputs::V0 {
+                                wasm_bytes: code.len(),
+                            },
+                            ContractCodeEntryExt::V1(v1) => VersionedContractCodeCostInputs::V1(
+                                v1.cost_inputs.metered_clone(host.as_budget())?,
+                            ),
+                        };
+                        self.parse_and_cache_module(host, hash, code, code_cost_inputs)?;
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
+
+    pub fn parse_and_cache_module(
+        &mut self,
+        host: &Host,
+        contract_id: &Hash,
+        wasm: &[u8],
+        cost_inputs: VersionedContractCodeCostInputs,
+    ) -> Result<(), HostError> {
+        if self.modules.contains_key(contract_id, host)? {
+            return Err(host.err(
+                ScErrorType::Context,
+                ScErrorCode::InternalError,
+                "module cache already contains contract",
+                &[],
+            ));
+        }
+        let parsed_module = Rc::new(ParsedModule::new(host, &self.engine, &wasm, cost_inputs)?);
+        self.modules =
+            self.modules
+                .insert(contract_id.metered_clone(host)?, parsed_module, host)?;
+        Ok(())
+    }
+
+    pub fn get_module(
+        &self,
+        host: &Host,
+        contract_id: &Hash,
+    ) -> Result<Rc<ParsedModule>, HostError> {
+        if let Some(m) = self.modules.get(contract_id, host)? {
+            return Ok(m.clone());
+        } else {
+            Err(host.err(
+                ScErrorType::Context,
+                ScErrorCode::InternalError,
+                "module cache missing contract",
+                &[],
+            ))
+        }
+    }
+}
diff --git a/soroban-env-host/src/vm/parsed_module.rs b/soroban-env-host/src/vm/parsed_module.rs
new file mode 100644
index 00000000..eff96007
--- /dev/null
+++ b/soroban-env-host/src/vm/parsed_module.rs
@@ -0,0 +1,577 @@
+use crate::{
+    err,
+    meta::{self, get_ledger_protocol_version},
+    xdr::{ContractCostType, Limited, ReadXdr, ScEnvMetaEntry, ScErrorCode, ScErrorType},
+    Host, HostError, DEFAULT_XDR_RW_LIMITS,
+};
+
+use wasmi::{Engine, Module};
+
+use super::{ModuleCache, MAX_VM_ARGS};
+use std::io::Cursor;
+
+#[derive(Debug, Clone)]
+pub enum VersionedContractCodeCostInputs {
+    V0 {
+        wasm_bytes: usize,
+    },
+    #[cfg(feature = "next")]
+    V1(crate::xdr::ContractCodeCostInputs),
+}
+
+impl VersionedContractCodeCostInputs {
+    pub fn is_v0(&self) -> bool {
+        match self {
+            Self::V0 { .. } => true,
+            #[cfg(feature = "next")]
+            Self::V1(_) => false,
+        }
+    }
+    pub fn charge_for_parsing(&self, host: &Host) -> Result<(), HostError> {
+        match self {
+            Self::V0 { wasm_bytes } => {
+                host.charge_budget(ContractCostType::VmInstantiation, Some(*wasm_bytes as u64))?;
+            }
+            #[cfg(feature = "next")]
+            Self::V1(inputs) => {
+                host.charge_budget(
+                    ContractCostType::ParseWasmInstructions,
+                    Some(inputs.n_instructions as u64),
+                )?;
+                host.charge_budget(
+                    ContractCostType::ParseWasmFunctions,
+                    Some(inputs.n_functions as u64),
+                )?;
+                host.charge_budget(
+                    ContractCostType::ParseWasmGlobals,
+                    Some(inputs.n_globals as u64),
+                )?;
+                host.charge_budget(
+                    ContractCostType::ParseWasmTableEntries,
+                    Some(inputs.n_table_entries as u64),
+                )?;
+                host.charge_budget(
+                    ContractCostType::ParseWasmTypes,
+                    Some(inputs.n_types as u64),
+                )?;
+                host.charge_budget(
+                    ContractCostType::ParseWasmDataSegments,
+                    Some(inputs.n_data_segments as u64),
+                )?;
+                host.charge_budget(
+                    ContractCostType::ParseWasmElemSegments,
+                    Some(inputs.n_elem_segments as u64),
+                )?;
+                host.charge_budget(
+                    ContractCostType::ParseWasmImports,
+                    Some(inputs.n_imports as u64),
+                )?;
+                host.charge_budget(
+                    ContractCostType::ParseWasmExports,
+                    Some(inputs.n_exports as u64),
+                )?;
+                host.charge_budget(
+                    ContractCostType::ParseWasmDataSegmentBytes,
+                    Some(inputs.n_data_segment_bytes as u64),
+                )?;
+            }
+        }
+        Ok(())
+    }
+    pub fn charge_for_instantiation(&self, _host: &Host) -> Result<(), HostError> {
+        match self {
+            Self::V0 { wasm_bytes } => {
+                // Before soroban supported cached instantiation, the full cost
+                // of parsing-and-instantiation was charged to the
+                // VmInstantiation cost type and we already charged it by the
+                // time we got here, in `charge_for_parsing` above. At-and-after
+                // the protocol that enabled cached instantiation, the
+                // VmInstantiation cost type was repurposed to only cover the
+                // cost of parsing, so we have to charge the "second half" cost
+                // of instantiaiton separately here.
+                if _host.get_ledger_protocol_version()? >= ModuleCache::MIN_LEDGER_VERSION {
+                    _host.charge_budget(
+                        ContractCostType::VmCachedInstantiation,
+                        Some(*wasm_bytes as u64),
+                    )?;
+                }
+            }
+            #[cfg(feature = "next")]
+            Self::V1(inputs) => {
+                _host.charge_budget(ContractCostType::InstantiateWasmInstructions, None)?;
+                _host.charge_budget(
+                    ContractCostType::InstantiateWasmFunctions,
+                    Some(inputs.n_functions as u64),
+                )?;
+                _host.charge_budget(
+                    ContractCostType::InstantiateWasmGlobals,
+                    Some(inputs.n_globals as u64),
+                )?;
+                _host.charge_budget(
+                    ContractCostType::InstantiateWasmTableEntries,
+                    Some(inputs.n_table_entries as u64),
+                )?;
+                _host.charge_budget(ContractCostType::InstantiateWasmTypes, None)?;
+                _host.charge_budget(
+                    ContractCostType::InstantiateWasmDataSegments,
+                    Some(inputs.n_data_segments as u64),
+                )?;
+                _host.charge_budget(
+                    ContractCostType::InstantiateWasmElemSegments,
+                    Some(inputs.n_elem_segments as u64),
+                )?;
+                _host.charge_budget(
+                    ContractCostType::InstantiateWasmImports,
+                    Some(inputs.n_imports as u64),
+                )?;
+                _host.charge_budget(
+                    ContractCostType::InstantiateWasmExports,
+                    Some(inputs.n_exports as u64),
+                )?;
+                _host.charge_budget(
+                    ContractCostType::InstantiateWasmDataSegmentBytes,
+                    Some(inputs.n_data_segment_bytes as u64),
+                )?;
+            }
+        }
+        Ok(())
+    }
+}
+
+/// A [ParsedModule] contains the parsed [wasmi::Module] for a given Wasm blob,
+/// as well as a protocol number and set of [ContractCodeCostInputs] extracted
+/// from the module when it was parsed.
+pub struct ParsedModule {
+    pub module: Module,
+    pub proto_version: u32,
+    pub cost_inputs: VersionedContractCodeCostInputs,
+}
+
+impl ParsedModule {
+    pub fn new(
+        host: &Host,
+        engine: &Engine,
+        wasm: &[u8],
+        cost_inputs: VersionedContractCodeCostInputs,
+    ) -> Result<Self, HostError> {
+        cost_inputs.charge_for_parsing(host)?;
+        let (module, proto_version) = Self::parse_wasm(host, engine, wasm)?;
+        Ok(Self {
+            module,
+            proto_version,
+            cost_inputs,
+        })
+    }
+
+    #[cfg(any(test, feature = "testutils"))]
+    pub fn new_with_isolated_engine(
+        host: &Host,
+        wasm: &[u8],
+        cost_inputs: VersionedContractCodeCostInputs,
+    ) -> Result<Self, HostError> {
+        use crate::budget::AsBudget;
+        let config = crate::vm::get_wasmi_config(host.as_budget())?;
+        let engine = Engine::new(&config);
+        cost_inputs.charge_for_parsing(host)?;
+        let (module, proto_version) = Self::parse_wasm(host, &engine, wasm)?;
+        Ok(Self {
+            module,
+            proto_version,
+            cost_inputs,
+        })
+    }
+
+    /// Parse the Wasm blob into a [Module] and its protocol number, checking its interface version
+    fn parse_wasm(host: &Host, engine: &Engine, wasm: &[u8]) -> Result<(Module, u32), HostError> {
+        let module = {
+            let _span0 = tracy_span!("parse module");
+            host.map_err(Module::new(&engine, wasm))?
+        };
+
+        Self::check_max_args(host, &module)?;
+        let interface_version = Self::check_meta_section(host, &module)?;
+        let contract_proto = get_ledger_protocol_version(interface_version);
+
+        Ok((module, contract_proto))
+    }
+
+    fn check_contract_interface_version(
+        host: &Host,
+        interface_version: u64,
+    ) -> Result<(), HostError> {
+        let want_proto = {
+            let ledger_proto = host.get_ledger_protocol_version()?;
+            let env_proto = get_ledger_protocol_version(meta::INTERFACE_VERSION);
+            if ledger_proto <= env_proto {
+                // ledger proto should be before or equal to env proto
+                ledger_proto
+            } else {
+                return Err(err!(
+                    host,
+                    (ScErrorType::Context, ScErrorCode::InternalError),
+                    "ledger protocol number is ahead of supported env protocol number",
+                    ledger_proto,
+                    env_proto
+                ));
+            }
+        };
+
+        // Not used when "next" is enabled
+        #[cfg(not(feature = "next"))]
+        let got_pre = meta::get_pre_release_version(interface_version);
+
+        let got_proto = get_ledger_protocol_version(interface_version);
+
+        if got_proto < want_proto {
+            // Old protocols are finalized, we only support contracts
+            // with similarly finalized (zero) prerelease numbers.
+            //
+            // Note that we only enable this check if the "next" feature isn't enabled
+            // because a "next" stellar-core can still run a "curr" test using non-finalized
+            // test Wasms. The "next" feature isn't safe for production and is meant to
+            // simulate the protocol version after the one currently supported in
+            // stellar-core, so bypassing this check for "next" is safe.
+            #[cfg(not(feature = "next"))]
+            if got_pre != 0 {
+                return Err(err!(
+                    host,
+                    (ScErrorType::WasmVm, ScErrorCode::InvalidInput),
+                    "contract pre-release number for old protocol is nonzero",
+                    got_pre
+                ));
+            }
+        } else if got_proto == want_proto {
+            // Relax this check as well for the "next" feature to allow for flexibility while testing.
+            // stellar-core can pass in an older protocol version, in which case the pre-release version
+            // will not match up with the "next" feature (The "next" pre-release version is always 1).
+            #[cfg(not(feature = "next"))]
+            {
+                // Current protocol might have a nonzero prerelease number; we will
+                // allow it only if it matches the current prerelease exactly.
+                let want_pre = meta::get_pre_release_version(meta::INTERFACE_VERSION);
+                if want_pre != got_pre {
+                    return Err(err!(
+                        host,
+                        (ScErrorType::WasmVm, ScErrorCode::InvalidInput),
+                        "contract pre-release number for current protocol does not match host",
+                        got_pre,
+                        want_pre
+                    ));
+                }
+            }
+        } else {
+            // Future protocols we don't allow. It might be nice (in the sense
+            // of "allowing uploads of a future-protocol contract that will go
+            // live as soon as the network upgrades to it") but there's a risk
+            // that the "future" protocol semantics baked in to a contract
+            // differ from the final semantics chosen by the network, so to be
+            // conservative we avoid even allowing this.
+            return Err(err!(
+                host,
+                (ScErrorType::WasmVm, ScErrorCode::InvalidInput),
+                "contract protocol number is newer than host",
+                got_proto
+            ));
+        }
+        Ok(())
+    }
+
+    fn module_custom_section(m: &Module, name: impl AsRef<str>) -> Option<&[u8]> {
+        m.custom_sections().iter().find_map(|s| {
+            if &*s.name == name.as_ref() {
+                Some(&*s.data)
+            } else {
+                None
+            }
+        })
+    }
+
+    /// Returns the raw bytes content of a named custom section from the Wasm
+    /// module loaded into the [Vm], or `None` if no such custom section exists.
+    pub fn custom_section(&self, name: impl AsRef<str>) -> Option<&[u8]> {
+        Self::module_custom_section(&self.module, name)
+    }
+
+    fn check_meta_section(host: &Host, m: &Module) -> Result<u64, HostError> {
+        if let Some(env_meta) = Self::module_custom_section(m, meta::ENV_META_V0_SECTION_NAME) {
+            let mut limits = DEFAULT_XDR_RW_LIMITS;
+            limits.len = env_meta.len();
+            let mut cursor = Limited::new(Cursor::new(env_meta), limits);
+            if let Some(env_meta_entry) = ScEnvMetaEntry::read_xdr_iter(&mut cursor).next() {
+                let ScEnvMetaEntry::ScEnvMetaKindInterfaceVersion(v) =
+                    host.map_err(env_meta_entry)?;
+                Self::check_contract_interface_version(host, v)?;
+                Ok(v)
+            } else {
+                Err(host.err(
+                    ScErrorType::WasmVm,
+                    ScErrorCode::InvalidInput,
+                    "contract missing environment interface version",
+                    &[],
+                ))
+            }
+        } else {
+            Err(host.err(
+                ScErrorType::WasmVm,
+                ScErrorCode::InvalidInput,
+                "contract missing metadata section",
+                &[],
+            ))
+        }
+    }
+
+    fn check_max_args(host: &Host, m: &Module) -> Result<(), HostError> {
+        for e in m.exports() {
+            match e.ty() {
+                wasmi::ExternType::Func(f) => {
+                    if f.results().len() > MAX_VM_ARGS {
+                        return Err(err!(
+                            host,
+                            (ScErrorType::WasmVm, ScErrorCode::InvalidInput),
+                            "Too many return values in Wasm export",
+                            f.results().len()
+                        ));
+                    }
+                    if f.params().len() > MAX_VM_ARGS {
+                        return Err(err!(
+                            host,
+                            (ScErrorType::WasmVm, ScErrorCode::InvalidInput),
+                            "Too many arguments Wasm export",
+                            f.params().len()
+                        ));
+                    }
+                }
+                _ => (),
+            }
+        }
+        Ok(())
+    }
+
+    // Do a second, manual parse of the Wasm blob to extract cost parameters we're
+    // interested in.
+    #[cfg(feature = "next")]
+    pub fn extract_refined_contract_cost_inputs(
+        host: &Host,
+        wasm: &[u8],
+    ) -> Result<crate::xdr::ContractCodeCostInputs, HostError> {
+        use wasmparser::{ElementItems, ElementKind, Parser, Payload::*, TableInit};
+
+        if !Parser::is_core_wasm(wasm) {
+            return Err(host.err(
+                ScErrorType::WasmVm,
+                ScErrorCode::InvalidInput,
+                "unsupported non-core wasm module",
+                &[],
+            ));
+        }
+
+        let mut costs = crate::xdr::ContractCodeCostInputs {
+            ext: crate::xdr::ExtensionPoint::V0,
+            n_instructions: 0,
+            n_functions: 0,
+            n_globals: 0,
+            n_table_entries: 0,
+            n_types: 0,
+            n_data_segments: 0,
+            n_elem_segments: 0,
+            n_imports: 0,
+            n_exports: 0,
+            n_data_segment_bytes: 0,
+        };
+
+        let parser = Parser::new(0);
+        let mut elements: u32 = 0;
+        let mut available_memory: u32 = 0;
+        for section in parser.parse_all(wasm) {
+            let section = host.map_err(section)?;
+            match section {
+                // Ignored sections.
+                Version { .. }
+                | DataCountSection { .. }
+                | CustomSection(_)
+                | CodeSectionStart { .. }
+                | End(_) => (),
+
+                // Component-model stuff or other unsupported sections. Error out.
+                StartSection { .. }
+                | ModuleSection { .. }
+                | InstanceSection(_)
+                | CoreTypeSection(_)
+                | ComponentSection { .. }
+                | ComponentInstanceSection(_)
+                | ComponentAliasSection(_)
+                | ComponentTypeSection(_)
+                | ComponentCanonicalSection(_)
+                | ComponentStartSection { .. }
+                | ComponentImportSection(_)
+                | ComponentExportSection(_)
+                | TagSection(_)
+                | UnknownSection { .. } => {
+                    return Err(host.err(
+                        ScErrorType::WasmVm,
+                        ScErrorCode::InvalidInput,
+                        "unsupported wasm section type",
+                        &[],
+                    ))
+                }
+
+                MemorySection(s) => {
+                    for mem in s {
+                        let mem = host.map_err(mem)?;
+                        if mem.memory64 {
+                            return Err(host.err(
+                                ScErrorType::WasmVm,
+                                ScErrorCode::InvalidInput,
+                                "unsupported 64-bit memory",
+                                &[],
+                            ));
+                        }
+                        if mem.shared {
+                            return Err(host.err(
+                                ScErrorType::WasmVm,
+                                ScErrorCode::InvalidInput,
+                                "unsupported shared memory",
+                                &[],
+                            ));
+                        }
+                        if mem
+                            .initial
+                            .saturating_mul(crate::vm::WASM_STD_MEM_PAGE_SIZE_IN_BYTES as u64)
+                            > u32::MAX as u64
+                        {
+                            return Err(host.err(
+                                ScErrorType::WasmVm,
+                                ScErrorCode::InvalidInput,
+                                "unsupported memory size",
+                                &[],
+                            ));
+                        }
+                        available_memory = available_memory.saturating_add(
+                            (mem.initial as u32)
+                                .saturating_mul(crate::vm::WASM_STD_MEM_PAGE_SIZE_IN_BYTES),
+                        );
+                    }
+                }
+
+                TypeSection(s) => costs.n_types = costs.n_types.saturating_add(s.count()),
+                ImportSection(s) => costs.n_imports = costs.n_imports.saturating_add(s.count()),
+                FunctionSection(s) => {
+                    costs.n_functions = costs.n_functions.saturating_add(s.count())
+                }
+                TableSection(s) => {
+                    for table in s {
+                        let table = host.map_err(table)?;
+                        costs.n_table_entries =
+                            costs.n_table_entries.saturating_add(table.ty.initial);
+                        match table.init {
+                            TableInit::RefNull => (),
+                            TableInit::Expr(ref expr) => {
+                                Self::check_const_expr_simple(&host, &expr)?;
+                            }
+                        }
+                    }
+                }
+                GlobalSection(s) => {
+                    costs.n_globals = costs.n_globals.saturating_add(s.count());
+                    for global in s {
+                        let global = host.map_err(global)?;
+                        Self::check_const_expr_simple(&host, &global.init_expr)?;
+                    }
+                }
+                ExportSection(s) => costs.n_exports = costs.n_exports.saturating_add(s.count()),
+                ElementSection(s) => {
+                    costs.n_elem_segments = costs.n_elem_segments.saturating_add(s.count());
+                    for elem in s {
+                        let elem = host.map_err(elem)?;
+                        match elem.kind {
+                            ElementKind::Declared | ElementKind::Passive => (),
+                            ElementKind::Active { offset_expr, .. } => {
+                                Self::check_const_expr_simple(&host, &offset_expr)?
+                            }
+                        }
+                        match elem.items {
+                            ElementItems::Functions(fs) => {
+                                elements = elements.saturating_add(fs.count());
+                            }
+                            ElementItems::Expressions(_, exprs) => {
+                                elements = elements.saturating_add(exprs.count());
+                                for expr in exprs {
+                                    let expr = host.map_err(expr)?;
+                                    Self::check_const_expr_simple(&host, &expr)?;
+                                }
+                            }
+                        }
+                    }
+                }
+                DataSection(s) => {
+                    costs.n_data_segments = costs.n_data_segments.saturating_add(s.count());
+                    for d in s {
+                        let d = host.map_err(d)?;
+                        if d.data.len() > u32::MAX as usize {
+                            return Err(host.err(
+                                ScErrorType::WasmVm,
+                                ScErrorCode::InvalidInput,
+                                "data segment exceeds u32::MAX",
+                                &[],
+                            ));
+                        }
+                        costs.n_data_segment_bytes = costs
+                            .n_data_segment_bytes
+                            .saturating_add(d.data.len() as u32);
+                        match d.kind {
+                            wasmparser::DataKind::Active { offset_expr, .. } => {
+                                Self::check_const_expr_simple(&host, &offset_expr)?
+                            }
+                            wasmparser::DataKind::Passive => (),
+                        }
+                    }
+                }
+                CodeSectionEntry(s) => {
+                    let ops = host.map_err(s.get_operators_reader())?;
+                    for _op in ops {
+                        costs.n_instructions = costs.n_instructions.saturating_add(1);
+                    }
+                }
+            }
+        }
+        if costs.n_data_segment_bytes > available_memory {
+            return Err(err!(
+                host,
+                (ScErrorType::WasmVm, ScErrorCode::InvalidInput),
+                "data segment(s) content exceeds memory size",
+                costs.n_data_segment_bytes,
+                available_memory
+            ));
+        }
+        if elements > costs.n_table_entries {
+            return Err(err!(
+                host,
+                (ScErrorType::WasmVm, ScErrorCode::InvalidInput),
+                "elem segments(s) content exceeds table size",
+                elements,
+                costs.n_table_entries
+            ));
+        }
+        Ok(costs)
+    }
+
+    #[cfg(feature = "next")]
+    fn check_const_expr_simple(host: &Host, expr: &wasmparser::ConstExpr) -> Result<(), HostError> {
+        use wasmparser::Operator::*;
+        let mut op = expr.get_operators_reader();
+        while !op.eof() {
+            match host.map_err(op.read())? {
+                I32Const { .. } | I64Const { .. } | RefFunc { .. } | RefNull { .. } | End => (),
+                _ => {
+                    return Err(host.err(
+                        ScErrorType::WasmVm,
+                        ScErrorCode::InvalidInput,
+                        "unsupported complex Wasm constant expression",
+                        &[],
+                    ))
+                }
+            }
+        }
+        Ok(())
+    }
+}
diff --git a/soroban-synth-wasm/Cargo.toml b/soroban-synth-wasm/Cargo.toml
index 4b1ea2eb..6729abf8 100644
--- a/soroban-synth-wasm/Cargo.toml
+++ b/soroban-synth-wasm/Cargo.toml
@@ -12,7 +12,7 @@ publish = false
 
 [dependencies]
 wasm-encoder = "=0.36.2"
-wasmparser = "=0.116.1"
+wasmparser = { workspace = true }
 soroban-env-common = { workspace = true }
 arbitrary = { version = "=1.3.2", features = ["derive"] }
 soroban-env-macros = { workspace = true }
diff --git a/soroban-synth-wasm/src/mod_emitter.rs b/soroban-synth-wasm/src/mod_emitter.rs
index 76ae0aa0..ea504e58 100644
--- a/soroban-synth-wasm/src/mod_emitter.rs
+++ b/soroban-synth-wasm/src/mod_emitter.rs
@@ -225,6 +225,12 @@ impl ModEmitter {
         }
     }
 
+    #[cfg(feature = "adversarial")]
+    pub fn add_raw_fn_type(&mut self, params: &[ValType], results: &[ValType]) {
+        self.types
+            .function(params.iter().cloned(), results.iter().cloned());
+    }
+
     #[cfg(feature = "adversarial")]
     pub fn add_fn_type_no_check(&mut self, arity: Arity, ret: Arity) -> TypeRef {
         let params: Vec<_> = std::iter::repeat(ValType::I64)


Test patch:
diff --git a/soroban-env-host/src/e2e_testutils.rs b/soroban-env-host/src/e2e_testutils.rs
index 284262a7..ef749f57 100644
--- a/soroban-env-host/src/e2e_testutils.rs
+++ b/soroban-env-host/src/e2e_testutils.rs
@@ -1,3 +1,5 @@
+#[cfg(feature = "next")]
+use crate::xdr::ContractCodeEntryExt;
 use crate::xdr::{
     AccountEntry, AccountEntryExt, AccountId, ContractCodeEntry, ContractDataDurability,
     ContractDataEntry, ContractExecutable, ContractIdPreimage, ContractIdPreimageFromAddress,
@@ -47,7 +49,10 @@ pub fn bytes_sc_val(bytes: &[u8]) -> ScVal {
 
 pub fn wasm_entry(wasm: &[u8]) -> LedgerEntry {
     ledger_entry(LedgerEntryData::ContractCode(ContractCodeEntry {
+        #[cfg(not(feature = "next"))]
         ext: ExtensionPoint::V0,
+        #[cfg(feature = "next")]
+        ext: ContractCodeEntryExt::V0,
         hash: get_wasm_hash(wasm).try_into().unwrap(),
         code: wasm.try_into().unwrap(),
     }))
diff --git a/soroban-env-host/src/test/budget_metering.rs b/soroban-env-host/src/test/budget_metering.rs
index 396d1f4f..ac623a98 100644
--- a/soroban-env-host/src/test/budget_metering.rs
+++ b/soroban-env-host/src/test/budget_metering.rs
@@ -383,7 +383,8 @@ fn total_amount_charged_from_random_inputs() -> Result<(), HostError> {
     }
 
     let actual = format!("{:?}", host.as_budget());
-    expect![[r#"
+    #[cfg(not(feature = "next"))]
+    let expected = expect![[r#"
         =====================================================================================================================================================================
         Cpu limit: 100000000; used: 13060190
         Mem limit: 41943040; used: 273960
@@ -419,8 +420,68 @@ fn total_amount_charged_from_random_inputs() -> Result<(), HostError> {
         Shadow mem limit: 41943040; used: 273960
         =====================================================================================================================================================================
 
-    "#]]
-    .assert_eq(&actual);
+    "#]];
+    #[cfg(feature = "next")]
+    let expected = expect![
+        r#"
+    =====================================================================================================================================================================
+    Cpu limit: 100000000; used: 13060190
+    Mem limit: 41943040; used: 273960
+    =====================================================================================================================================================================
+    CostType                 iterations     input          cpu_insns      mem_bytes      const_term_cpu      lin_term_cpu        const_term_mem      lin_term_mem        
+    WasmInsnExec             246            None           984            0              4                   0                   0                   0                   
+    MemAlloc                 1              Some(152)      453            168            434                 16                  16                  128                 
+    MemCpy                   1              Some(65)       50             0              42                  16                  0                   0                   
+    MemCmp                   1              Some(74)       53             0              44                  16                  0                   0                   
+    DispatchHostFunction     176            None           54560          0              310                 0                   0                   0                   
+    VisitObject              97             None           5917           0              61                  0                   0                   0                   
+    ValSer                   1              Some(49)       241            389            230                 29                  242                 384                 
+    ValDeser                 1              Some(103)      62271          309            59052               4001                0                   384                 
+    ComputeSha256Hash        1              Some(193)      14310          0              3738                7012                0                   0                   
+    ComputeEd25519PubKey     226            None           9097178        0              40253               0                   0                   0                   
+    VerifyEd25519Sig         1              Some(227)      384738         0              377524              4068                0                   0                   
+    VmInstantiation          1              Some(147)      503770         135880         451626              45405               130065              5064                
+    VmCachedInstantiation    1              Some(147)      503770         135880         451626              45405               130065              5064                
+    InvokeVmFunction         47             None           91556          658            1948                0                   14                  0                   
+    ComputeKeccak256Hash     1              Some(1)        3812           0              3766                5969                0                   0                   
+    ComputeEcdsaSecp256k1Sig 1              None           710            0              710                 0                   0                   0                   
+    RecoverEcdsaSecp256k1Key 1              None           2315295        181            2315295             0                   181                 0                   
+    Int256AddSub             1              None           4404           99             4404                0                   99                  0                   
+    Int256Mul                1              None           4947           99             4947                0                   99                  0                   
+    Int256Div                1              None           4911           99             4911                0                   99                  0                   
+    Int256Pow                1              None           4286           99             4286                0                   99                  0                   
+    Int256Shift              1              None           913            99             913                 0                   99                  0                   
+    ChaCha20DrawBytes        1              Some(1)        1061           0              1058                501                 0                   0                   
+    ParseWasmInstructions    0              Some(0)        0              0              72736               25420               17564               6457                
+    ParseWasmFunctions       0              Some(0)        0              0              0                   536688              0                   47464               
+    ParseWasmGlobals         0              Some(0)        0              0              0                   176902              0                   13420               
+    ParseWasmTableEntries    0              Some(0)        0              0              0                   29639               0                   6285                
+    ParseWasmTypes           0              Some(0)        0              0              0                   1048891             0                   64670               
+    ParseWasmDataSegments    0              Some(0)        0              0              0                   236970              0                   29074               
+    ParseWasmElemSegments    0              Some(0)        0              0              0                   317249              0                   48095               
+    ParseWasmImports         0              Some(0)        0              0              0                   694667              0                   102890              
+    ParseWasmExports         0              Some(0)        0              0              0                   427037              0                   36394               
+    ParseWasmDataSegmentBytes0              Some(0)        0              0              66075               28                  17580               257                 
+    InstantiateWasmInstructions0              None           0              0              25059               0                   70192               0                   
+    InstantiateWasmFunctions 0              Some(0)        0              0              0                   7503                0                   14613               
+    InstantiateWasmGlobals   0              Some(0)        0              0              0                   10761               0                   6833                
+    InstantiateWasmTableEntries0              Some(0)        0              0              0                   3211                0                   1025                
+    InstantiateWasmTypes     0              None           0              0              0                   0                   0                   0                   
+    InstantiateWasmDataSegments0              Some(0)        0              0              0                   16370               0                   129632              
+    InstantiateWasmElemSegments0              Some(0)        0              0              0                   28309               0                   13665               
+    InstantiateWasmImports   0              Some(0)        0              0              0                   683461              0                   77273               
+    InstantiateWasmExports   0              Some(0)        0              0              0                   297065              0                   9176                
+    InstantiateWasmDataSegmentBytes0              Some(0)        0              0              25191               14                  69256               126                 
+    =====================================================================================================================================================================
+    Internal details (diagnostics info, does not affect fees) 
+    Total # times meter was called: 23
+    Shadow cpu limit: 100000000; used: 13060190
+    Shadow mem limit: 41943040; used: 273960
+    =====================================================================================================================================================================
+
+    "#
+    ];
+    expected.assert_eq(&actual);
 
     assert_eq!(
         host.as_budget().get_cpu_insns_consumed()?,
diff --git a/soroban-env-host/src/test/bytes.rs b/soroban-env-host/src/test/bytes.rs
index 5f7d1a51..855b840c 100644
--- a/soroban-env-host/src/test/bytes.rs
+++ b/soroban-env-host/src/test/bytes.rs
@@ -490,8 +490,12 @@ fn instantiate_oversized_bytes_from_linear_memory() -> Result<(), HostError> {
         U32Val::from(100).to_val().get_payload()
     );
 
-    // constructing a big map will cause budget limit exceeded error
-    let wasm_long = wasm::wasm_module_with_large_bytes_from_linear_memory(480000, 7);
+    // constructing a big bytes will cause budget limit exceeded error
+    #[cfg(not(feature = "next"))]
+    const TOO_BIG: u32 = 480000;
+    #[cfg(feature = "next")]
+    const TOO_BIG: u32 = 8_000_000;
+    let wasm_long = wasm::wasm_module_with_large_bytes_from_linear_memory(TOO_BIG, 7);
     host.budget_ref().reset_unlimited()?;
     let contract_id_obj2 = host.register_test_contract_wasm(&wasm_long.as_slice());
     host.budget_ref().reset_default()?;
diff --git a/soroban-env-host/src/test/hostile.rs b/soroban-env-host/src/test/hostile.rs
index e6873adf..e0c1417a 100644
--- a/soroban-env-host/src/test/hostile.rs
+++ b/soroban-env-host/src/test/hostile.rs
@@ -527,6 +527,61 @@ fn excessive_logging() -> Result<(), HostError> {
     host.enable_debug()?;
     let contract_id_obj = host.register_test_contract_wasm(wasm.as_slice());
 
+    #[cfg(feature = "next")]
+    let expected_budget = expect![[r#"
+        =======================================================
+        Cpu limit: 2000000; used: 284819
+        Mem limit: 500000; used: 252830
+        =======================================================
+        CostType                 cpu_insns      mem_bytes      
+        WasmInsnExec             300            0              
+        MemAlloc                 15750          67248          
+        MemCpy                   2345           0              
+        MemCmp                   696            0              
+        DispatchHostFunction     310            0              
+        VisitObject              244            0              
+        ValSer                   0              0              
+        ValDeser                 0              0              
+        ComputeSha256Hash        3738           0              
+        ComputeEd25519PubKey     0              0              
+        VerifyEd25519Sig         0              0              
+        VmInstantiation          0              0              
+        VmCachedInstantiation    0              0              
+        InvokeVmFunction         1948           14             
+        ComputeKeccak256Hash     0              0              
+        ComputeEcdsaSecp256k1Sig 0              0              
+        RecoverEcdsaSecp256k1Key 0              0              
+        Int256AddSub             0              0              
+        Int256Mul                0              0              
+        Int256Div                0              0              
+        Int256Pow                0              0              
+        Int256Shift              0              0              
+        ChaCha20DrawBytes        0              0              
+        ParseWasmInstructions    74324          17967          
+        ParseWasmFunctions       4192           370            
+        ParseWasmGlobals         1382           104            
+        ParseWasmTableEntries    29639          6285           
+        ParseWasmTypes           8194           505            
+        ParseWasmDataSegments    0              0              
+        ParseWasmElemSegments    0              0              
+        ParseWasmImports         5427           803            
+        ParseWasmExports         6672           568            
+        ParseWasmDataSegmentBytes66075          17580          
+        InstantiateWasmInstructions25059          70192          
+        InstantiateWasmFunctions 58             114            
+        InstantiateWasmGlobals   84             53             
+        InstantiateWasmTableEntries3211           1025           
+        InstantiateWasmTypes     0              0              
+        InstantiateWasmDataSegments0              0              
+        InstantiateWasmElemSegments0              0              
+        InstantiateWasmImports   5339           603            
+        InstantiateWasmExports   4641           143            
+        InstantiateWasmDataSegmentBytes25191          69256          
+        =======================================================
+
+    "#]];
+
+    #[cfg(not(feature = "next"))]
     let expected_budget = expect![[r#"
         =======================================================
         Cpu limit: 2000000; used: 522315
diff --git a/soroban-env-host/src/test/invocation.rs b/soroban-env-host/src/test/invocation.rs
index dd418a11..fe58cc5c 100644
--- a/soroban-env-host/src/test/invocation.rs
+++ b/soroban-env-host/src/test/invocation.rs
@@ -67,9 +67,9 @@ fn invoke_alloc() -> Result<(), HostError> {
     // pages or about 1.3 MiB, plus the initial 17 pages (1.1MiB) plus some more
     // slop from general host machinery allocations, plus allocating a VM once
     // during upload and once during execution we get around 2.5MiB. Call
-    // is "less than 4MiB".
+    // is "less than 5MiB".
     assert!(used_bytes > (128 * 4096));
-    assert!(used_bytes < 0x40_0000);
+    assert!(used_bytes < 0x50_0000);
     Ok(())
 }
 
diff --git a/soroban-env-host/src/test/lifecycle.rs b/soroban-env-host/src/test/lifecycle.rs
index 10ecd31b..bc4f6417 100644
--- a/soroban-env-host/src/test/lifecycle.rs
+++ b/soroban-env-host/src/test/lifecycle.rs
@@ -566,3 +566,380 @@ fn test_large_contract() {
     assert!(err.error.is_type(ScErrorType::Budget));
     assert!(err.error.is_code(ScErrorCode::ExceededLimit));
 }
+
+#[cfg(feature = "next")]
+#[allow(dead_code)]
+mod cap_54_55_56 {
+
+    use super::*;
+    use crate::{
+        storage::{FootprintMap, StorageMap},
+        test::observe::ObservedHost,
+        testutils::wasm::wasm_module_with_a_bit_of_everything,
+        vm::ModuleCache,
+        xdr::{
+            ContractCostType::{self, *},
+            LedgerEntry, LedgerKey,
+        },
+        AddressObject, HostError,
+    };
+    use std::rc::Rc;
+
+    const V_NEW: u32 = ModuleCache::MIN_LEDGER_VERSION;
+    const V_OLD: u32 = V_NEW - 1;
+    const NEW_COST_TYPES: &'static [ContractCostType] = &[
+        ParseWasmInstructions,
+        ParseWasmFunctions,
+        ParseWasmGlobals,
+        ParseWasmTableEntries,
+        ParseWasmTypes,
+        ParseWasmDataSegments,
+        ParseWasmElemSegments,
+        ParseWasmImports,
+        ParseWasmExports,
+        ParseWasmDataSegmentBytes,
+        InstantiateWasmInstructions,
+        InstantiateWasmFunctions,
+        InstantiateWasmGlobals,
+        InstantiateWasmTableEntries,
+        InstantiateWasmTypes,
+        InstantiateWasmDataSegments,
+        InstantiateWasmElemSegments,
+        InstantiateWasmImports,
+        InstantiateWasmExports,
+        InstantiateWasmDataSegmentBytes,
+    ];
+
+    fn new_host_with_protocol_and_uploaded_contract(
+        hostname: &'static str,
+        proto: u32,
+    ) -> Result<(ObservedHost, AddressObject), HostError> {
+        let host = Host::test_host_with_recording_footprint();
+        let host = ObservedHost::new(hostname, host);
+        host.with_mut_ledger_info(|ledger_info| ledger_info.protocol_version = proto)?;
+        let contract_addr_obj =
+            host.register_test_contract_wasm(&wasm_module_with_a_bit_of_everything(proto));
+        Ok((host, contract_addr_obj))
+    }
+
+    struct ContractAndWasmEntries {
+        contract_key: Rc<LedgerKey>,
+        contract_entry: Rc<LedgerEntry>,
+        wasm_key: Rc<LedgerKey>,
+        wasm_entry: Rc<LedgerEntry>,
+    }
+
+    impl ContractAndWasmEntries {
+        fn from_contract_addr(
+            host: &Host,
+            contract_addr_obj: AddressObject,
+        ) -> Result<Self, HostError> {
+            let contract_id = host.contract_id_from_address(contract_addr_obj)?;
+            Self::from_contract_id(host, contract_id)
+        }
+        fn reload(self, host: &Host) -> Result<Self, HostError> {
+            host.with_mut_storage(|storage| {
+                let budget = host.budget_cloned();
+                let contract_entry = storage.get(&self.contract_key, &budget)?;
+                let wasm_entry = storage.get(&self.wasm_key, &budget)?;
+                Ok(ContractAndWasmEntries {
+                    contract_key: self.contract_key,
+                    contract_entry,
+                    wasm_key: self.wasm_key,
+                    wasm_entry,
+                })
+            })
+        }
+        fn from_contract_id(host: &Host, contract_id: Hash) -> Result<Self, HostError> {
+            let contract_key = host.contract_instance_ledger_key(&contract_id)?;
+            let wasm_hash = get_contract_wasm_ref(host, contract_id);
+            let wasm_key = host.contract_code_ledger_key(&wasm_hash)?;
+
+            host.with_mut_storage(|storage| {
+                let budget = host.budget_cloned();
+                let contract_entry = storage.get(&contract_key, &budget)?;
+                let wasm_entry = storage.get(&wasm_key, &budget)?;
+                Ok(ContractAndWasmEntries {
+                    contract_key,
+                    contract_entry,
+                    wasm_key,
+                    wasm_entry,
+                })
+            })
+        }
+        fn read_only_footprint(&self, budget: &Budget) -> Footprint {
+            Footprint(
+                FootprintMap::new()
+                    .insert(self.contract_key.clone(), AccessType::ReadOnly, budget)
+                    .unwrap()
+                    .insert(self.wasm_key.clone(), AccessType::ReadOnly, budget)
+                    .unwrap(),
+            )
+        }
+        fn wasm_writing_footprint(&self, budget: &Budget) -> Footprint {
+            Footprint(
+                FootprintMap::new()
+                    .insert(self.contract_key.clone(), AccessType::ReadOnly, budget)
+                    .unwrap()
+                    .insert(self.wasm_key.clone(), AccessType::ReadWrite, budget)
+                    .unwrap(),
+            )
+        }
+        fn storage_map(&self, budget: &Budget) -> StorageMap {
+            StorageMap::new()
+                .insert(
+                    self.contract_key.clone(),
+                    Some((self.contract_entry.clone(), Some(99999))),
+                    budget,
+                )
+                .unwrap()
+                .insert(
+                    self.wasm_key.clone(),
+                    Some((self.wasm_entry.clone(), Some(99999))),
+                    budget,
+                )
+                .unwrap()
+        }
+        fn read_only_storage(&self, budget: &Budget) -> Storage {
+            Storage::with_enforcing_footprint_and_map(
+                self.read_only_footprint(budget),
+                self.storage_map(budget),
+            )
+        }
+        fn wasm_writing_storage(&self, budget: &Budget) -> Storage {
+            Storage::with_enforcing_footprint_and_map(
+                self.wasm_writing_footprint(budget),
+                self.storage_map(budget),
+            )
+        }
+    }
+
+    fn upload_and_get_contract_and_wasm_entries(
+        upload_hostname: &'static str,
+        upload_proto: u32,
+    ) -> Result<ContractAndWasmEntries, HostError> {
+        let (host, contract) =
+            new_host_with_protocol_and_uploaded_contract(upload_hostname, upload_proto)?;
+        ContractAndWasmEntries::from_contract_addr(&host, contract)
+    }
+
+    fn upload_and_call(
+        upload_hostname: &'static str,
+        upload_proto: u32,
+        call_hostname: &'static str,
+        call_proto: u32,
+    ) -> Result<(Budget, Storage), HostError> {
+        // Phase 1: upload contract, tear down host, "close the ledger" and possibly change protocol.
+        let (host, contract) =
+            new_host_with_protocol_and_uploaded_contract(upload_hostname, upload_proto)?;
+        let contract_id = host.contract_id_from_address(contract)?;
+        let realhost = host.clone();
+        drop(host);
+        let (storage, _events) = realhost.try_finish()?;
+
+        // Phase 2: build new host with previous ledger output as storage, call contract. Possibly on new protocol.
+        let host = Host::with_storage_and_budget(storage, Budget::default());
+        host.enable_debug()?;
+        let host = ObservedHost::new(call_hostname, host);
+        host.set_ledger_info(LedgerInfo {
+            protocol_version: call_proto,
+            ..Default::default()
+        })?;
+        let contract = host.add_host_object(crate::xdr::ScAddress::Contract(contract_id))?;
+        let _ = host.call(
+            contract,
+            Symbol::try_from_small_str("test").unwrap(),
+            host.vec_new()?,
+        )?;
+        let realhost = host.clone();
+        drop(host);
+        let budget = realhost.budget_cloned();
+        let (storage, _events) = realhost.try_finish()?;
+        Ok((budget, storage))
+    }
+
+    fn code_entry_has_cost_inputs(entry: &Rc<LedgerEntry>) -> bool {
+        match &entry.data {
+            LedgerEntryData::ContractCode(cce) => match &cce.ext {
+                crate::xdr::ContractCodeEntryExt::V1(_v1) => return true,
+                _ => (),
+            },
+            _ => panic!("expected LedgerEntryData::ContractCode"),
+        }
+        false
+    }
+
+    // Test that running on protocol vOld only charges the VmInstantiation cost
+    // type.
+    #[test]
+    fn test_v_old_only_charges_vm_instantiation() -> Result<(), HostError> {
+        let (budget, _storage) = upload_and_call(
+            "test_v_old_only_charges_vminstantiation_upload",
+            V_OLD,
+            "test_v_old_only_charges_vm_instantiation_call",
+            V_OLD,
+        )?;
+        assert_ne!(budget.get_tracker(VmInstantiation)?.cpu, 0);
+        assert_eq!(budget.get_tracker(VmCachedInstantiation)?.cpu, 0);
+        for ct in NEW_COST_TYPES {
+            assert_eq!(budget.get_tracker(*ct)?.cpu, 0);
+        }
+        Ok(())
+    }
+
+    // Test that running on protocol vNew on a ContractCode LE that does not have
+    // ContractCodeCostInputs charges the VmInstantiation and VmCachedInstantiation
+    // cost types.
+    #[test]
+    fn test_v_new_no_contract_code_cost_inputs() -> Result<(), HostError> {
+        let (budget, _storage) = upload_and_call(
+            "test_v_new_no_contract_code_cost_inputs_upload",
+            V_OLD,
+            "test_v_new_no_contract_code_cost_inputs_call",
+            V_NEW,
+        )?;
+        assert_ne!(budget.get_tracker(VmInstantiation)?.cpu, 0);
+        assert_ne!(budget.get_tracker(VmCachedInstantiation)?.cpu, 0);
+        for ct in NEW_COST_TYPES {
+            assert_eq!(budget.get_tracker(*ct)?.cpu, 0);
+        }
+        Ok(())
+    }
+
+    // Test that running on protocol vNew does add ContractCodeCostInputs to a
+    // newly uploaded ContractCode LE.
+    #[test]
+    fn test_v_new_gets_contract_code_cost_inputs() -> Result<(), HostError> {
+        let entries = upload_and_get_contract_and_wasm_entries(
+            "test_v_new_gets_contract_code_cost_inputs_upload",
+            V_NEW,
+        )?;
+        assert!(code_entry_has_cost_inputs(&entries.wasm_entry));
+        Ok(())
+    }
+
+    // Test that running on protocol vNew on a ContractCode LE that does have
+    // ContractCodeCostInputs charges the new cost model types nonzero costs
+    // (both parsing and instantiation).
+    #[test]
+    fn test_v_new_with_contract_code_cost_inputs_causes_nonzero_costs() -> Result<(), HostError> {
+        let (budget, _storage) = upload_and_call(
+            "test_v_new_with_contract_code_cost_inputs_causes_nonzero_costs_upload",
+            V_NEW,
+            "test_v_new_with_contract_code_cost_inputs_causes_nonzero_costs_call",
+            V_NEW,
+        )?;
+        assert_eq!(budget.get_tracker(VmInstantiation)?.cpu, 0);
+        assert_eq!(budget.get_tracker(VmCachedInstantiation)?.cpu, 0);
+        for ct in NEW_COST_TYPES {
+            if *ct == InstantiateWasmTypes {
+                // This is a zero-cost type in the current calibration of the
+                // new model -- and exceptional case in this test -- though we
+                // keep it in case it becomes nonzero at some point (it's
+                // credible that it would).
+                continue;
+            }
+            assert_ne!(budget.get_tracker(*ct)?.cpu, 0);
+        }
+        Ok(())
+    }
+
+    // Test that running on protocol vOld does not add ContractCodeCostInputs to a
+    // newly uploaded ContractCode LE.
+    #[test]
+    fn test_v_old_no_contract_code_cost_inputs() -> Result<(), HostError> {
+        let entries = upload_and_get_contract_and_wasm_entries(
+            "test_v_old_no_contract_code_cost_inputs_upload",
+            V_OLD,
+        )?;
+        assert!(!code_entry_has_cost_inputs(&entries.wasm_entry));
+        Ok(())
+    }
+
+    // Test that running on protocol vOld does not rewrite a ContractCode LE when it
+    // already exists.
+    #[test]
+    fn test_v_old_no_rewrite() -> Result<(), HostError> {
+        let entries =
+            upload_and_get_contract_and_wasm_entries("test_v_old_no_rewrite_upload", V_OLD)?;
+        // make a new storage map for a new run
+        let budget = Budget::default();
+        let storage = entries.read_only_storage(&budget);
+        let host = Host::with_storage_and_budget(storage, budget);
+        let host = ObservedHost::new("test_v_old_no_rewrite_call", host);
+        host.set_ledger_info(LedgerInfo {
+            protocol_version: V_OLD,
+            ..Default::default()
+        })?;
+        host.upload_contract_wasm(wasm_module_with_a_bit_of_everything(V_OLD))?;
+        Ok(())
+    }
+
+    // Test that running on protocol vNew does rewrite a ContractCode LE when it
+    // already exists but doesn't yet have ContractCodeCostInputs.
+    #[test]
+    fn test_v_new_rewrite() -> Result<(), HostError> {
+        let entries = upload_and_get_contract_and_wasm_entries("test_v_new_rewrite_upload", V_OLD)?;
+        assert!(!code_entry_has_cost_inputs(&entries.wasm_entry));
+
+        // make a new storage map for a new upload but with read-only footprint -- this should fail
+        let budget = Budget::default();
+        let storage = entries.read_only_storage(&budget);
+        let host = Host::with_storage_and_budget(storage, budget);
+        let host = ObservedHost::new("test_v_new_rewrite_call_fail", host);
+        host.set_ledger_info(LedgerInfo {
+            protocol_version: V_NEW,
+            ..Default::default()
+        })?;
+        let wasm_blob = match &entries.wasm_entry.data {
+            LedgerEntryData::ContractCode(cce) => cce.code.to_vec(),
+            _ => panic!("expected ContractCode"),
+        };
+        assert!(host.upload_contract_wasm(wasm_blob.clone()).is_err());
+        let entries = entries.reload(&host)?;
+        assert!(!code_entry_has_cost_inputs(&entries.wasm_entry));
+
+        // make a new storage map for a new upload but with read-write footprint -- this should pass
+        let budget = Budget::default();
+        let storage = entries.wasm_writing_storage(&budget);
+        let host = Host::with_storage_and_budget(storage, budget);
+        let host = ObservedHost::new("test_v_new_rewrite_call_succeed", host);
+        host.set_ledger_info(LedgerInfo {
+            protocol_version: V_NEW,
+            ..Default::default()
+        })?;
+        host.upload_contract_wasm(wasm_blob)?;
+        let entries = entries.reload(&host)?;
+        assert!(code_entry_has_cost_inputs(&entries.wasm_entry));
+
+        Ok(())
+    }
+
+    // Test that running on protocol vNew does not rewrite a ContractCode LE when it
+    // already exists and already has ContractCodeCostInputs.
+    #[test]
+    fn test_v_new_no_rewrite() -> Result<(), HostError> {
+        let entries =
+            upload_and_get_contract_and_wasm_entries("test_v_new_no_rewrite_upload", V_NEW)?;
+        assert!(code_entry_has_cost_inputs(&entries.wasm_entry));
+
+        // make a new storage map for a new upload but with read-only footprint -- this should pass
+        let budget = Budget::default();
+        let storage = entries.read_only_storage(&budget);
+        let host = Host::with_storage_and_budget(storage, budget);
+        let host = ObservedHost::new("test_v_new_no_rewrite_call_pass", host);
+        host.set_ledger_info(LedgerInfo {
+            protocol_version: V_NEW,
+            ..Default::default()
+        })?;
+        let wasm_blob = match &entries.wasm_entry.data {
+            LedgerEntryData::ContractCode(cce) => cce.code.to_vec(),
+            _ => panic!("expected ContractCode"),
+        };
+        host.upload_contract_wasm(wasm_blob.clone())?;
+        let entries = entries.reload(&host)?;
+        assert!(code_entry_has_cost_inputs(&entries.wasm_entry));
+
+        Ok(())
+    }
+}
diff --git a/soroban-env-host/src/test/map.rs b/soroban-env-host/src/test/map.rs
index 0c6558ee..4e2bb7e5 100644
--- a/soroban-env-host/src/test/map.rs
+++ b/soroban-env-host/src/test/map.rs
@@ -484,8 +484,12 @@ fn instantiate_oversized_map_from_linear_memory() -> Result<(), HostError> {
     );
 
     // constructing a big map will cause budget limit exceeded error
+    #[cfg(not(feature = "next"))]
+    const TOO_BIG: u32 = 20_000;
+    #[cfg(feature = "next")]
+    const TOO_BIG: u32 = 1_000_000;
     let wasm_long =
-        wasm::wasm_module_with_large_map_from_linear_memory(20000, U32Val::from(7).to_val());
+        wasm::wasm_module_with_large_map_from_linear_memory(TOO_BIG, U32Val::from(7).to_val());
     host.budget_ref().reset_unlimited()?;
     let contract_id_obj2 = host.register_test_contract_wasm(&wasm_long.as_slice());
     host.budget_ref().reset_default()?;
diff --git a/soroban-env-host/src/test/vec.rs b/soroban-env-host/src/test/vec.rs
index 637f44de..6481aaca 100644
--- a/soroban-env-host/src/test/vec.rs
+++ b/soroban-env-host/src/test/vec.rs
@@ -453,8 +453,12 @@ fn instantiate_oversized_vec_from_linear_memory() -> Result<(), HostError> {
     );
 
     // constructing a big map will cause budget limit exceeded error
+    #[cfg(not(feature = "next"))]
+    const TOO_BIG: u32 = 60_000;
+    #[cfg(feature = "next")]
+    const TOO_BIG: u32 = 1_000_000;
     let wasm_long =
-        wasm::wasm_module_with_large_vector_from_linear_memory(60000, U32Val::from(7).to_val());
+        wasm::wasm_module_with_large_vector_from_linear_memory(TOO_BIG, U32Val::from(7).to_val());
     host.budget_ref().reset_unlimited()?;
     let contract_id_obj2 = host.register_test_contract_wasm(&wasm_long.as_slice());
     host.budget_ref().reset_default()?;
diff --git a/soroban-env-host/src/testutils.rs b/soroban-env-host/src/testutils.rs
index 5e914e19..c305cfa8 100644
--- a/soroban-env-host/src/testutils.rs
+++ b/soroban-env-host/src/testutils.rs
@@ -33,7 +33,7 @@ where
     C: FnOnce() -> R + UnwindSafe,
 {
     thread_local! {
-        static TEST_CONTRACT_CALL_COUNT: Cell<u64> = Cell::new(0);
+        static TEST_CONTRACT_CALL_COUNT: Cell<u64> = const { Cell::new(0) };
     }
 
     static WRAP_PANIC_HOOK: Once = Once::new();
@@ -1110,4 +1110,30 @@ pub(crate) mod wasm {
         fe.call_func(f0);
         fe.finish_and_export("test").finish()
     }
+
+    #[cfg(feature = "next")]
+    pub(crate) fn wasm_module_with_a_bit_of_everything(wasm_proto: u32) -> Vec<u8> {
+        let mut me = ModEmitter::new();
+        let pre = get_pre_release_version(INTERFACE_VERSION);
+        me.custom_section(
+            &"contractenvmetav0",
+            interface_meta_with_custom_versions(wasm_proto, pre).as_slice(),
+        );
+        me.table(RefType::FUNCREF, 128, None);
+        me.memory(1, None, false, false);
+        me.global(wasm_encoder::ValType::I64, true, &ConstExpr::i64_const(42));
+        me.export("memory", wasm_encoder::ExportKind::Memory, 0);
+        let _f0 = me.import_func("t", "_", Arity(0));
+        let mut fe = me.func(Arity(0), 0);
+        fe.push(Operand::Const64(1));
+        fe.push(Operand::Const64(2));
+        fe.i64_add();
+        fe.drop();
+        fe.push(Symbol::try_from_small_str("pass").unwrap());
+        let (mut me, fid) = fe.finish();
+        me.export_func(fid, "test");
+        me.define_elem_funcs(&[fid]);
+        me.define_data_segment(0x1234, vec![0; 8]);
+        me.finish()
+    }
 }