2114923834

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/iotaledger/wallet.rs/commit/<a class=hub.com/iotaledger/wallet.rs/commit/d3aeda1621f19352964e42da9d0d36cd1a7fde1f">d3aeda162<a href="https://github.com/iotaledger/wallet.rs/commit/d3aeda1621f19352964e42da9d0d36cd1a7fde1f">">Merge </a><a class="double-link" href="https://github.com/iotaledger/wallet.rs/commit/<a class="double-link" href="https://github.com/iotaledger/wallet.rs/commit/f104514996723f16da661032d8e8c483bddf9d6d">f10451499</a>">f10451499</a><a href="https://github.com/iotaledger/wallet.rs/commit/d3aeda1621f19352964e42da9d0d36cd1a7fde1f"> into b7db45051">b7db45051</a>

Pull Request Pull Request #978: Sync foundry outputs

Run Details

33 of 33 new or added lines in 4 files covered. (100.0%)

711 of 3502 relevant lines covered (20.3%)

8584.36 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/account/operations/syncing/addresses.rs

// Copyright 2021 IOTA Stiftung
// SPDX-License-Identifier: Apache-2.0

use crate::account::{
    constants::PARALLEL_REQUESTS_AMOUNT, handle::AccountHandle, operations::syncing::SyncOptions,
    types::address::AddressWithBalance, OutputData,
};
#[cfg(feature = "events")]
use crate::events::types::WalletEvent;

use iota_client::{
    bee_message::{
        address::{Address, AliasAddress},
        output::{Output, OutputId},
    },
    node_api::indexer_api::query_parameters::QueryParameter,
};

use std::{collections::HashSet, time::Instant};

impl AccountHandle {
    /// Get the addresses that should be synced with the current known unspent output ids
    /// Also adds alias and nft addresses from unspent alias or nft outputs that have no Timelock, Expiration or
    /// StorageDepositReturn [`UnlockCondition`]
    pub(crate) async fn get_addresses_to_sync(&self, options: &SyncOptions) -> crate::Result<Vec<AddressWithBalance>> {
        log::debug!("[SYNC] get_addresses_to_sync");

        let mut addresses_before_syncing = self.list_addresses().await?;

        // If custom addresses are provided check if they are in the account and only use them
        if !options.addresses.is_empty() {
            let mut specific_addresses_to_sync = HashSet::new();
            for bech32_address in &options.addresses {
                let (_bech32_hrp, address) = Address::try_from_bech32(bech32_address)?;
                match addresses_before_syncing.iter().find(|a| a.address.inner == address) {
                    Some(address) => {
                        specific_addresses_to_sync.insert(address.clone());
                    }
                    None => return Err(crate::Error::AddressNotFoundInAccount(bech32_address.to_string())),
                }
            }
            addresses_before_syncing = specific_addresses_to_sync.into_iter().collect();
        }

        // Filter addresses when address_start_index is not 0 so we skip these addresses
        // If we force syncing, we want to sync all addresses
        if options.addresses.is_empty() && options.address_start_index != 0 && !options.force_syncing {
            addresses_before_syncing = addresses_before_syncing
                .into_iter()
                .filter(|a| a.key_index >= options.address_start_index)
                .collect();
        }

        // Check if selected addresses contains addresses with balance so we can correctly update them
        let addresses_with_balance = self.list_addresses_with_balance().await?;
        let mut addresses_with_old_output_ids = Vec::new();
        for address in addresses_before_syncing {
            let mut output_ids = Vec::new();
            // Add currently known unspent output ids, so we can later compare them with the new output ids and see if
            // one got spent (is missing in the new returned output ids)
            if let Some(address_with_balance) = addresses_with_balance.iter().find(|a| a.address == address.address) {
                output_ids = address_with_balance.output_ids.to_vec();
            }
            addresses_with_old_output_ids.push(AddressWithBalance {
                address: address.address,
                key_index: address.key_index,
                internal: address.internal,
                amount: 0,
                output_ids,
            })
        }

        Ok(addresses_with_old_output_ids)
    }

