First insertion of the composed operations code.

Author: MathieuDutSik

linera-chain/src/block_tracker.rs

@@ -170,13 +170,20 @@ impl<'resources, 'blobs> BlockExecutionTracker<'resources, 'blobs> {
 
     /// Returns a new TransactionTracker for the current transaction.
     fn new_transaction_tracker(&mut self) -> Result<TransactionTracker, ChainError> {
+        // Convert operation results to Vec<Vec<u8>>
+        let previous_operation_results = self.operation_results
+            .iter()
+            .map(|result| result.0.clone())
+            .collect();
+            
         Ok(TransactionTracker::new(
             self.local_time,
             self.transaction_index,
             self.next_application_index,
             self.next_chain_index,
             self.oracle_responses()?,
             &self.blobs,
+            previous_operation_results,
         ))
     }
 

linera-chain/src/data_types.rs

@@ -181,11 +181,14 @@ impl From<&Operation> for OperationMetadata {
             },
             Operation::User {
                 application_id,
-                bytes,
+                input,
             } => OperationMetadata {
                 operation_type: "User".to_string(),
                 application_id: Some(*application_id),
-                user_bytes_hex: Some(hex::encode(bytes)),
+                user_bytes_hex: Some(match input {
+                    linera_execution::OperationInput::Direct(bytes) => hex::encode(bytes),
+                    linera_execution::OperationInput::Composed => "composed".to_string(),
+                }),
                 system_bytes_hex: None,
             },
         }

linera-chain/src/unit_tests/chain_tests.rs

@@ -26,7 +26,7 @@ use linera_execution::{
     committee::{Committee, ValidatorState},
     test_utils::{ExpectedCall, MockApplication},
     BaseRuntime, ContractRuntime, ExecutionError, ExecutionRuntimeConfig, ExecutionRuntimeContext,
-    Operation, ResourceControlPolicy, ServiceRuntime, SystemOperation, TestExecutionRuntimeContext,
+    Operation, OperationInput, ResourceControlPolicy, ServiceRuntime, SystemOperation, TestExecutionRuntimeContext,
 };
 use linera_views::{
     context::{Context as _, MemoryContext, ViewContext},
@@ -313,11 +313,11 @@ async fn test_application_permissions() -> anyhow::Result<()> {
     application.expect_call(ExpectedCall::default_finalize());
     let app_operation = Operation::User {
         application_id,
-        bytes: b"foo".to_vec(),
+        input: OperationInput::Direct(b"foo".to_vec()),
     };
     let another_app_operation = Operation::User {
         application_id: another_app_id,
-        bytes: b"bar".to_vec(),
+        input: OperationInput::Direct(b"bar".to_vec()),
     };
 
     let valid_block = make_first_block(chain_id)
@@ -740,7 +740,7 @@ async fn prepare_test_with_dummy_mock_application(
 
     let block = make_first_block(chain_id).with_operation(Operation::User {
         application_id,
-        bytes: vec![],
+        input: OperationInput::Direct(vec![]),
     });
 
     Ok((application, application_id, chain, block, time))

linera-client/src/benchmark.rs

@@ -19,7 +19,7 @@ use linera_core::{
     client::{ChainClient, ChainClientError},
     Environment,
 };
-use linera_execution::{system::SystemOperation, Operation};
+use linera_execution::{system::SystemOperation, Operation, OperationInput};
 use linera_sdk::abis::fungible::{self, FungibleOperation};
 use num_format::{Locale, ToFormattedString};
 use prometheus_parse::{HistogramCount, Scrape, Value};
@@ -730,7 +730,7 @@ impl<Env: Environment> Benchmark<Env> {
         .expect("should serialize fungible token operation");
         Operation::User {
             application_id,
-            bytes,
+            input: OperationInput::Direct(bytes),
         }
     }
 }

linera-core/src/unit_tests/wasm_worker_tests.rs

@@ -28,7 +28,7 @@ use linera_chain::{
 };
 use linera_execution::{
     system::SystemOperation, test_utils::SystemExecutionState, ExecutionRuntimeContext, Operation,
-    OperationContext, ResourceController, TransactionTracker, WasmContractModule, WasmRuntime,
+    OperationInput, OperationContext, ResourceController, TransactionTracker, WasmContractModule, WasmRuntime,
 };
 use linera_storage::{DbStorage, Storage};
 #[cfg(feature = "dynamodb")]
@@ -258,7 +258,7 @@ where
         .with_timestamp(3)
         .with_operation(Operation::User {
             application_id,
-            bytes: user_operation.clone(),
+            input: OperationInput::Direct(user_operation.clone()),
         });
     let operation_context = OperationContext {
         chain_id: creator_chain.id(),
@@ -273,7 +273,7 @@ where
             operation_context,
             Operation::User {
                 application_id,
-                bytes: user_operation,
+                input: OperationInput::Direct(user_operation),
             },
             &mut TransactionTracker::new(
                 Timestamp::from(3),
@@ -282,6 +282,7 @@ where
                 0,
                 Some(vec![OracleResponse::Blob(application_description_blob_id)]),
                 &[],
+                vec![],
             ),
             &mut controller,
         )

linera-execution/src/evm/revm.rs

@@ -38,7 +38,7 @@ use crate::{
         database::{DatabaseRuntime, StorageStats, EVM_SERVICE_GAS_LIMIT},
     },
     BaseRuntime, ContractRuntime, ContractSyncRuntimeHandle, DataBlobHash, EvmExecutionError,
-    EvmRuntime, ExecutionError, ServiceRuntime, ServiceSyncRuntimeHandle, UserContract,
+    EvmRuntime, ExecutionError, OperationInput, ServiceRuntime, ServiceSyncRuntimeHandle, UserContract,
     UserContractInstance, UserContractModule, UserService, UserServiceInstance, UserServiceModule,
 };
 
@@ -1449,7 +1449,8 @@ where
             EvmQuery::Query(vec) => vec,
             EvmQuery::Mutation(operation) => {
                 let mut runtime = self.db.runtime.lock().expect("The lock should be possible");
-                runtime.schedule_operation(operation)?;
+                let input = OperationInput::Direct(operation);
+                runtime.schedule_operation(input)?;
                 return Ok(Vec::new());
             }
         };

