Account for published blobs in the resource at the end of transaction execution rather than upfront. (#3995)

## Motivation

We are charging for all published blobs before we execute any operation
in the block. This is wrong b/c we want the incoming grant to be able to
cover the fees. The grants are processed during operation processing
only so they are not taken into account at the beginning of the block
processing.

## Proposal

Move the published blobs' resources tracking to the end of processing a
transaction that published it. At this point any incoming grant is
already processed. In order to do so, we track blob IDs referenced
during transaction execution via `TransactionTracker.published_blobs`
and consume it in `BlockExecutionTracker::process_txn_outcome`.

Also, charge for publishing user-created blobs only.
`BlobType::Committee` is not accounted for.

## Test Plan

CI.

## Release Plan

- Nothing to do / These changes follow the usual release cycle.

## Links

- [reviewer
checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)

Author: deuszx

linera-base/src/data_types.rs

@@ -1374,6 +1374,11 @@ impl Blob {
     pub async fn load_data_blob_from_file(path: impl AsRef<Path>) -> io::Result<Self> {
         Ok(Self::new_data(fs::read(path)?))
     }
+
+    /// Returns whether the blob is of [`BlobType::Committee`] variant.
+    pub fn is_committee_blob(&self) -> bool {
+        self.content().blob_type().is_committee_blob()
+    }
 }
 
 impl Serialize for Blob {

linera-base/src/identifiers.rs

@@ -169,6 +169,21 @@ pub enum BlobType {
     ChainDescription,
 }
 
+impl BlobType {
+    /// Returns whether the blob is of [`BlobType::Committee`] variant.
+    pub fn is_committee_blob(&self) -> bool {
+        match self {
+            BlobType::Data
+            | BlobType::ContractBytecode
+            | BlobType::ServiceBytecode
+            | BlobType::EvmBytecode
+            | BlobType::ApplicationDescription
+            | BlobType::ChainDescription => false,
+            BlobType::Committee => true,
+        }
+    }
+}
+
 impl fmt::Display for BlobType {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "{:?}", self)

linera-chain/src/block_tracker.rs

@@ -1,12 +1,12 @@
 // Copyright (c) Zefchain Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::collections::BTreeSet;
+use std::collections::{BTreeMap, BTreeSet};
 
 use custom_debug_derive::Debug;
 use linera_base::{
     data_types::{Blob, Event, OracleResponse, Timestamp},
-    identifiers::{AccountOwner, BlobType, ChainId},
+    identifiers::{AccountOwner, BlobId, ChainId},
 };
 use linera_execution::{
     OutgoingMessage, ResourceController, ResourceTracker, SystemExecutionStateView,
@@ -23,7 +23,7 @@ use crate::{
 /// Tracks execution of transactions within a block.
 /// Captures the resource policy, produced messages, oracle responses and events.
 #[derive(Debug)]
-pub struct BlockExecutionTracker<'resources> {
+pub struct BlockExecutionTracker<'resources, 'blobs> {
     resource_controller: &'resources mut ResourceController<Option<AccountOwner>, ResourceTracker>,
     local_time: Timestamp,
     #[debug(skip_if = Option::is_none)]
@@ -44,17 +44,21 @@ pub struct BlockExecutionTracker<'resources> {
     // Index of the currently executed transaction in a block.
     transaction_index: u32,
 
+    // Blobs published in the block.
+    published_blobs: BTreeMap<BlobId, &'blobs Blob>,
+
     // We expect the number of outcomes to be equal to the number of transactions in the block.
     expected_outcomes_count: usize,
 }
 
-impl<'resources> BlockExecutionTracker<'resources> {
+impl<'resources, 'blobs> BlockExecutionTracker<'resources, 'blobs> {
     /// Creates a new BlockExecutionTracker.
     pub fn new(
         resource_controller: &'resources mut ResourceController<
             Option<AccountOwner>,
             ResourceTracker,
         >,
+        published_blobs: BTreeMap<BlobId, &'blobs Blob>,
         local_time: Timestamp,
         replaying_oracle_responses: Option<Vec<Vec<OracleResponse>>>,
         proposal: &ProposedBlock,
@@ -76,6 +80,7 @@ impl<'resources> BlockExecutionTracker<'resources> {
             messages: Vec::new(),
             operation_results: Vec::new(),
             transaction_index: 0,
+            published_blobs,
             expected_outcomes_count: proposal.incoming_bundles.len() + proposal.operations.len(),
         })
     }
@@ -149,11 +154,23 @@ impl<'resources> BlockExecutionTracker<'resources> {
             ))
             .with_execution_context(context)?;
 
+        // Account for blobs published by this transaction directly.
         for blob in &txn_outcome.blobs {
-            if blob.content().blob_type() == BlobType::Data {
+            resource_controller
+                .track_blob_published(blob)
+                .with_execution_context(context)?;
+        }
+
+        // Account for blobs published indirectly but referenced by the transaction.
+        for blob_id in &txn_outcome.blobs_published {
+            if let Some(blob) = self.published_blobs.get(blob_id) {
                 resource_controller
-                    .track_blob_published(blob.content())
+                    .track_blob_published(blob)
                     .with_execution_context(context)?;
+            } else {
+                return Err(ChainError::InternalError(format!(
+                    "Missing published blob {blob_id}"
+                )));
             }
         }
 

linera-chain/src/chain.rs

@@ -726,32 +726,28 @@ where
             block.authenticated_signer,
         );
 
+        for blob in published_blobs {
+            let blob_id = blob.id();
+            resource_controller
+                .policy()
+                .check_blob_size(blob.content())
+                .with_execution_context(ChainExecutionContext::Block)?;
+            chain.system.used_blobs.insert(&blob_id)?;
+        }
+
         // Execute each incoming bundle as a transaction, then each operation.
         // Collect messages, events and oracle responses, each as one list per transaction.
         let mut block_execution_tracker = BlockExecutionTracker::new(
             &mut resource_controller,
+            published_blobs
+                .iter()
+                .map(|blob| (blob.id(), blob))
+                .collect(),
             local_time,
             replaying_oracle_responses,
             block,
         )?;
 
-        for blob in published_blobs {
-            let blob_type = blob.content().blob_type();
-            if blob_type == BlobType::Data
-                || blob_type == BlobType::ContractBytecode
-                || blob_type == BlobType::ServiceBytecode
-                || blob_type == BlobType::EvmBytecode
-            {
-                block_execution_tracker
-                    .resource_controller_mut()
-                    .with_state(&mut chain.system)
-                    .await?
-                    .track_blob_published(blob.content())
-                    .with_execution_context(ChainExecutionContext::Block)?;
-            }
-            chain.system.used_blobs.insert(&blob.id())?;
-        }
-
         for transaction in block.transactions() {
             let chain_execution_context =
                 block_execution_tracker.chain_execution_context(&transaction);

linera-execution/src/resources.rs

@@ -7,7 +7,7 @@ use std::{sync::Arc, time::Duration};
 
 use custom_debug_derive::Debug;
 use linera_base::{
-    data_types::{Amount, ArithmeticError, BlobContent},
+    data_types::{Amount, ArithmeticError, Blob},
     ensure,
     identifiers::AccountOwner,
     vm::VmRuntime,
@@ -338,9 +338,12 @@ where
     }
 
     /// Tracks a number of blob bytes published.
-    pub fn track_blob_published(&mut self, content: &BlobContent) -> Result<(), ExecutionError> {
-        self.policy.check_blob_size(content)?;
-        let size = content.bytes().len() as u64;
+    pub fn track_blob_published(&mut self, blob: &Blob) -> Result<(), ExecutionError> {
+        self.policy.check_blob_size(blob.content())?;
+        let size = blob.content().bytes().len() as u64;
+        if blob.is_committee_blob() {
+            return Ok(());
+        }
         {
             let tracker = self.tracker.as_mut();
             tracker.blob_bytes_published = tracker

linera-execution/src/system.rs

@@ -430,7 +430,10 @@ where
                 );
                 match admin_operation {
                     AdminOperation::PublishCommitteeBlob { blob_hash } => {
-                        self.blob_published(&BlobId::new(blob_hash, BlobType::Committee))?;
+                        self.blob_published(
+                            &BlobId::new(blob_hash, BlobType::Committee),
+                            txn_tracker,
+                        )?;
                     }
                     AdminOperation::CreateCommittee { epoch, blob_hash } => {
                         self.check_next_epoch(epoch)?;
@@ -461,7 +464,7 @@ where
             }
             PublishModule { module_id } => {
                 for blob_id in module_id.bytecode_blob_ids() {
-                    self.blob_published(&blob_id)?;
+                    self.blob_published(&blob_id, txn_tracker)?;
                 }
             }
             CreateApplication {
@@ -488,7 +491,7 @@ where
                 new_application = Some((app_id, instantiation_argument));
             }
             PublishDataBlob { blob_hash } => {
-                self.blob_published(&BlobId::new(blob_hash, BlobType::Data))?;
+                self.blob_published(&BlobId::new(blob_hash, BlobType::Data), txn_tracker)?;
             }
             ReadBlob { blob_id } => {
                 let content = self.read_blob_content(blob_id).await?;
@@ -982,8 +985,13 @@ where
 
     /// Records a blob that is published in this block. This does not create an oracle entry, and
     /// the blob can be used without using an oracle in the future on this chain.
-    fn blob_published(&mut self, blob_id: &BlobId) -> Result<(), ExecutionError> {
+    fn blob_published(
+        &mut self,
+        blob_id: &BlobId,
+        txn_tracker: &mut TransactionTracker,
+    ) -> Result<(), ExecutionError> {
         self.used_blobs.insert(blob_id)?;
+        txn_tracker.add_published_blob(*blob_id);
         Ok(())
     }
 

linera-execution/src/transaction_tracker.rs

@@ -1,7 +1,10 @@
 // Copyright (c) Zefchain Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-use std::{collections::BTreeMap, mem, vec};
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    mem, vec,
+};
 
 use custom_debug_derive::Debug;
 use linera_base::{
@@ -34,11 +37,17 @@ pub struct TransactionTracker {
     /// Events recorded by contracts' `emit` calls.
     events: Vec<Event>,
     /// Blobs created by contracts.
+    ///
+    /// As of right now, blobs created by the contracts are one of the two:
+    /// - [`OpenChain`]
+    /// - [`CreateApplication`]
     blobs: BTreeMap<BlobId, Blob>,
     /// Operation result.
     operation_result: Option<Vec<u8>>,
     /// Streams that have been updated but not yet processed during this transaction.
     streams_to_process: BTreeMap<ApplicationId, AppStreamUpdates>,
+    /// Published blobs this transaction refers to by [`BlobId`].
+    blobs_published: BTreeSet<BlobId>,
 }
 
 /// The [`TransactionTracker`] contents after a transaction has finished.
@@ -57,6 +66,8 @@ pub struct TransactionOutcome {
     pub blobs: Vec<Blob>,
     /// Operation result.
     pub operation_result: Vec<u8>,
+    /// Blobs published by this transaction.
+    pub blobs_published: BTreeSet<BlobId>,
 }
 
 impl TransactionTracker {
@@ -146,6 +157,10 @@ impl TransactionTracker {
         self.blobs.insert(blob.id(), blob);
     }
 
+    pub fn add_published_blob(&mut self, blob_id: BlobId) {
+        self.blobs_published.insert(blob_id);
+    }
+
     pub fn created_blobs(&self) -> &BTreeMap<BlobId, Blob> {
         &self.blobs
     }
@@ -266,6 +281,7 @@ impl TransactionTracker {
             blobs,
             operation_result,
             streams_to_process,
+            blobs_published,
         } = self;
         ensure!(
             streams_to_process.is_empty(),
@@ -286,6 +302,7 @@ impl TransactionTracker {
             events,
             blobs: blobs.into_values().collect(),
             operation_result: operation_result.unwrap_or_default(),
+            blobs_published,
         })
     }
 }