    /// Get the current output ids for provided addresses and only returns addresses that have outputs now and return
    /// spent outputs separated
    pub(crate) async fn get_address_output_ids(
        &self,
        options: &SyncOptions,
        addresses_with_balance: Vec<AddressWithBalance>,
    ) -> crate::Result<(Vec<AddressWithBalance>, Vec<OutputId>)> {
        log::debug!("[SYNC] start get_address_output_ids");
        let address_output_ids_start_time = Instant::now();
        let account = self.read().await;

        #[cfg(feature = "events")]
        let (account_index, consolidation_threshold) =
            (account.index, account.account_options.output_consolidation_threshold);
        drop(account);

        let mut addresses_with_outputs = Vec::new();
        // spent outputs or alias/nft/foundries that don't get synced anymore, because of other sync options
        let mut spent_or_not_anymore_synced_outputs = Vec::new();
        // We split the addresses into chunks so we don't get timeouts if we have thousands
        for addresses_chunk in &mut addresses_with_balance
            .chunks(PARALLEL_REQUESTS_AMOUNT)
            .map(|x: &[AddressWithBalance]| x.to_vec())
        {
            let mut tasks = Vec::new();
            for address in addresses_chunk {
                let client = self.client.clone();
                let account_handle = self.clone();
                let sync_options = options.clone();
                tasks.push(async move {
                    tokio::spawn(async move {
                        let client = client;
                        // Get basic outputs
                        let mut output_ids = client
                            .output_ids(vec![QueryParameter::Address(address.address.to_bech32())])
                            .await?;

                        if sync_options.sync_aliases_and_nfts {
                            // Get nft outputs
                            output_ids.extend(
                                client
                                    .nfts_output_ids(vec![QueryParameter::Address(address.address.to_bech32())])
                                    .await?
                                    .into_iter(),
                            );
                            // Get alias outputs
                            let alias_output_ids = client
                                .aliases_output_ids(vec![
                                    QueryParameter::StateController(address.address.to_bech32()),
                                    QueryParameter::Governor(address.address.to_bech32()),
                                ])
                                .await?;
                            output_ids.extend(alias_output_ids.clone().into_iter());

                            // get possible foundries of the alias outputs
                            let (alias_output_responses, _already_known_balance) =
                                account_handle.get_outputs(alias_output_ids, false).await?;
                            let mut foundry_output_ids = Vec::new();
                            for alias_output_response in alias_output_responses {
                                let output = Output::try_from(&alias_output_response.output)?;
                                if let Output::Alias(alias_output) = output {
                                    let alias_address = Address::Alias(AliasAddress::from(*alias_output.alias_id()));
                                    foundry_output_ids.extend(
                                        client
                                            .foundries_output_ids(vec![QueryParameter::Address(
                                                alias_address.to_bech32(address.address.bech32_hrp.clone()),
                                            )])
                                            .await?
                                            .into_iter(),
                                    );
                                }
                            }
                            output_ids.extend(foundry_output_ids.into_iter());
                        }
                        crate::Result::Ok((address, output_ids))
                    })
                    .await
                });
            }
            let results = futures::future::try_join_all(tasks).await?;
            for res in results {
                let (mut address, output_ids): (AddressWithBalance, Vec<OutputId>) = res?;
                #[cfg(feature = "events")]
                if output_ids.len() > consolidation_threshold {
                    self.event_emitter
                        .lock()
                        .await
                        .emit(account_index, WalletEvent::ConsolidationRequired);
                }
                // only return addresses with outputs
                if !output_ids.is_empty() {
                    // outputs we had before, but now not anymore, got spent or are alias/nft/foundries that don't get
                    // synced anymore because of other sync options
                    for output_id in address.output_ids {
                        if !output_ids.contains(&output_id) {
                            spent_or_not_anymore_synced_outputs.push(output_id);
                        }
                    }
                    address.output_ids = output_ids;
                    addresses_with_outputs.push(address);
                } else {
                    // outputs we had before, but now not anymore, got spent or are alias/nft/foundries that don't get
                    // synced anymore because of other sync options
                    spent_or_not_anymore_synced_outputs.extend(address.output_ids.into_iter());
                }
            }
        }
        if options.sync_aliases_and_nfts {
            log::debug!("[SYNC] spent outputs: {:?}", spent_or_not_anymore_synced_outputs);
        } else {
            log::debug!(
                "[SYNC] spent or not anymore synced alias/nft/foundries outputs: {:?}",
                spent_or_not_anymore_synced_outputs
            );
        }
        log::debug!(
            "[SYNC] finished get_address_output_ids in {:.2?}",
            address_output_ids_start_time.elapsed()
        );
        Ok((addresses_with_outputs, spent_or_not_anymore_synced_outputs))
    }