linera-execution/src/execution.rs

@@ -132,6 +132,7 @@ where
             next_chain_index,
             None,
             &[],
+            vec![],  // No previous operation results for testing
         );
         txn_tracker.add_created_blob(blob);
         self.run_user_action(
@@ -310,8 +311,24 @@ where
             }
             Operation::User {
                 application_id,
-                bytes,
+                input,
             } => {
+                // Get the bytes for the operation based on the input type
+                let bytes = match input {
+                    crate::OperationInput::Direct(bytes) => bytes.clone(),
+                    crate::OperationInput::Composed => {
+                        // For composed operations, use the result from the last operation
+                        let previous_results = txn_tracker.previous_operation_results();
+                        if previous_results.is_empty() {
+                            return Err(ExecutionError::InternalError(
+                                "Composed operation requires a previous operation result, but none found"
+                            ));
+                        }
+                        // Get the last operation result as input
+                        previous_results.last().unwrap().clone()
+                    }
+                };
+                
                 self.run_user_action(
                     application_id,
                     UserAction::Operation(context, bytes),

linera-execution/src/lib.rs

@@ -49,6 +49,7 @@ use linera_base::{
     vm::VmRuntime,
 };
 use linera_views::{batch::Batch, ViewError};
+use linera_witty::{WitLoad, WitStore, WitType};
 use serde::{Deserialize, Serialize};
 use system::AdminOperation;
 use thiserror::Error;
@@ -679,7 +680,7 @@ pub trait ServiceRuntime: BaseRuntime {
     ) -> Result<Vec<u8>, ExecutionError>;
 
     /// Schedules an operation to be included in the block proposed after execution.
-    fn schedule_operation(&mut self, operation: Vec<u8>) -> Result<(), ExecutionError>;
+    fn schedule_operation(&mut self, input: OperationInput) -> Result<(), ExecutionError>;
 
     /// Checks if the service has exceeded its execution time limit.
     fn check_execution_time(&mut self) -> Result<(), ExecutionError>;
@@ -817,17 +818,29 @@ pub trait ContractRuntime: BaseRuntime {
     fn write_batch(&mut self, batch: Batch) -> Result<(), ExecutionError>;
 }
 
+/// Input for an operation, which can be either direct bytes or a composed operation.
+#[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize, WitType, WitLoad, WitStore)]
+pub enum OperationInput {
+    /// Direct input as bytes.
+    Direct(
+        #[serde(with = "serde_bytes")]
+        #[debug(with = "hex_debug")]
+        Vec<u8>
+    ),
+    /// Composed input that uses the result from the previous operation in the block.
+    /// If operation_results is empty, an error should be raised during execution.
+    Composed,
+}
+
 /// An operation to be executed in a block.
 #[derive(Debug, PartialEq, Eq, Hash, Clone, Serialize, Deserialize)]
 pub enum Operation {
     /// A system operation.
     System(Box<SystemOperation>),
-    /// A user operation (in serialized form).
+    /// A user operation.
     User {
         application_id: ApplicationId,
-        #[serde(with = "serde_bytes")]
-        #[debug(with = "hex_debug")]
-        bytes: Vec<u8>,
+        input: OperationInput,
     },
 }
 
@@ -1176,10 +1189,20 @@ impl Operation {
     ) -> Result<Self, bcs::Error> {
         Ok(Operation::User {
             application_id,
-            bytes: bcs::to_bytes(&operation)?,
+            input: OperationInput::Direct(bcs::to_bytes(&operation)?),
         })
     }
 
+    /// Creates a new composed user application operation that uses the result from the 
+    /// previous operation as input.
+    #[cfg(with_testing)]
+    pub fn user_composed(application_id: ApplicationId) -> Self {
+        Operation::User {
+            application_id,
+            input: OperationInput::Composed,
+        }
+    }
+
     /// Returns a reference to the [`SystemOperation`] in this [`Operation`], if this [`Operation`]
     /// is for the system application.
     pub fn as_system_operation(&self) -> Option<&SystemOperation> {

linera-execution/src/resources.rs

@@ -238,8 +238,16 @@ where
         self.update_balance(self.policy.operation)?;
         match operation {
             Operation::System(_) => Ok(()),
-            Operation::User { bytes, .. } => {
-                let size = bytes.len();
+            Operation::User { input, .. } => {
+                let size = match input {
+                    crate::OperationInput::Direct(bytes) => bytes.len(),
+                    crate::OperationInput::Composed => {
+                        // For composed operations, we don't know the size until execution time
+                        // Since the input comes from the previous operation result.
+                        // We'll use 0 for now but this might need to be revisited.
+                        0
+                    }
+                };
                 self.tracker.as_mut().operation_bytes = self
                     .tracker
                     .as_mut()

linera-execution/src/runtime.rs

@@ -33,7 +33,7 @@ use crate::{
     util::{ReceiverExt, UnboundedSenderExt},
     ApplicationDescription, ApplicationId, BaseRuntime, ContractRuntime, DataBlobHash,
     ExecutionError, FinalizeContext, Message, MessageContext, MessageKind, ModuleId, Operation,
-    OutgoingMessage, QueryContext, QueryOutcome, ServiceRuntime, TransactionTracker,
+    OperationInput, OutgoingMessage, QueryContext, QueryOutcome, ServiceRuntime, TransactionTracker,
     UserContractCode, UserContractInstance, UserServiceCode, UserServiceInstance,
     MAX_STREAM_NAME_LEN,
 };
@@ -1809,13 +1809,13 @@ impl ServiceRuntime for ServiceSyncRuntimeHandle {
         Ok(response)
     }
 
-    fn schedule_operation(&mut self, operation: Vec<u8>) -> Result<(), ExecutionError> {
+    fn schedule_operation(&mut self, input: OperationInput) -> Result<(), ExecutionError> {
         let mut this = self.inner();
         let application_id = this.current_application().id;
 
         this.scheduled_operations.push(Operation::User {
             application_id,
-            bytes: operation,
+            input,
         });
 
         Ok(())