1use 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
71pub 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
82pub type ContractSyncRuntimeHandle = SyncRuntimeHandle<UserContractInstance>;
84pub type ServiceSyncRuntimeHandle = SyncRuntimeHandle<UserServiceInstance>;
86
87#[derive(Debug)]
89pub struct SyncRuntimeInternal<UserInstance: WithContext> {
90 chain_id: ChainId,
92 height: BlockHeight,
95 round: Option<u32>,
97 #[debug(skip_if = Option::is_none)]
99 executing_message: Option<ExecutingMessage>,
100
101 execution_state_sender: ExecutionStateSender,
103
104 is_finalizing: bool,
108 applications_to_finalize: Vec<ApplicationId>,
110
111 preloaded_applications: HashMap<ApplicationId, (UserInstance::Code, ApplicationDescription)>,
113 loaded_applications: HashMap<ApplicationId, LoadedApplication<UserInstance>>,
115 call_stack: Vec<ApplicationStatus>,
117 active_applications: HashSet<ApplicationId>,
119 scheduled_operations: Vec<Operation>,
121
122 view_user_states: BTreeMap<ApplicationId, ViewUserState>,
124
125 deadline: Option<Instant>,
129
130 #[debug(skip_if = Option::is_none)]
132 refund_grant_to: Option<Account>,
133 resource_controller: ResourceController,
135 user_context: UserInstance::UserContext,
137 allow_application_logs: bool,
139}
140
141#[derive(Debug)]
143struct ApplicationStatus {
144 caller_id: Option<ApplicationId>,
146 id: ApplicationId,
148 description: ApplicationDescription,
150 signer: Option<AccountOwner>,
152}
153
154#[derive(Debug)]
156struct LoadedApplication<Instance> {
157 instance: Arc<Mutex<Instance>>,
158 description: ApplicationDescription,
159}
160
161impl<Instance> LoadedApplication<Instance> {
162 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 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#[derive(Debug, Default)]
212struct QueryManager<T> {
213 pending_queries: BTreeMap<u32, Promise<T>>,
215 query_count: u32,
217 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 contains_key_queries: QueryManager<bool>,
262 contains_keys_queries: QueryManager<Vec<bool>>,
264 read_value_queries: QueryManager<Option<Value>>,
266 read_multi_values_queries: QueryManager<Vec<Option<Value>>>,
268 find_keys_queries: QueryManager<Keys>,
270 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 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 fn current_application(&self) -> &ApplicationStatus {
358 self.call_stack
359 .last()
360 .expect("Call stack is unexpectedly empty")
361 }
362
363 fn push_application(&mut self, status: ApplicationStatus) {
367 self.active_applications.insert(status.id);
368 self.call_stack.push(status);
369 }
370
371 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 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 #[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 let (code, description) = match self.preloaded_applications.entry(id) {
414 #[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 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 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 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 signer: authenticated_owner,
480 });
481 Ok(application.instance)
482 }
483
484 fn finish_call(&mut self) {
486 self.pop_application();
487 }
488
489 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 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 let (code, description) = match self.preloaded_applications.entry(id) {
534 #[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(¶meters)?;
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 this.execution_state_sender
1015 .unbounded_send(ExecutionRequest::Log { message, level })
1016 .ok();
1017 }
1018}
1019
1020trait ContractOrServiceRuntime {
1023 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 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 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 #[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 #[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 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 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 pub fn new(execution_state_sender: ExecutionStateSender, context: QueryContext) -> Self {
1746 Self::new_with_deadline(execution_state_sender, context, None)
1747 }
1748
1749 pub fn new_with_deadline(
1751 execution_state_sender: ExecutionStateSender,
1752 context: QueryContext,
1753 deadline: Option<Instant>,
1754 ) -> Self {
1755 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 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 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 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 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 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 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 let application = this.load_service_instance(self.clone(), queried_id)?;
1885 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#[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#[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
1953pub 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}