1use std::{
9 collections::HashMap,
10 io,
11 path::{Path, PathBuf},
12 sync::Arc,
13};
14
15use cargo_toml::Manifest;
16use futures::future;
17use linera_base::{
18 crypto::{AccountPublicKey, AccountSecretKey},
19 data_types::{
20 Amount, ApplicationDescription, Blob, BlockHeight, Bytecode, ChainDescription,
21 CompressedBytecode, Epoch,
22 },
23 identifiers::{AccountOwner, ApplicationId, ChainId, ModuleId},
24 vm::VmRuntime,
25};
26use linera_chain::{types::ConfirmedBlockCertificate, ChainExecutionContext};
27use linera_core::{data_types::ChainInfoQuery, worker::WorkerError};
28use linera_execution::{
29 system::{SystemOperation, SystemQuery, SystemResponse},
30 ExecutionError, Operation, Query, QueryOutcome, QueryResponse,
31};
32use linera_storage::Storage as _;
33use serde::Serialize;
34use tokio::{fs, sync::Mutex};
35
36use super::{BlockBuilder, TestValidator};
37use crate::{ContractAbi, ServiceAbi};
38
39pub struct ActiveChain {
41 key_pair: AccountSecretKey,
42 description: ChainDescription,
43 tip: Arc<Mutex<Option<ConfirmedBlockCertificate>>>,
44 validator: TestValidator,
45}
46
47impl Clone for ActiveChain {
48 fn clone(&self) -> Self {
49 ActiveChain {
50 key_pair: self.key_pair.copy(),
51 description: self.description.clone(),
52 tip: self.tip.clone(),
53 validator: self.validator.clone(),
54 }
55 }
56}
57
58impl ActiveChain {
59 pub fn new(
65 key_pair: AccountSecretKey,
66 description: ChainDescription,
67 validator: TestValidator,
68 ) -> Self {
69 ActiveChain {
70 key_pair,
71 description,
72 tip: Arc::default(),
73 validator,
74 }
75 }
76
77 pub fn id(&self) -> ChainId {
79 self.description.id()
80 }
81
82 pub fn public_key(&self) -> AccountPublicKey {
84 self.key_pair.public()
85 }
86
87 pub fn key_pair(&self) -> &AccountSecretKey {
89 &self.key_pair
90 }
91
92 pub fn set_key_pair(&mut self, key_pair: AccountSecretKey) {
94 self.key_pair = key_pair
95 }
96
97 pub async fn epoch(&self) -> Epoch {
99 *self
100 .validator
101 .worker()
102 .chain_state_view(self.id())
103 .await
104 .expect("Failed to load chain")
105 .execution_state
106 .system
107 .epoch
108 .get()
109 }
110
111 pub async fn chain_balance(&self) -> Amount {
113 let query = Query::System(SystemQuery);
114
115 let QueryOutcome { response, .. } = self
116 .validator
117 .worker()
118 .query_application(self.id(), query)
119 .await
120 .expect("Failed to query chain's balance");
121
122 let QueryResponse::System(SystemResponse { balance, .. }) = response else {
123 panic!("Unexpected response from system application");
124 };
125
126 balance
127 }
128
129 pub async fn owner_balance(&self, owner: &AccountOwner) -> Option<Amount> {
131 let chain_state = self
132 .validator
133 .worker()
134 .chain_state_view(self.id())
135 .await
136 .expect("Failed to read chain state");
137
138 chain_state
139 .execution_state
140 .system
141 .balances
142 .get(owner)
143 .await
144 .expect("Failed to read owner balance")
145 }
146
147 pub async fn owner_balances(
149 &self,
150 owners: impl IntoIterator<Item = AccountOwner>,
151 ) -> HashMap<AccountOwner, Option<Amount>> {
152 let chain_state = self
153 .validator
154 .worker()
155 .chain_state_view(self.id())
156 .await
157 .expect("Failed to read chain state");
158
159 let mut balances = HashMap::new();
160
161 for owner in owners {
162 let balance = chain_state
163 .execution_state
164 .system
165 .balances
166 .get(&owner)
167 .await
168 .expect("Failed to read an owner's balance");
169
170 balances.insert(owner, balance);
171 }
172
173 balances
174 }
175
176 pub async fn accounts(&self) -> Vec<AccountOwner> {
178 let chain_state = self
179 .validator
180 .worker()
181 .chain_state_view(self.id())
182 .await
183 .expect("Failed to read chain state");
184
185 chain_state
186 .execution_state
187 .system
188 .balances
189 .indices()
190 .await
191 .expect("Failed to list accounts on the chain")
192 }
193
194 pub async fn all_owner_balances(&self) -> HashMap<AccountOwner, Amount> {
196 self.owner_balances(self.accounts().await)
197 .await
198 .into_iter()
199 .map(|(owner, balance)| {
200 (
201 owner,
202 balance.expect("`accounts` should only return accounts with non-zero balance"),
203 )
204 })
205 .collect()
206 }
207
208 pub async fn add_block(
213 &self,
214 block_builder: impl FnOnce(&mut BlockBuilder),
215 ) -> ConfirmedBlockCertificate {
216 self.try_add_block(block_builder)
217 .await
218 .expect("Failed to execute block.")
219 }
220
221 pub async fn add_block_with_blobs(
226 &self,
227 block_builder: impl FnOnce(&mut BlockBuilder),
228 blobs: Vec<Blob>,
229 ) -> ConfirmedBlockCertificate {
230 self.try_add_block_with_blobs(block_builder, blobs)
231 .await
232 .expect("Failed to execute block.")
233 }
234
235 pub async fn try_add_block(
240 &self,
241 block_builder: impl FnOnce(&mut BlockBuilder),
242 ) -> Result<ConfirmedBlockCertificate, WorkerError> {
243 self.try_add_block_with_blobs(block_builder, vec![]).await
244 }
245
246 async fn try_add_block_with_blobs(
254 &self,
255 block_builder: impl FnOnce(&mut BlockBuilder),
256 blobs: Vec<Blob>,
257 ) -> Result<ConfirmedBlockCertificate, WorkerError> {
258 let mut tip = self.tip.lock().await;
259 let mut block = BlockBuilder::new(
260 self.description.id(),
261 self.key_pair.public().into(),
262 self.epoch().await,
263 tip.as_ref(),
264 self.validator.clone(),
265 );
266
267 block_builder(&mut block);
268
269 let certificate = Box::pin(block.try_sign(&blobs)).await?;
271
272 let result = self
273 .validator
274 .worker()
275 .fully_handle_certificate_with_notifications(certificate.clone(), &())
276 .await;
277 if let Err(WorkerError::BlobsNotFound(_)) = &result {
278 self.validator.storage().maybe_write_blobs(&blobs).await?;
279 self.validator
280 .worker()
281 .fully_handle_certificate_with_notifications(certificate.clone(), &())
282 .await
283 .expect("Rejected certificate");
284 } else {
285 result.expect("Rejected certificate");
286 }
287
288 *tip = Some(certificate.clone());
289
290 Ok(certificate)
291 }
292
293 pub async fn handle_received_messages(&self) {
298 let chain_id = self.id();
299 let (information, _) = self
300 .validator
301 .worker()
302 .handle_chain_info_query(ChainInfoQuery::new(chain_id).with_pending_message_bundles())
303 .await
304 .expect("Failed to query chain's pending messages");
305 let messages = information.info.requested_pending_message_bundles;
306 if messages.is_empty() {
309 return;
310 }
311 self.add_block(|block| {
312 block.with_incoming_bundles(messages);
313 })
314 .await;
315 }
316
317 pub async fn handle_new_events(&self) {
321 let chain_id = self.id();
322 let worker = self.validator.worker();
323 let subscription_map = worker
324 .chain_state_view(chain_id)
325 .await
326 .expect("Failed to query chain state view")
327 .execution_state
328 .system
329 .event_subscriptions
330 .index_values()
331 .await
332 .expect("Failed to query chain's event subscriptions");
333 let futures = subscription_map
335 .into_iter()
336 .map(|((chain_id, stream_id), subscriptions)| {
337 let worker = worker.clone();
338 async move {
339 worker
340 .chain_state_view(chain_id)
341 .await
342 .expect("Failed to query chain state view")
343 .execution_state
344 .stream_event_counts
345 .get(&stream_id)
346 .await
347 .expect("Failed to query chain's event counts")
348 .filter(|next_index| *next_index > subscriptions.next_index)
349 .map(|next_index| (chain_id, stream_id, next_index))
350 }
351 });
352 let updates = future::join_all(futures)
353 .await
354 .into_iter()
355 .flatten()
356 .collect::<Vec<_>>();
357 assert!(!updates.is_empty(), "No new events to process");
358
359 self.add_block(|block| {
360 block.with_system_operation(SystemOperation::UpdateStreams(updates));
361 })
362 .await;
363 }
364
365 pub async fn publish_current_module<Abi, Parameters, InstantiationArgument>(
371 &self,
372 ) -> ModuleId<Abi, Parameters, InstantiationArgument> {
373 self.publish_bytecode_files_in(".").await
374 }
375
376 pub async fn publish_bytecode_files_in<Abi, Parameters, InstantiationArgument>(
382 &self,
383 repository_path: impl AsRef<Path>,
384 ) -> ModuleId<Abi, Parameters, InstantiationArgument> {
385 let repository_path = fs::canonicalize(repository_path)
386 .await
387 .expect("Failed to obtain absolute application repository path");
388 Self::build_bytecode_files_in(&repository_path).await;
389 let (contract, service) = self.find_bytecode_files_in(&repository_path).await;
390 let contract_blob = Blob::new_contract_bytecode(contract);
391 let service_blob = Blob::new_service_bytecode(service);
392 let contract_blob_hash = contract_blob.id().hash;
393 let service_blob_hash = service_blob.id().hash;
394 let vm_runtime = VmRuntime::Wasm;
395
396 let module_id = ModuleId::new(contract_blob_hash, service_blob_hash, vm_runtime);
397
398 let certificate = self
399 .add_block_with_blobs(
400 |block| {
401 block.with_system_operation(SystemOperation::PublishModule { module_id });
402 },
403 vec![contract_blob, service_blob],
404 )
405 .await;
406
407 let block = certificate.inner().block();
408 assert_eq!(block.messages().len(), 1);
409 assert_eq!(block.messages()[0].len(), 0);
410
411 module_id.with_abi()
412 }
413
414 async fn build_bytecode_files_in(repository: &Path) {
416 let output = std::process::Command::new("cargo")
417 .args(["build", "--release", "--target", "wasm32-unknown-unknown"])
418 .current_dir(repository)
419 .output()
420 .expect("Failed to build Wasm binaries");
421
422 if !output.status.success() {
423 panic!(
424 "Failed to build bytecode binaries.\nstdout: {}\nstderr: {}",
425 String::from_utf8_lossy(&output.stdout),
426 String::from_utf8_lossy(&output.stderr)
427 );
428 }
429 }
430
431 async fn find_bytecode_files_in(
437 &self,
438 repository: &Path,
439 ) -> (CompressedBytecode, CompressedBytecode) {
440 let manifest_path = repository.join("Cargo.toml");
441 let cargo_manifest =
442 Manifest::from_path(manifest_path).expect("Failed to load Cargo.toml manifest");
443
444 let binaries = cargo_manifest
445 .bin
446 .into_iter()
447 .filter_map(|binary| binary.name)
448 .filter(|name| name.ends_with("service") || name.ends_with("contract"))
449 .collect::<Vec<_>>();
450
451 assert_eq!(
452 binaries.len(),
453 2,
454 "Could not figure out contract and service bytecode binaries.\
455 Please specify them manually using `publish_module`."
456 );
457
458 let (contract_binary, service_binary) = if binaries[0].ends_with("contract") {
459 (&binaries[0], &binaries[1])
460 } else {
461 (&binaries[1], &binaries[0])
462 };
463
464 let base_path = self
465 .find_output_directory_of(repository)
466 .await
467 .expect("Failed to look for output binaries");
468 let contract_path = base_path.join(format!("{}.wasm", contract_binary));
469 let service_path = base_path.join(format!("{}.wasm", service_binary));
470
471 let contract = Bytecode::load_from_file(contract_path)
472 .await
473 .expect("Failed to load contract bytecode from file");
474 let service = Bytecode::load_from_file(service_path)
475 .await
476 .expect("Failed to load service bytecode from file");
477
478 tokio::task::spawn_blocking(move || (contract.compress(), service.compress()))
479 .await
480 .expect("Failed to compress bytecode files")
481 }
482
483 async fn find_output_directory_of(&self, repository: &Path) -> Result<PathBuf, io::Error> {
490 let output_sub_directory = Path::new("target/wasm32-unknown-unknown/release");
491 let mut current_directory = repository;
492 let mut output_path = current_directory.join(output_sub_directory);
493
494 while !fs::try_exists(&output_path).await? {
495 current_directory = current_directory.parent().unwrap_or_else(|| {
496 panic!(
497 "Failed to find Wasm binary output directory in {}",
498 repository.display()
499 )
500 });
501
502 output_path = current_directory.join(output_sub_directory);
503 }
504
505 Ok(output_path)
506 }
507
508 pub async fn get_tip_height(&self) -> BlockHeight {
510 self.tip
511 .lock()
512 .await
513 .as_ref()
514 .expect("Block was not successfully added")
515 .inner()
516 .block()
517 .header
518 .height
519 }
520
521 pub async fn create_application<Abi, Parameters, InstantiationArgument>(
532 &mut self,
533 module_id: ModuleId<Abi, Parameters, InstantiationArgument>,
534 parameters: Parameters,
535 instantiation_argument: InstantiationArgument,
536 required_application_ids: Vec<ApplicationId>,
537 ) -> ApplicationId<Abi>
538 where
539 Abi: ContractAbi,
540 Parameters: Serialize,
541 InstantiationArgument: Serialize,
542 {
543 let parameters = serde_json::to_vec(¶meters).unwrap();
544 let instantiation_argument = serde_json::to_vec(&instantiation_argument).unwrap();
545
546 let creation_certificate = self
547 .add_block(|block| {
548 block.with_system_operation(SystemOperation::CreateApplication {
549 module_id: module_id.forget_abi(),
550 parameters: parameters.clone(),
551 instantiation_argument,
552 required_application_ids: required_application_ids.clone(),
553 });
554 })
555 .await;
556
557 let block = creation_certificate.inner().block();
558 assert_eq!(block.messages().len(), 1);
559 assert!(block.messages()[0].is_empty());
560
561 let description = ApplicationDescription {
562 module_id: module_id.forget_abi(),
563 creator_chain_id: block.header.chain_id,
564 block_height: block.header.height,
565 application_index: 0,
566 parameters,
567 required_application_ids,
568 };
569
570 ApplicationId::<()>::from(&description).with_abi()
571 }
572
573 pub async fn is_closed(&self) -> bool {
575 let chain = self
576 .validator
577 .worker()
578 .chain_state_view(self.id())
579 .await
580 .expect("Failed to load chain");
581 *chain.execution_state.system.closed.get()
582 }
583
584 pub async fn query<Abi>(
588 &self,
589 application_id: ApplicationId<Abi>,
590 query: Abi::Query,
591 ) -> QueryOutcome<Abi::QueryResponse>
592 where
593 Abi: ServiceAbi,
594 {
595 self.try_query(application_id, query)
596 .await
597 .expect("Failed to execute application service query")
598 }
599
600 pub async fn try_query<Abi>(
604 &self,
605 application_id: ApplicationId<Abi>,
606 query: Abi::Query,
607 ) -> Result<QueryOutcome<Abi::QueryResponse>, TryQueryError>
608 where
609 Abi: ServiceAbi,
610 {
611 let query_bytes = serde_json::to_vec(&query)?;
612
613 let QueryOutcome {
614 response,
615 operations,
616 } = self
617 .validator
618 .worker()
619 .query_application(
620 self.id(),
621 Query::User {
622 application_id: application_id.forget_abi(),
623 bytes: query_bytes,
624 },
625 )
626 .await?;
627
628 let deserialized_response = match response {
629 QueryResponse::User(bytes) => {
630 serde_json::from_slice(&bytes).expect("Failed to deserialize query response")
631 }
632 QueryResponse::System(_) => {
633 unreachable!("User query returned a system response")
634 }
635 };
636
637 Ok(QueryOutcome {
638 response: deserialized_response,
639 operations,
640 })
641 }
642
643 pub async fn graphql_query<Abi>(
647 &self,
648 application_id: ApplicationId<Abi>,
649 query: impl Into<async_graphql::Request>,
650 ) -> QueryOutcome<serde_json::Value>
651 where
652 Abi: ServiceAbi<Query = async_graphql::Request, QueryResponse = async_graphql::Response>,
653 {
654 let query = query.into();
655 let query_str = query.query.clone();
656
657 self.try_graphql_query(application_id, query)
658 .await
659 .unwrap_or_else(|error| panic!("Service query {query_str:?} failed: {error}"))
660 }
661
662 pub async fn try_graphql_query<Abi>(
666 &self,
667 application_id: ApplicationId<Abi>,
668 query: impl Into<async_graphql::Request>,
669 ) -> Result<QueryOutcome<serde_json::Value>, TryGraphQLQueryError>
670 where
671 Abi: ServiceAbi<Query = async_graphql::Request, QueryResponse = async_graphql::Response>,
672 {
673 let query = query.into();
674 let QueryOutcome {
675 response,
676 operations,
677 } = self.try_query(application_id, query).await?;
678
679 if !response.errors.is_empty() {
680 return Err(TryGraphQLQueryError::Service(response.errors));
681 }
682 let json_response = response.data.into_json()?;
683
684 Ok(QueryOutcome {
685 response: json_response,
686 operations,
687 })
688 }
689
690 pub async fn graphql_mutation<Abi>(
695 &self,
696 application_id: ApplicationId<Abi>,
697 query: impl Into<async_graphql::Request>,
698 ) -> ConfirmedBlockCertificate
699 where
700 Abi: ServiceAbi<Query = async_graphql::Request, QueryResponse = async_graphql::Response>,
701 {
702 self.try_graphql_mutation(application_id, query)
703 .await
704 .expect("Failed to execute service GraphQL mutation")
705 }
706
707 pub async fn try_graphql_mutation<Abi>(
712 &self,
713 application_id: ApplicationId<Abi>,
714 query: impl Into<async_graphql::Request>,
715 ) -> Result<ConfirmedBlockCertificate, TryGraphQLMutationError>
716 where
717 Abi: ServiceAbi<Query = async_graphql::Request, QueryResponse = async_graphql::Response>,
718 {
719 let QueryOutcome { operations, .. } = self.try_graphql_query(application_id, query).await?;
720
721 let certificate = self
722 .try_add_block(|block| {
723 for operation in operations {
724 match operation {
725 Operation::User {
726 application_id,
727 bytes,
728 } => {
729 block.with_raw_operation(application_id, bytes);
730 }
731 Operation::System(system_operation) => {
732 block.with_system_operation(*system_operation);
733 }
734 }
735 }
736 })
737 .await?;
738
739 Ok(certificate)
740 }
741}
742
743#[derive(Debug, thiserror::Error)]
745pub enum TryQueryError {
746 #[error("Failed to serialize query request")]
748 Serialization(#[from] serde_json::Error),
749
750 #[error("Failed to execute service query")]
752 Execution(#[from] WorkerError),
753}
754
755#[derive(Debug, thiserror::Error)]
757pub enum TryGraphQLQueryError {
758 #[error("Failed to serialize GraphQL query request")]
760 RequestSerialization(#[source] serde_json::Error),
761
762 #[error("Failed to execute service query")]
764 Execution(#[from] WorkerError),
765
766 #[error("Unexpected non-JSON service query response")]
768 ResponseDeserialization(#[from] serde_json::Error),
769
770 #[error("Service returned errors: {_0:#?}")]
772 Service(Vec<async_graphql::ServerError>),
773}
774
775impl From<TryQueryError> for TryGraphQLQueryError {
776 fn from(query_error: TryQueryError) -> Self {
777 match query_error {
778 TryQueryError::Serialization(error) => {
779 TryGraphQLQueryError::RequestSerialization(error)
780 }
781 TryQueryError::Execution(error) => TryGraphQLQueryError::Execution(error),
782 }
783 }
784}
785
786impl TryGraphQLQueryError {
787 pub fn expect_execution_error(self) -> ExecutionError {
793 let TryGraphQLQueryError::Execution(worker_error) = self else {
794 panic!("Expected an `ExecutionError`. Got: {self:#?}");
795 };
796
797 worker_error.expect_execution_error(ChainExecutionContext::Query)
798 }
799}
800
801#[derive(Debug, thiserror::Error)]
803pub enum TryGraphQLMutationError {
804 #[error(transparent)]
806 Query(#[from] TryGraphQLQueryError),
807
808 #[error("Failed to propose block with operations scheduled by the GraphQL mutation")]
810 Proposal(#[from] WorkerError),
811}
812
813impl TryGraphQLMutationError {
814 pub fn expect_proposal_execution_error(self, transaction_index: u32) -> ExecutionError {
820 let TryGraphQLMutationError::Proposal(proposal_error) = self else {
821 panic!("Expected an `ExecutionError` during the block proposal. Got: {self:#?}");
822 };
823
824 proposal_error.expect_execution_error(ChainExecutionContext::Operation(transaction_index))
825 }
826}