feat(engine)!: charge storage as a transaction writes it

crates/engine/src/fees/fee_module.rs

@@ -45,6 +45,10 @@ impl FeeModule {
         Ok((base_bytes, premium))
     }
 
+    fn storage_cost_for(&self, bytes: u64) -> Option<u64> {
+        Some(self.fee_table.per_byte_storage_cost().checked_mul(bytes)? / self.fee_table.storage_cost_divisor())
+    }
+
     /// Charges everything that is a function of the state being persisted, rather than of the work
     /// done to produce it: storage, the template-publish premium, substate slots, the metering of
     /// WASM and native execution, and the exhaust burn over the resulting total.
@@ -89,12 +93,9 @@ impl FeeModule {
             .checked_add(receipt_bytes)
             .ok_or_else(|| RuntimeModuleError::Overflow("Overflow accumulating storage bytes".to_string()))?;
 
-        let cost = self
-            .fee_table
-            .per_byte_storage_cost()
-            .checked_mul(total_storage as u64)
+        let storage_cost = self
+            .storage_cost_for(total_storage as u64)
             .ok_or_else(|| RuntimeModuleError::Overflow("Overflow calculating storage cost".to_string()))?;
-        let storage_cost = cost / self.fee_table.storage_cost_divisor();
 
         let template_base_cost = self
             .fee_table
@@ -214,6 +215,17 @@ impl<TStore: StateReader> RuntimeModule<TStore> for FeeModule {
         Ok(())
     }
 
+    fn on_storage_written(&self, track: &mut StateTracker<TStore>, bytes: u64) -> Result<(), RuntimeModuleError> {
+        let mut state = track.chargeable_state();
+        let fee_state = state.fee_state_mut();
+        fee_state.accumulate_storage_bytes(bytes);
+        let cost = self
+            .storage_cost_for(fee_state.accumulated_storage_bytes())
+            .ok_or_else(|| RuntimeModuleError::Overflow("Overflow calculating storage cost".to_string()))?;
+        fee_state.set_charge(FeeSource::Storage, cost);
+        Ok(())
+    }
+
     fn on_before_finalize(&self, track: &mut StateTracker<TStore>) -> Result<(), RuntimeModuleError> {
         self.charge_finalization_fees(&mut track.chargeable_state())
     }

crates/engine/src/runtime/fee_state.rs

@@ -16,6 +16,11 @@ pub struct FeeState {
     fee_payments: Vec<(ResourceContainer, VaultId)>,
     running_payments_total: u64,
     fee_charges: FeeBreakdown,
+    /// Substate growth metered during execution, in encoded bytes. Charged as the transaction runs so
+    /// that one which cannot pay for the state it is writing stops there rather than at finalization,
+    /// and so the fee intent's own storage is covered by the checkpoint's paid-in-full check. The
+    /// resulting charge is provisional: finalization recomputes it over the state actually persisted.
+    accumulated_storage_bytes: u64,
     /// Raw Wasmer metering points consumed across every WASM invocation in this transaction.
     /// Summed across invocations so the divisor in `FeeModule::on_before_finalize` rounds once
     /// against the total — dividing per-call would let small invocations round to zero and let a
@@ -99,6 +104,17 @@ impl FeeState {
         self.fee_charges.set(source, amount)
     }
 
+    /// Bytes of substate growth metered so far. Summed across the transaction so that the storage
+    /// divisor rounds once against the total — rounding each drain separately would let a
+    /// transaction write in sub-divisor increments and be charged nothing for any of them.
+    pub fn accumulate_storage_bytes(&mut self, bytes: u64) {
+        self.accumulated_storage_bytes = self.accumulated_storage_bytes.saturating_add(bytes);
+    }
+
+    pub fn accumulated_storage_bytes(&self) -> u64 {
+        self.accumulated_storage_bytes
+    }
+
     pub fn accumulate_wasm_points(&mut self, points: u64) {
         self.accumulated_wasm_points = self.accumulated_wasm_points.saturating_add(points);
     }

crates/engine/src/runtime/impl.rs

@@ -3361,6 +3361,33 @@ where
         })
     }
 
