Skip to main content

linera_execution/
runtime.rs

1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::{
5    collections::{hash_map, BTreeMap, HashMap, HashSet},
6    mem,
7    ops::{Deref, DerefMut},
8    sync::{Arc, Mutex},
9};
10
11use custom_debug_derive::Debug;
12use linera_base::{
13    data_types::{
14        Amount, ApplicationPermissions, ArithmeticError, Blob, BlockHeight, Bytecode,
15        SendMessageRequest, Timestamp,
16    },
17    ensure, http,
18    identifiers::{
19        Account, AccountOwner, ChainId, EventId, GenericApplicationId, StreamId, StreamName,
20    },
21    ownership::ChainOwnership,
22    time::Instant,
23    vm::VmRuntime,
24};
25use linera_views::batch::Batch;
26use oneshot::Receiver;
27use tracing::instrument;
28
29use crate::{
30    execution::UserAction,
31    execution_state_actor::{ExecutionRequest, ExecutionStateSender},
32    resources::ResourceController,
33    system::CreateApplicationResult,
34    util::{ReceiverExt, UnboundedSenderExt},
35    ApplicationDescription, ApplicationId, BaseRuntime, ContractRuntime, DataBlobHash,
36    ExecutionError, FinalizeContext, Message, MessageContext, MessageKind, ModuleId, Operation,
37    OutgoingMessage, QueryContext, QueryOutcome, ServiceRuntime, UserContractCode,
38    UserContractInstance, UserServiceCode, UserServiceInstance, MAX_STREAM_NAME_LEN,
39};
40
41#[cfg(test)]
42#[path = "unit_tests/runtime_tests.rs"]
43mod tests;
44
45pub trait WithContext {
46    type UserContext;
47    type Code;
48}
49
50impl WithContext for UserContractInstance {
51    type UserContext = Timestamp;
52    type Code = UserContractCode;
53}
54
55impl WithContext for UserServiceInstance {
56    type UserContext = ();
57    type Code = UserServiceCode;
58}
59
60#[cfg(test)]
61impl WithContext for Arc<dyn std::any::Any + Send + Sync> {
62    type UserContext = ();
63    type Code = ();
64}
65
66#[derive(Debug)]
67pub struct SyncRuntime<UserInstance: WithContext>(Option<SyncRuntimeHandle<UserInstance>>);
68
69pub type ContractSyncRuntime = SyncRuntime<UserContractInstance>;
70
71/// The synchronous runtime used to execute service queries.
72pub struct ServiceSyncRuntime {
73    runtime: SyncRuntime<UserServiceInstance>,
74    current_context: QueryContext,
75}
76
77#[derive(Debug)]
78pub struct SyncRuntimeHandle<UserInstance: WithContext>(
79    Arc<Mutex<SyncRuntimeInternal<UserInstance>>>,
80);
81
82/// A handle to the synchronous runtime used when executing contracts.
83pub type ContractSyncRuntimeHandle = SyncRuntimeHandle<UserContractInstance>;
84/// A handle to the synchronous runtime used when executing services.
85pub type ServiceSyncRuntimeHandle = SyncRuntimeHandle<UserServiceInstance>;
86
87/// Runtime data tracked during the execution of a transaction on the synchronous thread.
88#[derive(Debug)]
89pub struct SyncRuntimeInternal<UserInstance: WithContext> {
90    /// The current chain ID.
91    chain_id: ChainId,
92    /// The height of the next block that will be added to this chain. During operations
93    /// and messages, this is the current block height.
94    height: BlockHeight,
95    /// The current consensus round. Only available during block validation in multi-leader rounds.
96    round: Option<u32>,
97    /// The current message being executed, if there is one.
98    #[debug(skip_if = Option::is_none)]
99    executing_message: Option<ExecutingMessage>,
100
101    /// How to interact with the storage view of the execution state.
102    execution_state_sender: ExecutionStateSender,
103
104    /// If applications are being finalized.
105    ///
106    /// If [`true`], disables cross-application calls.
107    is_finalizing: bool,
108    /// Applications that need to be finalized.
109    applications_to_finalize: Vec<ApplicationId>,
110
111    /// Application instances loaded in this transaction.
112    preloaded_applications: HashMap<ApplicationId, (UserInstance::Code, ApplicationDescription)>,
113    /// Application instances loaded in this transaction.
114    loaded_applications: HashMap<ApplicationId, LoadedApplication<UserInstance>>,
115    /// The current stack of application descriptions.
116    call_stack: Vec<ApplicationStatus>,
117    /// The set of the IDs of the applications that are in the `call_stack`.
118    active_applications: HashSet<ApplicationId>,
119    /// The operations scheduled during this query.
120    scheduled_operations: Vec<Operation>,
121
122    /// Track application states based on views.
123    view_user_states: BTreeMap<ApplicationId, ViewUserState>,
124
125    /// The deadline this runtime should finish executing.
126    ///
127    /// Used to limit the execution time of services running as oracles.
128    deadline: Option<Instant>,
129
130    /// Where to send a refund for the unused part of the grant after execution, if any.
131    #[debug(skip_if = Option::is_none)]
132    refund_grant_to: Option<Account>,
133    /// Controller to track fuel and storage consumption.
134    resource_controller: ResourceController,
135    /// Additional context for the runtime.
136    user_context: UserInstance::UserContext,
137    /// Whether contract log messages should be output.
138    allow_application_logs: bool,
139}
140
141/// The runtime status of an application.
142#[derive(Debug)]
143struct ApplicationStatus {
144    /// The caller application ID, if forwarded during the call.
145    caller_id: Option<ApplicationId>,
146    /// The application ID.
147    id: ApplicationId,
148    /// The application description.
149    description: ApplicationDescription,
150    /// The authenticated owner for the execution thread, if any.
151    signer: Option<AccountOwner>,
152}
153
154/// A loaded application instance.
155#[derive(Debug)]
156struct LoadedApplication<Instance> {
157    instance: Arc<Mutex<Instance>>,
158    description: ApplicationDescription,
159}
160
161impl<Instance> LoadedApplication<Instance> {
162    /// Creates a new [`LoadedApplication`] entry from the `instance` and its `description`.
163    fn new(instance: Instance, description: ApplicationDescription) -> Self {
164        LoadedApplication {
165            instance: Arc::new(Mutex::new(instance)),
166            description,
167        }
168    }
169}
170
171impl<Instance> Clone for LoadedApplication<Instance> {
172    // Manual implementation is needed to prevent the derive macro from adding an `Instance: Clone`
173    // bound
174    fn clone(&self) -> Self {
175        LoadedApplication {
176            instance: self.instance.clone(),
177            description: self.description.clone(),
178        }
179    }
180}
181
182#[derive(Debug)]
183enum Promise<T> {
184    Ready(T),
185    Pending(Receiver<T>),
186}
187
188impl<T> Promise<T> {
189    fn force(&mut self) -> Result<(), ExecutionError> {
190        if let Promise::Pending(receiver) = self {
191            let value = receiver
192                .recv_ref()
193                .map_err(|oneshot::RecvError| ExecutionError::MissingRuntimeResponse)?;
194            *self = Promise::Ready(value);
195        }
196        Ok(())
197    }
198
199    fn read(self) -> Result<T, ExecutionError> {
200        match self {
201            Promise::Pending(receiver) => {
202                let value = receiver.recv_response()?;
203                Ok(value)
204            }
205            Promise::Ready(value) => Ok(value),
206        }
207    }
208}
209
210/// Manages a set of pending queries returning values of type `T`.
211#[derive(Debug, Default)]
212struct QueryManager<T> {
213    /// The queries in progress.
214    pending_queries: BTreeMap<u32, Promise<T>>,
215    /// The number of queries ever registered so far. Used for the index of the next query.
216    query_count: u32,
217    /// The number of active queries.
218    active_query_count: u32,
219}
220
221impl<T> QueryManager<T> {
222    fn register(&mut self, receiver: Receiver<T>) -> Result<u32, ExecutionError> {
223        let id = self.query_count;
224        self.pending_queries.insert(id, Promise::Pending(receiver));
225        self.query_count = self
226            .query_count
227            .checked_add(1)
228            .ok_or(ArithmeticError::Overflow)?;
229        self.active_query_count = self
230            .active_query_count
231            .checked_add(1)
232            .ok_or(ArithmeticError::Overflow)?;
233        Ok(id)
234    }
235
236    fn wait(&mut self, id: u32) -> Result<T, ExecutionError> {
237        let promise = self
238            .pending_queries
239            .remove(&id)
240            .ok_or(ExecutionError::InvalidPromise)?;
241        let value = promise.read()?;
242        self.active_query_count -= 1;
243        Ok(value)
244    }
245
246    fn force_all(&mut self) -> Result<(), ExecutionError> {
247        for promise in self.pending_queries.values_mut() {
248            promise.force()?;
249        }
250        Ok(())
251    }
252}
253
254type Keys = Vec<Vec<u8>>;
255type Value = Vec<u8>;
256type KeyValues = Vec<(Vec<u8>, Vec<u8>)>;
257
258#[derive(Debug, Default)]
259struct ViewUserState {
260    /// The contains-key queries in progress.
261    contains_key_queries: QueryManager<bool>,
262    /// The contains-keys queries in progress.
263    contains_keys_queries: QueryManager<Vec<bool>>,
264    /// The read-value queries in progress.
265    read_value_queries: QueryManager<Option<Value>>,
266    /// The read-multi-values queries in progress.
267    read_multi_values_queries: QueryManager<Vec<Option<Value>>>,
268    /// The find-keys queries in progress.
269    find_keys_queries: QueryManager<Keys>,
270    /// The find-key-values queries in progress.
271    find_key_values_queries: QueryManager<KeyValues>,
272}
273
274impl ViewUserState {
275    fn force_all_pending_queries(&mut self) -> Result<(), ExecutionError> {
276        self.contains_key_queries.force_all()?;
277        self.contains_keys_queries.force_all()?;
278        self.read_value_queries.force_all()?;
279        self.read_multi_values_queries.force_all()?;
280        self.find_keys_queries.force_all()?;
281        self.find_key_values_queries.force_all()?;
282        Ok(())
283    }
284}
285
286impl<UserInstance: WithContext> Deref for SyncRuntime<UserInstance> {
287    type Target = SyncRuntimeHandle<UserInstance>;
288
289    fn deref(&self) -> &Self::Target {
290        self.0.as_ref().expect(
291            "`SyncRuntime` should not be used after its `inner` contents have been moved out",
292        )
293    }
294}
295
296impl<UserInstance: WithContext> DerefMut for SyncRuntime<UserInstance> {
297    fn deref_mut(&mut self) -> &mut Self::Target {
298        self.0.as_mut().expect(
299            "`SyncRuntime` should not be used after its `inner` contents have been moved out",
300        )
301    }
302}
303
304impl<UserInstance: WithContext> Drop for SyncRuntime<UserInstance> {
305    fn drop(&mut self) {
306        // Ensure the `loaded_applications` are cleared to prevent circular references in
307        // the runtime
308        if let Some(handle) = self.0.take() {
309            handle.inner().loaded_applications.clear();
310        }
311    }
312}
313
314impl<UserInstance: WithContext> SyncRuntimeInternal<UserInstance> {
315    #[expect(clippy::too_many_arguments)]
316    fn new(
317        chain_id: ChainId,
318        height: BlockHeight,
319        round: Option<u32>,
320        executing_message: Option<ExecutingMessage>,
321        execution_state_sender: ExecutionStateSender,
322        deadline: Option<Instant>,
323        refund_grant_to: Option<Account>,
324        resource_controller: ResourceController,
325        user_context: UserInstance::UserContext,
326        allow_application_logs: bool,
327    ) -> Self {
328        Self {
329            chain_id,
330            height,
331            round,
332            executing_message,
333            execution_state_sender,
334            is_finalizing: false,
335            applications_to_finalize: Vec::new(),
336            preloaded_applications: HashMap::new(),
337            loaded_applications: HashMap::new(),
338            call_stack: Vec::new(),
339            active_applications: HashSet::new(),
340            view_user_states: BTreeMap::new(),
341            deadline,
342            refund_grant_to,
343            resource_controller,
344            scheduled_operations: Vec::new(),
345            user_context,
346            allow_application_logs,
347        }
348    }
349
350    /// Returns the [`ApplicationStatus`] of the current application.
351    ///
352    /// The current application is the last to be pushed to the `call_stack`.
353    ///
354    /// # Panics
355    ///
356    /// If the call stack is empty.
357    fn current_application(&self) -> &ApplicationStatus {
358        self.call_stack
359            .last()
360            .expect("Call stack is unexpectedly empty")
361    }
362
363    /// Inserts a new [`ApplicationStatus`] to the end of the `call_stack`.
364    ///
365    /// Ensures the application's ID is also tracked in the `active_applications` set.
366    fn push_application(&mut self, status: ApplicationStatus) {
367        self.active_applications.insert(status.id);
368        self.call_stack.push(status);
369    }
370
371    /// Removes the [`current_application`][`Self::current_application`] from the `call_stack`.
372    ///
373    /// Ensures the application's ID is also removed from the `active_applications` set.
374    ///
375    /// # Panics
376    ///
377    /// If the call stack is empty.
378    fn pop_application(&mut self) -> ApplicationStatus {
379        let status = self
380            .call_stack
381            .pop()
382            .expect("Can't remove application from empty call stack");
383        assert!(self.active_applications.remove(&status.id));
384        status
385    }
386
387    /// Ensures that a call to `application_id` is not-reentrant.
388    ///
389    /// Returns an error if there already is an entry for `application_id` in the call stack.
390    fn check_for_reentrancy(&self, application_id: ApplicationId) -> Result<(), ExecutionError> {
391        ensure!(
392            !self.active_applications.contains(&application_id),
393            ExecutionError::ReentrantCall(application_id)
394        );
395        Ok(())
396    }
397}
398
399impl SyncRuntimeInternal<UserContractInstance> {
400    /// Loads a contract instance, initializing it with this runtime if needed.
401    #[instrument(skip_all, fields(application_id = %id))]
402    fn load_contract_instance(
403        &mut self,
404        this: SyncRuntimeHandle<UserContractInstance>,
405        id: ApplicationId,
406    ) -> Result<LoadedApplication<UserContractInstance>, ExecutionError> {
407        match self.loaded_applications.entry(id) {
408            hash_map::Entry::Occupied(entry) => Ok(entry.get().clone()),
409
410            hash_map::Entry::Vacant(entry) => {
411                // First time actually using the application. Let's see if the code was
412                // pre-loaded.
413                let (code, description) = match self.preloaded_applications.entry(id) {
414                    // TODO(#2927): support dynamic loading of modules on the Web
415                    #[cfg(web)]
416                    hash_map::Entry::Vacant(_) => {
417                        drop(this);
418                        return Err(ExecutionError::UnsupportedDynamicApplicationLoad(Box::new(
419                            id,
420                        )));
421                    }
422                    #[cfg(not(web))]
423                    hash_map::Entry::Vacant(entry) => {
424                        let (code, description) = self
425                            .execution_state_sender
426                            .send_request(move |callback| ExecutionRequest::LoadContract {
427                                id,
428                                callback,
429                            })?
430                            .recv_response()?;
431                        entry.insert((code, description)).clone()
432                    }
433                    hash_map::Entry::Occupied(entry) => entry.get().clone(),
434                };
435                let instance = code.instantiate(this)?;
436
437                self.applications_to_finalize.push(id);
438                Ok(entry
439                    .insert(LoadedApplication::new(instance, description))
440                    .clone())
441            }
442        }
443    }
444
445    /// Configures the runtime for executing a call to a different contract.
446    fn prepare_for_call(
447        &mut self,
448        this: ContractSyncRuntimeHandle,
449        authenticated: bool,
450        callee_id: ApplicationId,
451    ) -> Result<Arc<Mutex<UserContractInstance>>, ExecutionError> {
452        self.check_for_reentrancy(callee_id)?;
453
454        ensure!(
455            !self.is_finalizing,
456            ExecutionError::CrossApplicationCallInFinalize {
457                caller_id: Box::new(self.current_application().id),
458                callee_id: Box::new(callee_id),
459            }
460        );
461
462        // Load the application.
463        let application = self.load_contract_instance(this, callee_id)?;
464
465        let caller = self.current_application();
466        let caller_id = caller.id;
467        let caller_signer = caller.signer;
468        // Make the call to user code.
469        let authenticated_owner = match caller_signer {
470            Some(signer) if authenticated => Some(signer),
471            _ => None,
472        };
473        let authenticated_caller_id = authenticated.then_some(caller_id);
474        self.push_application(ApplicationStatus {
475            caller_id: authenticated_caller_id,
476            id: callee_id,
477            description: application.description,
478            // Allow further nested calls to be authenticated if this one is.
479            signer: authenticated_owner,
480        });
481        Ok(application.instance)
482    }
483
484    /// Cleans up the runtime after the execution of a call to a different contract.
485    fn finish_call(&mut self) {
486        self.pop_application();
487    }
488
489    /// Runs the service in a separate thread as an oracle.
490    fn run_service_oracle_query(
491        &mut self,
492        application_id: ApplicationId,
493        query: Vec<u8>,
494    ) -> Result<Vec<u8>, ExecutionError> {
495        let timeout = self
496            .resource_controller
497            .remaining_service_oracle_execution_time()?;
498        let execution_start = Instant::now();
499        let deadline = Some(execution_start + timeout);
500        let response = self
501            .execution_state_sender
502            .send_request(|callback| ExecutionRequest::QueryServiceOracle {
503                deadline,
504                application_id,
505                next_block_height: self.height,
506                query,
507                callback,
508            })?
509            .recv_response()?;
510
511        self.resource_controller
512            .track_service_oracle_execution(execution_start.elapsed())?;
513        self.resource_controller
514            .track_service_oracle_response(response.len())?;
515
516        Ok(response)
517    }
518}
519
520impl SyncRuntimeInternal<UserServiceInstance> {
521    /// Initializes a service instance with this runtime.
522    fn load_service_instance(
523        &mut self,
524        this: SyncRuntimeHandle<UserServiceInstance>,
525        id: ApplicationId,
526    ) -> Result<LoadedApplication<UserServiceInstance>, ExecutionError> {
527        match self.loaded_applications.entry(id) {
528            hash_map::Entry::Occupied(entry) => Ok(entry.get().clone()),
529
530            hash_map::Entry::Vacant(entry) => {
531                // First time actually using the application. Let's see if the code was
532                // pre-loaded.
533                let (code, description) = match self.preloaded_applications.entry(id) {
534                    // TODO(#2927): support dynamic loading of modules on the Web
535                    #[cfg(web)]
536                    hash_map::Entry::Vacant(_) => {
537                        drop(this);
538                        return Err(ExecutionError::UnsupportedDynamicApplicationLoad(Box::new(
539                            id,
540                        )));
541                    }
542                    #[cfg(not(web))]
543                    hash_map::Entry::Vacant(entry) => {
544                        let (code, description) = self
545                            .execution_state_sender
546                            .send_request(move |callback| ExecutionRequest::LoadService {
547                                id,
548                                callback,
549                            })?
550                            .recv_response()?;
551                        entry.insert((code, description)).clone()
552                    }
553                    hash_map::Entry::Occupied(entry) => entry.get().clone(),
554                };
555                let instance = code.instantiate(this)?;
556
557                self.applications_to_finalize.push(id);
558                Ok(entry
559                    .insert(LoadedApplication::new(instance, description))
560                    .clone())
561            }
562        }
563    }
564}
565
566impl<UserInstance: WithContext> SyncRuntime<UserInstance> {
567    fn into_inner(mut self) -> Option<SyncRuntimeInternal<UserInstance>> {
568        let handle = self.0.take().expect(
569            "`SyncRuntime` should not be used after its `inner` contents have been moved out",
570        );
571        let runtime = Arc::into_inner(handle.0)?
572            .into_inner()
573            .expect("`SyncRuntime` should run in a single thread which should not panic");
574        Some(runtime)
575    }
576}
577
578impl<UserInstance: WithContext> From<SyncRuntimeInternal<UserInstance>>
579    for SyncRuntimeHandle<UserInstance>
580{
581    fn from(runtime: SyncRuntimeInternal<UserInstance>) -> Self {
582        SyncRuntimeHandle(Arc::new(Mutex::new(runtime)))
583    }
584}
585
586impl<UserInstance: WithContext> SyncRuntimeHandle<UserInstance> {
587    fn inner(&self) -> std::sync::MutexGuard<'_, SyncRuntimeInternal<UserInstance>> {
588        self.0
589            .try_lock()
590            .expect("Synchronous runtimes run on a single execution thread")
591    }
592}
593
594impl<UserInstance: WithContext> BaseRuntime for SyncRuntimeHandle<UserInstance>
595where
596    Self: ContractOrServiceRuntime,
597{
598    type Read = ();
599    type ReadValueBytes = u32;
600    type ContainsKey = u32;
601    type ContainsKeys = u32;
602    type ReadMultiValuesBytes = u32;
603    type FindKeysByPrefix = u32;
604    type FindKeyValuesByPrefix = u32;
605
606    fn chain_id(&mut self) -> Result<ChainId, ExecutionError> {
607        let mut this = self.inner();
608        let chain_id = this.chain_id;
609        this.resource_controller.track_runtime_chain_id()?;
610        Ok(chain_id)
611    }
612
613    fn block_height(&mut self) -> Result<BlockHeight, ExecutionError> {
614        let mut this = self.inner();
615        let height = this.height;
616        this.resource_controller.track_runtime_block_height()?;
617        Ok(height)
618    }
619
620    fn application_id(&mut self) -> Result<ApplicationId, ExecutionError> {
621        let mut this = self.inner();
622        let application_id = this.current_application().id;
623        this.resource_controller.track_runtime_application_id()?;
624        Ok(application_id)
625    }
626
627    fn application_creator_chain_id(&mut self) -> Result<ChainId, ExecutionError> {
628        let mut this = self.inner();
629        let application_creator_chain_id = this.current_application().description.creator_chain_id;
630        this.resource_controller.track_runtime_application_id()?;
631        Ok(application_creator_chain_id)
632    }
633
634    fn read_application_description(
635        &mut self,
636        application_id: ApplicationId,
637    ) -> Result<ApplicationDescription, ExecutionError> {
638        let mut this = self.inner();
639        let description = this
640            .execution_state_sender
641            .send_request(|callback| ExecutionRequest::ReadApplicationDescription {
642                application_id,
643                callback,
644            })?
645            .recv_response()?;
646        this.resource_controller
647            .track_runtime_application_description(&description)?;
648        Ok(description)
649    }
650
651    fn application_parameters(&mut self) -> Result<Vec<u8>, ExecutionError> {
652        let mut this = self.inner();
653        let parameters = this.current_application().description.parameters.clone();
654        this.resource_controller
655            .track_runtime_application_parameters(&parameters)?;
656        Ok(parameters)
657    }
658
659    fn read_system_timestamp(&mut self) -> Result<Timestamp, ExecutionError> {
660        let mut this = self.inner();
661        let timestamp = this
662            .execution_state_sender
663            .send_request(|callback| ExecutionRequest::SystemTimestamp { callback })?
664            .recv_response()?;
665        this.resource_controller.track_runtime_timestamp()?;
666        Ok(timestamp)
667    }
668
669    fn read_chain_balance(&mut self) -> Result<Amount, ExecutionError> {
670        let mut this = self.inner();
671        let balance = this
672            .execution_state_sender
673            .send_request(|callback| ExecutionRequest::ChainBalance { callback })?
674            .recv_response()?;
675        this.resource_controller.track_runtime_balance()?;
676        Ok(balance)
677    }
678
679    fn read_owner_balance(&mut self, owner: AccountOwner) -> Result<Amount, ExecutionError> {
680        let mut this = self.inner();
681        let balance = this
682            .execution_state_sender
683            .send_request(|callback| ExecutionRequest::OwnerBalance { owner, callback })?
684            .recv_response()?;
685        this.resource_controller.track_runtime_balance()?;
686        Ok(balance)
687    }
688
689    fn read_owner_balances(&mut self) -> Result<Vec<(AccountOwner, Amount)>, ExecutionError> {
690        let mut this = self.inner();
691        let owner_balances = this
692            .execution_state_sender
693            .send_request(|callback| ExecutionRequest::OwnerBalances { callback })?
694            .recv_response()?;
695        this.resource_controller
696            .track_runtime_owner_balances(&owner_balances)?;
697        Ok(owner_balances)
698    }
699
700    fn read_balance_owners(&mut self) -> Result<Vec<AccountOwner>, ExecutionError> {
701        let mut this = self.inner();
702        let owners = this
703            .execution_state_sender
704            .send_request(|callback| ExecutionRequest::BalanceOwners { callback })?
705            .recv_response()?;
706        this.resource_controller.track_runtime_owners(&owners)?;
707        Ok(owners)
708    }
709
710    fn read_allowance(
711        &mut self,
712        owner: AccountOwner,
713        spender: AccountOwner,
714    ) -> Result<Amount, ExecutionError> {
715        let this = self.inner();
716        let allowance = this
717            .execution_state_sender
718            .send_request(|callback| ExecutionRequest::Allowance {
719                owner,
720                spender,
721                callback,
722            })?
723            .recv_response()?;
724        Ok(allowance)
725    }
726
727    fn read_allowances(
728        &mut self,
729    ) -> Result<Vec<(AccountOwner, AccountOwner, Amount)>, ExecutionError> {
730        let this = self.inner();
731        let allowances = this
732            .execution_state_sender
733            .send_request(|callback| ExecutionRequest::Allowances { callback })?
734            .recv_response()?;
735        Ok(allowances)
736    }
737
738    fn chain_ownership(&mut self) -> Result<ChainOwnership, ExecutionError> {
739        let mut this = self.inner();
740        let chain_ownership = this
741            .execution_state_sender
742            .send_request(|callback| ExecutionRequest::ChainOwnership { callback })?
743            .recv_response()?;
744        this.resource_controller
745            .track_runtime_chain_ownership(&chain_ownership)?;
746        Ok(chain_ownership)
747    }
748
749    fn application_permissions(&mut self) -> Result<ApplicationPermissions, ExecutionError> {
750        let this = self.inner();
751        let application_permissions = this
752            .execution_state_sender
753            .send_request(|callback| ExecutionRequest::ApplicationPermissions { callback })?
754            .recv_response()?;
755        Ok(application_permissions)
756    }
757
758    fn contains_key_new(&mut self, key: Vec<u8>) -> Result<Self::ContainsKey, ExecutionError> {
759        let mut this = self.inner();
760        let id = this.current_application().id;
761        this.resource_controller.track_read_operation()?;
762        let receiver = this
763            .execution_state_sender
764            .send_request(move |callback| ExecutionRequest::ContainsKey { id, key, callback })?;
765        let state = this.view_user_states.entry(id).or_default();
766        state.contains_key_queries.register(receiver)
767    }
768
769    fn contains_key_wait(&mut self, promise: &Self::ContainsKey) -> Result<bool, ExecutionError> {
770        let mut this = self.inner();
771        let id = this.current_application().id;
772        let state = this.view_user_states.entry(id).or_default();
773        let value = state.contains_key_queries.wait(*promise)?;
774        Ok(value)
775    }
776
777    fn contains_keys_new(
778        &mut self,
779        keys: Vec<Vec<u8>>,
780    ) -> Result<Self::ContainsKeys, ExecutionError> {
781        let mut this = self.inner();
782        let id = this.current_application().id;
783        this.resource_controller.track_read_operation()?;
784        let receiver = this
785            .execution_state_sender
786            .send_request(move |callback| ExecutionRequest::ContainsKeys { id, keys, callback })?;
787        let state = this.view_user_states.entry(id).or_default();
788        state.contains_keys_queries.register(receiver)
789    }
790
791    fn contains_keys_wait(
792        &mut self,
793        promise: &Self::ContainsKeys,
794    ) -> Result<Vec<bool>, ExecutionError> {
795        let mut this = self.inner();
796        let id = this.current_application().id;
797        let state = this.view_user_states.entry(id).or_default();
798        let value = state.contains_keys_queries.wait(*promise)?;
799        Ok(value)
800    }
801
802    fn read_multi_values_bytes_new(
803        &mut self,
804        keys: Vec<Vec<u8>>,
805    ) -> Result<Self::ReadMultiValuesBytes, ExecutionError> {
806        let mut this = self.inner();
807        let id = this.current_application().id;
808        this.resource_controller.track_read_operation()?;
809        let receiver = this.execution_state_sender.send_request(move |callback| {
810            ExecutionRequest::ReadMultiValuesBytes { id, keys, callback }
811        })?;
812        let state = this.view_user_states.entry(id).or_default();
813        state.read_multi_values_queries.register(receiver)
814    }
815
816    fn read_multi_values_bytes_wait(
817        &mut self,
818        promise: &Self::ReadMultiValuesBytes,
819    ) -> Result<Vec<Option<Vec<u8>>>, ExecutionError> {
820        let mut this = self.inner();
821        let id = this.current_application().id;
822        let state = this.view_user_states.entry(id).or_default();
823        let values = state.read_multi_values_queries.wait(*promise)?;
824        for value in &values {
825            if let Some(value) = &value {
826                this.resource_controller
827                    .track_bytes_read(value.len() as u64)?;
828            }
829        }
830        Ok(values)
831    }
832
833    fn read_value_bytes_new(
834        &mut self,
835        key: Vec<u8>,
836    ) -> Result<Self::ReadValueBytes, ExecutionError> {
837        let mut this = self.inner();
838        let id = this.current_application().id;
839        this.resource_controller.track_read_operation()?;
840        let receiver = this
841            .execution_state_sender
842            .send_request(move |callback| ExecutionRequest::ReadValueBytes { id, key, callback })?;
843        let state = this.view_user_states.entry(id).or_default();
844        state.read_value_queries.register(receiver)
845    }
846
847    fn read_value_bytes_wait(
848        &mut self,
849        promise: &Self::ReadValueBytes,
850    ) -> Result<Option<Vec<u8>>, ExecutionError> {
851        let mut this = self.inner();
852        let id = this.current_application().id;
853        let value = {
854            let state = this.view_user_states.entry(id).or_default();
855            state.read_value_queries.wait(*promise)?
856        };
857        if let Some(value) = &value {
858            this.resource_controller
859                .track_bytes_read(value.len() as u64)?;
860        }
861        Ok(value)
862    }
863
864    fn find_keys_by_prefix_new(
865        &mut self,
866        key_prefix: Vec<u8>,
867    ) -> Result<Self::FindKeysByPrefix, ExecutionError> {
868        let mut this = self.inner();
869        let id = this.current_application().id;
870        this.resource_controller.track_read_operation()?;
871        let receiver = this.execution_state_sender.send_request(move |callback| {
872            ExecutionRequest::FindKeysByPrefix {
873                id,
874                key_prefix,
875                callback,
876            }
877        })?;
878        let state = this.view_user_states.entry(id).or_default();
879        state.find_keys_queries.register(receiver)
880    }
881
882    fn find_keys_by_prefix_wait(
883        &mut self,
884        promise: &Self::FindKeysByPrefix,
885    ) -> Result<Vec<Vec<u8>>, ExecutionError> {
886        let mut this = self.inner();
887        let id = this.current_application().id;
888        let keys = {
889            let state = this.view_user_states.entry(id).or_default();
890            state.find_keys_queries.wait(*promise)?
891        };
892        let mut read_size = 0;
893        for key in &keys {
894            read_size += key.len();
895        }
896        this.resource_controller
897            .track_bytes_read(read_size as u64)?;
898        Ok(keys)
899    }
900
901    fn find_key_values_by_prefix_new(
902        &mut self,
903        key_prefix: Vec<u8>,
904    ) -> Result<Self::FindKeyValuesByPrefix, ExecutionError> {
905        let mut this = self.inner();
906        let id = this.current_application().id;
907        this.resource_controller.track_read_operation()?;
908        let receiver = this.execution_state_sender.send_request(move |callback| {
909            ExecutionRequest::FindKeyValuesByPrefix {
910                id,
911                key_prefix,
912                callback,
913            }
914        })?;
915        let state = this.view_user_states.entry(id).or_default();
916        state.find_key_values_queries.register(receiver)
917    }
918
919    fn find_key_values_by_prefix_wait(
920        &mut self,
921        promise: &Self::FindKeyValuesByPrefix,
922    ) -> Result<Vec<(Vec<u8>, Vec<u8>)>, ExecutionError> {
923        let mut this = self.inner();
924        let id = this.current_application().id;
925        let state = this.view_user_states.entry(id).or_default();
926        let key_values = state.find_key_values_queries.wait(*promise)?;
927        let mut read_size = 0;
928        for (key, value) in &key_values {
929            read_size += key.len() + value.len();
930        }
931        this.resource_controller
932            .track_bytes_read(read_size as u64)?;
933        Ok(key_values)
934    }
935
936    fn perform_http_request(
937        &mut self,
938        request: http::Request,
939    ) -> Result<http::Response, ExecutionError> {
940        let mut this = self.inner();
941        let app_permissions = this
942            .execution_state_sender
943            .send_request(|callback| ExecutionRequest::GetApplicationPermissions { callback })?
944            .recv_response()?;
945
946        let app_id = this.current_application().id;
947        ensure!(
948            app_permissions.can_make_http_requests(&app_id),
949            ExecutionError::UnauthorizedApplication(app_id)
950        );
951
952        this.resource_controller.track_http_request()?;
953
954        this.execution_state_sender
955            .send_request(|callback| ExecutionRequest::PerformHttpRequest {
956                request,
957                http_responses_are_oracle_responses:
958                    Self::LIMIT_HTTP_RESPONSE_SIZE_TO_ORACLE_RESPONSE_SIZE,
959                callback,
960            })?
961            .recv_response()
962    }
963
964    fn assert_before(&mut self, timestamp: Timestamp) -> Result<(), ExecutionError> {
965        let this = self.inner();
966        this.execution_state_sender
967            .send_request(|callback| ExecutionRequest::AssertBefore {
968                timestamp,
969                callback,
970            })?
971            .recv_response()?
972    }
973
974    fn read_data_blob(&mut self, hash: DataBlobHash) -> Result<Vec<u8>, ExecutionError> {
975        let this = self.inner();
976        let blob_id = hash.into();
977        let content = this
978            .execution_state_sender
979            .send_request(|callback| ExecutionRequest::ReadBlobContent { blob_id, callback })?
980            .recv_response()?;
981        Ok(content.into_vec_or_clone())
982    }
983
984    fn assert_data_blob_exists(&mut self, hash: DataBlobHash) -> Result<(), ExecutionError> {
985        let this = self.inner();
986        let blob_id = hash.into();
987        this.execution_state_sender
988            .send_request(|callback| ExecutionRequest::AssertBlobExists { blob_id, callback })?
989            .recv_response()
990    }
991
992    fn has_empty_storage(&mut self, application: ApplicationId) -> Result<bool, ExecutionError> {
993        let this = self.inner();
994        this.execution_state_sender
995            .send_request(move |callback| ExecutionRequest::HasEmptyStorage {
996                application,
997                callback,
998            })?
999            .recv_response()
1000    }
1001
1002    fn maximum_blob_size(&mut self) -> Result<u64, ExecutionError> {
1003        Ok(self.inner().resource_controller.policy().maximum_blob_size)
1004    }
1005
1006    fn allow_application_logs(&mut self) -> Result<bool, ExecutionError> {
1007        Ok(self.inner().allow_application_logs)
1008    }
1009
1010    #[cfg(web)]
1011    fn send_log(&mut self, message: String, level: tracing::log::Level) {
1012        let this = self.inner();
1013        // Fire-and-forget: ignore errors since logging shouldn't affect execution.
1014        this.execution_state_sender
1015            .unbounded_send(ExecutionRequest::Log { message, level })
1016            .ok();
1017    }
1018}
1019
1020/// An extension trait to determine in compile time the different behaviors between contract and
1021/// services in the implementation of [`BaseRuntime`].
1022trait ContractOrServiceRuntime {
1023    /// Configured to `true` if the HTTP response size should be limited to the oracle response
1024    /// size.
1025    ///
1026    /// This is `false` for services, potentially allowing them to receive a larger HTTP response
1027    /// and only storing in the block a shorter oracle response.
1028    const LIMIT_HTTP_RESPONSE_SIZE_TO_ORACLE_RESPONSE_SIZE: bool;
1029}
1030
1031impl ContractOrServiceRuntime for ContractSyncRuntimeHandle {
1032    const LIMIT_HTTP_RESPONSE_SIZE_TO_ORACLE_RESPONSE_SIZE: bool = true;
1033}
1034
1035impl ContractOrServiceRuntime for ServiceSyncRuntimeHandle {
1036    const LIMIT_HTTP_RESPONSE_SIZE_TO_ORACLE_RESPONSE_SIZE: bool = false;
1037}
1038
1039impl<UserInstance: WithContext> Clone for SyncRuntimeHandle<UserInstance> {
1040    fn clone(&self) -> Self {
1041        SyncRuntimeHandle(self.0.clone())
1042    }
1043}
1044
1045impl ContractSyncRuntime {
1046    pub(crate) fn new(
1047        execution_state_sender: ExecutionStateSender,
1048        chain_id: ChainId,
1049        refund_grant_to: Option<Account>,
1050        resource_controller: ResourceController,
1051        action: &UserAction,
1052        allow_application_logs: bool,
1053    ) -> Self {
1054        SyncRuntime(Some(ContractSyncRuntimeHandle::from(
1055            SyncRuntimeInternal::new(
1056                chain_id,
1057                action.height(),
1058                action.round(),
1059                if let UserAction::Message(context, _) = action {
1060                    Some(context.into())
1061                } else {
1062                    None
1063                },
1064                execution_state_sender,
1065                None,
1066                refund_grant_to,
1067                resource_controller,
1068                action.timestamp(),
1069                allow_application_logs,
1070            ),
1071        )))
1072    }
1073
1074    /// Preloads the code of a contract into the runtime's memory.
1075    pub(crate) fn preload_contract(
1076        &self,
1077        id: ApplicationId,
1078        code: UserContractCode,
1079        description: ApplicationDescription,
1080    ) {
1081        let this = self
1082            .0
1083            .as_ref()
1084            .expect("contracts shouldn't be preloaded while the runtime is being dropped");
1085        let mut this_guard = this.inner();
1086
1087        if let hash_map::Entry::Vacant(entry) = this_guard.preloaded_applications.entry(id) {
1088            entry.insert((code, description));
1089        }
1090    }
1091
1092    /// Main entry point to start executing a user action.
1093    pub(crate) fn run_action(
1094        mut self,
1095        application_id: ApplicationId,
1096        chain_id: ChainId,
1097        action: UserAction,
1098    ) -> Result<(Option<Vec<u8>>, ResourceController), ExecutionError> {
1099        let result = self
1100            .deref_mut()
1101            .run_action(application_id, chain_id, action)?;
1102        let runtime = self
1103            .into_inner()
1104            .expect("Runtime clones should have been freed by now");
1105
1106        Ok((result, runtime.resource_controller))
1107    }
1108}
1109
1110impl ContractSyncRuntimeHandle {
1111    #[instrument(skip_all, fields(application_id = %application_id))]
1112    fn run_action(
1113        &self,
1114        application_id: ApplicationId,
1115        chain_id: ChainId,
1116        action: UserAction,
1117    ) -> Result<Option<Vec<u8>>, ExecutionError> {
1118        let finalize_context = FinalizeContext {
1119            authenticated_owner: action.signer(),
1120            chain_id,
1121            height: action.height(),
1122            round: action.round(),
1123        };
1124
1125        {
1126            let runtime = self.inner();
1127            assert_eq!(runtime.chain_id, chain_id);
1128            assert_eq!(runtime.height, action.height());
1129        }
1130
1131        let signer = action.signer();
1132        let closure = move |code: &mut UserContractInstance| match action {
1133            UserAction::Instantiate(_context, argument) => {
1134                code.instantiate(argument).map(|()| None)
1135            }
1136            UserAction::Operation(_context, operation) => {
1137                code.execute_operation(operation).map(Option::Some)
1138            }
1139            UserAction::Message(_context, message) => code.execute_message(message).map(|()| None),
1140            UserAction::ProcessStreams(_context, updates) => {
1141                code.process_streams(updates).map(|()| None)
1142            }
1143            UserAction::SummarizeEvents(_context, updates) => {
1144                code.summarize_events(updates).map(|()| None)
1145            }
1146        };
1147
1148        let result = self.execute(application_id, signer, closure)?;
1149        self.finalize(finalize_context)?;
1150        Ok(result)
1151    }
1152
1153    /// Notifies all loaded applications that execution is finalizing.
1154    #[instrument(skip_all)]
1155    fn finalize(&self, context: FinalizeContext) -> Result<(), ExecutionError> {
1156        let applications = mem::take(&mut self.inner().applications_to_finalize)
1157            .into_iter()
1158            .rev();
1159
1160        self.inner().is_finalizing = true;
1161
1162        for application in applications {
1163            self.execute(application, context.authenticated_owner, |contract| {
1164                contract.finalize().map(|_| None)
1165            })?;
1166            self.inner().loaded_applications.remove(&application);
1167        }
1168
1169        Ok(())
1170    }
1171
1172    /// Executes a `closure` with the contract code for the `application_id`.
1173    #[instrument(skip_all, fields(application_id = %application_id))]
1174    fn execute(
1175        &self,
1176        application_id: ApplicationId,
1177        signer: Option<AccountOwner>,
1178        closure: impl FnOnce(&mut UserContractInstance) -> Result<Option<Vec<u8>>, ExecutionError>,
1179    ) -> Result<Option<Vec<u8>>, ExecutionError> {
1180        let contract = {
1181            let mut runtime = self.inner();
1182            let application = runtime.load_contract_instance(self.clone(), application_id)?;
1183
1184            let status = ApplicationStatus {
1185                caller_id: None,
1186                id: application_id,
1187                description: application.description.clone(),
1188                signer,
1189            };
1190
1191            runtime.push_application(status);
1192
1193            application
1194        };
1195
1196        let result = closure(
1197            &mut contract
1198                .instance
1199                .try_lock()
1200                .expect("Application should not be already executing"),
1201        )?;
1202
1203        let mut runtime = self.inner();
1204        let application_status = runtime.pop_application();
1205        assert_eq!(application_status.caller_id, None);
1206        assert_eq!(application_status.id, application_id);
1207        assert_eq!(application_status.description, contract.description);
1208        assert_eq!(application_status.signer, signer);
1209        assert!(runtime.call_stack.is_empty());
1210
1211        Ok(result)
1212    }
1213}
1214
1215impl ContractRuntime for ContractSyncRuntimeHandle {
1216    fn authenticated_owner(&mut self) -> Result<Option<AccountOwner>, ExecutionError> {
1217        let this = self.inner();
1218        Ok(this.current_application().signer)
1219    }
1220
1221    fn message_is_bouncing(&mut self) -> Result<Option<bool>, ExecutionError> {
1222        Ok(self
1223            .inner()
1224            .executing_message
1225            .map(|metadata| metadata.is_bouncing))
1226    }
1227
1228    fn message_origin_chain_id(&mut self) -> Result<Option<ChainId>, ExecutionError> {
1229        Ok(self
1230            .inner()
1231            .executing_message
1232            .map(|metadata| metadata.origin))
1233    }
1234
1235    fn message_origin_timestamp(&mut self) -> Result<Option<Timestamp>, ExecutionError> {
1236        Ok(self
1237            .inner()
1238            .executing_message
1239            .map(|metadata| metadata.origin_timestamp))
1240    }
1241
1242    fn authenticated_caller_id(&mut self) -> Result<Option<ApplicationId>, ExecutionError> {
1243        let this = self.inner();
1244        if this.call_stack.len() <= 1 {
1245            return Ok(None);
1246        }
1247        Ok(this.current_application().caller_id)
1248    }
1249
1250    fn maximum_fuel_per_block(&mut self, vm_runtime: VmRuntime) -> Result<u64, ExecutionError> {
1251        Ok(match vm_runtime {
1252            VmRuntime::Wasm => {
1253                self.inner()
1254                    .resource_controller
1255                    .policy()
1256                    .maximum_wasm_fuel_per_block
1257            }
1258            VmRuntime::Evm => {
1259                self.inner()
1260                    .resource_controller
1261                    .policy()
1262                    .maximum_evm_fuel_per_block
1263            }
1264        })
1265    }
1266
1267    fn remaining_fuel(&mut self, vm_runtime: VmRuntime) -> Result<u64, ExecutionError> {
1268        Ok(self.inner().resource_controller.remaining_fuel(vm_runtime))
1269    }
1270
1271    fn consume_fuel(&mut self, fuel: u64, vm_runtime: VmRuntime) -> Result<(), ExecutionError> {
1272        let mut this = self.inner();
1273        this.resource_controller.track_fuel(fuel, vm_runtime)
1274    }
1275
1276    fn send_message(&mut self, message: SendMessageRequest<Vec<u8>>) -> Result<(), ExecutionError> {
1277        let mut this = self.inner();
1278        let application = this.current_application();
1279        let application_id = application.id;
1280        let authenticated_owner = application.signer;
1281        let mut refund_grant_to = this.refund_grant_to;
1282
1283        let grant = this
1284            .resource_controller
1285            .policy()
1286            .total_price(&message.grant)?;
1287        if grant.is_zero() {
1288            refund_grant_to = None;
1289        } else {
1290            this.resource_controller.track_grant(grant)?;
1291        }
1292        let kind = if message.is_tracked {
1293            MessageKind::Tracked
1294        } else {
1295            MessageKind::Simple
1296        };
1297
1298        this.execution_state_sender
1299            .send_request(|callback| ExecutionRequest::AddOutgoingMessage {
1300                message: OutgoingMessage {
1301                    destination: message.destination,
1302                    authenticated_owner,
1303                    refund_grant_to,
1304                    grant,
1305                    kind,
1306                    message: Message::User {
1307                        application_id,
1308                        bytes: message.message,
1309                    },
1310                },
1311                callback,
1312            })?
1313            .recv_response()?;
1314
1315        Ok(())
1316    }
1317
1318    fn transfer(
1319        &mut self,
1320        source: AccountOwner,
1321        destination: Account,
1322        amount: Amount,
1323    ) -> Result<(), ExecutionError> {
1324        let this = self.inner();
1325        let current_application = this.current_application();
1326        let application_id = current_application.id;
1327        let signer = current_application.signer;
1328
1329        this.execution_state_sender
1330            .send_request(|callback| ExecutionRequest::Transfer {
1331                source,
1332                destination,
1333                amount,
1334                signer,
1335                application_id,
1336                callback,
1337            })?
1338            .recv_response()?;
1339        Ok(())
1340    }
1341
1342    fn claim(
1343        &mut self,
1344        source: Account,
1345        destination: Account,
1346        amount: Amount,
1347    ) -> Result<(), ExecutionError> {
1348        let this = self.inner();
1349        let current_application = this.current_application();
1350        let application_id = current_application.id;
1351        let signer = current_application.signer;
1352
1353        this.execution_state_sender
1354            .send_request(|callback| ExecutionRequest::Claim {
1355                source,
1356                destination,
1357                amount,
1358                signer,
1359                application_id,
1360                callback,
1361            })?
1362            .recv_response()?;
1363        Ok(())
1364    }
1365
1366    fn approve(
1367        &mut self,
1368        owner: AccountOwner,
1369        spender: AccountOwner,
1370        amount: Amount,
1371    ) -> Result<(), ExecutionError> {
1372        let this = self.inner();
1373        let current_application = this.current_application();
1374        let application_id = current_application.id;
1375        let signer = current_application.signer;
1376
1377        this.execution_state_sender
1378            .send_request(|callback| ExecutionRequest::Approve {
1379                owner,
1380                spender,
1381                amount,
1382                signer,
1383                application_id,
1384                callback,
1385            })?
1386            .recv_response()?;
1387        Ok(())
1388    }
1389
1390    fn transfer_from(
1391        &mut self,
1392        owner: AccountOwner,
1393        spender: AccountOwner,
1394        destination: Account,
1395        amount: Amount,
1396    ) -> Result<(), ExecutionError> {
1397        let this = self.inner();
1398        let current_application = this.current_application();
1399        let application_id = current_application.id;
1400        let signer = current_application.signer;
1401
1402        this.execution_state_sender
1403            .send_request(|callback| ExecutionRequest::TransferFrom {
1404                owner,
1405                spender,
1406                destination,
1407                amount,
1408                signer,
1409                application_id,
1410                callback,
1411            })?
1412            .recv_response()?;
1413        Ok(())
1414    }
1415
1416    fn try_call_application(
1417        &mut self,
1418        authenticated: bool,
1419        callee_id: ApplicationId,
1420        argument: Vec<u8>,
1421    ) -> Result<Vec<u8>, ExecutionError> {
1422        let contract = self
1423            .inner()
1424            .prepare_for_call(self.clone(), authenticated, callee_id)?;
1425
1426        let value = contract
1427            .try_lock()
1428            .expect("Applications should not have reentrant calls")
1429            .execute_operation(argument)?;
1430
1431        self.inner().finish_call();
1432
1433        Ok(value)
1434    }
1435
1436    fn emit(&mut self, stream_name: StreamName, value: Vec<u8>) -> Result<u32, ExecutionError> {
1437        let mut this = self.inner();
1438        ensure!(
1439            stream_name.0.len() <= MAX_STREAM_NAME_LEN,
1440            ExecutionError::StreamNameTooLong
1441        );
1442        let application_id = GenericApplicationId::User(this.current_application().id);
1443        let stream_id = StreamId {
1444            stream_name,
1445            application_id,
1446        };
1447        let value_len = value.len() as u64;
1448        let index = this
1449            .execution_state_sender
1450            .send_request(|callback| ExecutionRequest::Emit {
1451                stream_id,
1452                value,
1453                callback,
1454            })?
1455            .recv_response()?;
1456        // TODO(#365): Consider separate event fee categories.
1457        this.resource_controller.track_bytes_written(value_len)?;
1458        Ok(index)
1459    }
1460
1461    fn read_event(
1462        &mut self,
1463        chain_id: ChainId,
1464        stream_name: StreamName,
1465        index: u32,
1466    ) -> Result<Vec<u8>, ExecutionError> {
1467        let mut this = self.inner();
1468        ensure!(
1469            stream_name.0.len() <= MAX_STREAM_NAME_LEN,
1470            ExecutionError::StreamNameTooLong
1471        );
1472        let application_id = GenericApplicationId::User(this.current_application().id);
1473        let stream_id = StreamId {
1474            stream_name,
1475            application_id,
1476        };
1477        let event_id = EventId {
1478            stream_id,
1479            index,
1480            chain_id,
1481        };
1482        let event = this
1483            .execution_state_sender
1484            .send_request(|callback| ExecutionRequest::ReadEvent { event_id, callback })?
1485            .recv_response()?;
1486        // TODO(#365): Consider separate event fee categories.
1487        this.resource_controller
1488            .track_bytes_read(event.len() as u64)?;
1489        Ok(event)
1490    }
1491
1492    fn subscribe_to_events(
1493        &mut self,
1494        chain_id: ChainId,
1495        application_id: ApplicationId,
1496        stream_name: StreamName,
1497    ) -> Result<(), ExecutionError> {
1498        let this = self.inner();
1499        ensure!(
1500            stream_name.0.len() <= MAX_STREAM_NAME_LEN,
1501            ExecutionError::StreamNameTooLong
1502        );
1503        let stream_id = StreamId {
1504            stream_name,
1505            application_id: application_id.into(),
1506        };
1507        let subscriber_app_id = this.current_application().id;
1508        this.execution_state_sender
1509            .send_request(|callback| ExecutionRequest::SubscribeToEvents {
1510                chain_id,
1511                stream_id,
1512                subscriber_app_id,
1513                callback,
1514            })?
1515            .recv_response()?;
1516        Ok(())
1517    }
1518
1519    fn unsubscribe_from_events(
1520        &mut self,
1521        chain_id: ChainId,
1522        application_id: ApplicationId,
1523        stream_name: StreamName,
1524    ) -> Result<(), ExecutionError> {
1525        let this = self.inner();
1526        ensure!(
1527            stream_name.0.len() <= MAX_STREAM_NAME_LEN,
1528            ExecutionError::StreamNameTooLong
1529        );
1530        let stream_id = StreamId {
1531            stream_name,
1532            application_id: application_id.into(),
1533        };
1534        let subscriber_app_id = this.current_application().id;
1535        this.execution_state_sender
1536            .send_request(|callback| ExecutionRequest::UnsubscribeFromEvents {
1537                chain_id,
1538                stream_id,
1539                subscriber_app_id,
1540                callback,
1541            })?
1542            .recv_response()?;
1543        Ok(())
1544    }
1545
1546    fn query_service(
1547        &mut self,
1548        application_id: ApplicationId,
1549        query: Vec<u8>,
1550    ) -> Result<Vec<u8>, ExecutionError> {
1551        let mut this = self.inner();
1552
1553        let app_permissions = this
1554            .execution_state_sender
1555            .send_request(|callback| ExecutionRequest::GetApplicationPermissions { callback })?
1556            .recv_response()?;
1557
1558        let app_id = this.current_application().id;
1559        ensure!(
1560            app_permissions.can_call_services(&app_id),
1561            ExecutionError::UnauthorizedApplication(app_id)
1562        );
1563
1564        this.resource_controller.track_service_oracle_call()?;
1565
1566        this.run_service_oracle_query(application_id, query)
1567    }
1568
1569    fn open_chain(
1570        &mut self,
1571        ownership: ChainOwnership,
1572        application_permissions: ApplicationPermissions,
1573        balance: Amount,
1574    ) -> Result<ChainId, ExecutionError> {
1575        let parent_id = self.inner().chain_id;
1576        let block_height = self.block_height()?;
1577
1578        let timestamp = self.inner().user_context;
1579
1580        let chain_id = self
1581            .inner()
1582            .execution_state_sender
1583            .send_request(|callback| ExecutionRequest::OpenChain {
1584                ownership,
1585                balance,
1586                parent_id,
1587                block_height,
1588                timestamp,
1589                application_permissions,
1590                callback,
1591            })?
1592            .recv_response()?;
1593
1594        Ok(chain_id)
1595    }
1596
1597    fn close_chain(&mut self) -> Result<(), ExecutionError> {
1598        let this = self.inner();
1599        let application_id = this.current_application().id;
1600        this.execution_state_sender
1601            .send_request(|callback| ExecutionRequest::CloseChain {
1602                application_id,
1603                callback,
1604            })?
1605            .recv_response()?
1606    }
1607
1608    fn change_ownership(&mut self, ownership: ChainOwnership) -> Result<(), ExecutionError> {
1609        let this = self.inner();
1610        let application_id = this.current_application().id;
1611        this.execution_state_sender
1612            .send_request(|callback| ExecutionRequest::ChangeOwnership {
1613                application_id,
1614                ownership,
1615                callback,
1616            })?
1617            .recv_response()?
1618    }
1619
1620    fn change_application_permissions(
1621        &mut self,
1622        application_permissions: ApplicationPermissions,
1623    ) -> Result<(), ExecutionError> {
1624        let this = self.inner();
1625        let application_id = this.current_application().id;
1626        this.execution_state_sender
1627            .send_request(|callback| ExecutionRequest::ChangeApplicationPermissions {
1628                application_id,
1629                application_permissions,
1630                callback,
1631            })?
1632            .recv_response()?
1633    }
1634
1635    fn peek_application_index(&mut self) -> Result<u32, ExecutionError> {
1636        let index = self
1637            .inner()
1638            .execution_state_sender
1639            .send_request(move |callback| ExecutionRequest::PeekApplicationIndex { callback })?
1640            .recv_response()?;
1641        Ok(index)
1642    }
1643
1644    fn create_application(
1645        &mut self,
1646        module_id: ModuleId,
1647        parameters: Vec<u8>,
1648        argument: Vec<u8>,
1649        required_application_ids: Vec<ApplicationId>,
1650    ) -> Result<ApplicationId, ExecutionError> {
1651        let chain_id = self.inner().chain_id;
1652        let block_height = self.block_height()?;
1653
1654        let CreateApplicationResult { app_id } = self
1655            .inner()
1656            .execution_state_sender
1657            .send_request(move |callback| ExecutionRequest::CreateApplication {
1658                chain_id,
1659                block_height,
1660                module_id,
1661                parameters,
1662                required_application_ids,
1663                callback,
1664            })?
1665            .recv_response()?;
1666
1667        let contract = self.inner().prepare_for_call(self.clone(), true, app_id)?;
1668
1669        contract
1670            .try_lock()
1671            .expect("Applications should not have reentrant calls")
1672            .instantiate(argument)?;
1673
1674        self.inner().finish_call();
1675
1676        Ok(app_id)
1677    }
1678
1679    fn create_data_blob(&mut self, bytes: Vec<u8>) -> Result<DataBlobHash, ExecutionError> {
1680        let blob = Blob::new_data(bytes);
1681        let blob_id = blob.id();
1682        let this = self.inner();
1683        this.execution_state_sender
1684            .send_request(|callback| ExecutionRequest::AddCreatedBlob { blob, callback })?
1685            .recv_response()?;
1686        Ok(DataBlobHash(blob_id.hash))
1687    }
1688
1689    fn publish_module(
1690        &mut self,
1691        contract: Bytecode,
1692        service: Bytecode,
1693        vm_runtime: VmRuntime,
1694        formats: Option<Vec<u8>>,
1695    ) -> Result<ModuleId, ExecutionError> {
1696        let (blobs, module_id) = crate::runtime::create_bytecode_blobs_sync(
1697            &contract,
1698            &service,
1699            vm_runtime,
1700            formats.as_deref(),
1701        );
1702        let this = self.inner();
1703        for blob in blobs {
1704            this.execution_state_sender
1705                .send_request(|callback| ExecutionRequest::AddCreatedBlob { blob, callback })?
1706                .recv_response()?;
1707        }
1708        Ok(module_id)
1709    }
1710
1711    fn validation_round(&mut self) -> Result<Option<u32>, ExecutionError> {
1712        let this = self.inner();
1713        let round = this.round;
1714        this.execution_state_sender
1715            .send_request(|callback| ExecutionRequest::ValidationRound { round, callback })?
1716            .recv_response()
1717    }
1718
1719    fn write_batch(&mut self, batch: Batch) -> Result<(), ExecutionError> {
1720        let mut this = self.inner();
1721        let id = this.current_application().id;
1722        let state = this.view_user_states.entry(id).or_default();
1723        state.force_all_pending_queries()?;
1724        this.resource_controller.track_write_operations(
1725            batch
1726                .num_operations()
1727                .try_into()
1728                .map_err(|_| ExecutionError::from(ArithmeticError::Overflow))?,
1729        )?;
1730        this.resource_controller
1731            .track_bytes_written(batch.size() as u64)?;
1732        this.execution_state_sender
1733            .send_request(|callback| ExecutionRequest::WriteBatch {
1734                id,
1735                batch,
1736                callback,
1737            })?
1738            .recv_response()?;
1739        Ok(())
1740    }
1741}
1742
1743impl ServiceSyncRuntime {
1744    /// Creates a new [`ServiceSyncRuntime`] ready to execute using a provided [`QueryContext`].
1745    pub fn new(execution_state_sender: ExecutionStateSender, context: QueryContext) -> Self {
1746        Self::new_with_deadline(execution_state_sender, context, None)
1747    }
1748
1749    /// Creates a new [`ServiceSyncRuntime`] ready to execute using a provided [`QueryContext`].
1750    pub fn new_with_deadline(
1751        execution_state_sender: ExecutionStateSender,
1752        context: QueryContext,
1753        deadline: Option<Instant>,
1754    ) -> Self {
1755        // Query the allow_application_logs setting from the execution state.
1756        let allow_application_logs = execution_state_sender
1757            .send_request(|callback| ExecutionRequest::AllowApplicationLogs { callback })
1758            .ok()
1759            .and_then(|receiver| receiver.recv_response().ok())
1760            .unwrap_or(false);
1761
1762        let runtime = SyncRuntime(Some(
1763            SyncRuntimeInternal::new(
1764                context.chain_id,
1765                context.next_block_height,
1766                None,
1767                None,
1768                execution_state_sender,
1769                deadline,
1770                None,
1771                ResourceController::default(),
1772                (),
1773                allow_application_logs,
1774            )
1775            .into(),
1776        ));
1777
1778        ServiceSyncRuntime {
1779            runtime,
1780            current_context: context,
1781        }
1782    }
1783
1784    /// Preloads the code of a service into the runtime's memory.
1785    pub(crate) fn preload_service(
1786        &self,
1787        id: ApplicationId,
1788        code: UserServiceCode,
1789        description: ApplicationDescription,
1790    ) {
1791        let this = self
1792            .runtime
1793            .0
1794            .as_ref()
1795            .expect("services shouldn't be preloaded while the runtime is being dropped");
1796        let mut this_guard = this.inner();
1797
1798        if let hash_map::Entry::Vacant(entry) = this_guard.preloaded_applications.entry(id) {
1799            entry.insert((code, description));
1800        }
1801    }
1802
1803    /// Runs the service runtime actor, waiting for `incoming_requests` to respond to.
1804    pub fn run(&mut self, incoming_requests: &std::sync::mpsc::Receiver<ServiceRuntimeRequest>) {
1805        while let Ok(request) = incoming_requests.recv() {
1806            let ServiceRuntimeRequest::Query {
1807                application_id,
1808                context,
1809                query,
1810                callback,
1811            } = request;
1812
1813            let result = self
1814                .prepare_for_query(context)
1815                .and_then(|()| self.run_query(application_id, query));
1816
1817            if let Err(err) = callback.send(result) {
1818                tracing::debug!(%err, "Receiver for query result has been dropped");
1819            }
1820        }
1821    }
1822
1823    /// Prepares the runtime to query an application.
1824    pub(crate) fn prepare_for_query(
1825        &mut self,
1826        new_context: QueryContext,
1827    ) -> Result<(), ExecutionError> {
1828        let expected_context = QueryContext {
1829            local_time: new_context.local_time,
1830            ..self.current_context
1831        };
1832
1833        if new_context != expected_context {
1834            let execution_state_sender = self.handle_mut().inner().execution_state_sender.clone();
1835            *self = ServiceSyncRuntime::new(execution_state_sender, new_context);
1836        } else {
1837            self.handle_mut()
1838                .inner()
1839                .execution_state_sender
1840                .send_request(|callback| ExecutionRequest::SetLocalTime {
1841                    local_time: new_context.local_time,
1842                    callback,
1843                })?
1844                .recv_response()?;
1845        }
1846        Ok(())
1847    }
1848
1849    /// Queries an application specified by its [`ApplicationId`].
1850    pub(crate) fn run_query(
1851        &mut self,
1852        application_id: ApplicationId,
1853        query: Vec<u8>,
1854    ) -> Result<QueryOutcome<Vec<u8>>, ExecutionError> {
1855        let this = self.handle_mut();
1856        let response = this.try_query_application(application_id, query)?;
1857        let operations = mem::take(&mut this.inner().scheduled_operations);
1858
1859        Ok(QueryOutcome {
1860            response,
1861            operations,
1862        })
1863    }
1864
1865    /// Obtains the [`SyncRuntimeHandle`] stored in this [`ServiceSyncRuntime`].
1866    fn handle_mut(&mut self) -> &mut ServiceSyncRuntimeHandle {
1867        self.runtime.0.as_mut().expect(
1868            "`SyncRuntimeHandle` should be available while `SyncRuntime` hasn't been dropped",
1869        )
1870    }
1871}
1872
1873impl ServiceRuntime for ServiceSyncRuntimeHandle {
1874    /// Note that queries are not available from writable contexts.
1875    fn try_query_application(
1876        &mut self,
1877        queried_id: ApplicationId,
1878        argument: Vec<u8>,
1879    ) -> Result<Vec<u8>, ExecutionError> {
1880        let service = {
1881            let mut this = self.inner();
1882
1883            // Load the application.
1884            let application = this.load_service_instance(self.clone(), queried_id)?;
1885            // Make the call to user code.
1886            this.push_application(ApplicationStatus {
1887                caller_id: None,
1888                id: queried_id,
1889                description: application.description,
1890                signer: None,
1891            });
1892            application.instance
1893        };
1894        let response = service
1895            .try_lock()
1896            .expect("Applications should not have reentrant calls")
1897            .handle_query(argument)?;
1898        self.inner().pop_application();
1899        Ok(response)
1900    }
1901
1902    fn schedule_operation(&mut self, operation: Vec<u8>) -> Result<(), ExecutionError> {
1903        let mut this = self.inner();
1904        let application_id = this.current_application().id;
1905
1906        this.scheduled_operations.push(Operation::User {
1907            application_id,
1908            bytes: operation,
1909        });
1910
1911        Ok(())
1912    }
1913
1914    fn check_execution_time(&mut self) -> Result<(), ExecutionError> {
1915        if let Some(deadline) = self.inner().deadline {
1916            if Instant::now() >= deadline {
1917                return Err(ExecutionError::MaximumServiceOracleExecutionTimeExceeded);
1918            }
1919        }
1920        Ok(())
1921    }
1922}
1923
1924/// A request to the service runtime actor.
1925#[allow(missing_docs)]
1926pub enum ServiceRuntimeRequest {
1927    Query {
1928        application_id: ApplicationId,
1929        context: QueryContext,
1930        query: Vec<u8>,
1931        callback: oneshot::Sender<Result<QueryOutcome<Vec<u8>>, ExecutionError>>,
1932    },
1933}
1934
1935/// The origin of the execution.
1936#[derive(Clone, Copy, Debug)]
1937struct ExecutingMessage {
1938    is_bouncing: bool,
1939    origin: ChainId,
1940    origin_timestamp: Timestamp,
1941}
1942
1943impl From<&MessageContext> for ExecutingMessage {
1944    fn from(context: &MessageContext) -> Self {
1945        ExecutingMessage {
1946            is_bouncing: context.is_bouncing,
1947            origin: context.origin,
1948            origin_timestamp: context.origin_timestamp,
1949        }
1950    }
1951}
1952
1953/// Creates a compressed contract and service bytecode synchronously, plus an
1954/// optional `ApplicationFormats` blob built from the BCS-encoded `Formats`
1955/// description bytes.
1956pub fn create_bytecode_blobs_sync(
1957    contract: &Bytecode,
1958    service: &Bytecode,
1959    vm_runtime: VmRuntime,
1960    formats: Option<&[u8]>,
1961) -> (Vec<Blob>, ModuleId) {
1962    let formats_blob = formats.map(Blob::new_application_formats);
1963    let formats_blob_hash = formats_blob.as_ref().map(|blob| blob.id().hash);
1964    let (mut blobs, module_id) = match vm_runtime {
1965        VmRuntime::Wasm => {
1966            let compressed_contract = contract.compress();
1967            let compressed_service = service.compress();
1968            let contract_blob = Blob::new_contract_bytecode(compressed_contract);
1969            let service_blob = Blob::new_service_bytecode(compressed_service);
1970            let module_id = ModuleId::new_with_formats(
1971                contract_blob.id().hash,
1972                service_blob.id().hash,
1973                vm_runtime,
1974                formats_blob_hash,
1975            );
1976            (vec![contract_blob, service_blob], module_id)
1977        }
1978        VmRuntime::Evm => {
1979            let compressed_contract = contract.compress();
1980            let evm_contract_blob = Blob::new_evm_bytecode(compressed_contract);
1981            let module_id = ModuleId::new_with_formats(
1982                evm_contract_blob.id().hash,
1983                evm_contract_blob.id().hash,
1984                vm_runtime,
1985                formats_blob_hash,
1986            );
1987            (vec![evm_contract_blob], module_id)
1988        }
1989    };
1990    if let Some(blob) = formats_blob {
1991        blobs.push(blob);
1992    }
1993    (blobs, module_id)
1994}