Repository: stellar/rs-soroban-env
Pull request #1001 (2023-08-17 15:58:44)

Tighten up `Map` and `Vector` and metering coverage
### What

resolves https://github.com/stellar/rs-soroban-env/issues/992, https://github.com/stellar/rs-soroban-env/issues/1002, #373 

### Why

[TODO: Why this change is being made. Include any context required to understand the why.]

### Known limitations

[TODO or N/A]



Patch:
diff --git a/soroban-env-common/env.json b/soroban-env-common/env.json
index 791eae75..bf016720 100644
--- a/soroban-env-common/env.json
+++ b/soroban-env-common/env.json
@@ -821,62 +821,38 @@
                 },
                 {
                     "export": "5",
-                    "name": "map_prev_key",
+                    "name": "map_key_by_pos",
                     "args": [
                         {
                             "name": "m",
                             "type": "MapObject"
                         },
                         {
-                            "name": "k",
-                            "type": "Val"
+                            "name": "i",
+                            "type": "U32Val"
                         }
                     ],
                     "return": "Val",
-                    "docs": "Given a key, find the first key less than itself in the map's sorted order. If such a key does not exist, return an ScError."
+                    "docs": "Get the key from a map at position `i`. If `i` is an invalid position, return ScError."
                 },
                 {
                     "export": "6",
-                    "name": "map_next_key",
+                    "name": "map_val_by_pos",
                     "args": [
                         {
                             "name": "m",
                             "type": "MapObject"
                         },
                         {
-                            "name": "k",
-                            "type": "Val"
+                            "name": "i",
+                            "type": "U32Val"
                         }
                     ],
                     "return": "Val",
-                    "docs": "Given a key, find the first key greater than itself in the map's sorted order. If such a key does not exist, return an ScError."
+                    "docs": "Get the value from a map at position `i`. If `i` is an invalid position, return ScError."
                 },
                 {
                     "export": "7",
-                    "name": "map_min_key",
-                    "args": [
-                        {
-                            "name": "m",
-                            "type": "MapObject"
-                        }
-                    ],
-                    "return": "Val",
-                    "docs": "Find the minimum key from a map. If the map is empty, return an ScError."
-                },
-                {
-                    "export": "8",
-                    "name": "map_max_key",
-                    "args": [
-                        {
-                            "name": "m",
-                            "type": "MapObject"
-                        }
-                    ],
-                    "return": "Val",
-                    "docs": "Find the maximum key from a map. If the map is empty, return an ScError."
-                },
-                {
-                    "export": "9",
                     "name": "map_keys",
                     "args": [
                         {
@@ -888,7 +864,7 @@
                     "docs": "Return a new vector containing all the keys in a map. The new vector is ordered in the original map's key-sorted order."
                 },
                 {
-                    "export": "a",
+                    "export": "8",
                     "name": "map_values",
                     "args": [
                         {
@@ -900,7 +876,7 @@
                     "docs": "Return a new vector containing all the values in a map. The new vector is ordered in the original map's key-sorted order."
                 },
                 {
-                    "export": "b",
+                    "export": "9",
                     "name": "map_new_from_linear_memory",
                     "args": [
                         {
@@ -920,7 +896,7 @@
                     "docs": "Return a new map initialized from a pair of equal-length arrays, one for keys and one for values, given by a pair of linear-memory addresses and a length in Vals."
                 },
                 {
-                    "export": "c",
+                    "export": "a",
                     "name": "map_unpack_to_linear_memory",
                     "args": [
                         {
@@ -952,14 +928,9 @@
                 {
                     "export": "_",
                     "name": "vec_new",
-                    "args": [
-                        {
-                            "name": "c",
-                            "type": "Val"
-                        }
-                    ],
+                    "args": [],
                     "return": "VecObject",
-                    "docs": "Creates a new vector with an optional capacity hint `c`. If `c` is `Void`, no hint is assumed and the new vector is empty. Otherwise, `c` is parsed as a `u32` that represents the initial capacity of the new vector."
+                    "docs": "Creates an empty new vector."
                 },
                 {
                     "export": "0",
diff --git a/soroban-env-common/src/meta.rs b/soroban-env-common/src/meta.rs
index f706e339..00d924fa 100644
--- a/soroban-env-common/src/meta.rs
+++ b/soroban-env-common/src/meta.rs
@@ -42,7 +42,7 @@ pub const ENV_META_V0_SECTION_NAME: &str = "contractenvmetav0";
 
 soroban_env_macros::generate_env_meta_consts!(
     ledger_protocol_version: 20,
-    pre_release_version: 54,
+    pre_release_version: 55,
 );
 
 pub fn get_ledger_protocol_version(interface_version: u64) -> u32 {
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 94cb3dad..a68072f7 100644
--- a/soroban-env-host/benches/common/cost_types/vm_ops.rs
+++ b/soroban-env-host/benches/common/cost_types/vm_ops.rs
@@ -40,7 +40,7 @@ impl HostCostMeasurement for VmInstantiationMeasure {
 
     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 idx = rng.gen_range(0..10) % util::TEST_WASMS.len();
         let wasm = util::TEST_WASMS[idx].into();
         VmInstantiationSample { id: Some(id), wasm }
     }
diff --git a/soroban-env-host/benches/common/measure.rs b/soroban-env-host/benches/common/measure.rs
index 347bf626..a83494f3 100644
--- a/soroban-env-host/benches/common/measure.rs
+++ b/soroban-env-host/benches/common/measure.rs
@@ -447,7 +447,7 @@ pub fn measure_cost_variation<HCM: HostCostMeasurement>(
                 1 => Some(HCM::new_best_case(host, &mut rng)),
                 2 => Some(HCM::new_worst_case(host, &mut rng, large_input)),
                 n if n < count => {
-                    let input = rng.gen_range(1, 2 + large_input);
+                    let input = rng.gen_range(1..2 + large_input);
                     Some(HCM::new_random_case(host, &mut rng, input))
                 }
                 _ => None,
diff --git a/soroban-env-host/src/auth.rs b/soroban-env-host/src/auth.rs
index 47bda02f..fcb69101 100644
--- a/soroban-env-host/src/auth.rs
+++ b/soroban-env-host/src/auth.rs
@@ -13,7 +13,7 @@ use soroban_env_common::{AddressObject, Compare, Symbol, TryFromVal, TryIntoVal,
 
 use crate::budget::{AsBudget, Budget};
 use crate::host::error::TryBorrowOrErr;
-use crate::host::metered_clone::{MeteredClone, MeteredContainer, MeteredIterator};
+use crate::host::metered_clone::{MeteredAlloc, MeteredClone, MeteredContainer, MeteredIterator};
 use crate::host::Frame;
 use crate::host_object::HostVec;
 use crate::native_contract::account_contract::{
@@ -345,14 +345,12 @@ impl AuthStackFrame {
             AuthStackFrame::Contract(contract_frame) => {
                 Ok(AuthorizedFunction::ContractFn(ContractFunction {
                     contract_address: contract_frame.contract_address,
-                    function_name: contract_frame
-                        .function_name
-                        .metered_clone(host.budget_ref())?,
+                    function_name: contract_frame.function_name.metered_clone(host)?,
                     args,
                 }))
             }
             AuthStackFrame::CreateContractHostFn(args) => Ok(
-                AuthorizedFunction::CreateContractHostFn(args.metered_clone(host.budget_ref())?),
+                AuthorizedFunction::CreateContractHostFn(args.metered_clone(host)?),
             ),
         }
     }
@@ -398,7 +396,7 @@ impl AuthorizedFunction {
             }
             AuthorizedFunction::CreateContractHostFn(create_contract_args) => {
                 Ok(SorobanAuthorizedFunction::CreateContractHostFn(
-                    create_contract_args.metered_clone(host.budget_ref())?,
+                    create_contract_args.metered_clone(host)?,
                 ))
             }
         }
@@ -435,7 +433,7 @@ impl AuthorizedInvocation {
         let sub_invocations = sub_invocations_xdr
             .into_iter()
             .map(|a| AuthorizedInvocation::from_xdr(host, a))
-            .metered_collect::<Result<Vec<_>, _>>(host.as_budget())??;
+            .metered_collect::<Result<Vec<_>, _>>(host)??;
         Ok(Self {
             function: AuthorizedFunction::from_xdr(host, xdr_invocation.function)?,
             sub_invocations,
@@ -472,9 +470,7 @@ impl AuthorizedInvocation {
                 .sub_invocations
                 .iter()
                 .map(|i| i.to_xdr(host))
-                .metered_collect::<Result<Vec<xdr::SorobanAuthorizedInvocation>, HostError>>(
-                    host.as_budget(),
-                )??
+                .metered_collect::<Result<Vec<xdr::SorobanAuthorizedInvocation>, HostError>>(host)??
                 .try_into()
                 .map_err(|_| HostError::from((ScErrorType::Auth, ScErrorCode::InternalError)))?,
         })
@@ -572,10 +568,7 @@ impl AuthorizationManager {
         host: &Host,
         auth_entries: Vec<SorobanAuthorizationEntry>,
     ) -> Result<Self, HostError> {
-        Vec::<AccountAuthorizationTracker>::charge_bulk_init_cpy(
-            auth_entries.len() as u64,
-            host.as_budget(),
-        )?;
+        Vec::<AccountAuthorizationTracker>::charge_bulk_init_cpy(auth_entries.len() as u64, host)?;
         let mut trackers = Vec::with_capacity(auth_entries.len());
         for auth_entry in auth_entries {
             trackers.push(RefCell::new(
@@ -657,7 +650,7 @@ impl AuthorizationManager {
         let auth_entries =
             host.visit_obj(auth_entries, |e: &HostVec| e.to_vec(host.budget_ref()))?;
         let mut trackers = self.try_borrow_invoker_contract_trackers_mut(host)?;
-        Vec::<Val>::charge_bulk_init_cpy(auth_entries.len() as u64, host.as_budget())?;
+        Vec::<Val>::charge_bulk_init_cpy(auth_entries.len() as u64, host)?;
         trackers.reserve(auth_entries.len());
         for e in auth_entries {
             trackers.push(InvokerContractAuthorizationTracker::new(host, e)?)
@@ -869,8 +862,7 @@ impl AuthorizationManager {
                 let len = self.try_borrow_account_trackers(host)?.len();
                 let mut snapshots = Vec::with_capacity(len);
                 Vec::<Option<AccountAuthorizationTrackerSnapshot>>::charge_bulk_init_cpy(
-                    len as u64,
-                    host.as_budget(),
+                    len as u64, host,
                 )?;
                 for t in self.try_borrow_account_trackers(host)?.iter() {
                     let sp = if let Ok(tracker) = t.try_borrow() {
@@ -900,9 +892,7 @@ impl AuthorizationManager {
                     .root_authorized_invocation
                     .snapshot(host.as_budget())
             })
-            .metered_collect::<Result<Vec<AuthorizedInvocationSnapshot>, HostError>>(
-                host.as_budget(),
-            )??;
+            .metered_collect::<Result<Vec<AuthorizedInvocationSnapshot>, HostError>>(host)??;
         let tracker_by_address_handle = match &self.mode {
             AuthorizationMode::Enforcing => None,
             AuthorizationMode::Recording(recording_info) => Some(
@@ -1005,7 +995,7 @@ impl AuthorizationManager {
         host: &Host,
         args: CreateContractArgs,
     ) -> Result<(), HostError> {
-        Vec::<CreateContractArgs>::charge_bulk_init_cpy(1, host.as_budget())?;
+        Vec::<CreateContractArgs>::charge_bulk_init_cpy(1, host)?;
         self.try_borrow_call_stack_mut(host)?
             .push(AuthStackFrame::CreateContractHostFn(args));
         self.push_tracker_frame(host)
@@ -1018,7 +1008,7 @@ impl AuthorizationManager {
         let _span = tracy_span!("push auth frame");
         let (contract_id, function_name) = match frame {
             Frame::ContractVM { vm, fn_name, .. } => {
-                (vm.contract_id.metered_clone(host.budget_ref())?, *fn_name)
+                (vm.contract_id.metered_clone(host)?, *fn_name)
             }
             // Skip the top-level host function stack frames as they don't
             // contain all the necessary information.
@@ -1026,13 +1016,13 @@ impl AuthorizationManager {
             // `push_create_contract_host_fn_frame`) functions instead to push
             // the frame with the required info.
             Frame::HostFunction(_) => return Ok(()),
-            Frame::Token(id, fn_name, ..) => (id.metered_clone(host.budget_ref())?, *fn_name),
+            Frame::Token(id, fn_name, ..) => (id.metered_clone(host)?, *fn_name),
             #[cfg(any(test, feature = "testutils"))]
-            Frame::TestContract(tc) => (tc.id.clone(), tc.func),
+            Frame::TestContract(tc) => (tc.id.metered_clone(host)?, tc.func),
         };
         // Currently only contracts might appear in the call stack.
         let contract_address = host.add_host_object(ScAddress::Contract(contract_id))?;
-        Vec::<ContractInvocation>::charge_bulk_init_cpy(1, host.as_budget())?;
+        Vec::<ContractInvocation>::charge_bulk_init_cpy(1, host)?;
         self.try_borrow_call_stack_mut(host)?
             .push(AuthStackFrame::Contract(ContractInvocation {
                 contract_address,
@@ -1623,9 +1613,7 @@ impl AccountAuthorizationTracker {
         })?;
         let payload_preimage =
             HashIdPreimage::SorobanAuthorization(HashIdPreimageSorobanAuthorization {
-                network_id: Hash(
-                    host.with_ledger_info(|li| li.network_id.metered_clone(host.budget_ref()))?,
-                ),
+                network_id: Hash(host.with_ledger_info(|li| li.network_id.metered_clone(host))?),
                 nonce,
                 signature_expiration_ledger: expiration_ledger,
                 invocation: self.root_invocation_to_xdr(host)?,
@@ -1773,10 +1761,10 @@ impl Host {
         let nonce_key_scval = ScVal::LedgerKeyNonce(ScNonceKey { nonce });
         let sc_address = self.scaddress_from_address(address)?;
         let nonce_key = self.storage_key_for_address(
-            sc_address.metered_clone(self.budget_ref())?,
-            nonce_key_scval.metered_clone(self.budget_ref())?,
+            sc_address.metered_clone(self)?,
+            nonce_key_scval.metered_clone(self)?,
             xdr::ContractDataDurability::Temporary,
-        );
+        )?;
         let expiration_ledger = expiration_ledger
             .max(self.get_min_expiration_ledger(xdr::ContractDataDurability::Temporary)?);
         self.with_mut_storage(|storage| {
@@ -1804,7 +1792,11 @@ impl Host {
                 data,
                 ext: LedgerEntryExt::V0,
             };
-            storage.put(&nonce_key, &Rc::new(entry), self.budget_ref())
+            storage.put(
+                &nonce_key,
+                &Rc::metered_new(entry, self)?,
+                self.budget_ref(),
+            )
         })
     }
 }
diff --git a/soroban-env-host/src/budget.rs b/soroban-env-host/src/budget.rs
index 2728b73f..0bab914a 100644
--- a/soroban-env-host/src/budget.rs
+++ b/soroban-env-host/src/budget.rs
@@ -882,7 +882,7 @@ impl Display for Budget {
     }
 }
 
-pub trait AsBudget {
+pub trait AsBudget: Clone {
     fn as_budget(&self) -> &Budget;
 }
 
@@ -892,6 +892,12 @@ impl AsBudget for Budget {
     }
 }
 
+impl AsBudget for &Budget {
+    fn as_budget(&self) -> &Budget {
+        self
+    }
+}
+
 impl AsBudget for Host {
     fn as_budget(&self) -> &Budget {
         self.budget_ref()
diff --git a/soroban-env-host/src/events/internal.rs b/soroban-env-host/src/events/internal.rs
index 9cb82153..3c88fd44 100644
--- a/soroban-env-host/src/events/internal.rs
+++ b/soroban-env-host/src/events/internal.rs
@@ -179,7 +179,7 @@ impl InternalEventsBuffer {
                     // the resulting buffer length may be different on different instances
                     // (due to diagnostic events) and we need a deterministic cost across all
                     // instances, the cost needs to be amortized and buffer size-independent.
-                    Vec::<HostEvent>::charge_bulk_init_cpy(1, host.as_budget())?;
+                    Vec::<HostEvent>::charge_bulk_init_cpy(1, host)?;
                     Ok(HostEvent {
                         event: c.to_xdr(host)?,
                         failed_call: e.1 == EventError::FromFailedCall,
diff --git a/soroban-env-host/src/host.rs b/soroban-env-host/src/host.rs
index 1f17439b..da44c295 100644
--- a/soroban-env-host/src/host.rs
+++ b/soroban-env-host/src/host.rs
@@ -54,6 +54,7 @@ use soroban_env_common::xdr::{
 
 use self::{
     frame::{Context, ContractReentryMode},
+    metered_clone::MeteredAlloc,
     metered_vector::MeteredVector,
     prng::Prng,
 };
@@ -253,6 +254,8 @@ impl Host {
     pub fn with_storage_and_budget(storage: Storage, budget: Budget) -> Self {
         #[cfg(all(not(target_family = "wasm"), feature = "tracy"))]
         let _client = tracy_client::Client::start();
+        // Here we are missing charge for the Rc::new but that is once for the
+        // lifetime of the host, so that should be okay.
         Self(Rc::new(HostImpl {
             source_account: RefCell::new(None),
             ledger: RefCell::new(None),
@@ -286,13 +289,13 @@ impl Host {
 
     #[cfg(test)]
     pub(crate) fn source_account_id(&self) -> Result<Option<AccountId>, HostError> {
-        Ok(self.try_borrow_source_account()?.clone())
+        Ok(self.try_borrow_source_account()?.metered_clone(self)?)
     }
 
     pub fn source_account_address(&self) -> Result<Option<AddressObject>, HostError> {
         if let Some(acc) = self.try_borrow_source_account()?.as_ref() {
             Ok(Some(self.add_host_object(ScAddress::Account(
-                acc.metered_clone(self.budget_ref())?,
+                acc.metered_clone(self)?,
             ))?))
         } else {
             Ok(None)
@@ -582,7 +585,7 @@ impl Host {
         let has_deployer = deployer.is_some();
         if has_deployer {
             self.try_borrow_authorization_manager()?
-                .push_create_contract_host_fn_frame(self, args.metered_clone(self.budget_ref())?)?;
+                .push_create_contract_host_fn_frame(self, args.metered_clone(self)?)?;
         }
         // Make sure that even in case of operation failure we still pop the
         // stack frame.
@@ -610,11 +613,10 @@ impl Host {
             )?;
         }
 
-        let id_preimage = self.get_full_contract_id_preimage(
-            args.contract_id_preimage.metered_clone(self.budget_ref())?,
-        )?;
+        let id_preimage =
+            self.get_full_contract_id_preimage(args.contract_id_preimage.metered_clone(self)?)?;
         let hash_id = Hash(self.metered_hash_xdr(&id_preimage)?);
-        self.create_contract_with_id(hash_id.metered_clone(self.budget_ref())?, args.executable)?;
+        self.create_contract_with_id(hash_id.metered_clone(self)?, args.executable)?;
         self.maybe_initialize_asset_token(&hash_id, &args.contract_id_preimage)?;
         self.add_host_object(ScAddress::Contract(hash_id))
     }
@@ -625,15 +627,12 @@ impl Host {
         salt: BytesObject,
     ) -> Result<Hash, HostError> {
         let contract_id_preimage = ContractIdPreimage::Address(ContractIdPreimageFromAddress {
-            address: self.visit_obj(deployer, |addr: &ScAddress| {
-                addr.metered_clone(self.budget_ref())
-            })?,
+            address: self.visit_obj(deployer, |addr: &ScAddress| addr.metered_clone(self))?,
             salt: self.u256_from_bytesobj_input("contract_id_salt", salt)?,
         });
 
-        let id_preimage = self.get_full_contract_id_preimage(
-            contract_id_preimage.metered_clone(self.budget_ref())?,
-        )?;
+        let id_preimage =
+            self.get_full_contract_id_preimage(contract_id_preimage.metered_clone(self)?)?;
         Ok(Hash(self.metered_hash_xdr(&id_preimage)?))
     }
 
@@ -739,10 +738,13 @@ impl Host {
                 )
             })?;
         let hash_obj = self.add_host_object(self.scbytes_from_slice(hash_bytes.as_slice())?)?;
-        let code_key = Rc::new(LedgerKey::ContractCode(LedgerKeyContractCode {
-            hash: Hash(hash_bytes.metered_clone(self.budget_ref())?),
-            body_type: ContractEntryBodyType::DataEntry,
-        }));
+        let code_key = Rc::metered_new(
+            LedgerKey::ContractCode(LedgerKeyContractCode {
+                hash: Hash(hash_bytes.metered_clone(self)?),
+                body_type: ContractEntryBodyType::DataEntry,
+            }),
+            self,
+        )?;
         if !self
             .try_borrow_storage_mut()?
             .has(&code_key, self.as_budget())
@@ -767,7 +769,7 @@ impl Host {
                 });
                 storage.put(
                     &code_key,
-                    &Host::ledger_entry_from_data(data),
+                    &Host::ledger_entry_from_data(self, data)?,
                     self.as_budget(),
                 )
             })?;
@@ -868,7 +870,7 @@ impl Host {
                 .try_borrow_storage_mut()?
                 .get(&key, self.as_budget())
                 .map_err(|e| self.decorate_contract_data_storage_error(e, k))?)
-            .metered_clone(self.as_budget())?;
+            .metered_clone(self)?;
 
             match current.data {
                 LedgerEntryData::ContractData(ref mut entry) => match entry.body {
@@ -897,7 +899,7 @@ impl Host {
                 }
             }
             self.try_borrow_storage_mut()?
-                .put(&key, &Rc::new(current), self.as_budget())
+                .put(&key, &Rc::metered_new(current, self)?, self.as_budget())
                 .map_err(|e| self.decorate_contract_data_storage_error(e, k))?;
         } else {
             let body = ContractDataEntryBody::DataEntry(ContractDataEntryData {
@@ -912,7 +914,11 @@ impl Host {
                 durability,
             });
             self.try_borrow_storage_mut()?
-                .put(&key, &Host::ledger_entry_from_data(data), self.as_budget())
+                .put(
+                    &key,
+                    &Host::ledger_entry_from_data(self, data)?,
+                    self.as_budget(),
+                )
                 .map_err(|e| self.decorate_contract_data_storage_error(e, k))?;
         }
 
@@ -932,11 +938,7 @@ impl Host {
         };
         // Need to copy the footprint out of the storage to allow mut borrow of
         // storage.
-        let footprint_map = self
-            .try_borrow_storage()?
-            .footprint
-            .0
-            .metered_clone(self.budget_ref())?;
+        let footprint_map = self.try_borrow_storage()?.footprint.0.metered_clone(self)?;
         for (key, _) in footprint_map.iter(self.budget_ref())? {
             match key.as_ref() {
                 LedgerKey::ContractData(_) | LedgerKey::ContractCode(_) => {
@@ -944,7 +946,7 @@ impl Host {
                         self.try_borrow_storage_mut()?
                             .bump_relative_to_entry_expiration(
                                 self,
-                                key.clone(),
+                                key.metered_clone(self)?,
                                 autobump_ledgers,
                             )?;
                     }
@@ -1099,6 +1101,8 @@ impl EnvBase for Host {
         assert!(Rc::ptr_eq(&self.0, &other.0));
     }
 
+    // This function is not being metered, it's not used anywhere so it's okay
+    // for now
     fn deep_clone(&self) -> Self {
         Host(Rc::new((*self.0).clone()))
     }
@@ -1160,7 +1164,7 @@ impl EnvBase for Host {
     }
 
     fn map_new_from_slices(&self, keys: &[&str], vals: &[Val]) -> Result<MapObject, HostError> {
-        Vec::<Symbol>::charge_bulk_init_cpy(keys.len() as u64, self.as_budget())?;
+        Vec::<Symbol>::charge_bulk_init_cpy(keys.len() as u64, self)?;
         // If only fallible iterators worked better in Rust, we would not need this Vec<...>.
         let mut key_syms: Vec<Symbol> = Vec::with_capacity(keys.len());
         for k in keys.iter() {
@@ -1183,9 +1187,6 @@ impl EnvBase for Host {
         keys: &[&str],
         vals: &mut [Val],
     ) -> Result<Void, HostError> {
-        // Main costs are already covered by `visit_obj` and `check_symbol_matches`. Here
-        // we charge shallow copy of the values.
-        metered_clone::charge_shallow_copy::<Val>(keys.len() as u64, self.as_budget())?;
         if keys.len() != vals.len() {
             return Err(self.err(
                 ScErrorType::Object,
@@ -1209,6 +1210,7 @@ impl EnvBase for Host {
                 self.check_symbol_matches(ik.as_bytes(), sym)?;
             }
 
+            metered_clone::charge_shallow_copy::<Val>(keys.len() as u64, self)?;
             for (iv, mv) in vals.iter_mut().zip(hm.values(self)?) {
                 *iv = *mv;
             }
@@ -1235,7 +1237,7 @@ impl EnvBase for Host {
                     &[],
                 ));
             }
-            metered_clone::charge_shallow_copy::<Val>(hv.len() as u64, self.as_budget())?;
+            metered_clone::charge_shallow_copy::<Val>(hv.len() as u64, self)?;
             vals.copy_from_slice(hv.as_slice());
             Ok(())
         })?;
@@ -1310,7 +1312,7 @@ impl VmCallerEnv for Host {
         self.add_host_object(ScAddress::Contract(invoking_contract_hash))
     }
 
-    // Metered: covered by `visit` and `metered_cmp`.
+    // Metered: covered by `visit`.
     fn obj_cmp(&self, _vmcaller: &mut VmCaller<Host>, a: Val, b: Val) -> Result<i64, HostError> {
         let res = match {
             match (Object::try_from(a), Object::try_from(b)) {
@@ -1702,59 +1704,27 @@ impl VmCallerEnv for Host {
         self.visit_obj(m, move |hm: &HostMap| Ok(hm.contains_key(&k, self)?.into()))
     }
 
-    fn map_prev_key(
+    fn map_key_by_pos(
         &self,
         _vmcaller: &mut VmCaller<Host>,
         m: MapObject,
-        k: Val,
+        i: U32Val,
     ) -> Result<Val, HostError> {
+        let i: u32 = i.into();
         self.visit_obj(m, |hm: &HostMap| {
-            if let Some((pk, _)) = hm.get_prev(&k, self)? {
-                Ok(*pk)
-            } else {
-                // We return Ok(error) here to indicate "the end of iteration".
-                Ok(
-                    Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds)
-                        .to_val(),
-                )
-            }
+            hm.get_at_index(i as usize, self).map(|r| r.0)
         })
     }
 
-    fn map_next_key(
+    fn map_val_by_pos(
         &self,
         _vmcaller: &mut VmCaller<Host>,
         m: MapObject,
-        k: Val,
+        i: U32Val,
     ) -> Result<Val, HostError> {
+        let i: u32 = i.into();
         self.visit_obj(m, |hm: &HostMap| {
-            if let Some((pk, _)) = hm.get_next(&k, self)? {
-                Ok(*pk)
-            } else {
-                // We return Ok(error) here to indicate "the end of iteration".
-                Ok(
-                    Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds)
-                        .to_val(),
-                )
-            }
-        })
-    }
-
-    fn map_min_key(&self, _vmcaller: &mut VmCaller<Host>, m: MapObject) -> Result<Val, HostError> {
-        self.visit_obj(m, |hm: &HostMap| match hm.get_min(self)? {
-            Some((pk, pv)) => Ok(*pk),
-            None => Ok(
-                Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds).to_val(),
-            ),
-        })
-    }
-
-    fn map_max_key(&self, _vmcaller: &mut VmCaller<Host>, m: MapObject) -> Result<Val, HostError> {
-        self.visit_obj(m, |hm: &HostMap| match hm.get_max(self)? {
-            Some((pk, pv)) => Ok(*pk),
-            None => Ok(
-                Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds).to_val(),
-            ),
+            hm.get_at_index(i as usize, self).map(|r| r.1)
         })
     }
 
@@ -1793,7 +1763,7 @@ impl VmCallerEnv for Host {
             pos: keys_pos,
             len,
         } = self.decode_vmslice(keys_pos, len)?;
-        Vec::<Symbol>::charge_bulk_init_cpy(len as u64, self.as_budget())?;
+        Vec::<Symbol>::charge_bulk_init_cpy(len as u64, self)?;
         let mut key_syms: Vec<Symbol> = Vec::with_capacity(len as usize);
         self.metered_vm_scan_slices_in_linear_memory(
             vmcaller,
@@ -1811,7 +1781,7 @@ impl VmCallerEnv for Host {
 
         // Step 2: extract all val Vals.
         let vals_pos: u32 = vals_pos.into();
-        Vec::<Val>::charge_bulk_init_cpy(len as u64, self.as_budget())?;
+        Vec::<Val>::charge_bulk_init_cpy(len as u64, self)?;
         let mut vals: Vec<Val> = vec![Val::VOID.into(); len as usize];
         self.metered_vm_read_vals_from_linear_memory::<8, Val>(
             vmcaller,
@@ -1891,14 +1861,7 @@ impl VmCallerEnv for Host {
         Ok(Val::VOID)
     }
 
-    fn vec_new(&self, _vmcaller: &mut VmCaller<Host>, c: Val) -> Result<VecObject, HostError> {
-        // NB: we ignore capacity because vectors are immutable
-        // and there's no reuse, we always size them exactly.
-        let _capacity: usize = if c.is_void() {
-            0
-        } else {
-            self.usize_from_rawval_u32_input("c", c)?
-        };
+    fn vec_new(&self, _vmcaller: &mut VmCaller<Host>) -> Result<VecObject, HostError> {
         self.add_host_object(HostVec::new())
     }
 
@@ -2102,7 +2065,7 @@ impl VmCallerEnv for Host {
         len: U32Val,
     ) -> Result<VecObject, HostError> {
         let VmSlice { vm, pos, len } = self.decode_vmslice(vals_pos, len)?;
-        Vec::<Val>::charge_bulk_init_cpy(len as u64, self.as_budget())?;
+        Vec::<Val>::charge_bulk_init_cpy(len as u64, self)?;
         let mut vals: Vec<Val> = vec![Val::VOID.to_val(); len as usize];
         self.metered_vm_read_vals_from_linear_memory::<8, Val>(
             vmcaller,
@@ -2293,7 +2256,7 @@ impl VmCallerEnv for Host {
         let contract_id = self.get_current_contract_id_internal()?;
         let key = self.contract_instance_ledger_key(&contract_id)?;
         self.try_borrow_storage_mut()?
-            .bump(self, key.clone(), min.into())
+            .bump(self, key.metered_clone(self)?, min.into())
             .map_err(|e| self.decorate_contract_instance_storage_error(e, &contract_id))?;
         match self
             .retrieve_contract_instance_from_storage(&key)?
@@ -2319,7 +2282,7 @@ impl VmCallerEnv for Host {
         let contract_id = self.contract_id_from_address(contract)?;
         let key = self.contract_instance_ledger_key(&contract_id)?;
         self.try_borrow_storage_mut()?
-            .bump(self, key.clone(), min.into())
+            .bump(self, key.metered_clone(self)?, min.into())
             .map_err(|e| self.decorate_contract_instance_storage_error(e, &contract_id))?;
         match self
             .retrieve_contract_instance_from_storage(&key)?
@@ -2345,9 +2308,7 @@ impl VmCallerEnv for Host {
         salt: BytesObject,
     ) -> Result<AddressObject, HostError> {
         let contract_id_preimage = ContractIdPreimage::Address(ContractIdPreimageFromAddress {
-            address: self.visit_obj(deployer, |addr: &ScAddress| {
-                addr.metered_clone(self.budget_ref())
-            })?,
+            address: self.visit_obj(deployer, |addr: &ScAddress| addr.metered_clone(self))?,
             salt: self.u256_from_bytesobj_input("contract_id_salt", salt)?,
         });
         let executable =
@@ -2404,9 +2365,8 @@ impl VmCallerEnv for Host {
         _vmcaller: &mut VmCaller<Host>,
         wasm: BytesObject,
     ) -> Result<BytesObject, HostError> {
-        let wasm_vec = self.visit_obj(wasm, |bytes: &ScBytes| {
-            bytes.as_vec().metered_clone(self.budget_ref())
-        })?;
+        let wasm_vec =
+            self.visit_obj(wasm, |bytes: &ScBytes| bytes.as_vec().metered_clone(self))?;
         self.upload_contract_wasm(wasm_vec)
     }
 
@@ -2656,7 +2616,7 @@ impl VmCallerEnv for Host {
         let i: u32 = iv.into();
         let u = self.u8_from_u32val_input("u", u)?;
         let vnew = self.visit_obj(b, move |hv: &ScBytes| {
-            let mut vnew: Vec<u8> = hv.metered_clone(&self.0.budget)?.into();
+            let mut vnew: Vec<u8> = hv.metered_clone(self)?.into();
             match vnew.get_mut(i as usize) {
                 None => Err(self.err(
                     ScErrorType::Object,
@@ -2704,12 +2664,12 @@ impl VmCallerEnv for Host {
         let i: u32 = i.into();
         let vnew = self.visit_obj(b, move |hv: &ScBytes| {
             self.validate_index_lt_bound(i, hv.len())?;
-            let mut vnew: Vec<u8> = hv.metered_clone(&self.0.budget)?.into();
+            let mut vnew: Vec<u8> = hv.metered_clone(self)?.into();
             // len > i has been verified above but use saturating_sub just in case
             let n_elts = (hv.len() as u64).saturating_sub(i as u64);
             // remove elements incurs the cost of moving bytes, it does not incur
             // allocation/deallocation
-            metered_clone::charge_shallow_copy::<u8>(n_elts, self.as_budget())?;
+            metered_clone::charge_shallow_copy::<u8>(n_elts, self)?;
             vnew.remove(i as usize);
             Ok(ScBytes(vnew.try_into()?))
         })?;
@@ -2758,8 +2718,7 @@ impl VmCallerEnv for Host {
             // we allocate the new vector to be able to hold `len + 1` bytes, so that the push
             // will not trigger a reallocation, causing data to be cloned twice.
             let len = hv.len().saturating_add(1);
-            metered_clone::charge_heap_alloc::<u8>(len as u64, self.as_budget())?;
-            metered_clone::charge_shallow_copy::<u8>(len as u64, self.as_budget())?;
+            Vec::<u8>::charge_bulk_init_cpy(len as u64, self)?;
             let mut vnew: Vec<u8> = Vec::with_capacity(len);
             vnew.extend_from_slice(hv.as_slice());
             vnew.push(u);
@@ -2775,7 +2734,7 @@ impl VmCallerEnv for Host {
         b: BytesObject,
     ) -> Result<BytesObject, HostError> {
         let vnew = self.visit_obj(b, move |hv: &ScBytes| {
-            let mut vnew: Vec<u8> = hv.metered_clone(self.as_budget())?.into();
+            let mut vnew: Vec<u8> = hv.metered_clone(self)?.into();
             // Popping will not trigger reallocation. Here we don't charge anything since this is
             // just a `len` reduction.
             if vnew.pop().is_none() {
@@ -2845,12 +2804,9 @@ impl VmCallerEnv for Host {
             // we allocate the new vector to be able to hold `len + 1` bytes, so that the push
             // will not trigger a reallocation, causing data to be cloned twice.
             let len = hv.len().saturating_add(1);
-            metered_clone::charge_heap_alloc::<u8>(len as u64, self.as_budget())?;
-            metered_clone::charge_shallow_copy::<u8>(len as u64, self.as_budget())?;
+            Vec::<u8>::charge_bulk_init_cpy(len as u64, self)?;
             let mut vnew: Vec<u8> = Vec::with_capacity(len);
             vnew.extend_from_slice(hv.as_slice());
-            // insert will cause the memcpy by shifting all the values at and after `i`.
-            metered_clone::charge_shallow_copy::<u8>(len as u64, self.as_budget())?;
             vnew.insert(i as usize, u);
             Ok(ScBytes(vnew.try_into()?))
         })?;
@@ -2873,8 +2829,7 @@ impl VmCallerEnv for Host {
                 // we already checked above that `len` will not overflow, here using
                 // saturating_add just in case.
                 let len = sb1.len().saturating_add(sb2.len());
-                metered_clone::charge_heap_alloc::<u8>(len as u64, self.as_budget())?;
-                metered_clone::charge_shallow_copy::<u8>(len as u64, self.as_budget())?;
+                Vec::<u8>::charge_bulk_init_cpy(len as u64, self)?;
                 let mut vnew: Vec<u8> = Vec::with_capacity(len);
                 vnew.extend_from_slice(sb1.as_slice());
                 vnew.extend_from_slice(sb2.as_slice());
@@ -2985,7 +2940,7 @@ impl VmCallerEnv for Host {
 
         let get_host_val_tuple = |id: &Hash, function: &Symbol| -> Result<[Val; 2], HostError> {
             let addr_val = self
-                .add_host_object(ScAddress::Contract(id.metered_clone(self.as_budget())?))?
+                .add_host_object(ScAddress::Contract(id.metered_clone(self)?))?
                 .into();
             let function_val = (*function).into();
             Ok([addr_val, function_val])
@@ -3066,7 +3021,7 @@ impl VmCallerEnv for Host {
                 #[cfg(any(test, feature = "testutils"))]
                 Frame::TestContract(c) => &c.args,
             };
-            args.metered_clone(self.budget_ref())
+            args.metered_clone(self)
         })?;
 
         Ok(self
@@ -3109,7 +3064,7 @@ impl VmCallerEnv for Host {
         _vmcaller: &mut VmCaller<Self::VmUserState>,
         address: AddressObject,
     ) -> Result<Val, Self::Error> {
-        let addr = self.visit_obj(address, |addr: &ScAddress| Ok(addr.clone()))?;
+        let addr = self.visit_obj(address, |addr: &ScAddress| addr.metered_clone(self))?;
         match addr {
             ScAddress::Account(AccountId(PublicKey::PublicKeyTypeEd25519(pk))) => Ok(self
                 .add_host_object(ScBytes(self.metered_slice_to_vec(&pk.0)?.try_into()?))?
@@ -3123,7 +3078,7 @@ impl VmCallerEnv for Host {
         _vmcaller: &mut VmCaller<Self::VmUserState>,
         address: AddressObject,
     ) -> Result<Val, Self::Error> {
-        let addr = self.visit_obj(address, |addr: &ScAddress| Ok(addr.clone()))?;
+        let addr = self.visit_obj(address, |addr: &ScAddress| addr.metered_clone(self))?;
         match addr {
             ScAddress::Account(_) => Ok(().into()),
             ScAddress::Contract(Hash(h)) => Ok(self
diff --git a/soroban-env-host/src/host/conversion.rs b/soroban-env-host/src/host/conversion.rs
index ff6ba481..6e62741e 100644
--- a/soroban-env-host/src/host/conversion.rs
+++ b/soroban-env-host/src/host/conversion.rs
@@ -1,6 +1,8 @@
 use std::rc::Rc;
 
-use super::metered_clone::{charge_shallow_copy, MeteredClone, MeteredContainer, MeteredIterator};
+use super::metered_clone::{
+    charge_shallow_copy, MeteredAlloc, MeteredClone, MeteredContainer, MeteredIterator,
+};
 use crate::budget::AsBudget;
 use crate::err;
 use crate::host_object::{HostMap, HostObject, HostVec};
@@ -66,7 +68,7 @@ impl Host {
 
     pub(crate) fn u256_from_account(&self, account_id: &AccountId) -> Result<Uint256, HostError> {
         let crate::xdr::PublicKey::PublicKeyTypeEd25519(ed25519) =
-            account_id.metered_clone(&self.0.budget)?.0;
+            account_id.metered_clone(self)?.0;
         Ok(ed25519)
     }
 
@@ -165,13 +167,16 @@ impl Host {
         contract_address: ScAddress,
         key: ScVal,
         durability: ContractDataDurability,
-    ) -> Rc<LedgerKey> {
-        Rc::new(LedgerKey::ContractData(LedgerKeyContractData {
-            contract: contract_address,
-            key,
-            durability: durability,
-            body_type: ContractEntryBodyType::DataEntry,
-        }))
+    ) -> Result<Rc<LedgerKey>, HostError> {
+        Rc::metered_new(
+            LedgerKey::ContractData(LedgerKeyContractData {
+                contract: contract_address,
+                key,
+                durability: durability,
+                body_type: ContractEntryBodyType::DataEntry,
+            }),
+            self,
+        )
     }
 
     pub(crate) fn storage_key_from_scval(
@@ -180,7 +185,7 @@ impl Host {
         durability: ContractDataDurability,
     ) -> Result<Rc<LedgerKey>, HostError> {
         let contract_id = self.get_current_contract_id_internal()?;
-        Ok(self.storage_key_for_address(ScAddress::Contract(contract_id), key, durability))
+        self.storage_key_for_address(ScAddress::Contract(contract_id), key, durability)
     }
 
     // Notes on metering: covered by components.
@@ -225,10 +230,7 @@ impl Host {
     }
 
     pub(crate) fn call_args_from_obj(&self, args: VecObject) -> Result<Vec<Val>, HostError> {
-        self.visit_obj(args, |hv: &HostVec| {
-            // Metering: free
-            Ok(hv.iter().cloned().collect())
-        })
+        self.visit_obj(args, |hv: &HostVec| hv.to_vec(self.as_budget()))
     }
 
     // Metering: covered by rawvals_to_vec
@@ -245,7 +247,7 @@ impl Host {
         raw_vals
             .iter()
             .map(|v| self.from_host_val(*v))
-            .metered_collect::<Result<Vec<ScVal>, HostError>>(self.as_budget())??
+            .metered_collect::<Result<Vec<ScVal>, HostError>>(self)??
             .try_into()
             .map_err(|_| {
                 err!(
@@ -280,7 +282,7 @@ impl Host {
         sc_vals
             .iter()
             .map(|scv| self.to_host_val(scv))
-            .metered_collect::<Result<Vec<Val>, HostError>>(self.as_budget())?
+            .metered_collect::<Result<Vec<Val>, HostError>>(self)?
     }
 
     pub(crate) fn bytesobj_from_internal_contract_id(
@@ -301,7 +303,7 @@ impl Host {
     }
 
     pub(crate) fn metered_slice_to_vec(&self, s: &[u8]) -> Result<Vec<u8>, HostError> {
-        Vec::<u8>::charge_bulk_init_cpy(s.len() as u64, self.as_budget())?;
+        Vec::<u8>::charge_bulk_init_cpy(s.len() as u64, self)?;
         Ok(s.to_vec())
     }
 
@@ -318,13 +320,11 @@ impl Host {
         &self,
         address: AddressObject,
     ) -> Result<ScAddress, HostError> {
-        self.visit_obj(address, |addr: &ScAddress| {
-            addr.metered_clone(self.budget_ref())
-        })
+        self.visit_obj(address, |addr: &ScAddress| addr.metered_clone(self))
     }
 
     pub(crate) fn host_map_to_scmap(&self, map: &HostMap) -> Result<ScMap, HostError> {
-        Vec::<ScMapEntry>::charge_bulk_init_cpy(map.len() as u64, self.as_budget())?;
+        Vec::<ScMapEntry>::charge_bulk_init_cpy(map.len() as u64, self)?;
         let mut mv = Vec::with_capacity(map.len());
         for (k, v) in map.iter(self)? {
             let key = self.from_host_val(*k)?;
@@ -404,7 +404,7 @@ impl Host {
             self.visit_obj_untyped(objref, |ho| {
                 let val = match ho {
                     HostObject::Vec(vv) => {
-                        Vec::<ScVal>::charge_bulk_init_cpy(vv.len() as u64, self.as_budget())?;
+                        Vec::<ScVal>::charge_bulk_init_cpy(vv.len() as u64, self)?;
                         let sv = vv.iter().map(|e| self.from_host_val(*e)).collect::<Result<
                             Vec<ScVal>,
                             HostError,
@@ -414,33 +414,31 @@ impl Host {
                     }
                     HostObject::Map(mm) => ScVal::Map(Some(self.host_map_to_scmap(mm)?)),
                     HostObject::U64(u) => {
-                        charge_shallow_copy::<u64>(1, self.as_budget())?;
+                        charge_shallow_copy::<u64>(1, self)?;
                         ScVal::U64(*u)
                     }
                     HostObject::I64(i) => {
-                        charge_shallow_copy::<i64>(1, self.as_budget())?;
+                        charge_shallow_copy::<i64>(1, self)?;
                         ScVal::I64(*i)
                     }
-                    HostObject::TimePoint(tp) => {
-                        ScVal::Timepoint(tp.metered_clone(self.as_budget())?)
-                    }
-                    HostObject::Duration(d) => ScVal::Duration(d.metered_clone(self.as_budget())?),
+                    HostObject::TimePoint(tp) => ScVal::Timepoint(tp.metered_clone(self)?),
+                    HostObject::Duration(d) => ScVal::Duration(d.metered_clone(self)?),
                     HostObject::U128(u) => {
-                        charge_shallow_copy::<u128>(1, self.as_budget())?;
+                        charge_shallow_copy::<u128>(1, self)?;
                         ScVal::U128(UInt128Parts {
                             hi: int128_helpers::u128_hi(*u),
                             lo: int128_helpers::u128_lo(*u),
                         })
                     }
                     HostObject::I128(i) => {
-                        charge_shallow_copy::<i128>(1, self.as_budget())?;
+                        charge_shallow_copy::<i128>(1, self)?;
                         ScVal::I128(Int128Parts {
                             hi: int128_helpers::i128_hi(*i),
                             lo: int128_helpers::i128_lo(*i),
                         })
                     }
                     HostObject::U256(u) => {
-                        charge_shallow_copy::<u128>(2, self.as_budget())?;
+                        charge_shallow_copy::<u128>(2, self)?;
                         let (hi_hi, hi_lo, lo_hi, lo_lo) = u256_into_pieces(*u);
                         ScVal::U256(UInt256Parts {
                             hi_hi,
@@ -450,7 +448,7 @@ impl Host {
                         })
                     }
                     HostObject::I256(i) => {
-                        charge_shallow_copy::<i128>(2, self.as_budget())?;
+                        charge_shallow_copy::<i128>(2, self)?;
                         let (hi_hi, hi_lo, lo_hi, lo_lo) = i256_into_pieces(*i);
                         ScVal::I256(Int256Parts {
                             hi_hi,
@@ -459,12 +457,10 @@ impl Host {
                             lo_lo,
                         })
                     }
-                    HostObject::Bytes(b) => ScVal::Bytes(b.metered_clone(self.as_budget())?),
-                    HostObject::String(s) => ScVal::String(s.metered_clone(self.as_budget())?),
-                    HostObject::Symbol(s) => ScVal::Symbol(s.metered_clone(self.as_budget())?),
-                    HostObject::Address(addr) => {
-                        ScVal::Address(addr.metered_clone(self.as_budget())?)
-                    } // For any future `HostObject` types we add, make sure to add some metering.
+                    HostObject::Bytes(b) => ScVal::Bytes(b.metered_clone(self)?),
+                    HostObject::String(s) => ScVal::String(s.metered_clone(self)?),
+                    HostObject::Symbol(s) => ScVal::Symbol(s.metered_clone(self)?),
+                    HostObject::Address(addr) => ScVal::Address(addr.metered_clone(self)?), // For any future `HostObject` types we add, make sure to add some metering.
                 };
                 Ok(ScValObject::unchecked_from_val(val))
             })
@@ -478,7 +474,7 @@ impl Host {
             // below. There is no otherwise ubiquitous metering for ScVal->Val conversion,
             // since most of them happens in the "common" crate with no access to the host.
             ScVal::Vec(Some(v)) => {
-                Vec::<Val>::charge_bulk_init_cpy(v.len() as u64, self.as_budget())?;
+                Vec::<Val>::charge_bulk_init_cpy(v.len() as u64, self)?;
                 let mut vv = Vec::with_capacity(v.len());
                 for e in v.iter() {
                     vv.push(self.to_host_val(e)?)
@@ -486,7 +482,7 @@ impl Host {
                 Ok(self.add_host_object(HostVec::from_vec(vv)?)?.into())
             }
             ScVal::Map(Some(m)) => {
-                Vec::<(Val, Val)>::charge_bulk_init_cpy(m.len() as u64, self.as_budget())?;
+                Vec::<(Val, Val)>::charge_bulk_init_cpy(m.len() as u64, self)?;
                 let mut mm = Vec::with_capacity(m.len());
                 for pair in m.iter() {
                     let k = self.to_host_val(&pair.key)?;
@@ -508,55 +504,43 @@ impl Host {
                 &[],
             )),
             ScVal::U64(u) => {
-                charge_shallow_copy::<u64>(1, self.as_budget())?;
+                charge_shallow_copy::<u64>(1, self)?;
                 Ok(self.add_host_object(*u)?.into())
             }
             ScVal::I64(i) => {
-                charge_shallow_copy::<i64>(1, self.as_budget())?;
+                charge_shallow_copy::<i64>(1, self)?;
                 Ok(self.add_host_object(*i)?.into())
             }
-            ScVal::Timepoint(t) => Ok(self
-                .add_host_object(t.metered_clone(self.as_budget())?)?
-                .into()),
-            ScVal::Duration(d) => Ok(self
-                .add_host_object(d.metered_clone(self.as_budget())?)?
-                .into()),
+            ScVal::Timepoint(t) => Ok(self.add_host_object(t.metered_clone(self)?)?.into()),
+            ScVal::Duration(d) => Ok(self.add_host_object(d.metered_clone(self)?)?.into()),
             ScVal::U128(u) => {
-                charge_shallow_copy::<u128>(1, self.as_budget())?;
+                charge_shallow_copy::<u128>(1, self)?;
                 Ok(self
                     .add_host_object(int128_helpers::u128_from_pieces(u.hi, u.lo))?
                     .into())
             }
             ScVal::I128(i) => {
-                charge_shallow_copy::<i128>(1, self.as_budget())?;
+                charge_shallow_copy::<i128>(1, self)?;
                 Ok(self
                     .add_host_object(int128_helpers::i128_from_pieces(i.hi, i.lo))?
                     .into())
             }
             ScVal::U256(u) => {
-                charge_shallow_copy::<u128>(2, self.as_budget())?;
+                charge_shallow_copy::<u128>(2, self)?;
                 Ok(self
                     .add_host_object(u256_from_pieces(u.hi_hi, u.hi_lo, u.lo_hi, u.lo_lo))?
                     .into())
             }
             ScVal::I256(i) => {
-                charge_shallow_copy::<i128>(2, self.as_budget())?;
+                charge_shallow_copy::<i128>(2, self)?;
                 Ok(self
                     .add_host_object(i256_from_pieces(i.hi_hi, i.hi_lo, i.lo_hi, i.lo_lo))?
                     .into())
             }
-            ScVal::Bytes(b) => Ok(self
-                .add_host_object(b.metered_clone(self.as_budget())?)?
-                .into()),
-            ScVal::String(s) => Ok(self
-                .add_host_object(s.metered_clone(self.as_budget())?)?
-                .into()),
-            ScVal::Symbol(s) => Ok(self
-                .add_host_object(s.metered_clone(self.as_budget())?)?
-                .into()),
-            ScVal::Address(addr) => Ok(self
-                .add_host_object(addr.metered_clone(self.as_budget())?)?
-                .into()),
+            ScVal::Bytes(b) => Ok(self.add_host_object(b.metered_clone(self)?)?.into()),
+            ScVal::String(s) => Ok(self.add_host_object(s.metered_clone(self)?)?.into()),
+            ScVal::Symbol(s) => Ok(self.add_host_object(s.metered_clone(self)?)?.into()),
+            ScVal::Address(addr) => Ok(self.add_host_object(addr.metered_clone(self)?)?.into()),
             ScVal::Bool(_)
             | ScVal::Void
             | ScVal::Error(_)
diff --git a/soroban-env-host/src/host/data_helper.rs b/soroban-env-host/src/host/data_helper.rs
index 9314501e..955a2980 100644
--- a/soroban-env-host/src/host/data_helper.rs
+++ b/soroban-env-host/src/host/data_helper.rs
@@ -17,21 +17,23 @@ use crate::xdr::{
 };
 use crate::{err, Host, HostError};
 
-use super::metered_clone::MeteredClone;
+use super::metered_clone::{MeteredAlloc, MeteredClone};
 
 impl Host {
-    // Notes on metering: free
     pub fn contract_instance_ledger_key(
         &self,
         contract_id: &Hash,
     ) -> Result<Rc<LedgerKey>, HostError> {
-        let contract_id = contract_id.metered_clone(self.as_budget())?;
-        Ok(Rc::new(LedgerKey::ContractData(LedgerKeyContractData {
-            key: ScVal::LedgerKeyContractInstance,
-            durability: ContractDataDurability::Persistent,
-            body_type: ContractEntryBodyType::DataEntry,
-            contract: ScAddress::Contract(contract_id),
-        })))
+        let contract_id = contract_id.metered_clone(self)?;
+        Rc::metered_new(
+            LedgerKey::ContractData(LedgerKeyContractData {
+                key: ScVal::LedgerKeyContractInstance,
+                durability: ContractDataDurability::Persistent,
+                body_type: ContractEntryBodyType::DataEntry,
+                contract: ScAddress::Contract(contract_id),
+            }),
+            self,
+        )
     }
 
     // Notes on metering: retrieving from storage covered. Rest are free.
@@ -43,7 +45,7 @@ impl Host {
         match &entry.data {
             LedgerEntryData::ContractData(ContractDataEntry { body, .. }) => match body {
                 ContractDataEntryBody::DataEntry(data) => match &data.val {
-                    ScVal::ContractInstance(instance) => instance.metered_clone(self.as_budget()),
+                    ScVal::ContractInstance(instance) => instance.metered_clone(self),
                     other => Err(err!(
                         self,
                         (ScErrorType::Storage, ScErrorCode::InternalError),
@@ -70,11 +72,14 @@ impl Host {
         &self,
         wasm_hash: &Hash,
     ) -> Result<Rc<LedgerKey>, HostError> {
-        let wasm_hash = wasm_hash.metered_clone(self.as_budget())?;
-        Ok(Rc::new(LedgerKey::ContractCode(LedgerKeyContractCode {
-            hash: wasm_hash,
-            body_type: ContractEntryBodyType::DataEntry,
-        })))
+        let wasm_hash = wasm_hash.metered_clone(self)?;
+        Rc::metered_new(
+            LedgerKey::ContractCode(LedgerKeyContractCode {
+                hash: wasm_hash,
+                body_type: ContractEntryBodyType::DataEntry,
+            }),
+            self,
+        )
     }
 
     pub(crate) fn retrieve_wasm_from_storage(&self, wasm_hash: &Hash) -> Result<BytesM, HostError> {
@@ -86,7 +91,7 @@ impl Host {
             .data
         {
             LedgerEntryData::ContractCode(e) => match &e.body {
-                ContractCodeEntryBody::DataEntry(code) => code.metered_clone(self.as_budget()),
+                ContractCodeEntryBody::DataEntry(code) => code.metered_clone(self),
                 _ => Err(err!(
                     self,
                     (ScErrorType::Storage, ScErrorCode::InternalError),
@@ -127,7 +132,7 @@ impl Host {
             .map_err(|e| self.decorate_contract_instance_storage_error(e, &contract_id))?
         {
             let mut current = (*self.try_borrow_storage_mut()?.get(&key, self.as_budget())?)
-                .metered_clone(&self.0.budget)?;
+                .metered_clone(self)?;
 
             match current.data {
                 LedgerEntryData::ContractData(ref mut entry) => {
@@ -143,11 +148,11 @@ impl Host {
                 }
             }
             self.try_borrow_storage_mut()?
-                .put(&key, &Rc::new(current), self.as_budget())
+                .put(&key, &Rc::metered_new(current, self)?, self.as_budget())
                 .map_err(|e| self.decorate_contract_instance_storage_error(e, &contract_id))?;
         } else {
             let data = LedgerEntryData::ContractData(ContractDataEntry {
-                contract: ScAddress::Contract(contract_id.metered_clone(self.as_budget())?),
+                contract: ScAddress::Contract(contract_id.metered_clone(self)?),
                 key: ScVal::LedgerKeyContractInstance,
                 body,
                 durability: ContractDataDurability::Persistent,
@@ -155,7 +160,11 @@ impl Host {
                     .get_min_expiration_ledger(ContractDataDurability::Persistent)?,
             });
             self.try_borrow_storage_mut()?
-                .put(key, &Host::ledger_entry_from_data(data), self.as_budget())
+                .put(
+                    key,
+                    &Host::ledger_entry_from_data(self, data)?,
+                    self.as_budget(),
+                )
                 .map_err(|e| self.decorate_contract_instance_storage_error(e, &contract_id))?;
         }
         Ok(())
@@ -175,9 +184,9 @@ impl Host {
 
     // notes on metering: `get` from storage is covered. Rest are free.
     pub fn load_account(&self, account_id: AccountId) -> Result<AccountEntry, HostError> {
-        let acc = self.to_account_key(account_id);
+        let acc = self.to_account_key(account_id)?;
         self.with_mut_storage(|storage| match &storage.get(&acc, self.as_budget())?.data {
-            LedgerEntryData::Account(ae) => ae.metered_clone(self.as_budget()),
+            LedgerEntryData::Account(ae) => ae.metered_clone(self),
             e => Err(err!(
                 self,
                 (ScErrorType::Storage, ScErrorCode::InternalError),
@@ -187,8 +196,8 @@ impl Host {
         })
     }
 
-    pub(crate) fn to_account_key(&self, account_id: AccountId) -> Rc<LedgerKey> {
-        Rc::new(LedgerKey::Account(LedgerKeyAccount { account_id }))
+    pub(crate) fn to_account_key(&self, account_id: AccountId) -> Result<Rc<LedgerKey>, HostError> {
+        Rc::metered_new(LedgerKey::Account(LedgerKeyAccount { account_id }), self)
     }
 
     pub(crate) fn create_asset_4(&self, asset_code: [u8; 4], issuer: AccountId) -> TrustLineAsset {
@@ -215,11 +224,11 @@ impl Host {
         &self,
         account_id: AccountId,
         asset: TrustLineAsset,
-    ) -> Rc<LedgerKey> {
-        Rc::new(LedgerKey::Trustline(LedgerKeyTrustLine {
-            account_id,
-            asset,
-        }))
+    ) -> Result<Rc<LedgerKey>, HostError> {
+        Rc::metered_new(
+            LedgerKey::Trustline(LedgerKeyTrustLine { account_id, asset }),
+            self,
+        )
     }
 
     pub(crate) fn get_signer_weight_from_account(
@@ -229,7 +238,7 @@ impl Host {
     ) -> Result<u8, HostError> {
         if account.account_id
             == AccountId(PublicKey::PublicKeyTypeEd25519(
-                target_signer.metered_clone(&self.0.budget)?,
+                target_signer.metered_clone(self)?,
             ))
         {
             // Target signer is the master key, so return the master weight
@@ -263,14 +272,20 @@ impl Host {
         }
     }
 
-    pub(crate) fn ledger_entry_from_data(data: LedgerEntryData) -> Rc<LedgerEntry> {
-        Rc::new(LedgerEntry {
-            // This is modified to the appropriate value on the core side during
-            // commiting the ledger transaction.
-            last_modified_ledger_seq: 0,
-            data,
-            ext: LedgerEntryExt::V0,
-        })
+    pub(crate) fn ledger_entry_from_data(
+        &self,
+        data: LedgerEntryData,
+    ) -> Result<Rc<LedgerEntry>, HostError> {
+        Rc::metered_new(
+            LedgerEntry {
+                // This is modified to the appropriate value on the core side during
+                // commiting the ledger transaction.
+                last_modified_ledger_seq: 0,
+                data,
+                ext: LedgerEntryExt::V0,
+            },
+            self,
+        )
     }
 
     pub(crate) fn contract_id_from_scaddress(&self, address: ScAddress) -> Result<Hash, HostError> {
@@ -290,7 +305,7 @@ impl Host {
         address: AddressObject,
     ) -> Result<Hash, HostError> {
         self.visit_obj(address, |addr: &ScAddress| {
-            self.contract_id_from_scaddress(addr.metered_clone(self.budget_ref())?)
+            self.contract_id_from_scaddress(addr.metered_clone(self)?)
         })
     }
 }
diff --git a/soroban-env-host/src/host/declared_size.rs b/soroban-env-host/src/host/declared_size.rs
index 03491a8b..662b7608 100644
--- a/soroban-env-host/src/host/declared_size.rs
+++ b/soroban-env-host/src/host/declared_size.rs
@@ -6,9 +6,10 @@ use crate::{
     events::{EventError, HostEvent, InternalContractEvent, InternalEvent},
     host::{frame::Context, Events},
     host_object::HostObject,
+    native_contract::base_types::Address,
     storage::AccessType,
     xdr::{
-        AccountEntry, AccountId, BytesM, ClaimableBalanceEntry, ConfigSettingEntry,
+        AccountEntry, AccountId, Asset, BytesM, ClaimableBalanceEntry, ConfigSettingEntry,
         ContractCodeEntry, ContractDataDurability, ContractEntryBodyType, ContractEvent,
         ContractExecutable, ContractIdPreimage, CreateContractArgs, DataEntry, Duration,
         ExtensionPoint, Hash, LedgerEntry, LedgerEntryExt, LedgerKey, LedgerKeyAccount,
@@ -103,6 +104,7 @@ impl_declared_size_type!(I256, 32);
 impl_declared_size_type!(HostObject, 48);
 impl_declared_size_type!(HostError, 16);
 impl_declared_size_type!(Context, 512);
+impl_declared_size_type!(Address, 16);
 // xdr types
 impl_declared_size_type!(TimePoint, 8);
 impl_declared_size_type!(Duration, 8);
@@ -162,6 +164,7 @@ impl_declared_size_type!(AccountAuthorizationTracker, 232);
 impl_declared_size_type!(InvokerContractAuthorizationTracker, 192);
 impl_declared_size_type!(AccountAuthorizationTrackerSnapshot, 40);
 impl_declared_size_type!(ContractInvocation, 16);
+impl_declared_size_type!(Asset, 45);
 
 // composite types
 
@@ -299,6 +302,7 @@ mod test {
         expect!["512"].assert_eq(size_of::<Context>().to_string().as_str());
         #[cfg(target_arch = "aarch64")]
         expect!["488"].assert_eq(size_of::<Context>().to_string().as_str());
+        expect!["16"].assert_eq(size_of::<Address>().to_string().as_str());
         // xdr types
         expect!["8"].assert_eq(size_of::<TimePoint>().to_string().as_str());
         expect!["8"].assert_eq(size_of::<Duration>().to_string().as_str());
@@ -376,6 +380,7 @@ mod test {
                 .as_str(),
         );
         expect!["16"].assert_eq(size_of::<ContractInvocation>().to_string().as_str());
+        expect!["45"].assert_eq(size_of::<Asset>().to_string().as_str());
         // composite types
         expect!["16"].assert_eq(size_of::<&[ScVal]>().to_string().as_str());
         expect!["72"].assert_eq(size_of::<(Val, ScVal)>().to_string().as_str());
@@ -464,6 +469,7 @@ mod test {
         assert_mem_size_le_declared_size!(HostObject);
         assert_mem_size_le_declared_size!(HostError);
         assert_mem_size_le_declared_size!(Context);
+        assert_mem_size_le_declared_size!(Address);
         // xdr types
         assert_mem_size_le_declared_size!(TimePoint);
         assert_mem_size_le_declared_size!(Duration);
@@ -520,6 +526,7 @@ mod test {
         assert_mem_size_le_declared_size!(InvokerContractAuthorizationTracker);
         assert_mem_size_le_declared_size!(AccountAuthorizationTrackerSnapshot);
         assert_mem_size_le_declared_size!(ContractInvocation);
+        assert_mem_size_le_declared_size!(Asset);
         // composite types
         assert_mem_size_le_declared_size!(&[ScVal]);
         assert_mem_size_le_declared_size!((Val, ScVal));
diff --git a/soroban-env-host/src/host/frame.rs b/soroban-env-host/src/host/frame.rs
index 2717fba9..8ecafcc1 100644
--- a/soroban-env-host/src/host/frame.rs
+++ b/soroban-env-host/src/host/frame.rs
@@ -24,7 +24,7 @@ use crate::Vm;
 
 use super::{
     invoker_type::InvokerType,
-    metered_clone::{self, MeteredClone, MeteredContainer},
+    metered_clone::{MeteredClone, MeteredContainer, MeteredIterator},
     prng::Prng,
 };
 
@@ -136,7 +136,7 @@ impl Host {
         Vec::<Context>::charge_bulk_init_cpy(1, self.as_budget())?;
         self.try_borrow_context_mut()?.push(ctx);
         Ok(RollbackPoint {
-            storage: self.try_borrow_storage()?.map.clone(),
+            storage: self.try_borrow_storage()?.map.metered_clone(self)?,
             events: self.try_borrow_events()?.vec.len(),
             auth: auth_snapshot,
         })
@@ -363,11 +363,11 @@ impl Host {
     /// frame at its top.
     pub(crate) fn get_current_contract_id_opt_internal(&self) -> Result<Option<Hash>, HostError> {
         self.with_current_frame(|frame| match frame {
-            Frame::ContractVM { vm, .. } => Ok(Some(vm.contract_id.metered_clone(&self.0.budget)?)),
+            Frame::ContractVM { vm, .. } => Ok(Some(vm.contract_id.metered_clone(self)?)),
             Frame::HostFunction(_) => Ok(None),
-            Frame::Token(id, ..) => Ok(Some(id.metered_clone(&self.0.budget)?)),
+            Frame::Token(id, ..) => Ok(Some(id.metered_clone(self)?)),
             #[cfg(any(test, feature = "testutils"))]
-            Frame::TestContract(tc) => Ok(Some(tc.id.clone())),
+            Frame::TestContract(tc) => Ok(Some(tc.id.metered_clone(self)?)),
         })
     }
 
@@ -392,16 +392,16 @@ impl Host {
         // the previous frame must exist and must be a contract
         let hash = match frames.as_slice() {
             [.., c2, _] => match &c2.frame {
-                Frame::ContractVM { vm, .. } => Ok(vm.contract_id.metered_clone(&self.0.budget)?),
+                Frame::ContractVM { vm, .. } => Ok(vm.contract_id.metered_clone(self)?),
                 Frame::HostFunction(_) => Err(self.err(
                     ScErrorType::Context,
                     ScErrorCode::UnexpectedType,
                     "invoker is not a contract",
                     &[],
                 )),
-                Frame::Token(id, ..) => Ok(id.clone()),
+                Frame::Token(id, ..) => Ok(id.metered_clone(self)?),
                 #[cfg(any(test, feature = "testutils"))]
-                Frame::TestContract(tc) => Ok(tc.id.clone()), // no metering
+                Frame::TestContract(tc) => Ok(tc.id.metered_clone(self)?), // no metering
             },
             _ => Err(self.err(
                 ScErrorType::Context,
@@ -467,20 +467,18 @@ impl Host {
         let instance = self
             .retrieve_contract_instance_from_storage(&storage_key)
             .map_err(|e| self.decorate_contract_instance_storage_error(e, &id))?;
+        Vec::<Val>::charge_bulk_init_cpy(args.len() as u64, self.as_budget())?;
+        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.0.budget)?,
-                    code_entry.as_slice(),
-                )?;
+                let vm = Vm::new(self, id.metered_clone(self)?, code_entry.as_slice())?;
                 let relative_objects = Vec::new();
                 self.with_frame(
                     Frame::ContractVM {
-                        vm: vm.clone(),
+                        vm: Rc::clone(&vm),
                         fn_name: *func,
-                        args: args.to_vec(),
+                        args: args_vec,
                         instance,
                         relative_objects,
                     },
@@ -488,7 +486,7 @@ impl Host {
                 )
             }
             ContractExecutable::Token => self.with_frame(
-                Frame::Token(id.clone(), *func, args.to_vec(), instance),
+                Frame::Token(id.metered_clone(self)?, *func, args_vec, instance),
                 || {
                     use crate::native_contract::{NativeContract, Token};
                     Token.call(func, self, args)
@@ -703,11 +701,7 @@ impl Host {
                 self.with_frame(Frame::HostFunction(hf_type), || {
                     let deployer: Option<AddressObject> = match &args.contract_id_preimage {
                         ContractIdPreimage::Address(preimage_from_addr) => Some(
-                            self.add_host_object(
-                                preimage_from_addr
-                                    .address
-                                    .metered_clone(self.budget_ref())?,
-                            )?,
+                            self.add_host_object(preimage_from_addr.address.metered_clone(self)?)?,
                         ),
                         ContractIdPreimage::Asset(_) => None,
                     };
@@ -753,13 +747,9 @@ impl Host {
             storage_map.as_ref().map_or_else(
                 || Ok(vec![]),
                 |m| {
-                    metered_clone::charge_heap_alloc::<(Val, Val)>(
-                        m.len() as u64,
-                        self.budget_ref(),
-                    )?;
                     m.iter()
                         .map(|i| Ok((self.to_host_val(&i.key)?, self.to_host_val(&i.val)?)))
-                        .collect::<Result<Vec<(Val, Val)>, HostError>>()
+                        .metered_collect::<Result<Vec<(Val, Val)>, HostError>>(self)?
                 },
             )?,
             self,
@@ -775,7 +765,7 @@ impl Host {
                 }
                 let executable = match &ctx.frame {
                     Frame::ContractVM { instance, .. } => {
-                        instance.executable.metered_clone(self.budget_ref())?
+                        instance.executable.metered_clone(self)?
                     }
                     Frame::HostFunction(_) => {
                         return Err(self.err(
@@ -785,9 +775,7 @@ impl Host {
                             &[],
                         ))
                     }
-                    Frame::Token(_, _, _, instance) => {
-                        instance.executable.metered_clone(self.budget_ref())?
-                    }
+                    Frame::Token(_, _, _, instance) => instance.executable.metered_clone(self)?,
                     // Mock executable for test contract 'instances'. This is
                     // just a placeholder - it's not used for actually calling
                     // the test contracts.
diff --git a/soroban-env-host/src/host/mem_helper.rs b/soroban-env-host/src/host/mem_helper.rs
index fd7945de..c75ee7fd 100644
--- a/soroban-env-host/src/host/mem_helper.rs
+++ b/soroban-env-host/src/host/mem_helper.rs
@@ -19,7 +19,7 @@ impl Host {
         let len: u32 = len.into();
         self.with_current_frame(|frame| match frame {
             Frame::ContractVM { vm, .. } => {
-                let vm = vm.clone();
+                let vm = Rc::clone(&vm);
                 Ok(VmSlice { vm, pos, len })
             }
             _ => Err(self.err(
@@ -298,7 +298,7 @@ impl Host {
     ) -> Result<HOT::Wrapper, HostError> {
         let obj_pos: u32 = obj_pos.into();
         let mut obj_new: Vec<u8> = self
-            .visit_obj(obj, move |hv: &HOT| hv.metered_clone(&self.0.budget))?
+            .visit_obj(obj, move |hv: &HOT| hv.metered_clone(self))?
             .into();
         let obj_end = obj_pos
             .checked_add(len)
diff --git a/soroban-env-host/src/host/metered_clone.rs b/soroban-env-host/src/host/metered_clone.rs
index b54d4bf0..ae95b98f 100644
--- a/soroban-env-host/src/host/metered_clone.rs
+++ b/soroban-env-host/src/host/metered_clone.rs
@@ -1,24 +1,23 @@
 use std::{iter::FromIterator, mem, rc::Rc};
 
-use soroban_env_common::xdr::{DepthLimiter, SorobanAuthorizationEntry};
-
 use crate::{
-    budget::Budget,
+    budget::AsBudget,
     events::{EventError, HostEvent, InternalContractEvent, InternalEvent},
     host::Events,
     host_object::HostObject,
+    native_contract::base_types::Address,
     storage::AccessType,
     xdr::{
-        AccountEntry, AccountId, BytesM, ClaimableBalanceEntry, ConfigSettingEntry,
+        AccountEntry, AccountId, Asset, BytesM, ClaimableBalanceEntry, ConfigSettingEntry,
         ContractCodeEntry, ContractCodeEntryBody, ContractCostType, ContractDataEntryBody,
         ContractEvent, ContractEventBody, ContractEventType, ContractExecutable,
-        ContractIdPreimage, CreateContractArgs, DataEntry, Duration, Hash, LedgerEntry,
-        LedgerEntryData, LedgerEntryExt, LedgerKey, LedgerKeyAccount, LedgerKeyClaimableBalance,
-        LedgerKeyConfigSetting, LedgerKeyContractCode, LedgerKeyData, LedgerKeyLiquidityPool,
-        LedgerKeyOffer, LedgerKeyTrustLine, LiquidityPoolEntry, OfferEntry, PublicKey, ScAddress,
-        ScBytes, ScContractInstance, ScErrorCode, ScErrorType, ScMap, ScMapEntry, ScNonceKey,
-        ScString, ScSymbol, ScVal, ScVec, SorobanAuthorizedInvocation, StringM, TimePoint,
-        TrustLineAsset, TrustLineEntry, Uint256,
+        ContractIdPreimage, CreateContractArgs, DataEntry, DepthLimiter, Duration, Hash,
+        LedgerEntry, LedgerEntryData, LedgerEntryExt, LedgerKey, LedgerKeyAccount,
+        LedgerKeyClaimableBalance, LedgerKeyConfigSetting, LedgerKeyContractCode, LedgerKeyData,
+        LedgerKeyLiquidityPool, LedgerKeyOffer, LedgerKeyTrustLine, LiquidityPoolEntry, OfferEntry,
+        PublicKey, ScAddress, ScBytes, ScContractInstance, ScErrorCode, ScErrorType, ScMap,
+        ScMapEntry, ScNonceKey, ScString, ScSymbol, ScVal, ScVec, SorobanAuthorizationEntry,
+        SorobanAuthorizedInvocation, StringM, TimePoint, TrustLineAsset, TrustLineEntry, Uint256,
     },
     AddressObject, Bool, BytesObject, DurationObject, DurationSmall, DurationVal, Error, HostError,
     I128Object, I128Small, I128Val, I256Object, I256Small, I256Val, I32Val, I64Object, I64Small,
@@ -36,7 +35,7 @@ use super::declared_size::DeclaredSizeForMetering;
 // might cause metering to differ across compilations, causing problems in concensus and replay.
 pub(crate) fn charge_shallow_copy<T: DeclaredSizeForMetering>(
     n_elts: u64,
-    budget: &Budget,
+    budget: impl AsBudget,
 ) -> Result<(), HostError> {
     // Ideally we would want a static assertion. However, it does not work due to rust restrictions
     // (e.g. see https://github.com/rust-lang/rust/issues/57775). Here we make a runtime assertion
@@ -49,7 +48,7 @@ pub(crate) fn charge_shallow_copy<T: DeclaredSizeForMetering>(
         std::mem::size_of::<T>(),
         T::DECLARED_SIZE
     );
-    budget.charge(
+    budget.as_budget().charge(
         ContractCostType::HostMemCpy,
         Some(n_elts.saturating_mul(T::DECLARED_SIZE)),
     )
@@ -59,7 +58,7 @@ pub(crate) fn charge_shallow_copy<T: DeclaredSizeForMetering>(
 // not just metered clone, e.g. Box::<T>::new().
 pub(crate) fn charge_heap_alloc<T: DeclaredSizeForMetering>(
     n_elts: u64,
-    budget: &Budget,
+    budget: impl AsBudget,
 ) -> Result<(), HostError> {
     // Here we make a runtime assertion that the type's size is below its promised element size for
     // budget charging.
@@ -69,12 +68,29 @@ pub(crate) fn charge_heap_alloc<T: DeclaredSizeForMetering>(
         std::mem::size_of::<T>(),
         T::DECLARED_SIZE
     );
-    budget.charge(
+    budget.as_budget().charge(
         ContractCostType::HostMemAlloc,
         Some(n_elts.saturating_mul(T::DECLARED_SIZE)),
     )
 }
 
+pub trait MeteredAlloc<T: MeteredClone>: Sized {
+    fn metered_new(value: T, budget: impl AsBudget) -> Result<Self, HostError>;
+
+    fn metered_new_from_ref(value: &T, budget: impl AsBudget) -> Result<Self, HostError>;
+}
+
+impl<T: MeteredClone> MeteredAlloc<T> for Rc<T> {
+    fn metered_new(value: T, budget: impl AsBudget) -> Result<Self, HostError> {
+        charge_heap_alloc::<T>(1, budget)?;
+        Ok(Rc::new(value))
+    }
+
+    fn metered_new_from_ref(value: &T, budget: impl AsBudget) -> Result<Self, HostError> {
+        Self::metered_new(value.metered_clone(budget.clone())?, budget)
+    }
+}
+
 /// Represents a collection type which can be created from an iterator, and provides
 /// a method for bulk charging its creation cost.
 pub trait MeteredContainer: FromIterator<Self::Item> + DeclaredSizeForMetering
@@ -84,9 +100,9 @@ where
     type Item: DeclaredSizeForMetering;
     /// Charges the cost for bulk initializing the container and copying `len` elements
     /// into it.
-    fn charge_bulk_init_cpy(len: u64, budget: &Budget) -> Result<(), HostError> {
-        charge_shallow_copy::<Self>(1, budget)?;
-        charge_heap_alloc::<Self::Item>(len, budget)?;
+    fn charge_bulk_init_cpy(len: u64, budget: impl AsBudget) -> Result<(), HostError> {
+        charge_shallow_copy::<Self>(1, budget.clone())?;
+        charge_heap_alloc::<Self::Item>(len, budget.clone())?;
         charge_shallow_copy::<Self::Item>(len, budget)
     }
 }
@@ -114,7 +130,7 @@ where
 {
     fn metered_collect<B: MeteredContainer<Item = Self::Item>>(
         self,
-        budget: &Budget,
+        budget: impl AsBudget,
     ) -> Result<B, HostError> {
         if let (_, Some(ub)) = self.size_hint() {
             B::charge_bulk_init_cpy(ub as u64, budget)?;
@@ -140,33 +156,42 @@ pub trait MeteredClone: Clone + DeclaredSizeForMetering {
     // Called to clone the substructures of Self. The default implementation is a no-op and is
     // _only_ appropriate when Self is a shallow type (with no substructure). If you override
     // Self::IS_SHALLOW and set it to false, you should override this method also (it will actually
-    // panic if you don't). This charge does not include shallow copying of `Self` because that
+    // Err if you don't). This charge does not include shallow copying of `Self` because that
     // should be taken care of by the caller beforehand, e.g. in `metered_clone`.
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
-        assert!(Self::IS_SHALLOW);
-        Ok(())
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
+        if Self::IS_SHALLOW {
+            Ok(())
+        } else {
+            Err((ScErrorType::Object, ScErrorCode::InternalError).into())
+        }
     }
 
     // Called when cloning the substructures of a slice of Self. When Self::IS_SHALLOW, this is a
     // no-op. If Self::IS_SHALLOW is false, this method will do element-wise charging for
     // substructure, assuming it is paired with element-wise cloning.
-    fn bulk_charge_for_substructure(clones: &[Self], budget: &Budget) -> Result<(), HostError> {
+    fn bulk_charge_for_substructure(
+        clones: &[Self],
+        budget: impl AsBudget,
+    ) -> Result<(), HostError> {
         if Self::IS_SHALLOW {
             // If we're shallow, just return.
             Ok(())
         } else {
             for elt in clones {
-                elt.charge_for_substructure(budget)?;
+                elt.charge_for_substructure(budget.clone())?;
             }
             Ok(())
         }
     }
 
     // Convenience method for charging for a deep clone knowing the type is not shallow.
-    fn charge_deep_clone(&self, budget: &Budget) -> Result<(), HostError> {
-        debug_assert!(!Self::IS_SHALLOW);
-        charge_shallow_copy::<Self>(1, budget)?;
-        self.charge_for_substructure(budget)
+    fn charge_deep_clone(&self, budget: impl AsBudget) -> Result<(), HostError> {
+        if Self::IS_SHALLOW {
+            Err((ScErrorType::Object, ScErrorCode::InternalError).into())
+        } else {
+            charge_shallow_copy::<Self>(1, budget.clone())?;
+            self.charge_for_substructure(budget)
+        }
     }
 
     // Composite helper handles metering before clone. It first charges for the shallow footprint
@@ -175,8 +200,8 @@ pub trait MeteredClone: Clone + DeclaredSizeForMetering {
     // care of here. Typically overriding `IS_SHALLOW` and `charge_for_substructure` should
     // propertly take care of custom non-trivial cloning. Thus this method should not need to be
     // overriden.
-    fn metered_clone(&self, budget: &Budget) -> Result<Self, HostError> {
-        charge_shallow_copy::<Self>(1, budget)?;
+    fn metered_clone(&self, budget: impl AsBudget) -> Result<Self, HostError> {
+        charge_shallow_copy::<Self>(1, budget.clone())?;
         self.charge_for_substructure(budget)?;
         Ok(self.clone())
     }
@@ -236,6 +261,7 @@ impl MeteredClone for SymbolSmallIter {}
 impl MeteredClone for U256 {}
 impl MeteredClone for I256 {}
 impl MeteredClone for HostObject {}
+impl MeteredClone for Address {}
 // xdr types
 impl MeteredClone for TimePoint {}
 impl MeteredClone for Duration {}
@@ -271,20 +297,35 @@ impl MeteredClone for CreateContractArgs {}
 impl MeteredClone for ContractIdPreimage {}
 impl MeteredClone for SorobanAuthorizedInvocation {}
 impl MeteredClone for SorobanAuthorizationEntry {}
+impl MeteredClone for Asset {}
 // composite types
+// cloning Rc is just a ref-count bump
 impl<T> MeteredClone for Rc<T> {}
+// cloning a slice is just cloning the reference
 impl<T> MeteredClone for &[T] {}
+
 impl<K, V> MeteredClone for (K, V)
 where
     K: MeteredClone,
     V: MeteredClone,
 {
+    const IS_SHALLOW: bool = K::IS_SHALLOW && V::IS_SHALLOW;
+
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
+        if !K::IS_SHALLOW {
+            K::charge_for_substructure(&self.0, budget.clone())?;
+        }
+        if !V::IS_SHALLOW {
+            V::charge_for_substructure(&self.1, budget)?;
+        }
+        Ok(())
+    }
 }
 
 impl<C: MeteredClone, const N: usize> MeteredClone for [C; N] {
     const IS_SHALLOW: bool = C::IS_SHALLOW;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         C::bulk_charge_for_substructure(self.as_slice(), budget)
     }
 }
@@ -292,9 +333,10 @@ impl<C: MeteredClone, const N: usize> MeteredClone for [C; N] {
 impl MeteredClone for ScVal {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         // This is the depth limit checkpoint for `ScVal` cloning.
-        budget.clone().with_limited_depth(|_| {
+        let mut budget_clone = budget.as_budget().clone();
+        budget_clone.with_limited_depth(|_| {
             match self {
                 ScVal::Vec(Some(v)) => ScVec::charge_for_substructure(v, budget),
                 ScVal::Map(Some(m)) => ScMap::charge_for_substructure(m, budget),
@@ -332,7 +374,7 @@ impl MeteredClone for ScVal {
 impl MeteredClone for ScValObject {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         <Self as AsRef<ScVal>>::as_ref(self).charge_for_substructure(budget)
     }
 }
@@ -340,8 +382,8 @@ impl MeteredClone for ScValObject {
 impl MeteredClone for ScMapEntry {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
-        self.key.charge_for_substructure(budget)?;
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
+        self.key.charge_for_substructure(budget.clone())?;
         self.val.charge_for_substructure(budget)
     }
 }
@@ -349,7 +391,7 @@ impl MeteredClone for ScMapEntry {
 impl MeteredClone for ScVec {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         self.0.charge_for_substructure(budget) // self.0 will be deref'ed into Vec
     }
 }
@@ -357,7 +399,7 @@ impl MeteredClone for ScVec {
 impl MeteredClone for ScMap {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         self.0.charge_for_substructure(budget) // self.0 will be deref'ed into Vec
     }
 }
@@ -365,7 +407,7 @@ impl MeteredClone for ScMap {
 impl<const C: u32> MeteredClone for BytesM<C> {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         <Self as AsRef<Vec<u8>>>::as_ref(self).charge_for_substructure(budget)
     }
 }
@@ -373,7 +415,7 @@ impl<const C: u32> MeteredClone for BytesM<C> {
 impl<const C: u32> MeteredClone for StringM<C> {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         <Self as AsRef<Vec<u8>>>::as_ref(self).charge_for_substructure(budget)
     }
 }
@@ -381,7 +423,7 @@ impl<const C: u32> MeteredClone for StringM<C> {
 impl MeteredClone for ScBytes {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         <Self as AsRef<Vec<u8>>>::as_ref(self).charge_for_substructure(budget)
     }
 }
@@ -389,7 +431,7 @@ impl MeteredClone for ScBytes {
 impl MeteredClone for ScString {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         <Self as AsRef<Vec<u8>>>::as_ref(self).charge_for_substructure(budget)
     }
 }
@@ -397,7 +439,7 @@ impl MeteredClone for ScString {
 impl MeteredClone for ScSymbol {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         <Self as AsRef<Vec<u8>>>::as_ref(self).charge_for_substructure(budget)
     }
 }
@@ -405,11 +447,11 @@ impl MeteredClone for ScSymbol {
 impl<C: MeteredClone> MeteredClone for Vec<C> {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         // first take care of Vec clone, which involves allocating memory and shallowly cloning the
         // data type into it, then deep clone any substructure.
-        charge_heap_alloc::<C>(self.len() as u64, budget)?;
-        charge_shallow_copy::<C>(self.len() as u64, budget)?;
+        charge_heap_alloc::<C>(self.len() as u64, budget.clone())?;
+        charge_shallow_copy::<C>(self.len() as u64, budget.clone())?;
         C::bulk_charge_for_substructure(self.as_slice(), budget)?;
         Ok(())
     }
@@ -418,10 +460,10 @@ impl<C: MeteredClone> MeteredClone for Vec<C> {
 impl<C: MeteredClone> MeteredClone for Box<C> {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         // first take care of the Box clone: allocating memory and shallow cloning of data type.
-        charge_heap_alloc::<C>(1, budget)?;
-        charge_shallow_copy::<C>(1, budget)?;
+        charge_heap_alloc::<C>(1, budget.clone())?;
+        charge_shallow_copy::<C>(1, budget.clone())?;
         // then take care of any substructure clone (recursively)
         let inner: &C = self;
         inner.charge_for_substructure(budget)
@@ -431,7 +473,7 @@ impl<C: MeteredClone> MeteredClone for Box<C> {
 impl<C: MeteredClone> MeteredClone for Option<C> {
     const IS_SHALLOW: bool = C::IS_SHALLOW;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         match self {
             Some(elt) => elt.charge_for_substructure(budget),
             None => Ok(()),
@@ -442,14 +484,14 @@ impl<C: MeteredClone> MeteredClone for Option<C> {
 impl MeteredClone for ContractEvent {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         if let ContractEventType::Diagnostic = self.type_ {
             // Diagnostic events shouldn't be `metered_clone`d
             Err((ScErrorType::Events, ScErrorCode::InternalError).into())
         } else {
-            self.contract_id.charge_for_substructure(budget)?;
+            self.contract_id.charge_for_substructure(budget.clone())?;
             let ContractEventBody::V0(event) = &self.body;
-            event.topics.charge_for_substructure(budget)?;
+            event.topics.charge_for_substructure(budget.clone())?;
             event.data.charge_for_substructure(budget)
         }
     }
@@ -458,7 +500,7 @@ impl MeteredClone for ContractEvent {
 impl MeteredClone for HostEvent {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         self.event.charge_for_substructure(budget)
     }
 }
@@ -466,7 +508,7 @@ impl MeteredClone for HostEvent {
 impl MeteredClone for Events {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         self.0.charge_for_substructure(budget)
     }
 }
@@ -474,7 +516,7 @@ impl MeteredClone for Events {
 impl MeteredClone for InternalEvent {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         match self {
             InternalEvent::Contract(c) => c.charge_for_substructure(budget),
             InternalEvent::Diagnostic(_) => {
@@ -487,7 +529,7 @@ impl MeteredClone for InternalEvent {
 impl MeteredClone for ScContractInstance {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         self.storage.charge_for_substructure(budget)
     }
 }
@@ -495,7 +537,7 @@ impl MeteredClone for ScContractInstance {
 impl MeteredClone for LedgerKey {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         match self {
             LedgerKey::ContractData(d) => d.key.charge_for_substructure(budget),
             LedgerKey::Account(_)
@@ -513,12 +555,12 @@ impl MeteredClone for LedgerKey {
 impl MeteredClone for LedgerEntry {
     const IS_SHALLOW: bool = false;
 
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         use LedgerEntryData::*;
         match &self.data {
             ContractData(d) => {
                 if let ContractDataEntryBody::DataEntry(e) = &d.body {
-                    e.val.charge_for_substructure(budget)?;
+                    e.val.charge_for_substructure(budget.clone())?;
                 }
                 d.key.charge_for_substructure(budget)
             }
diff --git a/soroban-env-host/src/host/metered_map.rs b/soroban-env-host/src/host/metered_map.rs
index 0af96b97..5a0cd67b 100644
--- a/soroban-env-host/src/host/metered_map.rs
+++ b/soroban-env-host/src/host/metered_map.rs
@@ -4,10 +4,12 @@ use super::{declared_size::DeclaredSizeForMetering, MeteredClone};
 use crate::{
     budget::{AsBudget, Budget},
     xdr::ContractCostType,
-    Compare, Host, HostError,
+    Compare, Error, Host, HostError,
 };
 use std::{borrow::Borrow, cmp::Ordering, marker::PhantomData};
 
+const MAP_OOB: Error = Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds);
+
 pub struct MeteredOrdMap<K, V, Ctx> {
     pub(crate) map: Vec<(K, V)>,
     ctx: PhantomData<Ctx>,
@@ -85,7 +87,7 @@ where
 
     pub fn from_map(map: Vec<(K, V)>, ctx: &Ctx) -> Result<Self, HostError> {
         if u32::try_from(map.len()).is_err() {
-            return Err((ScErrorType::Object, ScErrorCode::ExceededLimit).into());
+            return Err(MAP_OOB.into());
         }
         // Allocation cost already paid for by caller, here just checks that input
         // has sorted and unique keys.
@@ -113,14 +115,18 @@ where
     ) -> Result<Self, HostError> {
         let _span = tracy_span!("new map");
         if let (_, Some(sz)) = iter.size_hint() {
-            // It's possible we temporarily go over-budget here before charging, but
-            // only by the cost of temporarily allocating twice the size of our largest
-            // possible object. In exchange we get to batch all charges associated with
-            // the clone into one (when A::IS_SHALLOW==true).
-            let map: Vec<(K, V)> = iter.collect();
-            map.charge_deep_clone(ctx.as_budget())?;
-            // Delegate to from_map here to recheck sort order.
-            Self::from_map(map, ctx)
+            if u32::try_from(sz).is_err() {
+                Err(MAP_OOB.into())
+            } else {
+                // It's possible we temporarily go over-budget here before charging, but
+                // only by the cost of temporarily allocating twice the size of our largest
+                // possible object. In exchange we get to batch all charges associated with
+                // the clone into one (when A::IS_SHALLOW==true).
+                let map: Vec<(K, V)> = iter.collect();
+                map.charge_deep_clone(ctx.as_budget())?;
+                // Delegate to from_map here to recheck sort order.
+                Self::from_map(map, ctx)
+            }
         } else {
             // This is a logic error, we should never get here.
             Err((ScErrorType::Object, ScErrorCode::InternalError).into())
@@ -163,7 +169,7 @@ where
                 // take(1) + new + skip(2)
                 // [0] + new + [2]
                 if replace_pos == usize::MAX - 1 {
-                    return Err((ScErrorType::Value, ScErrorCode::ArithDomain).into());
+                    return Err(MAP_OOB.into());
                 }
                 let init = self.map.iter().take(replace_pos).cloned();
                 let fini = self.map.iter().skip(replace_pos + 1).cloned();
@@ -174,10 +180,14 @@ where
                 // [0,1,2] insert_pos == 1
                 // take(1) + new + skip(1)
                 // [0] new [1, 2]
-                let init = self.map.iter().take(insert_pos).cloned();
-                let fini = self.map.iter().skip(insert_pos).cloned();
-                let iter = init.chain([(key, value)].into_iter()).chain(fini);
-                Self::from_exact_iter(iter, ctx)
+                if self.len() == u32::MAX as usize {
+                    return Err(MAP_OOB.into());
+                } else {
+                    let init = self.map.iter().take(insert_pos).cloned();
+                    let fini = self.map.iter().skip(insert_pos).cloned();
+                    let iter = init.chain([(key, value)].into_iter()).chain(fini);
+                    Self::from_exact_iter(iter, ctx)
+                }
             }
         }
     }
@@ -196,6 +206,13 @@ where
         }
     }
 
+    pub fn get_at_index(&self, index: usize, ctx: &Ctx) -> Result<&(K, V), HostError> {
+        self.charge_access(1, ctx)?;
+        self.map.get(index).ok_or_else(|| {
+            Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds).into()
+        })
+    }
+
     /// Returns a `Some((new_self, val))` pair where `new_self` no longer
     /// contains an entry for `key`, if the key existed, otherwise `None` if
     /// `key` didn't exist (in which case there's no need to clone).
@@ -323,7 +340,7 @@ where
     V: MeteredClone,
     Ctx: AsBudget,
 {
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         self.map.charge_for_substructure(budget)
     }
 }
diff --git a/soroban-env-host/src/host/metered_vector.rs b/soroban-env-host/src/host/metered_vector.rs
index d0ea2fba..edb95de8 100644
--- a/soroban-env-host/src/host/metered_vector.rs
+++ b/soroban-env-host/src/host/metered_vector.rs
@@ -64,18 +64,23 @@ where
     }
 
     pub fn from_array(buf: &[A], budget: &Budget) -> Result<Self, HostError> {
-        // we may temporarily go over budget here.
-        let vec: Vec<A> = buf.into();
-        vec.charge_deep_clone(budget)?;
-        Self::from_vec(vec)
+        if u32::try_from(buf.len()).is_err() {
+            Err(VEC_OOB.into())
+        } else {
+            // we may temporarily go over budget here.
+            let vec: Vec<A> = buf.into();
+            vec.charge_deep_clone(budget)?;
+            Self::from_vec(vec)
+        }
     }
 
     // No meter charge, assuming allocation cost has been covered by the caller from the outside.
     pub fn from_vec(vec: Vec<A>) -> Result<Self, HostError> {
         if u32::try_from(vec.len()).is_err() {
-            return Err((ScErrorType::Object, ScErrorCode::ExceededLimit).into());
+            Err(VEC_OOB.into())
+        } else {
+            Ok(Self { vec })
         }
-        Ok(Self { vec })
     }
 
     pub fn as_slice(&self) -> &[A] {
@@ -96,13 +101,17 @@ where
     ) -> Result<Self, HostError> {
         let _span = tracy_span!("new vec");
         if let (_, Some(sz)) = iter.size_hint() {
-            // It's possible we temporarily go over-budget here before charging, but
-            // only by the cost of temporarily allocating twice the size of our largest
-            // possible object. In exchange we get to batch all charges associated with
-            // the clone into one (when A::IS_SHALLOW==true).
-            let vec: Vec<A> = iter.collect();
-            vec.charge_deep_clone(budget)?;
-            Self::from_vec(vec)
+            if u32::try_from(sz).is_err() {
+                Err(VEC_OOB.into())
+            } else {
+                // It's possible we temporarily go over-budget here before charging, but
+                // only by the cost of temporarily allocating twice the size of our largest
+                // possible object. In exchange we get to batch all charges associated with
+                // the clone into one (when A::IS_SHALLOW==true).
+                let vec: Vec<A> = iter.collect();
+                vec.charge_deep_clone(budget)?;
+                Self::from_vec(vec)
+            }
         } else {
             // This is a logic error, we should never get here.
             Err((ScErrorType::Object, ScErrorCode::InternalError).into())
@@ -127,8 +136,12 @@ where
     }
 
     pub fn push_front(&self, value: A, budget: &Budget) -> Result<Self, HostError> {
-        let iter = [value].into_iter().chain(self.vec.iter().cloned());
-        Self::from_exact_iter(iter, budget)
+        if self.len() == u32::MAX as usize {
+            Err(VEC_OOB.into())
+        } else {
+            let iter = [value].into_iter().chain(self.vec.iter().cloned());
+            Self::from_exact_iter(iter, budget)
+        }
     }
 
     pub fn pop_front(&self, budget: &Budget) -> Result<Self, HostError> {
@@ -141,27 +154,36 @@ where
     }
 
     pub fn push_back(&self, value: A, budget: &Budget) -> Result<Self, HostError> {
-        let iter = self.vec.iter().cloned().chain([value].into_iter());
-        Self::from_exact_iter(iter, budget)
+        if self.len() == u32::MAX as usize {
+            Err(VEC_OOB.into())
+        } else {
+            let iter = self.vec.iter().cloned().chain([value].into_iter());
+            Self::from_exact_iter(iter, budget)
+        }
     }
 
-    fn err_overflow() -> HostError {
-        (ScErrorType::Value, ScErrorCode::ArithDomain).into()
+    fn err_oob() -> HostError {
+        VEC_OOB.into()
     }
 
-    fn add_or_overflow(x: usize, y: usize) -> Result<usize, HostError> {
-        x.checked_add(y).ok_or_else(|| Self::err_overflow())
+    fn add_or_err(x: usize, y: usize) -> Result<usize, HostError> {
+        let sz = x.checked_add(y).ok_or_else(|| Self::err_oob())?;
+        if u32::try_from(sz).is_err() {
+            Err(Self::err_oob())
+        } else {
+            Ok(sz)
+        }
     }
 
-    fn sub_or_overflow(x: usize, y: usize) -> Result<usize, HostError> {
-        x.checked_sub(y).ok_or_else(|| Self::err_overflow())
+    fn sub_or_err(x: usize, y: usize) -> Result<usize, HostError> {
+        x.checked_sub(y).ok_or_else(|| Self::err_oob())
     }
 
     pub fn pop_back(&self, budget: &Budget) -> Result<Self, HostError> {
         if self.vec.is_empty() {
             Err(VEC_OOB.into())
         } else {
-            let count = Self::sub_or_overflow(self.vec.len(), 1)?;
+            let count = Self::sub_or_err(self.vec.len(), 1)?;
             let iter = self.vec.iter().take(count).cloned();
             Self::from_exact_iter(iter, budget)
         }
@@ -173,7 +195,7 @@ where
         } else {
             // [0, 1, 2]
             // del 1 => take(1) + skip(2)
-            let skip = Self::add_or_overflow(idx, 1)?;
+            let skip = Self::add_or_err(idx, 1)?;
             let init = self.vec.iter().take(idx).cloned();
             let fini = self.vec.iter().skip(skip).cloned();
             Self::from_exact_iter(init.chain(fini), budget)
@@ -191,6 +213,9 @@ where
     }
 
     pub fn insert(&self, index: usize, value: A, budget: &Budget) -> Result<Self, HostError> {
+        if self.vec.len() == u32::MAX as usize {
+            return Err(VEC_OOB.into());
+        }
         let len = self.vec.len();
         if index > len {
             Err(VEC_OOB.into())
@@ -207,6 +232,7 @@ where
     }
 
     pub fn append(&self, other: &Self, budget: &Budget) -> Result<Self, HostError> {
+        Self::add_or_err(self.len(), other.len())?;
         let iter = self.vec.iter().cloned().chain(other.vec.iter().cloned());
         Self::from_exact_iter(iter, budget)
     }
@@ -223,9 +249,8 @@ where
         F: Fn(&A) -> Result<Ordering, HostError>,
     {
         self.charge_scan(budget)?;
-        let iter = self.vec.iter();
         // this is similar logic to `iter.position(f)` but is fallible
-        for (i, val) in iter.enumerate() {
+        for (i, val) in self.vec.iter().enumerate() {
             if f(val)? == Ordering::Equal {
                 return Ok(Some(i));
             }
@@ -238,11 +263,8 @@ where
         F: Fn(&A) -> Result<Ordering, HostError>,
     {
         self.charge_scan(budget)?;
-        let mut i = self.vec.len();
-        let mut iter = self.vec.iter();
         // this is similar logic to `iter.rposition(f)` but is fallible
-        while let Some(val) = iter.next_back() {
-            i -= 1;
+        for (i, val) in self.vec.iter().enumerate().rev() {
             if f(val)? == Ordering::Equal {
                 return Ok(Some(i));
             }
@@ -286,13 +308,13 @@ where
     {
         // The closure evaluation is not metered here, it is assumed to be taken care of outside.
         // Here just covers the cost of cloning a Vec.
+        self.charge_deep_clone(budget)?;
         let mut vec = Vec::with_capacity(self.len());
         for (i, v) in self.vec.iter_mut().enumerate() {
             if f(i, v)? {
                 vec.push(v.clone());
             }
         }
-        vec.charge_deep_clone(budget)?;
         Self::from_vec(vec)
     }
 
@@ -320,7 +342,7 @@ impl<A> MeteredClone for MeteredVector<A>
 where
     A: MeteredClone,
 {
-    fn charge_for_substructure(&self, budget: &Budget) -> Result<(), HostError> {
+    fn charge_for_substructure(&self, budget: impl AsBudget) -> Result<(), HostError> {
         self.vec.charge_for_substructure(budget)
     }
 }
diff --git a/soroban-env-host/src/host/num.rs b/soroban-env-host/src/host/num.rs
index d483805d..a90b451a 100644
--- a/soroban-env-host/src/host/num.rs
+++ b/soroban-env-host/src/host/num.rs
@@ -19,7 +19,7 @@ macro_rules! impl_wrapping_obj_to_num {
             _vmcaller: &mut VmCaller<Host>,
             obj: <$data as HostObjectType>::Wrapper,
         ) -> Result<$num, HostError> {
-            self.visit_obj(obj, |t: &$data| Ok(t.clone().into()))
+            self.visit_obj(obj, |t: &$data| Ok(t.metered_clone(self)?.into()))
         }
     };
 }
diff --git a/soroban-env-host/src/host_object.rs b/soroban-env-host/src/host_object.rs
index 3ec46ce7..77a40f3f 100644
--- a/soroban-env-host/src/host_object.rs
+++ b/soroban-env-host/src/host_object.rs
@@ -7,7 +7,7 @@ use soroban_env_common::{
 };
 
 use crate::{
-    budget::{AsBudget, Budget},
+    budget::Budget,
     host::metered_clone::{self, MeteredClone},
     HostError,
 };
@@ -293,7 +293,7 @@ impl Host {
                 ))
             } else {
                 // Push a new entry into the relative-objects vector.
-                metered_clone::charge_heap_alloc::<Object>(1, self.as_budget())?;
+                metered_clone::charge_heap_alloc::<Object>(1, self)?;
                 let index = self.with_current_frame_relative_object_table(|table| {
                     let index = table.len();
                     table.push(obj);
@@ -318,7 +318,7 @@ impl Host {
         let handle = index_to_handle(self, index, false)?;
         // charge for the new host object, which is just the amortized cost of a single
         // `HostObject` allocation
-        metered_clone::charge_heap_alloc::<HostObject>(1, self.as_budget())?;
+        metered_clone::charge_heap_alloc::<HostObject>(1, self)?;
         self.try_borrow_objects_mut()?.push(HOT::inject(hot));
         Ok(HOT::new_from_handle(handle))
     }
diff --git a/soroban-env-host/src/native_contract/base_types.rs b/soroban-env-host/src/native_contract/base_types.rs
index e711e9dd..12d96bde 100644
--- a/soroban-env-host/src/native_contract/base_types.rs
+++ b/soroban-env-host/src/native_contract/base_types.rs
@@ -427,7 +427,7 @@ impl TryFromVal<Host, &std::vec::Vec<Val>> for Vec {
 
 impl Vec {
     pub fn new(env: &Host) -> Result<Self, HostError> {
-        let vec = env.vec_new(().into())?;
+        let vec = env.vec_new()?;
         Ok(Self {
             host: env.clone(),
             object: vec,
diff --git a/soroban-env-host/src/native_contract/invoker_contract_auth.rs b/soroban-env-host/src/native_contract/invoker_contract_auth.rs
index 4a7d5fdf..f00b40e9 100644
--- a/soroban-env-host/src/native_contract/invoker_contract_auth.rs
+++ b/soroban-env-host/src/native_contract/invoker_contract_auth.rs
@@ -86,7 +86,7 @@ impl InvokerContractAuthEntry {
                 let function = AuthorizedFunction::CreateContractHostFn(CreateContractArgs {
                     contract_id_preimage: ContractIdPreimage::Address(
                         ContractIdPreimageFromAddress {
-                            address: invoker_contract_addr.metered_clone(host.budget_ref())?,
+                            address: invoker_contract_addr.metered_clone(host)?,
                             salt: host.u256_from_bytesobj_input(
                                 "salt",
                                 create_contract_fn.salt.as_object(),
diff --git a/soroban-env-host/src/native_contract/token/allowance.rs b/soroban-env-host/src/native_contract/token/allowance.rs
index e5b08f56..bfbdd72b 100644
--- a/soroban-env-host/src/native_contract/token/allowance.rs
+++ b/soroban-env-host/src/native_contract/token/allowance.rs
@@ -1,4 +1,4 @@
-use crate::host::Host;
+use crate::host::{metered_clone::MeteredClone, Host};
 use crate::native_contract::base_types::Address;
 use crate::native_contract::contract_error::ContractError;
 use crate::native_contract::storage_utils::StorageUtils;
@@ -115,8 +115,8 @@ fn write_allowance_amount(
     amount: i128,
 ) -> Result<(), HostError> {
     let key = DataKey::Allowance(AllowanceDataKey {
-        from: from.clone(),
-        spender: spender.clone(),
+        from: from.metered_clone(e)?,
+        spender: spender.metered_clone(e)?,
     });
 
     let allowance: AllowanceValue = e
@@ -132,7 +132,7 @@ pub fn spend_allowance(
     spender: Address,
     amount: i128,
 ) -> Result<(), HostError> {
-    let allowance = read_allowance(e, from.clone(), spender.clone())?;
+    let allowance = read_allowance(e, from.metered_clone(e)?, spender.metered_clone(e)?)?;
     if allowance < amount {
         return Err(err!(
             e,
diff --git a/soroban-env-host/src/native_contract/token/balance.rs b/soroban-env-host/src/native_contract/token/balance.rs
index 5b5702f0..70857cc2 100644
--- a/soroban-env-host/src/native_contract/token/balance.rs
+++ b/soroban-env-host/src/native_contract/token/balance.rs
@@ -1,4 +1,5 @@
 use crate::budget::AsBudget;
+use crate::host::metered_clone::MeteredClone;
 use crate::host::Host;
 use crate::native_contract::base_types::{Address, BytesN};
 use crate::native_contract::contract_error::ContractError;
@@ -75,7 +76,7 @@ fn write_balance(e: &Host, addr: Address, balance: BalanceValue) -> Result<(), H
 
 // Metering: covered by components.
 pub fn receive_balance(e: &Host, addr: Address, amount: i128) -> Result<(), HostError> {
-    if !is_authorized(e, addr.clone())? {
+    if !is_authorized(e, addr.metered_clone(e)?)? {
         return Err(e.error(
             ContractError::BalanceDeauthorizedError.into(),
             "balance is deauthorized",
@@ -95,7 +96,7 @@ pub fn receive_balance(e: &Host, addr: Address, amount: i128) -> Result<(), Host
             Ok(transfer_classic_balance(e, acc_id, i64_amount)?)
         }
         ScAddress::Contract(_) => {
-            let key = DataKey::Balance(addr.clone());
+            let key = DataKey::Balance(addr.metered_clone(e)?);
             let mut balance = if let Some(raw_balance) =
                 StorageUtils::try_get(e, key.try_into_val(e)?, StorageType::Persistent)?
             {
@@ -143,7 +144,7 @@ pub fn spend_balance_no_authorization_check(
         ScAddress::Contract(_) => {
             // If a balance exists, calculate new amount and write the existing authorized state as is because
             // this can be used to clawback when deauthorized.
-            let key = DataKey::Balance(addr.clone());
+            let key = DataKey::Balance(addr.metered_clone(e)?);
             if let Some(raw_balance) =
                 StorageUtils::try_get(e, key.try_into_val(e)?, StorageType::Persistent)?
             {
@@ -183,7 +184,7 @@ pub fn spend_balance_no_authorization_check(
 
 // Metering: covered by components.
 pub fn spend_balance(e: &Host, addr: Address, amount: i128) -> Result<(), HostError> {
-    if !is_authorized(e, addr.clone())? {
+    if !is_authorized(e, addr.metered_clone(e)?)? {
         return Err(e.error(
             ContractError::BalanceDeauthorizedError.into(),
             "balance is deauthorized",
@@ -225,7 +226,7 @@ pub fn write_authorization(e: &Host, addr: Address, authorize: bool) -> Result<(
     match addr.to_sc_address()? {
         ScAddress::Account(acc_id) => set_authorization(e, acc_id, authorize),
         ScAddress::Contract(_) => {
-            let key = DataKey::Balance(addr.clone());
+            let key = DataKey::Balance(addr.metered_clone(e)?);
             if let Some(raw_balance) =
                 StorageUtils::try_get(e, key.try_into_val(e)?, StorageType::Persistent)?
             {
@@ -278,7 +279,10 @@ pub fn check_clawbackable(e: &Host, addr: Address) -> Result<(), HostError> {
             AssetInfo::AlphaNum4(asset) => {
                 let issuer_account_id = e.account_id_from_bytesobj(asset.issuer.into())?;
                 validate_trustline(
-                    e.create_asset_4(asset.asset_code.to_array()?, issuer_account_id.clone()),
+                    e.create_asset_4(
+                        asset.asset_code.to_array()?,
+                        issuer_account_id.metered_clone(e)?,
+                    ),
                     issuer_account_id,
                     acc_id,
                 )
@@ -286,7 +290,10 @@ pub fn check_clawbackable(e: &Host, addr: Address) -> Result<(), HostError> {
             AssetInfo::AlphaNum12(asset) => {
                 let issuer_account_id = e.account_id_from_bytesobj(asset.issuer.into())?;
                 validate_trustline(
-                    e.create_asset_12(asset.asset_code.to_array()?, issuer_account_id.clone()),
+                    e.create_asset_12(
+                        asset.asset_code.to_array()?,
+                        issuer_account_id.metered_clone(e)?,
+                    ),
                     issuer_account_id,
                     acc_id,
                 )
@@ -337,7 +344,10 @@ pub fn transfer_classic_balance(e: &Host, to_key: AccountId, amount: i64) -> Res
         AssetInfo::AlphaNum4(asset) => {
             let issuer_account_id = e.account_id_from_bytesobj(asset.issuer.into())?;
             transfer_trustline_balance_safe(
-                e.create_asset_4(asset.asset_code.to_array()?, issuer_account_id.clone()),
+                e.create_asset_4(
+                    asset.asset_code.to_array()?,
+                    issuer_account_id.metered_clone(e)?,
+                ),
                 issuer_account_id,
                 to_key,
             )?
@@ -345,7 +355,10 @@ pub fn transfer_classic_balance(e: &Host, to_key: AccountId, amount: i64) -> Res
         AssetInfo::AlphaNum12(asset) => {
             let issuer_account_id = e.account_id_from_bytesobj(asset.issuer.into())?;
             transfer_trustline_balance_safe(
-                e.create_asset_12(asset.asset_code.to_array()?, issuer_account_id.clone()),
+                e.create_asset_12(
+                    asset.asset_code.to_array()?,
+                    issuer_account_id.metered_clone(e)?,
+                ),
                 issuer_account_id,
                 to_key,
             )?
@@ -373,7 +386,10 @@ fn get_classic_balance(e: &Host, to_key: AccountId) -> Result<(i64, i64), HostEr
         AssetInfo::AlphaNum4(asset) => {
             let issuer_account_id = e.account_id_from_bytesobj(asset.issuer.into())?;
             get_trustline_balance_safe(
-                e.create_asset_4(asset.asset_code.to_array()?, issuer_account_id.clone()),
+                e.create_asset_4(
+                    asset.asset_code.to_array()?,
+                    issuer_account_id.metered_clone(e)?,
+                ),
                 issuer_account_id,
                 to_key,
             )
@@ -382,7 +398,10 @@ fn get_classic_balance(e: &Host, to_key: AccountId) -> Result<(i64, i64), HostEr
         AssetInfo::AlphaNum12(asset) => {
             let issuer_account_id = e.account_id_from_bytesobj(asset.issuer.into())?;
             get_trustline_balance_safe(
-                e.create_asset_12(asset.asset_code.to_array()?, issuer_account_id.clone()),
+                e.create_asset_12(
+                    asset.asset_code.to_array()?,
+                    issuer_account_id.metered_clone(e)?,
+                ),
                 issuer_account_id,
                 to_key,
             )
@@ -392,7 +411,7 @@ fn get_classic_balance(e: &Host, to_key: AccountId) -> Result<(i64, i64), HostEr
 
 // Metering: *mostly* covered by components. The arithmetics are free.
 fn transfer_account_balance(e: &Host, account_id: AccountId, amount: i64) -> Result<(), HostError> {
-    let lk = e.to_account_key(account_id);
+    let lk = e.to_account_key(account_id)?;
 
     e.with_mut_storage(|storage| {
         let mut le = storage
@@ -400,7 +419,7 @@ fn transfer_account_balance(e: &Host, account_id: AccountId, amount: i64) -> Res
             .map_err(|_| e.error(ContractError::AccountMissingError.into(), "account missing", &[]))?;
 
         let mut ae = match &le.data {
-            LedgerEntryData::Account(ae) => Ok(ae.clone()),
+            LedgerEntryData::Account(ae) => Ok(ae.metered_clone(e)?),
             _ => Err(e.error(ContractError::InternalError.into(), "unexpected entry found", &[])),
         }?;
 
@@ -411,7 +430,7 @@ fn transfer_account_balance(e: &Host, account_id: AccountId, amount: i64) -> Res
         };
         if new_balance >= min_balance && new_balance <= max_balance {
             ae.balance = new_balance;
-            le = Host::ledger_entry_from_data(LedgerEntryData::Account(ae));
+            le = Host::ledger_entry_from_data(e, LedgerEntryData::Account(ae))?;
             storage.put(&lk, &le, e.as_budget())
         } else {
             Err(err!(
@@ -433,14 +452,14 @@ fn transfer_trustline_balance(
     asset: TrustLineAsset,
     amount: i64,
 ) -> Result<(), HostError> {
-    let lk = e.to_trustline_key(account_id, asset);
+    let lk = e.to_trustline_key(account_id, asset)?;
     e.with_mut_storage(|storage| {
         let mut le = storage.get(&lk, e.as_budget()).map_err(|_| {
             e.error(ContractError::TrustlineMissingError.into(), "trustline missing", &[])
         })?;
 
         let mut tl = match &le.data {
-            LedgerEntryData::Trustline(tl) => Ok(tl.clone()),
+            LedgerEntryData::Trustline(tl) => Ok(tl.metered_clone(e)?),
             _ => Err(e.error(ContractError::InternalError.into(), "unexpected entry found", &[])),
         }?;
 
@@ -451,7 +470,7 @@ fn transfer_trustline_balance(
         };
         if new_balance >= min_balance && new_balance <= max_balance {
             tl.balance = new_balance;
-            le = Host::ledger_entry_from_data(LedgerEntryData::Trustline(tl));
+            le = Host::ledger_entry_from_data(e, LedgerEntryData::Trustline(tl))?;
             storage.put(&lk, &le, e.as_budget())
         } else {
             Err(err!(
@@ -469,7 +488,7 @@ fn transfer_trustline_balance(
 // TODO: Metering analysis
 //returns (total balance, spendable balance)
 fn get_account_balance(e: &Host, account_id: AccountId) -> Result<(i64, i64), HostError> {
-    let lk = e.to_account_key(account_id);
+    let lk = e.to_account_key(account_id)?;
 
     e.with_mut_storage(|storage| {
         let le = storage.get(&lk, e.as_budget()).map_err(|_| {
@@ -537,7 +556,7 @@ fn get_trustline_balance(
     account_id: AccountId,
     asset: TrustLineAsset,
 ) -> Result<(i64, i64), HostError> {
-    let lk = e.to_trustline_key(account_id, asset);
+    let lk = e.to_trustline_key(account_id, asset)?;
     e.with_mut_storage(|storage| {
         let le = storage.get(&lk, e.as_budget()).map_err(|_| {
             e.error(
@@ -548,7 +567,7 @@ fn get_trustline_balance(
         })?;
 
         let tl = match &le.data {
-            LedgerEntryData::Trustline(tl) => Ok(tl.clone()),
+            LedgerEntryData::Trustline(tl) => Ok(tl.metered_clone(e)?),
             _ => Err(e.error(
                 ContractError::InternalError.into(),
                 "unexpected entry found",
@@ -611,7 +630,10 @@ fn is_account_authorized(e: &Host, account_id: AccountId) -> Result<bool, HostEr
         AssetInfo::AlphaNum4(asset) => {
             let issuer_account_id = e.account_id_from_bytesobj(asset.issuer.into())?;
             is_trustline_authorized_safe(
-                e.create_asset_4(asset.asset_code.to_array()?, issuer_account_id.clone()),
+                e.create_asset_4(
+                    asset.asset_code.to_array()?,
+                    issuer_account_id.metered_clone(e)?,
+                ),
                 issuer_account_id,
                 account_id,
             )
@@ -619,7 +641,10 @@ fn is_account_authorized(e: &Host, account_id: AccountId) -> Result<bool, HostEr
         AssetInfo::AlphaNum12(asset) => {
             let issuer_account_id = e.account_id_from_bytesobj(asset.issuer.into())?;
             is_trustline_authorized_safe(
-                e.create_asset_12(asset.asset_code.to_array()?, issuer_account_id.clone()),
+                e.create_asset_12(
+                    asset.asset_code.to_array()?,
+                    issuer_account_id.metered_clone(e)?,
+                ),
                 issuer_account_id,
                 account_id,
             )
@@ -633,7 +658,7 @@ fn get_trustline_flags(
     account_id: AccountId,
     asset: TrustLineAsset,
 ) -> Result<u32, HostError> {
-    let lk = e.to_trustline_key(account_id, asset);
+    let lk = e.to_trustline_key(account_id, asset)?;
     e.with_mut_storage(|storage| {
         let le = storage.get(&lk, e.as_budget()).map_err(|_| {
             e.error(
@@ -689,7 +714,10 @@ fn set_authorization(e: &Host, to_key: AccountId, authorize: bool) -> Result<(),
         AssetInfo::AlphaNum4(asset) => {
             let issuer_account_id = e.account_id_from_bytesobj(asset.issuer.into())?;
             set_trustline_authorization_safe(
-                e.create_asset_4(asset.asset_code.to_array()?, issuer_account_id.clone()),
+                e.create_asset_4(
+                    asset.asset_code.to_array()?,
+                    issuer_account_id.metered_clone(e)?,
+                ),
                 issuer_account_id,
                 to_key,
             )
@@ -697,7 +725,10 @@ fn set_authorization(e: &Host, to_key: AccountId, authorize: bool) -> Result<(),
         AssetInfo::AlphaNum12(asset) => {
             let issuer_account_id = e.account_id_from_bytesobj(asset.issuer.into())?;
             set_trustline_authorization_safe(
-                e.create_asset_12(asset.asset_code.to_array()?, issuer_account_id.clone()),
+                e.create_asset_12(
+                    asset.asset_code.to_array()?,
+                    issuer_account_id.metered_clone(e)?,
+                ),
                 issuer_account_id,
                 to_key,
             )
@@ -712,7 +743,7 @@ fn set_trustline_authorization(
     asset: TrustLineAsset,
     authorize: bool,
 ) -> Result<(), HostError> {
-    let lk = e.to_trustline_key(account_id, asset);
+    let lk = e.to_trustline_key(account_id, asset)?;
     e.with_mut_storage(|storage| {
         let mut le = storage.get(&lk, e.as_budget()).map_err(|_| {
             e.error(
@@ -723,7 +754,7 @@ fn set_trustline_authorization(
         })?;
 
         let mut tl = match &le.data {
-            LedgerEntryData::Trustline(tl) => Ok(tl.clone()),
+            LedgerEntryData::Trustline(tl) => Ok(tl.metered_clone(e)?),
             _ => Err(e.error(
                 ContractError::InternalError.into(),
                 "unexpected entry found",
@@ -740,7 +771,7 @@ fn set_trustline_authorization(
             tl.flags &= !(TrustLineFlags::AuthorizedFlag as u32);
             tl.flags |= TrustLineFlags::AuthorizedToMaintainLiabilitiesFlag as u32;
         }
-        le = Host::ledger_entry_from_data(LedgerEntryData::Trustline(tl));
+        le = Host::ledger_entry_from_data(e, LedgerEntryData::Trustline(tl))?;
         storage.put(&lk, &le, e.as_budget())
     })
 }
diff --git a/soroban-env-host/src/native_contract/token/contract.rs b/soroban-env-host/src/native_contract/token/contract.rs
index e0ee005b..a54270a8 100644
--- a/soroban-env-host/src/native_contract/token/contract.rs
+++ b/soroban-env-host/src/native_contract/token/contract.rs
@@ -1,4 +1,4 @@
-use crate::host::Host;
+use crate::host::{metered_clone::MeteredClone, Host};
 use crate::native_contract::base_types::{Address, Bytes, BytesN, String};
 use crate::native_contract::contract_error::ContractError;
 use crate::native_contract::token::allowance::{read_allowance, spend_allowance, write_allowance};
@@ -123,7 +123,7 @@ impl TokenTrait for Token {
         let asset: Asset = e.metered_from_xdr_obj(asset_bytes.into())?;
 
         let curr_contract_id = e.get_current_contract_id_internal()?;
-        let expected_contract_id = e.get_asset_contract_id_hash(asset.clone())?;
+        let expected_contract_id = e.get_asset_contract_id_hash(asset.metered_clone(e)?)?;
         if curr_contract_id != expected_contract_id {
             return Err(e.error(
                 ContractError::InternalError.into(),
@@ -201,7 +201,13 @@ impl TokenTrait for Token {
 
         e.bump_current_contract_instance_and_code(INSTANCE_BUMP_AMOUNT.into())?;
 
-        write_allowance(e, from.clone(), spender.clone(), amount, expiration_ledger)?;
+        write_allowance(
+            e,
+            from.metered_clone(e)?,
+            spender.metered_clone(e)?,
+            amount,
+            expiration_ledger,
+        )?;
         event::approve(e, from, spender, amount, expiration_ledger)?;
         Ok(())
     }
@@ -234,8 +240,8 @@ impl TokenTrait for Token {
 
         e.bump_current_contract_instance_and_code(INSTANCE_BUMP_AMOUNT.into())?;
 
-        spend_balance(e, from.clone(), amount)?;
-        receive_balance(e, to.clone(), amount)?;
+        spend_balance(e, from.metered_clone(e)?, amount)?;
+        receive_balance(e, to.metered_clone(e)?, amount)?;
         event::transfer(e, from, to, amount)?;
         Ok(())
     }
@@ -254,9 +260,9 @@ impl TokenTrait for Token {
 
         e.bump_current_contract_instance_and_code(INSTANCE_BUMP_AMOUNT.into())?;
 
-        spend_allowance(e, from.clone(), spender, amount)?;
-        spend_balance(e, from.clone(), amount)?;
-        receive_balance(e, to.clone(), amount)?;
+        spend_allowance(e, from.metered_clone(e)?, spender, amount)?;
+        spend_balance(e, from.metered_clone(e)?, amount)?;
+        receive_balance(e, to.metered_clone(e)?, amount)?;
         event::transfer(e, from, to, amount)?;
         Ok(())
     }
@@ -270,7 +276,7 @@ impl TokenTrait for Token {
 
         e.bump_current_contract_instance_and_code(INSTANCE_BUMP_AMOUNT.into())?;
 
-        spend_balance(e, from.clone(), amount)?;
+        spend_balance(e, from.metered_clone(e)?, amount)?;
         event::burn(e, from, amount)?;
         Ok(())
     }
@@ -284,8 +290,8 @@ impl TokenTrait for Token {
 
         e.bump_current_contract_instance_and_code(INSTANCE_BUMP_AMOUNT.into())?;
 
-        spend_allowance(e, from.clone(), spender, amount)?;
-        spend_balance(e, from.clone(), amount)?;
+        spend_allowance(e, from.metered_clone(e)?, spender, amount)?;
+        spend_balance(e, from.metered_clone(e)?, amount)?;
         event::burn(e, from, amount)?;
         Ok(())
     }
@@ -294,13 +300,13 @@ impl TokenTrait for Token {
     fn clawback(e: &Host, from: Address, amount: i128) -> Result<(), HostError> {
         let _span = tracy_span!("native token clawback");
         check_nonnegative_amount(e, amount)?;
-        check_clawbackable(e, from.clone())?;
+        check_clawbackable(e, from.metered_clone(e)?)?;
         let admin = read_administrator(e)?;
         admin.require_auth()?;
 
         e.bump_current_contract_instance_and_code(INSTANCE_BUMP_AMOUNT.into())?;
 
-        spend_balance_no_authorization_check(e, from.clone(), amount)?;
+        spend_balance_no_authorization_check(e, from.metered_clone(e)?, amount)?;
         event::clawback(e, admin, from, amount)?;
         Ok(())
     }
@@ -313,7 +319,7 @@ impl TokenTrait for Token {
 
         e.bump_current_contract_instance_and_code(INSTANCE_BUMP_AMOUNT.into())?;
 
-        write_authorization(e, addr.clone(), authorize)?;
+        write_authorization(e, addr.metered_clone(e)?, authorize)?;
         event::set_authorized(e, admin, addr, authorize)?;
         Ok(())
     }
@@ -327,7 +333,7 @@ impl TokenTrait for Token {
 
         e.bump_current_contract_instance_and_code(INSTANCE_BUMP_AMOUNT.into())?;
 
-        receive_balance(e, to.clone(), amount)?;
+        receive_balance(e, to.metered_clone(e)?, amount)?;
         event::mint(e, admin, to, amount)?;
         Ok(())
     }
@@ -340,7 +346,7 @@ impl TokenTrait for Token {
 
         e.bump_current_contract_instance_and_code(INSTANCE_BUMP_AMOUNT.into())?;
 
-        write_administrator(e, new_admin.clone())?;
+        write_administrator(e, new_admin.metered_clone(e)?)?;
         event::set_admin(e, admin, new_admin)?;
         Ok(())
     }
diff --git a/soroban-env-host/src/storage.rs b/soroban-env-host/src/storage.rs
index 226d64e4..6c0ffebc 100644
--- a/soroban-env-host/src/storage.rs
+++ b/soroban-env-host/src/storage.rs
@@ -15,7 +15,7 @@ use soroban_env_common::{Compare, Val};
 
 use crate::budget::Budget;
 use crate::host::ledger_info_helper::{get_entry_expiration, set_entry_expiration};
-use crate::host::metered_clone::MeteredClone;
+use crate::host::metered_clone::{MeteredAlloc, MeteredClone};
 use crate::xdr::{LedgerEntry, LedgerKey};
 use crate::Host;
 use crate::{host::metered_map::MeteredOrdMap, HostError};
@@ -333,11 +333,13 @@ impl Storage {
         })?;
 
         if new_expiration > old_expiration {
-            let mut new_entry = (*old_entry).metered_clone(host.budget_ref())?;
+            let mut new_entry = (*old_entry).metered_clone(host)?;
             set_entry_expiration(&mut new_entry, new_expiration);
-            self.map = self
-                .map
-                .insert(key, Some(Rc::new(new_entry)), host.budget_ref())?;
+            self.map = self.map.insert(
+                key,
+                Some(Rc::metered_new(new_entry, host)?),
+                host.budget_ref(),
+            )?;
         }
         Ok(())
     }
@@ -376,11 +378,13 @@ impl Storage {
         );
 
         if new_expiration > old_expiration {
-            let mut new_entry = (*old_entry).metered_clone(host.budget_ref())?;
+            let mut new_entry = (*old_entry).metered_clone(host)?;
             set_entry_expiration(&mut new_entry, new_expiration);
-            self.map = self
-                .map
-                .insert(key, Some(Rc::new(new_entry)), host.budget_ref())?;
+            self.map = self.map.insert(
+                key,
+                Some(Rc::metered_new(new_entry, host)?),
+                host.budget_ref(),
+            )?;
         }
         Ok(())
     }
@@ -428,12 +432,15 @@ mod test_footprint {
         budget.reset_unlimited()?;
         let mut fp = Footprint::default();
         // record when key not exist
-        let key = Rc::new(LedgerKey::ContractData(LedgerKeyContractData {
-            contract: ScAddress::Contract([0; 32].into()),
-            key: ScVal::I32(0),
-            durability: ContractDataDurability::Persistent,
-            body_type: ContractEntryBodyType::DataEntry,
-        }));
+        let key = Rc::metered_new(
+            LedgerKey::ContractData(LedgerKeyContractData {
+                contract: ScAddress::Contract([0; 32].into()),
+                key: ScVal::I32(0),
+                durability: ContractDataDurability::Persistent,
+                body_type: ContractEntryBodyType::DataEntry,
+            }),
+            &budget,
+        )?;
         fp.record_access(&key, AccessType::ReadOnly, &budget)?;
         assert_eq!(fp.0.contains_key::<LedgerKey>(&key, &budget)?, true);
         assert_eq!(
@@ -457,28 +464,37 @@ mod test_footprint {
     #[test]
     fn footprint_enforce_access() -> Result<(), HostError> {
         let budget = Budget::default();
-        let key = Rc::new(LedgerKey::ContractData(LedgerKeyContractData {
-            contract: ScAddress::Contract([0; 32].into()),
-            key: ScVal::I32(0),
-            durability: ContractDataDurability::Persistent,
-            body_type: ContractEntryBodyType::DataEntry,
-        }));
+        let key = Rc::metered_new(
+            LedgerKey::ContractData(LedgerKeyContractData {
+                contract: ScAddress::Contract([0; 32].into()),
+                key: ScVal::I32(0),
+                durability: ContractDataDurability::Persistent,
+                body_type: ContractEntryBodyType::DataEntry,
+            }),
+            &budget,
+        )?;
 
         // Key not in footprint. Only difference is type_
-        let key2 = Rc::new(LedgerKey::ContractData(LedgerKeyContractData {
-            contract: ScAddress::Contract([0; 32].into()),
-            key: ScVal::I32(0),
-            durability: ContractDataDurability::Temporary,
-            body_type: ContractEntryBodyType::DataEntry,
-        }));
+        let key2 = Rc::metered_new(
+            LedgerKey::ContractData(LedgerKeyContractData {
+                contract: ScAddress::Contract([0; 32].into()),
+                key: ScVal::I32(0),
+                durability: ContractDataDurability::Temporary,
+                body_type: ContractEntryBodyType::DataEntry,
+            }),
+            &budget,
+        )?;
 
         // Key not in footprint. Only difference is body_type
-        let key3 = Rc::new(LedgerKey::ContractData(LedgerKeyContractData {
-            contract: ScAddress::Contract([0; 32].into()),
-            key: ScVal::I32(0),
-            durability: ContractDataDurability::Persistent,
-            body_type: ContractEntryBodyType::ExpirationExtension,
-        }));
+        let key3 = Rc::metered_new(
+            LedgerKey::ContractData(LedgerKeyContractData {
+                contract: ScAddress::Contract([0; 32].into()),
+                key: ScVal::I32(0),
+                durability: ContractDataDurability::Persistent,
+                body_type: ContractEntryBodyType::ExpirationExtension,
+            }),
+            &budget,
+        )?;
 
         let om = [(Rc::clone(&key), AccessType::ReadOnly)].into();
         let mom = MeteredOrdMap::from_map(om, &budget)?;
@@ -502,12 +518,15 @@ mod test_footprint {
     fn footprint_enforce_access_not_exist() -> Result<(), HostError> {
         let budget = Budget::default();
         let mut fp = Footprint::default();
-        let key = Rc::new(LedgerKey::ContractData(LedgerKeyContractData {
-            contract: ScAddress::Contract([0; 32].into()),
-            key: ScVal::I32(0),
-            durability: ContractDataDurability::Persistent,
-            body_type: ContractEntryBodyType::DataEntry,
-        }));
+        let key = Rc::metered_new(
+            LedgerKey::ContractData(LedgerKeyContractData {
+                contract: ScAddress::Contract([0; 32].into()),
+                key: ScVal::I32(0),
+                durability: ContractDataDurability::Persistent,
+                body_type: ContractEntryBodyType::DataEntry,
+            }),
+            &budget,
+        )?;
         let res = fp.enforce_access(&key, AccessType::ReadOnly, &budget);
         assert!(HostError::result_matches_err(
             res,
@@ -519,12 +538,15 @@ mod test_footprint {
     #[test]
     fn footprint_attempt_to_write_readonly_entry() -> Result<(), HostError> {
         let budget = Budget::default();
-        let key = Rc::new(LedgerKey::ContractData(LedgerKeyContractData {
-            contract: ScAddress::Contract([0; 32].into()),
-            key: ScVal::I32(0),
-            durability: ContractDataDurability::Persistent,
-            body_type: ContractEntryBodyType::DataEntry,
-        }));
+        let key = Rc::metered_new(
+            LedgerKey::ContractData(LedgerKeyContractData {
+                contract: ScAddress::Contract([0; 32].into()),
+                key: ScVal::I32(0),
+                durability: ContractDataDurability::Persistent,
+                body_type: ContractEntryBodyType::DataEntry,
+            }),
+            &budget,
+        )?;
         let om = [(Rc::clone(&key), AccessType::ReadOnly)].into();
         let mom = MeteredOrdMap::from_map(om, &budget)?;
         let mut fp = Footprint(mom);
diff --git a/soroban-env-host/src/vm.rs b/soroban-env-host/src/vm.rs
index f2e0f17d..9679dd46 100644
--- a/soroban-env-host/src/vm.rs
+++ b/soroban-env-host/src/vm.rs
@@ -214,6 +214,7 @@ impl Vm {
         // Here we do _not_ supply the store with any fuel. Fuel is supplied
         // right before the VM is being run, i.e., before crossing the host->VM
         // boundary.
+        // Missing metering for the Rc, but this is once per Vm so should be okay
         Ok(Rc::new(Self {
             contract_id,
             module,


Test patch:
diff --git a/soroban-env-host/src/e2e_invoke.rs b/soroban-env-host/src/e2e_invoke.rs
index 75fa917b..3b2d2462 100644
--- a/soroban-env-host/src/e2e_invoke.rs
+++ b/soroban-env-host/src/e2e_invoke.rs
@@ -21,7 +21,7 @@ use crate::{
     fees::LedgerEntryRentChange,
     host::{
         ledger_info_helper::{get_entry_expiration, get_key_durability, set_entry_expiration},
-        metered_clone::{MeteredClone, MeteredContainer},
+        metered_clone::{MeteredAlloc, MeteredClone, MeteredIterator},
         metered_xdr::{metered_from_xdr_with_budget, metered_write_xdr},
     },
     storage::{AccessType, Footprint, FootprintMap, SnapshotSource, Storage, StorageMap},
@@ -255,7 +255,7 @@ pub fn invoke_host_function<T: AsRef<[u8]>, I: ExactSizeIterator<Item = T>>(
     let footprint = build_storage_footprint_from_xdr(&budget, resources.footprint)?;
     let map =
         build_storage_map_from_xdr_ledger_entries(&budget, &footprint, encoded_ledger_entries)?;
-    let init_storage_map = map.metered_clone(&budget)?;
+    let init_storage_map = map.metered_clone(budget)?;
 
     let storage = Storage::with_enforcing_footprint_and_map(footprint, map);
     let host = Host::with_storage_and_budget(storage, budget.clone());
@@ -322,7 +322,7 @@ pub fn encode_contract_events(budget: &Budget, events: &Events) -> Result<Vec<Ve
             metered_write_xdr(budget, &e.event, &mut buf)?;
             Ok(buf)
         })
-        .collect()
+        .metered_collect::<Result<Vec<Vec<u8>>, HostError>>(budget)?
 }
 
 fn extract_diagnostic_events(events: &Events, diagnostic_events: &mut Vec<DiagnosticEvent>) {
@@ -374,7 +374,7 @@ fn ledger_entry_to_ledger_key(le: &LedgerEntry, budget: &Budget) -> Result<Ledge
             },
         })),
         LedgerEntryData::ContractCode(code) => Ok(LedgerKey::ContractCode(LedgerKeyContractCode {
-            hash: code.hash.clone(),
+            hash: code.hash.metered_clone(budget)?,
             body_type: match &code.body {
                 ContractCodeEntryBody::DataEntry(_) => ContractEntryBodyType::DataEntry,
                 ContractCodeEntryBody::ExpirationExtension => {
@@ -402,14 +402,22 @@ fn build_storage_footprint_from_xdr(
 ) -> Result<Footprint, HostError> {
     let mut footprint_map = FootprintMap::new();
 
-    for key in footprint.read_write.to_vec() {
+    for key in footprint.read_write.as_vec() {
         validate_footprint_key(&key)?;
-        footprint_map = footprint_map.insert(Rc::new(key), AccessType::ReadWrite, budget)?;
+        footprint_map = footprint_map.insert(
+            Rc::metered_new_from_ref(key, budget)?,
+            AccessType::ReadWrite,
+            budget,
+        )?;
     }
 
-    for key in footprint.read_only.to_vec() {
+    for key in footprint.read_only.as_vec() {
         validate_footprint_key(&key)?;
-        footprint_map = footprint_map.insert(Rc::new(key), AccessType::ReadOnly, budget)?;
+        footprint_map = footprint_map.insert(
+            Rc::metered_new_from_ref(key, budget)?,
+            AccessType::ReadOnly,
+            budget,
+        )?;
     }
     Ok(Footprint(footprint_map))
 }
@@ -421,11 +429,11 @@ fn build_storage_map_from_xdr_ledger_entries<T: AsRef<[u8]>, I: ExactSizeIterato
 ) -> Result<StorageMap, HostError> {
     let mut map = StorageMap::new();
     for buf in encoded_ledger_entries {
-        let le = Rc::new(metered_from_xdr_with_budget::<LedgerEntry>(
-            buf.as_ref(),
-            &budget,
-        )?);
-        let key = Rc::new(ledger_entry_to_ledger_key(&le, budget)?);
+        let le = Rc::metered_new(
+            metered_from_xdr_with_budget::<LedgerEntry>(buf.as_ref(), budget)?,
+            budget,
+        )?;
+        let key = Rc::metered_new(ledger_entry_to_ledger_key(&le, budget)?, budget)?;
         if !footprint.0.contains_key::<LedgerKey>(&key, budget)? {
             return Err(Error::from_type_and_code(
                 ScErrorType::Storage,
@@ -439,7 +447,7 @@ fn build_storage_map_from_xdr_ledger_entries<T: AsRef<[u8]>, I: ExactSizeIterato
     // Add non-existing entries from the footprint to the storage.
     for k in footprint.0.keys(budget)? {
         if !map.contains_key::<LedgerKey>(k, budget)? {
-            map = map.insert(k.clone(), None, budget)?;
+            map = map.insert(Rc::clone(k), None, budget)?;
         }
     }
     Ok(map)
@@ -450,13 +458,11 @@ impl Host {
         &self,
         encoded_contract_auth_entries: I,
     ) -> Result<Vec<SorobanAuthorizationEntry>, HostError> {
-        Vec::<SorobanAuthorizationEntry>::charge_bulk_init_cpy(
-            encoded_contract_auth_entries.len() as u64,
-            self.as_budget(),
-        )?;
         encoded_contract_auth_entries
             .map(|buf| self.metered_from_xdr::<SorobanAuthorizationEntry>(buf.as_ref()))
-            .collect()
+            .metered_collect::<Result<Vec<SorobanAuthorizationEntry>, HostError>>(
+                self.as_budget(),
+            )?
     }
 }
 
@@ -468,7 +474,7 @@ struct StorageMapSnapshotSource<'a> {
 impl<'a> SnapshotSource for StorageMapSnapshotSource<'a> {
     fn get(&self, key: &Rc<LedgerKey>) -> Result<Rc<LedgerEntry>, HostError> {
         if let Some(Some(value)) = self.map.get::<Rc<LedgerKey>>(key, self.budget)? {
-            Ok(value.clone())
+            Ok(Rc::clone(value))
         } else {
             Err(Error::from_type_and_code(ScErrorType::Storage, ScErrorCode::InternalError).into())
         }
diff --git a/soroban-env-host/src/native_contract/testutils.rs b/soroban-env-host/src/native_contract/testutils.rs
index e9aa2df5..79ba1dd7 100644
--- a/soroban-env-host/src/native_contract/testutils.rs
+++ b/soroban-env-host/src/native_contract/testutils.rs
@@ -270,7 +270,7 @@ pub(crate) fn create_account(
     sponsorships: Option<(u32, u32)>,
     flags: u32,
 ) {
-    let key = host.to_account_key(account_id.clone());
+    let key = host.to_account_key(account_id.clone()).unwrap();
     let account_id = match key.as_ref() {
         LedgerKey::Account(acc) => acc.account_id.clone(),
         _ => unreachable!(),
@@ -321,7 +321,7 @@ pub(crate) fn create_account(
 
     host.add_ledger_entry(
         &key,
-        &Host::ledger_entry_from_data(LedgerEntryData::Account(acc_entry)),
+        &Host::ledger_entry_from_data(host, LedgerEntryData::Account(acc_entry)).unwrap(),
     )
     .unwrap();
 }
diff --git a/soroban-env-host/src/test/auth.rs b/soroban-env-host/src/test/auth.rs
index 639ee607..fcda0f75 100644
--- a/soroban-env-host/src/test/auth.rs
+++ b/soroban-env-host/src/test/auth.rs
@@ -216,11 +216,14 @@ impl AuthTest {
 
     fn read_nonce_expiration(&self, address: &Address, nonce: i64) -> Option<u32> {
         let nonce_key_scval = ScVal::LedgerKeyNonce(ScNonceKey { nonce });
-        let nonce_key = self.host.storage_key_for_address(
-            address.to_sc_address().unwrap(),
-            nonce_key_scval,
-            ContractDataDurability::Temporary,
-        );
+        let nonce_key = self
+            .host
+            .storage_key_for_address(
+                address.to_sc_address().unwrap(),
+                nonce_key_scval,
+                ContractDataDurability::Temporary,
+            )
+            .unwrap();
         self.host
             .with_mut_storage(|storage| {
                 if !storage.has(&nonce_key, self.host.budget_ref())? {
diff --git a/soroban-env-host/src/test/bytes.rs b/soroban-env-host/src/test/bytes.rs
index 01a0ac39..59c1d0f4 100644
--- a/soroban-env-host/src/test/bytes.rs
+++ b/soroban-env-host/src/test/bytes.rs
@@ -160,7 +160,7 @@ fn linear_memory_operations() -> Result<(), HostError> {
     // tests bytes_copy_{to,from}_linear_memory
     {
         let obj0 = host.test_bin_obj(&[1, 2, 3, 4])?;
-        let mut args = host.vec_new(Val::from_void().into())?;
+        let mut args = host.vec_new()?;
         args = host.vec_push_back(args, obj0.to_val())?;
         let obj: BytesObject = host
             .call(id_obj, Symbol::try_from_small_str("bin_inc").unwrap(), args)?
diff --git a/soroban-env-host/src/test/event.rs b/soroban-env-host/src/test/event.rs
index 029ea420..e56705d1 100644
--- a/soroban-env-host/src/test/event.rs
+++ b/soroban-env-host/src/test/event.rs
@@ -20,7 +20,7 @@ impl ContractFunctionSet for ContractWithSingleEvent {
         // Add a contract event
         let mut data = host.map_new().unwrap();
         data = host.map_put(data, 1_u32.into(), 2_u32.into()).unwrap();
-        let mut topics = host.vec_new(().into()).unwrap();
+        let mut topics = host.vec_new().unwrap();
         topics = host.vec_push_back(topics, 0u32.into()).unwrap();
         topics = host.vec_push_back(topics, 1u32.into()).unwrap();
         Some(host.contract_event(topics, data.to_val()).unwrap().into())
diff --git a/soroban-env-host/src/test/invocation.rs b/soroban-env-host/src/test/invocation.rs
index 00f0c3ba..448ec221 100644
--- a/soroban-env-host/src/test/invocation.rs
+++ b/soroban-env-host/src/test/invocation.rs
@@ -231,7 +231,7 @@ fn native_invoke_return_err_variants() -> Result<(), HostError> {
     host.register_test_contract(addr, Rc::new(ReturnContractError))?;
 
     let sym = Symbol::try_from_small_str("go")?;
-    let args = host.vec_new(Val::VOID.into())?;
+    let args = host.vec_new()?;
     let err = ReturnContractError::ERR;
 
     // We want a call to return `Err(Error)` not `Ok(Error)`
diff --git a/soroban-env-host/src/test/map.rs b/soroban-env-host/src/test/map.rs
index 25e7d35c..44d1a897 100644
--- a/soroban-env-host/src/test/map.rs
+++ b/soroban-env-host/src/test/map.rs
@@ -5,7 +5,7 @@ use soroban_env_common::{
         AccountId, LedgerEntry, LedgerKey, LedgerKeyAccount, PublicKey, ScErrorCode, ScErrorType,
         Uint256,
     },
-    MapObject,
+    MapObject, U32Val,
 };
 
 use crate::{
@@ -13,6 +13,8 @@ use crate::{
     Env, Error, Host, HostError, Symbol, TryFromVal, Val,
 };
 
+const MAP_OOB: Error = Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds);
+
 #[test]
 fn map_put_has_and_get() -> Result<(), HostError> {
     let host = Host::default();
@@ -101,7 +103,7 @@ fn map_put_insert_and_remove() -> Result<(), HostError> {
 }
 
 #[test]
-fn map_prev_and_next() -> Result<(), HostError> {
+fn map_iter_by_index() -> Result<(), HostError> {
     let host = Host::default();
     let scmap: ScMap = host.map_err(
         vec![
@@ -118,65 +120,38 @@ fn map_prev_and_next() -> Result<(), HostError> {
     )?;
     let scobj = ScVal::Map(Some(scmap));
     let obj: MapObject = host.to_host_val(&scobj)?.try_into()?;
-    // prev
-    {
-        assert_eq!(
-            host.map_prev_key(obj, 0_u32.into())?.get_payload(),
-            Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds)
-                .to_val()
-                .get_payload()
-        );
-        assert_eq!(
-            host.map_prev_key(obj, 1_u32.into())?.get_payload(),
-            Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds)
-                .to_val()
-                .get_payload()
-        );
-        assert_eq!(
-            host.map_prev_key(obj, 2_u32.into())?.get_payload(),
-            Val::from_u32(1).to_val().get_payload()
-        );
-        assert_eq!(
-            host.map_prev_key(obj, 4_u32.into())?.get_payload(),
-            Val::from_u32(1).to_val().get_payload()
-        );
-        assert_eq!(
-            host.map_prev_key(obj, 5_u32.into())?.get_payload(),
-            Val::from_u32(4).to_val().get_payload()
-        );
-    }
-    // next
-    {
-        assert_eq!(
-            host.map_next_key(obj, 5_u32.into())?.get_payload(),
-            Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds)
-                .to_val()
-                .get_payload()
-        );
-        assert_eq!(
-            host.map_next_key(obj, 4_u32.into())?.get_payload(),
-            Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds)
-                .to_val()
-                .get_payload()
-        );
-        assert_eq!(
-            host.map_next_key(obj, 3_u32.into())?.get_payload(),
-            Val::from_u32(4).to_val().get_payload()
-        );
-        assert_eq!(
-            host.map_next_key(obj, 1_u32.into())?.get_payload(),
-            Val::from_u32(4).to_val().get_payload()
-        );
-        assert_eq!(
-            host.map_next_key(obj, 0_u32.into())?.get_payload(),
-            Val::from_u32(1).to_val().get_payload()
-        );
-    }
+    // get keys
+    assert_eq!(
+        host.map_key_by_pos(obj, U32Val::from(0))?.get_payload(),
+        U32Val::from(1).to_val().get_payload()
+    );
+    assert_eq!(
+        host.map_key_by_pos(obj, U32Val::from(1))?.get_payload(),
+        U32Val::from(4).to_val().get_payload()
+    );
+    assert!(HostError::result_matches_err(
+        host.map_key_by_pos(obj, U32Val::from(2)),
+        MAP_OOB
+    ));
+    // get vals
+    assert_eq!(
+        host.map_val_by_pos(obj, U32Val::from(0))?.get_payload(),
+        U32Val::from(2).to_val().get_payload()
+    );
+    assert_eq!(
+        host.map_val_by_pos(obj, U32Val::from(1))?.get_payload(),
+        U32Val::from(8).to_val().get_payload()
+    );
+    assert!(HostError::result_matches_err(
+        host.map_val_by_pos(obj, U32Val::from(2)),
+        MAP_OOB
+    ));
+
     Ok(())
 }
 
 #[test]
-fn map_prev_and_next_heterogeneous() -> Result<(), HostError> {
+fn hetro_map_iter_by_index() -> Result<(), HostError> {
     let host = Host::default();
     let scmap: ScMap = host.map_err(
         vec![ScMapEntry {
@@ -195,59 +170,62 @@ fn map_prev_and_next_heterogeneous() -> Result<(), HostError> {
     let obj_vec = host.to_host_val(&scobj_vec)?;
 
     let mut test_map = host.map_new()?;
-    test_map = host.map_put(test_map, 2u32.into(), 4u32.into())?;
-    test_map = host.map_put(test_map, obj_map, 4u32.into())?;
-    test_map = host.map_put(test_map, obj_vec, 4u32.into())?;
+    test_map = host.map_put(test_map, 2u32.into(), 1u32.into())?;
+    test_map = host.map_put(test_map, obj_map, 2u32.into())?;
+    test_map = host.map_put(test_map, obj_vec, 3u32.into())?;
     test_map = host.map_put(test_map, sym.into(), 4u32.into())?;
     // The key ordering should be [u32, symbol, vec, map]
-    // prev
     {
+        assert!(HostError::result_matches_err(
+            host.map_key_by_pos(test_map, U32Val::from(4)),
+            MAP_OOB
+        ));
         assert_eq!(
-            host.map_prev_key(test_map, 0_u32.into())?.get_payload(),
-            Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds)
-                .to_val()
-                .get_payload()
-        );
-        assert_eq!(
-            host.map_prev_key(test_map, 4_u32.into())?.get_payload(),
-            Val::from_u32(2).to_val().get_payload()
-        );
-        assert_eq!(
-            host.map_prev_key(test_map, sym.into())?.get_payload(),
+            host.map_key_by_pos(test_map, U32Val::from(0))?
+                .get_payload(),
             Val::from_u32(2).to_val().get_payload()
         );
         assert_eq!(
-            host.map_prev_key(test_map, obj_vec)?.get_payload(),
+            host.map_key_by_pos(test_map, U32Val::from(1))?
+                .get_payload(),
             sym.as_val().get_payload()
         );
         assert_eq!(
-            host.map_prev_key(test_map, obj_map)?.get_payload(),
+            host.map_key_by_pos(test_map, U32Val::from(2))?
+                .get_payload(),
             obj_vec.get_payload()
         );
-    }
-    // next
-    {
         assert_eq!(
-            host.map_next_key(test_map, 0_u32.into())?.get_payload(),
-            Val::from_u32(2).to_val().get_payload()
+            host.map_key_by_pos(test_map, U32Val::from(3))?
+                .get_payload(),
+            obj_map.get_payload()
         );
+    }
+    // The val ordering should be [1u32, 4u32, 3u32, 2u32]
+    {
+        assert!(HostError::result_matches_err(
+            host.map_val_by_pos(test_map, U32Val::from(4)),
+            MAP_OOB
+        ));
         assert_eq!(
-            host.map_next_key(test_map, 4_u32.into())?.get_payload(),
-            sym.as_val().get_payload()
+            host.map_val_by_pos(test_map, U32Val::from(0))?
+                .get_payload(),
+            Val::from_u32(1).to_val().get_payload()
         );
         assert_eq!(
-            host.map_next_key(test_map, sym.into())?.get_payload(),
-            obj_vec.get_payload()
+            host.map_val_by_pos(test_map, U32Val::from(1))?
+                .get_payload(),
+            Val::from_u32(4).to_val().get_payload()
         );
         assert_eq!(
-            host.map_next_key(test_map, obj_vec)?.get_payload(),
-            obj_map.get_payload()
+            host.map_val_by_pos(test_map, U32Val::from(2))?
+                .get_payload(),
+            Val::from_u32(3).to_val().get_payload()
         );
         assert_eq!(
-            host.map_next_key(test_map, obj_map)?.get_payload(),
-            Error::from_type_and_code(ScErrorType::Object, ScErrorCode::IndexBounds)
-                .to_val()
-                .get_payload()
+            host.map_val_by_pos(test_map, U32Val::from(3))?
+                .get_payload(),
+            Val::from_u32(2).to_val().get_payload()
         );
     }
 
diff --git a/soroban-env-host/src/test/token.rs b/soroban-env-host/src/test/token.rs
index 058fe439..ca3f15ee 100644
--- a/soroban-env-host/src/test/token.rs
+++ b/soroban-env-host/src/test/token.rs
@@ -178,8 +178,10 @@ impl TokenTest {
                 match entry.data.clone() {
                     LedgerEntryData::Account(mut account) => {
                         account.flags = new_flags;
-                        let update =
-                            Host::ledger_entry_from_data(LedgerEntryData::Account(account));
+                        let update = Host::ledger_entry_from_data(
+                            &self.host,
+                            LedgerEntryData::Account(account),
+                        )?;
                         s.put(key, &update, self.host.as_budget())
                     }
                     _ => unreachable!(),
@@ -214,7 +216,8 @@ impl TokenTest {
         };
         let key = self
             .host
-            .to_trustline_key(account_id.clone(), asset.clone());
+            .to_trustline_key(account_id.clone(), asset.clone())
+            .unwrap();
         let ext = if let Some((buying, selling)) = liabilities {
             TrustLineEntryExt::V1({
                 TrustLineEntryV1 {
@@ -237,7 +240,11 @@ impl TokenTest {
         self.host
             .add_ledger_entry(
                 &key,
-                &Host::ledger_entry_from_data(LedgerEntryData::Trustline(trustline_entry)),
+                &Host::ledger_entry_from_data(
+                    &self.host,
+                    LedgerEntryData::Trustline(trustline_entry),
+                )
+                .unwrap(),
             )
             .unwrap();
 
@@ -251,8 +258,10 @@ impl TokenTest {
                 match entry.data.clone() {
                     LedgerEntryData::Trustline(mut trustline) => {
                         trustline.flags = new_flags;
-                        let update =
-                            Host::ledger_entry_from_data(LedgerEntryData::Trustline(trustline));
+                        let update = Host::ledger_entry_from_data(
+                            &self.host,
+                            LedgerEntryData::Trustline(trustline),
+                        )?;
                         s.put(key, &update, self.host.as_budget())
                     }
                     _ => unreachable!(),
@@ -1048,7 +1057,8 @@ fn test_clawback_on_contract() {
 
     let issuer_ledger_key = test
         .host
-        .to_account_key(signing_key_to_account_id(&test.issuer_key));
+        .to_account_key(signing_key_to_account_id(&test.issuer_key))
+        .unwrap();
 
     let user_1 = generate_bytes_array();
     let user_2 = generate_bytes_array();
@@ -1114,7 +1124,8 @@ fn test_auth_revocable_on_contract() {
 
     let issuer_ledger_key = test
         .host
-        .to_account_key(signing_key_to_account_id(&test.issuer_key));
+        .to_account_key(signing_key_to_account_id(&test.issuer_key))
+        .unwrap();
 
     let user_1 = generate_bytes_array();
     let user_1_addr = contract_id_to_address(&test.host, user_1);
@@ -1164,7 +1175,8 @@ fn test_auth_required() {
 
     let issuer_ledger_key = test
         .host
-        .to_account_key(signing_key_to_account_id(&test.issuer_key));
+        .to_account_key(signing_key_to_account_id(&test.issuer_key))
+        .unwrap();
 
     let user_1 = generate_bytes_array();
     let user_2 = generate_bytes_array();
diff --git a/soroban-env-host/src/test/tuple.rs b/soroban-env-host/src/test/tuple.rs
index 9efd59b5..622d3a1d 100644
--- a/soroban-env-host/src/test/tuple.rs
+++ b/soroban-env-host/src/test/tuple.rs
@@ -8,7 +8,7 @@ fn tuple_conversions() -> Result<(), HostError> {
 
     let val: Val = (1u32, 1i32).try_into_val(&host)?;
 
-    let mut obj = host.vec_new(Val::VOID.into())?;
+    let mut obj = host.vec_new()?;
     obj = host.vec_push_back(obj, (1u32).into())?;
     obj = host.vec_push_back(obj, (1i32).into())?;
 
diff --git a/soroban-env-host/src/test/util.rs b/soroban-env-host/src/test/util.rs
index 24cdad7b..fa329239 100644
--- a/soroban-env-host/src/test/util.rs
+++ b/soroban-env-host/src/test/util.rs
@@ -136,7 +136,11 @@ impl Host {
             signers: Default::default(),
             ext: xdr::AccountEntryExt::V0,
         };
-        let le = Host::ledger_entry_from_data(LedgerEntryData::Account(account_entry));
+        let le = Rc::new(LedgerEntry {
+            last_modified_ledger_seq: 0,
+            data: LedgerEntryData::Account(account_entry),
+            ext: xdr::LedgerEntryExt::V0,
+        });
         (lk, le)
     }
 
diff --git a/soroban-env-host/src/test/vec.rs b/soroban-env-host/src/test/vec.rs
index b9d87cd2..8dd7a018 100644
--- a/soroban-env-host/src/test/vec.rs
+++ b/soroban-env-host/src/test/vec.rs
@@ -28,15 +28,13 @@ fn vec_as_seen_by_host() -> Result<(), HostError> {
 }
 
 #[test]
-fn vec_new_with_capacity() -> Result<(), HostError> {
+fn vec_new() -> Result<(), HostError> {
     let host = Host::default();
-    host.vec_new(Val::from_void().to_val())?;
-    host.vec_new(5_u32.into())?;
-    let code = (ScErrorType::Value, ScErrorCode::UnexpectedType);
-    let res = host.vec_new(5_i32.into());
-    assert!(HostError::result_matches_err(res, code));
-    let res = host.vec_new(Val::from_bool(true).to_val());
-    assert!(HostError::result_matches_err(res, code));
+    let v = host.vec_new()?;
+    assert_eq!(
+        host.vec_len(v)?.to_val().get_payload(),
+        U32Val::from(0).to_val().get_payload()
+    );
     Ok(())
 }
 
diff --git a/soroban-env-host/tests/integration.rs b/soroban-env-host/tests/integration.rs
index c7eae35c..69bd2a63 100644
--- a/soroban-env-host/tests/integration.rs
+++ b/soroban-env-host/tests/integration.rs
@@ -7,12 +7,12 @@ fn vec_as_seen_by_user() -> Result<(), HostError> {
     let host = Host::default();
     let int1 = host.obj_from_i64(5)?;
 
-    let vec1a = host.vec_new(Val::from_void().into())?;
+    let vec1a = host.vec_new()?;
     let vec1b = host.vec_push_back(vec1a, *int1.as_ref())?;
 
     assert_ne!(vec1a.as_val().get_payload(), vec1b.as_val().get_payload());
 
-    let vec2a = host.vec_new(Val::from_void().into())?;
+    let vec2a = host.vec_new()?;
     let vec2b = host.vec_push_back(vec2a, *int1.as_ref())?;
 
     assert_ne!(vec2a.as_val().get_payload(), vec2b.as_val().get_payload());
diff --git a/soroban-test-wasms/wasm-workspace/Cargo.lock b/soroban-test-wasms/wasm-workspace/Cargo.lock
index 17bef94b..fad67ab4 100644
--- a/soroban-test-wasms/wasm-workspace/Cargo.lock
+++ b/soroban-test-wasms/wasm-workspace/Cargo.lock
@@ -138,7 +138,7 @@ dependencies = [
  "num-bigint",
  "proc-macro2",
  "quote",
- "syn 2.0.18",
+ "syn",
 ]
 
 [[package]]
@@ -227,7 +227,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1586fa608b1dab41f667475b4a41faec5ba680aee428bfa5de4ea520fdc6e901"
 dependencies = [
  "quote",
- "syn 2.0.18",
+ "syn",
 ]
 
 [[package]]
@@ -268,7 +268,7 @@ checksum = "83fdaf97f4804dcebfa5862639bc9ce4121e82140bec2a987ac5140294865b5b"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.18",
+ "syn",
 ]
 
 [[package]]
@@ -292,7 +292,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "strsim",
- "syn 2.0.18",
+ "syn",
 ]
 
 [[package]]
@@ -303,7 +303,7 @@ checksum = "29a358ff9f12ec09c3e61fef9b5a9902623a695a46a917b07f269bff1445611a"
 dependencies = [
  "darling_core",
  "quote",
- "syn 2.0.18",
+ "syn",
 ]
 
 [[package]]
@@ -324,7 +324,7 @@ checksum = "53e0efad4403bfc52dc201159c4b842a246a14b98c64b55dfd0f2d89729dfeb8"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.18",
+ "syn",
 ]
 
 [[package]]
@@ -789,13 +789,13 @@ dependencies = [
 
 [[package]]
 name = "num-derive"
-version = "0.3.3"
+version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "876a53fff98e03a936a674b29568b0e605f06b29372c2489ff4de23f1949743d"
+checksum = "9e6a0fd4f737c707bd9086cc16c925f294943eb62eb71499e9fd4cf71f8b9f4e"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 1.0.109",
+ "syn",
 ]
 
 [[package]]
@@ -873,7 +873,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3b69d39aab54d069e7f2fe8cb970493e7834601ca2d8c65fd7bbd183578080d1"
 dependencies = [
  "proc-macro2",
- "syn 2.0.18",
+ "syn",
 ]
 
 [[package]]
@@ -1033,7 +1033,7 @@ checksum = "d9735b638ccc51c28bf6914d90a2e9725b377144fc612c49a611fddd1b631d68"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.18",
+ "syn",
 ]
 
 [[package]]
@@ -1072,7 +1072,7 @@ dependencies = [
  "darling",
  "proc-macro2",
  "quote",
- "syn 2.0.18",
+ "syn",
 ]
 
 [[package]]
@@ -1139,7 +1139,6 @@ dependencies = [
  "crate-git-revision",
  "ethnum",
  "num-derive",
- "num-integer",
  "num-traits",
  "serde",
  "soroban-env-macros",
@@ -1191,8 +1190,7 @@ dependencies = [
  "serde",
  "serde_json",
  "stellar-xdr",
- "syn 2.0.18",
- "thiserror",
+ "syn",
 ]
 
 [[package]]
@@ -1213,7 +1211,7 @@ dependencies = [
  "itertools",
  "proc-macro2",
  "quote",
- "syn 2.0.18",
+ "syn",
 ]
 
 [[package]]
@@ -1247,7 +1245,7 @@ dependencies = [
  "soroban-spec",
  "soroban-spec-rust",
  "stellar-xdr",
- "syn 2.0.18",
+ "syn",
 ]
 
 [[package]]
@@ -1270,7 +1268,7 @@ dependencies = [
  "sha2 0.9.9",
  "soroban-spec",
  "stellar-xdr",
- "syn 2.0.18",
+ "syn",
  "thiserror",
 ]
 
@@ -1349,17 +1347,6 @@ version = "2.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "81cdd64d312baedb58e21336b31bc043b77e01cc99033ce76ef539f78e965ebc"
 
-[[package]]
-name = "syn"
-version = "1.0.109"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237"
-dependencies = [
- "proc-macro2",
- "quote",
- "unicode-ident",
-]
-
 [[package]]
 name = "syn"
 version = "2.0.18"
@@ -1395,7 +1382,7 @@ checksum = "f9456a42c5b0d803c8cd86e73dd7cc9edd429499f37a3550d286d5e86720569f"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.18",
+ "syn",
 ]
 
 [[package]]
@@ -1476,7 +1463,7 @@ dependencies = [
  "once_cell",
  "proc-macro2",
  "quote",
- "syn 2.0.18",
+ "syn",
  "wasm-bindgen-shared",
 ]
 
@@ -1498,7 +1485,7 @@ checksum = "54681b18a46765f095758388f2d0cf16eb8d4169b639ab575a8f5693af210c7b"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.18",
+ "syn",
  "wasm-bindgen-backend",
  "wasm-bindgen-shared",
 ]
@@ -1648,5 +1635,5 @@ checksum = "ce36e65b0d2999d2aafac989fb249189a141aee1f53c612c1f37d72631959f69"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.18",
+ "syn",
 ]