    /// Get outputs from addresses
    pub(crate) async fn get_addresses_outputs(
        &self,
        addresses_with_balance: Vec<AddressWithBalance>,
    ) -> crate::Result<(Vec<AddressWithBalance>, Vec<OutputData>)> {
        log::debug!("[SYNC] start get_addresses_outputs");
        let address_outputs_start_time = Instant::now();

        let mut addresses_with_outputs = Vec::new();
        let mut outputs_data = Vec::new();

        // We split the addresses into chunks so we don't get timeouts if we have thousands
        for addresses_chunk in &mut addresses_with_balance
            .chunks(PARALLEL_REQUESTS_AMOUNT)
            .map(|x: &[AddressWithBalance]| x.to_vec())
        {
            let mut tasks = Vec::new();
            for mut address in addresses_chunk {
                let account_handle = self.clone();
                tasks.push(async move {
                    tokio::spawn(async move {
                        let (output_responses, already_known_balance) =
                            account_handle.get_outputs(address.output_ids.clone(), false).await?;
                        let outputs = account_handle
                            .output_response_to_output_data(output_responses, &address)
                            .await?;
                        // Add balance from new outputs together with balance from already known outputs
                        address.amount =
                            outputs.iter().map(|output| output.amount).sum::<u64>() + already_known_balance;
                        crate::Result::Ok((address, outputs))
                    })
                    .await
                });
            }
            let results = futures::future::try_join_all(tasks).await?;
            for res in results {
                let (address, outputs): (AddressWithBalance, Vec<OutputData>) = res?;
                addresses_with_outputs.push(address);
                outputs_data.extend(outputs.into_iter());
            }
        }
        log::debug!(
            "[SYNC] finished get_address_output_ids in {:.2?}",
            address_outputs_start_time.elapsed()
        );
        Ok((addresses_with_outputs, outputs_data))
    }
}