+    fn charge_storage_written(&mut self) -> Result<(), RuntimeError> {
+        let bytes = self.tracker.take_storage_bytes_written()?;
+        if bytes > 0 {
+            for module in self.modules.iter() {
+                module.on_storage_written(&mut self.tracker, bytes)?;
+            }
+        }
+
+        // Within the fee intent a transaction may still be sourcing its fee, so its writes are not
+        // tested against payments here — `checkpoint_fee_intent` does that once the intent ends, and
+        // now sees the storage those instructions accrued. Past the checkpoint the payment is what it
+        // is, and charges only grow, so a transaction that is already over cannot come back.
+        if self.tracker.is_fee_state_dry_run() || !self.tracker.has_fee_checkpoint() {
+            return Ok(());
+        }
+
+        let charges = self.tracker.total_fee_charges();
+        let payments = self.tracker.total_fee_payments();
+        if payments < charges {
+            return Err(RuntimeError::InsufficientFeesPaid {
+                required_fee: charges,
+                fees_paid: payments,
+            });
+        }
+        Ok(())
+    }
+
     fn checkpoint_fee_intent(&mut self) -> Result<(), RuntimeError> {
         if !self.tracker.is_fee_state_dry_run() && self.tracker.total_fee_payments() < self.tracker.total_fee_charges()
         {

crates/engine/src/runtime/mod.rs

@@ -191,6 +191,11 @@ pub trait RuntimeInterface {
         max_amount: Option<Amount>,
     ) -> Result<(), RuntimeError>;
 
+    /// Meters the substate growth since the last call and charges for it. Called at instruction
+    /// boundaries, so a transaction that writes more state than it has paid for stops at the next
+    /// boundary instead of running to completion and being rejected at finalization.
+    fn charge_storage_written(&mut self) -> Result<(), RuntimeError>;
+
     fn checkpoint_fee_intent(&mut self) -> Result<(), RuntimeError>;
     fn finalize(&mut self) -> Result<FinalizeResult, RuntimeError>;
     fn finalize_failure(&mut self, reason: RejectReason) -> Result<FinalizeResult, RuntimeError>;

crates/engine/src/runtime/module.rs

@@ -32,6 +32,12 @@ pub trait RuntimeModule<TStore>: Send + Sync {
         Ok(())
     }
 
+    /// Invoked at instruction boundaries with the encoded bytes of substate growth since the last
+    /// call, so that storage can be charged as a transaction writes rather than only once it ends.
+    fn on_storage_written(&self, _track: &mut StateTracker<TStore>, _bytes: u64) -> Result<(), RuntimeModuleError> {
+        Ok(())
+    }
+
     /// Invoked at the start of finalization, against the live working state — before it is known
     /// whether the transaction commits or falls back to a fee-intent commit. Charges added here
     /// decide that outcome: they are what the paid-in-full check sees.

crates/engine/src/runtime/state_store.rs

@@ -4,6 +4,7 @@
 use std::{collections::HashMap, mem};
 
 use indexmap::{IndexMap, IndexSet};
+use tari_bor::{ByteCounter, encode_into_writer};
 use tari_engine_types::{
     Utxo,
     component::Component,
@@ -36,6 +37,18 @@ pub struct WorkingStateStore<TStore> {
 
     downed_utxos: IndexSet<UtxoAddress>,
     downed_confidential_outputs: IndexSet<ConfidentialOutputAddress>,
+
+    /// Substates written since storage was last metered. Only these are re-encoded when the running
+    /// tally is drained, so a transaction pays the encoding cost of what it touched rather than of
+    /// everything it has ever touched.
+    ///
+    /// Entries are added wherever a caller *could* have written, not where one provably did: an
+    /// over-marked substate re-encodes to the same size and contributes nothing.
+    written_since_metered: IndexSet<SubstateId>,
+    /// Encoded size each substate has already been metered for, so that repeated writes to one
+    /// substate are charged for its growth rather than for its whole size again.
+    metered_bytes: HashMap<SubstateId, usize>,
+
     /// The underlying state store that is used to load substates that are not in the working state maps.
     state_store: TStore,
 }
@@ -48,10 +61,44 @@ impl<TStore: StateReader> WorkingStateStore<TStore> {
             locked_substates: LockedSubstates::default(),
             downed_utxos: IndexSet::default(),
             downed_confidential_outputs: IndexSet::default(),
+            written_since_metered: IndexSet::default(),
+            metered_bytes: HashMap::new(),
             state_store,
         }
     }
 
+    fn mark_written(&mut self, id: &SubstateId) {
+        self.written_since_metered.insert(id.clone());
+    }
+
+    /// Encoded bytes newly occupied by substates written since the last call.
+    ///
+    /// Only growth counts: a substate that shrank does not give bytes back. The running tally exists
+    /// to bound what a transaction may write before it has paid, not to settle the bill — the charge
+    /// finalization computes over the persisted state replaces it.
+    pub fn take_storage_bytes_written(&mut self) -> Result<u64, RuntimeError> {
+        let mut accrued = 0u64;
+        for id in mem::take(&mut self.written_since_metered) {
+            let Some(substate) = self.new_substates.get(&id) else {
+                // Written and then removed again — a UTXO downed after being mutated, say. Nothing
+                // is left to persist, and the bytes it did occupy are not clawed back.
+                continue;
+            };
+            let mut counter = ByteCounter::new();
+            encode_into_writer(substate, &mut counter).map_err(|e| RuntimeError::InvariantError {
+                function: "WorkingStateStore::take_storage_bytes_written",
+                details: format!("Failed to encode substate {id}: {e}"),
+            })?;
+            let size = counter.get();
+            let metered = self.metered_bytes.entry(id).or_insert(0);
+            if size > *metered {
+                accrued = accrued.saturating_add((size - *metered) as u64);
+                *metered = size;
+            }
+        }
+        Ok(accrued)
+    }
+
     pub fn try_lock(&mut self, id: SubstateId, lock_flag: LockFlag) -> Result<LockId, RuntimeError> {
         if !self.exists(&id)? {
             return Err(RuntimeError::SubstateNotFound { id: id.clone() });
@@ -87,8 +134,9 @@ impl<TStore: StateReader> WorkingStateStore<TStore> {
         if let Some(mut substate) = self.loaded_substates.remove(lock.substate_id()) {
             return match callback(lock.substate_id(), substate.substate_value_mut())? {
                 Some(ret) => {
-                    self.new_substates
-                        .insert(lock.substate_id().clone(), substate.into_substate_value());
+                    let id = lock.substate_id().clone();
+                    self.new_substates.insert(id.clone(), substate.into_substate_value());
+                    self.mark_written(&id);
                     Ok(Some(ret))
                 },
                 None => {
@@ -108,7 +156,10 @@ impl<TStore: StateReader> WorkingStateStore<TStore> {
                 })?;
 
         // Since the substate is already mutated, we don't really care if the callback mutates it again or not
-        callback(lock.substate_id(), substate_mut)
+        let id = lock.substate_id().clone();
+        let ret = callback(&id, substate_mut);
+        self.mark_written(&id);
+        ret
     }
 
     pub fn get_locked_substate(&self, lock_id: LockId) -> Result<(SubstateId, &SubstateValue), RuntimeError> {
@@ -132,6 +183,10 @@ impl<TStore: StateReader> WorkingStateStore<TStore> {
                 .insert(address.clone(), substate.into_substate_value());
         }
 
+        // The caller holds a write lock and receives a mutable reference, so the write completes out
+        // of sight of this store. Mark it here — the size is read back when storage is next metered.
+        self.written_since_metered.insert(address.clone());
+
         if let Some(substate_mut) = self.new_substates.get_mut(address) {
             return Ok(substate_mut);
         }
@@ -152,7 +207,8 @@ impl<TStore: StateReader> WorkingStateStore<TStore> {
         if self.exists(&id)? {
             return Err(RuntimeError::DuplicateSubstate { address: id });
         }
-        self.new_substates.insert(id, value);
+        self.new_substates.insert(id.clone(), value);
+        self.mark_written(&id);
         Ok(())
     }
 

crates/engine/src/runtime/tracker.rs

@@ -493,6 +493,16 @@ impl<TStore: StateReader> StateTracker<TStore> {
         })
     }
 
+    /// Whether the fee intent has ended. Past that point the transaction has paid what its fee
+    /// instructions charged, and any further work is funded by that payment alone.
+    pub fn has_fee_checkpoint(&self) -> bool {
+        self.fee_checkpoint.is_some()
+    }
+
+    pub fn take_storage_bytes_written(&mut self) -> Result<u64, RuntimeError> {
+        self.write_with(|state| state.take_storage_bytes_written())
+    }
+
     /// The live working state, as a module may charge against it.
     ///
     /// The fee module uses this to compute its finalization charges before the state to persist has

crates/engine/src/runtime/working_state.rs

@@ -1203,6 +1203,10 @@ impl<TStore: StateReader> WorkingState<TStore> {
         self.last_instruction_output = Some(output);
     }
 
+    pub fn take_storage_bytes_written(&mut self) -> Result<u64, RuntimeError> {
+        self.store.take_storage_bytes_written()
+    }
+
     /// Counts substates in the to-persist set that did not previously exist in the state store.
     /// Used by the fee module to charge a slot-allocation premium on top of per-byte storage.
     pub fn count_newly_created_substates(&self) -> Result<usize, RuntimeError> {

crates/engine/src/transaction/processor.rs

@@ -325,8 +325,15 @@ where
             .into_iter()
             .enumerate()
             .map(|(idx, instruction)| {
-                Self::process_instruction(template_provider, &mut runtime, instruction, blobs)
-                    .map_err(|e| TransactionError::new(idx + 1, e))
+                let result = Self::process_instruction(template_provider, &mut runtime, instruction, blobs)
+                    .map_err(|e| TransactionError::new(idx + 1, e))?;
+                // Charge for what this instruction wrote before starting the next one, so a
+                // transaction that cannot pay for the state it is producing stops here.
+                runtime
+                    .interface_mut()
+                    .charge_storage_written()
+                    .map_err(|e| TransactionError::new(idx + 1, e.into()))?;
+                Ok(result)
             })
             .collect();
 

crates/engine/tests/fees.rs

@@ -266,6 +266,8 @@ fn an_unaffordable_fee_intent_commits_nothing() {
         )
         .unwrap();
 
+    // Rejected outright rather than committed as a fee intent: the checkpoint is what a fee-intent
+    // commit falls back to, so a fee intent that cannot pay for its own writes leaves nothing behind.
     assert!(
         matches!(
             result.finalize.result,