Implement composed operations

examples/counter/src/contract.rs

@@ -45,11 +45,11 @@ impl Contract for CounterContract {
         self.state.value.set(value);
     }
 
-    async fn execute_operation(&mut self, operation: CounterOperation) -> u64 {
-        let CounterOperation::Increment(operation) = operation;
-        let new_value = self.state.value.get() + operation;
+    async fn execute_operation(&mut self, operation: CounterOperation) -> CounterOperation {
+        let CounterOperation::Increment(increment) = operation;
+        let new_value = self.state.value.get() + increment;
         self.state.value.set(new_value);
-        new_value
+        operation
     }
 
     async fn execute_message(&mut self, _message: ()) {
@@ -84,8 +84,8 @@ mod tests {
 
         let expected_value = initial_value + increment;
 
-        assert_eq!(response, expected_value);
-        assert_eq!(*counter.state.value.get(), initial_value + increment);
+        assert_eq!(response, CounterOperation::Increment(increment));
+        assert_eq!(*counter.state.value.get(), expected_value);
     }
 
     #[test]
@@ -115,7 +115,7 @@ mod tests {
 
         let expected_value = initial_value + increment;
 
-        assert_eq!(response, expected_value);
+        assert_eq!(response, CounterOperation::Increment(increment));
         assert_eq!(*counter.state.value.get(), expected_value);
     }
 

examples/counter/src/lib.rs

@@ -12,15 +12,15 @@ use serde::{Deserialize, Serialize};
 
 pub struct CounterAbi;
 
-#[derive(Debug, Deserialize, Serialize, GraphQLMutationRoot)]
+#[derive(Debug, Deserialize, PartialEq, Eq, Serialize, GraphQLMutationRoot)]
 pub enum CounterOperation {
     /// Increment the counter by the given value
     Increment(u64),
 }
 
 impl ContractAbi for CounterAbi {
     type Operation = CounterOperation;
-    type Response = u64;
+    type Response = CounterOperation;
 }
 
 impl ServiceAbi for CounterAbi {

linera-chain/src/block_tracker.rs

@@ -170,13 +170,21 @@ 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,8 @@ 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 +314,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 +741,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::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_client_tests.rs

@@ -166,8 +166,10 @@ where
 
     let increment = 5_u64;
     let counter_operation = counter::CounterOperation::Increment(increment);
+    let operation1 = Operation::user(application_id, &counter_operation)?;
+    let operation2 = Operation::user_composed(application_id.forget_abi());
     creator
-        .execute_operation(Operation::user(application_id, &counter_operation)?)
+        .execute_operations(vec![operation1, operation2], vec![])
         .await
         .unwrap();
 
@@ -177,9 +179,10 @@ where
         .await
         .unwrap();
 
+    // We do an increment by 5 and then immediately reiterate it.
     let expected = QueryOutcome {
         response: async_graphql::Response::new(
-            async_graphql::Value::from_json(json!({"value": 15})).unwrap(),
+            async_graphql::Value::from_json(json!({"value": 20})).unwrap(),
         ),
         operations: vec![],
     };

linera-core/src/unit_tests/wasm_worker_tests.rs

@@ -28,7 +28,8 @@ use linera_chain::{
 };
 use linera_execution::{
     system::SystemOperation, test_utils::SystemExecutionState, ExecutionRuntimeContext, Operation,
-    OperationContext, ResourceController, TransactionTracker, WasmContractModule, WasmRuntime,
+    OperationContext, OperationInput, ResourceController, TransactionTracker, WasmContractModule,
+    WasmRuntime,
 };
 use linera_storage::{DbStorage, Storage};
 #[cfg(feature = "dynamodb")]
@@ -258,7 +259,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 +274,7 @@ where
             operation_context,
             Operation::User {
                 application_id,
-                bytes: user_operation,
+                input: OperationInput::Direct(user_operation),
             },
             &mut TransactionTracker::new(
                 Timestamp::from(3),
@@ -282,6 +283,7 @@ where
                 0,
                 Some(vec![OracleResponse::Blob(application_description_blob_id)]),
                 &[],
+                vec![],
             ),
             &mut controller,
         )
@@ -291,6 +293,7 @@ where
         .system
         .used_blobs
         .insert(&application_description_blob_id)?;
+    let result = counter::CounterOperation::Increment(5);
     let run_block_proposal = ConfirmedBlock::new(
         BlockExecutionOutcome {
             messages: vec![Vec::new()],
@@ -300,7 +303,7 @@ where
             blobs: vec![Vec::new()],
             state_hash: creator_state.crypto_hash().await?,
             oracle_responses: vec![vec![]],
-            operation_results: vec![OperationResult(bcs::to_bytes(&15u64)?)],
+            operation_results: vec![OperationResult(bcs::to_bytes(&result)?)],
         }
         .with(run_block),
     );

linera-execution/src/evm/revm.rs

@@ -38,8 +38,9 @@ use crate::{
         database::{DatabaseRuntime, StorageStats, EVM_SERVICE_GAS_LIMIT},
     },
     BaseRuntime, ContractRuntime, ContractSyncRuntimeHandle, DataBlobHash, EvmExecutionError,
-    EvmRuntime, ExecutionError, ServiceRuntime, ServiceSyncRuntimeHandle, UserContract,
-    UserContractInstance, UserContractModule, UserService, UserServiceInstance, UserServiceModule,
+    EvmRuntime, ExecutionError, OperationInput, ServiceRuntime, ServiceSyncRuntimeHandle,
+    UserContract, UserContractInstance, UserContractModule, UserService, UserServiceInstance,
+    UserServiceModule,
 };
 
 /// This is the selector of the `execute_message` that should be called
@@ -1449,7 +1450,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,22 @@ 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::ComposedOperationCannotBeFirst);
+                        }
+                        // 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;
@@ -210,7 +211,8 @@ pub enum ExecutionError {
     DecompressionError(#[from] DecompressionError),
     #[error("The given promise is invalid or was polled once already")]
     InvalidPromise,
-
+    #[error("Composed operation cannot be first")]
+    ComposedOperationCannotBeFirst,
     #[error("Attempted to perform a reentrant call to application {0}")]
     ReentrantCall(ApplicationId),
     #[error(
@@ -679,7 +681,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 +819,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 +1190,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()