1	// Copyright 2021 IOTA Stiftung
2	// SPDX-License-Identifier: Apache-2.0
3
4	use crate::account::{
5	constants::PARALLEL_REQUESTS_AMOUNT, handle::AccountHandle, operations::syncing::SyncOptions,
6	types::address::AddressWithBalance, OutputData,
7	};
8	#[cfg(feature = "events")]
9	use crate::events::types::WalletEvent;
10
11	use iota_client::{
12	bee_message::{
13	address::{Address, AliasAddress},
14	output::{Output, OutputId},
15	},
16	node_api::indexer_api::query_parameters::QueryParameter,
17	};
18
19	use std::{collections::HashSet, time::Instant};
20
21	impl AccountHandle {
22	/// Get the addresses that should be synced with the current known unspent output ids
23	/// Also adds alias and nft addresses from unspent alias or nft outputs that have no Timelock, Expiration or
24	/// StorageDepositReturn [`UnlockCondition`]
25	pub(crate) async fn get_addresses_to_sync(&self, options: &SyncOptions) -> crate::Result<Vec<AddressWithBalance>> {	×
26	log::debug!("[SYNC] get_addresses_to_sync");	×
27
28	let mut addresses_before_syncing = self.list_addresses().await?;	×
29
30	// If custom addresses are provided check if they are in the account and only use them
31	if !options.addresses.is_empty() {	×
32	let mut specific_addresses_to_sync = HashSet::new();	×
33	for bech32_address in &options.addresses {	×
34	let (_bech32_hrp, address) = Address::try_from_bech32(bech32_address)?;	×
35	match addresses_before_syncing.iter().find(\|a\| a.address.inner == address) {	×
36	Some(address) => {	×
37	specific_addresses_to_sync.insert(address.clone());	×
38	}	×
39	None => return Err(crate::Error::AddressNotFoundInAccount(bech32_address.to_string())),	×
40	}
41	}
42	addresses_before_syncing = specific_addresses_to_sync.into_iter().collect();	×
43	}	×
44
45	// Filter addresses when address_start_index is not 0 so we skip these addresses
46	// If we force syncing, we want to sync all addresses
47	if options.addresses.is_empty() && options.address_start_index != 0 && !options.force_syncing {	×
48	addresses_before_syncing = addresses_before_syncing	×
49	.into_iter()	×
50	.filter(\|a\| a.key_index >= options.address_start_index)	×
51	.collect();	×
52	}	×
53
54	// Check if selected addresses contains addresses with balance so we can correctly update them
55	let addresses_with_balance = self.list_addresses_with_balance().await?;	×
56	let mut addresses_with_old_output_ids = Vec::new();	×
57	for address in addresses_before_syncing {	×
58	let mut output_ids = Vec::new();	×
59	// Add currently known unspent output ids, so we can later compare them with the new output ids and see if
60	// one got spent (is missing in the new returned output ids)
61	if let Some(address_with_balance) = addresses_with_balance.iter().find(\|a\| a.address == address.address) {	×
62	output_ids = address_with_balance.output_ids.to_vec();	×
63	}	×
64	addresses_with_old_output_ids.push(AddressWithBalance {	×
65	address: address.address,	×
66	key_index: address.key_index,	×
67	internal: address.internal,	×
68	amount: 0,	×
69	output_ids,	×
70	})	×
71	}
72
73	Ok(addresses_with_old_output_ids)	×
74	}	×
75
76	/// Get the current output ids for provided addresses and only returns addresses that have outputs now and return
77	/// spent outputs separated
78	pub(crate) async fn get_address_output_ids(	×
79	&self,	×
80	options: &SyncOptions,	×
81	addresses_with_balance: Vec<AddressWithBalance>,	×
82	) -> crate::Result<(Vec<AddressWithBalance>, Vec<OutputId>)> {	×
83	log::debug!("[SYNC] start get_address_output_ids");	×
84	let address_output_ids_start_time = Instant::now();	×
85	let account = self.read().await;	×
86
87	#[cfg(feature = "events")]
88	let (account_index, consolidation_threshold) =	×
89	(account.index, account.account_options.output_consolidation_threshold);	×
90	drop(account);	×
91		×
92	let mut addresses_with_outputs = Vec::new();	×
93	// spent outputs or alias/nft/foundries that don't get synced anymore, because of other sync options	×
94	let mut spent_or_not_anymore_synced_outputs = Vec::new();	×
95	// We split the addresses into chunks so we don't get timeouts if we have thousands
96	for addresses_chunk in &mut addresses_with_balance	×
97	.chunks(PARALLEL_REQUESTS_AMOUNT)	×
98	.map(\|x: &[AddressWithBalance]\| x.to_vec())	×
99	{
100	let mut tasks = Vec::new();	×
101	for address in addresses_chunk {	×
102	let client = self.client.clone();	×
103	let account_handle = self.clone();	×
104	let sync_options = options.clone();	×
105	tasks.push(async move {	×
106	tokio::spawn(async move {	×
107	let client = client;	×
108	// Get basic outputs
109	let mut output_ids = client	×
110	.output_ids(vec![QueryParameter::Address(address.address.to_bech32())])	×
111	.await?;	×
112
113	if sync_options.sync_aliases_and_nfts {	×
114	// Get nft outputs
115	output_ids.extend(
116	client	×
117	.nfts_output_ids(vec![QueryParameter::Address(address.address.to_bech32())])	×
118	.await?	×
119	.into_iter(),	×
120	);
121	// Get alias outputs
122	let alias_output_ids = client	×
123	.aliases_output_ids(vec![	×
124	QueryParameter::StateController(address.address.to_bech32()),	×
125	QueryParameter::Governor(address.address.to_bech32()),	×
126	])	×
127	.await?;	×
128	output_ids.extend(alias_output_ids.clone().into_iter());	×
129
130	// get possible foundries of the alias outputs
131	let (alias_output_responses, _already_known_balance) =	×
132	account_handle.get_outputs(alias_output_ids, false).await?;	×
133	let mut foundry_output_ids = Vec::new();	×
134	for alias_output_response in alias_output_responses {	×
135	let output = Output::try_from(&alias_output_response.output)?;	×
136	if let Output::Alias(alias_output) = output {	×
137	let alias_address = Address::Alias(AliasAddress::from(*alias_output.alias_id()));	×
138	foundry_output_ids.extend(	×
139	client	×
140	.foundries_output_ids(vec![QueryParameter::Address(	×
141	alias_address.to_bech32(address.address.bech32_hrp.clone()),	×
142	)])	×
143	.await?	×
144	.into_iter(),	×
145	);
146	}	×
147	}
148	output_ids.extend(foundry_output_ids.into_iter());	×
149	}	×
150	crate::Result::Ok((address, output_ids))	×
151	})	×
152	.await	×
153	});	×
154	}	×
155	let results = futures::future::try_join_all(tasks).await?;	×
156	for res in results {	×
157	let (mut address, output_ids): (AddressWithBalance, Vec<OutputId>) = res?;	×
158	#[cfg(feature = "events")]
159	if output_ids.len() > consolidation_threshold {	×
160	self.event_emitter	×
161	.lock()	×
162	.await	×
163	.emit(account_index, WalletEvent::ConsolidationRequired);	×
164	}	×
165	// only return addresses with outputs
166	if !output_ids.is_empty() {	×
167	// outputs we had before, but now not anymore, got spent or are alias/nft/foundries that don't get
168	// synced anymore because of other sync options
169	for output_id in address.output_ids {	×
170	if !output_ids.contains(&output_id) {	×
171	spent_or_not_anymore_synced_outputs.push(output_id);	×
172	}	×
173	}
174	address.output_ids = output_ids;	×
175	addresses_with_outputs.push(address);	×
176	} else {	×
177	// outputs we had before, but now not anymore, got spent or are alias/nft/foundries that don't get	×
178	// synced anymore because of other sync options	×
179	spent_or_not_anymore_synced_outputs.extend(address.output_ids.into_iter());	×
180	}	×
181	}
182	}
183	if options.sync_aliases_and_nfts {	×
184	log::debug!("[SYNC] spent outputs: {:?}", spent_or_not_anymore_synced_outputs);	×
185	} else {
186	log::debug!(
187	"[SYNC] spent or not anymore synced alias/nft/foundries outputs: {:?}",	×
188	spent_or_not_anymore_synced_outputs
189	);
190	}
191	log::debug!(
192	"[SYNC] finished get_address_output_ids in {:.2?}",	×
193	address_output_ids_start_time.elapsed()	×
194	);
195	Ok((addresses_with_outputs, spent_or_not_anymore_synced_outputs))	×
196	}	×
197
198	/// Get outputs from addresses
199	pub(crate) async fn get_addresses_outputs(	×
200	&self,	×
201	addresses_with_balance: Vec<AddressWithBalance>,	×
202	) -> crate::Result<(Vec<AddressWithBalance>, Vec<OutputData>)> {	×
203	log::debug!("[SYNC] start get_addresses_outputs");	×
204	let address_outputs_start_time = Instant::now();	×
205		×
206	let mut addresses_with_outputs = Vec::new();	×
207	let mut outputs_data = Vec::new();	×
208
209	// We split the addresses into chunks so we don't get timeouts if we have thousands
210	for addresses_chunk in &mut addresses_with_balance	×
211	.chunks(PARALLEL_REQUESTS_AMOUNT)	×
212	.map(\|x: &[AddressWithBalance]\| x.to_vec())	×
213	{
214	let mut tasks = Vec::new();	×
215	for mut address in addresses_chunk {	×
216	let account_handle = self.clone();	×
217	tasks.push(async move {	×
218	tokio::spawn(async move {	×
219	let (output_responses, already_known_balance) =	×
220	account_handle.get_outputs(address.output_ids.clone(), false).await?;	×
221	let outputs = account_handle	×
222	.output_response_to_output_data(output_responses, &address)	×
223	.await?;	×
224	// Add balance from new outputs together with balance from already known outputs
225	address.amount =	×
226	outputs.iter().map(\|output\| output.amount).sum::<u64>() + already_known_balance;	×
227	crate::Result::Ok((address, outputs))	×
228	})	×
229	.await	×
230	});	×
231	}	×
232	let results = futures::future::try_join_all(tasks).await?;	×
233	for res in results {	×
234	let (address, outputs): (AddressWithBalance, Vec<OutputData>) = res?;	×
235	addresses_with_outputs.push(address);	×
236	outputs_data.extend(outputs.into_iter());	×
237	}
238	}
239	log::debug!(
240	"[SYNC] finished get_address_output_ids in {:.2?}",	×
241	address_outputs_start_time.elapsed()	×
242	);
243	Ok((addresses_with_outputs, outputs_data))	×
244	}	×
245	}

iotaledger / wallet.rs / 2114923834

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous