Skip to content

Commit

Permalink
feat: shutdown peers gracefully
Browse files Browse the repository at this point in the history 
Signed-off-by: 0x009922 <[email protected]>
  • Loading branch information
@0x009922
0x009922 committed Oct 1, 2024
1 parent a104eaf commit 9d61ee4
Show file tree
Hide file tree
Showing 5 changed files with 191 additions and 135 deletions.
15 changes: 14 additions & 1 deletion Cargo.lock
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

18 changes: 9 additions & 9 deletions crates/iroha/tests/integration/extra_functional/restart_peer.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -20,24 +20,24 @@ async fn restarted_peer_should_restore_its_state() -> Result<()> {
let client = peer_a.client();
let asset_definition_clone = asset_definition_id.clone();
spawn_blocking(move || {
let tx = client.build_transaction(
[
InstructionBox::from(Register::asset_definition(AssetDefinition::numeric(
client
.submit_all_blocking::<InstructionBox>([
Register::asset_definition(AssetDefinition::numeric(
asset_definition_clone.clone(),
))),
))
.into(),
Mint::asset_numeric(
quantity,
AssetId::new(asset_definition_clone, ALICE_ID.clone()),
)
.into(),
],
<_>::default(),
);
client.submit_transaction_blocking(&tx).unwrap();
])
.unwrap();
})
.await?;

network.ensure_blocks(2).await?;

// shutdown all
network.shutdown().await;

// restart another one, **without a genesis** even
Expand Down
175 changes: 85 additions & 90 deletions crates/iroha/tests/integration/extra_functional/unregister_peer.rs
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,7 @@
use eyre::{eyre, Result};
use std::time::Duration;

use assert_matches::assert_matches;
use eyre::Result;
use iroha::{
client,
client::Client,
Expand All @@ -7,96 +10,107 @@ use iroha::{
use iroha_test_network::{NetworkBuilder, NetworkPeer};
use iroha_test_samples::gen_account_in;
use nonzero_ext::nonzero;
use tokio::{task::spawn_blocking, time::sleep};

#[tokio::test]
async fn network_stable_after_add_and_after_remove_peer() -> Result<()> {
const PIPELINE_TIME: Duration = Duration::from_millis(300);

#[test]
fn network_stable_after_add_and_after_remove_peer() -> Result<()> {
// Given a network
let (mut network, rt) = NetworkBuilder::new()
.with_default_pipeline_time()
let mut network = NetworkBuilder::new()
.with_pipeline_time(PIPELINE_TIME)
.with_peers(4)
.start_blocking()?;
let main_client = network.client();
let (account_id, asset_definition_id) = init(&main_client)?;
rt.block_on(async { network.ensure_blocks(2).await })?;
.with_genesis_instruction(SetParameter::new(Parameter::Block(
BlockParameter::MaxTransactions(nonzero!(1_u64)),
)))
.start()
.await?;
let client = network.client();

// When assets are minted
mint(
&main_client,
&asset_definition_id,
&account_id,
numeric!(100),
)?;
rt.block_on(async { network.ensure_blocks(3).await })?;
let (account, _account_keypair) = gen_account_in("domain");
let asset_def: AssetDefinitionId = "xor#domain".parse()?;
{
let client = client.clone();
let account = account.clone();
let asset_def = asset_def.clone();
spawn_blocking(move || {
client.submit_all_blocking::<InstructionBox>([
Register::domain(Domain::new("domain".parse()?)).into(),
Register::account(Account::new(account)).into(),
Register::asset_definition(AssetDefinition::numeric(asset_def)).into(),
])
})
.await??; // blocks=2
}

// When assets are minted
mint(&client, &asset_def, &account, numeric!(100)).await?;
network.ensure_blocks(3).await?;
// and a new peer is registered
let new_peer = NetworkPeer::generate();
let new_peer_id = new_peer.id();
let new_peer_client = new_peer.client();
network.add_peer(new_peer.clone());
rt.block_on(async { new_peer.start(network.config(), None).await });
main_client.submit_blocking(Register::peer(Peer::new(new_peer_id.clone())))?;
rt.block_on(async { network.ensure_blocks(4).await })?;
network.add_peer(&new_peer);
new_peer.start(network.config(), None).await;
{
let client = client.clone();
let id = new_peer_id.clone();
spawn_blocking(move || client.submit_blocking(Register::peer(Peer::new(id)))).await??;
}
network.ensure_blocks(4).await?;
// Then the new peer should already have the mint result.
check_assets(
&new_peer_client,
&account_id,
&asset_definition_id,
numeric!(100),
)?;
assert_eq!(
find_asset(&new_peer_client, &account, &asset_def).await?,
numeric!(100)
);

// Also, when a peer is unregistered
let remove_peer = Unregister::peer(new_peer_id);
main_client.submit_blocking(remove_peer)?;
rt.block_on(async { network.ensure_blocks(5).await })?;
// When a peer is unregistered
{
let client = client.clone();
spawn_blocking(move || client.submit_blocking(Unregister::peer(new_peer_id))).await??;
// blocks=6
}
network.remove_peer(&new_peer);
// We can mint without an error.
mint(
&main_client,
&asset_definition_id,
&account_id,
numeric!(200),
)?;
mint(&client, &asset_def, &account, numeric!(200)).await?;
// Assets are increased on the main network.
check_assets(
&main_client,
&account_id,
&asset_definition_id,
numeric!(300),
)?;
network.ensure_blocks(6).await?;
assert_eq!(
find_asset(&client, &account, &asset_def).await?,
numeric!(300)
);
// But not on the unregistered peer's network.
std::thread::sleep(network.pipeline_time());
check_assets(
&new_peer_client,
&account_id,
&asset_definition_id,
numeric!(100),
)?;
sleep(PIPELINE_TIME * 5).await;
assert_eq!(
find_asset(&new_peer_client, &account, &asset_def).await?,
numeric!(100)
);

Ok(())
}

fn check_assets(
async fn find_asset(
client: &Client,
account_id: &AccountId,
asset_definition_id: &AssetDefinitionId,
quantity: Numeric,
) -> Result<()> {
let account_id = account_id.clone();
let assets = client
.query(client::asset::all())
.filter_with(|asset| asset.id.account.eq(account_id.clone()))
.execute_all()?;
account: &AccountId,
asset_definition: &AssetDefinitionId,
) -> Result<Numeric> {
let account_id = account.clone();
let client = client.clone();
let asset = spawn_blocking(move || {
client
.query(client::asset::all())
.filter_with(|asset| asset.id.account.eq(account_id.clone()))
.execute_all()
})
.await??
.into_iter()
.find(|asset| asset.id().definition() == asset_definition)
.expect("asset should be there");

if assets.iter().any(|asset| {
asset.id().definition() == asset_definition_id
&& *asset.value() == AssetValue::Numeric(quantity)
}) {
Ok(())
} else {
Err(eyre!("assets mismatch"))
}
assert_matches!(asset.value(), AssetValue::Numeric(quantity) => Ok(quantity.clone()))
}

fn mint(
async fn mint(
client: &Client,
asset_definition_id: &AssetDefinitionId,
account_id: &AccountId,
Expand All @@ -106,26 +120,7 @@ fn mint(
quantity,
AssetId::new(asset_definition_id.clone(), account_id.clone()),
);
client.submit_blocking(mint_asset)?;
let client = client.clone();
spawn_blocking(move || client.submit_blocking(mint_asset)).await??;
Ok(())
}

fn init(client: &Client) -> Result<(AccountId, AssetDefinitionId)> {
let set_max_txns_in_block = SetParameter::new(Parameter::Block(
BlockParameter::MaxTransactions(nonzero!(1_u64)),
));

let create_domain = Register::domain(Domain::new("domain".parse()?));
let (account_id, _account_keypair) = gen_account_in("domain");
let create_account = Register::account(Account::new(account_id.clone()));
let asset_definition_id: AssetDefinitionId = "xor#domain".parse()?;
let create_asset =
Register::asset_definition(AssetDefinition::numeric(asset_definition_id.clone()));
client.submit_all_blocking::<InstructionBox>([
set_max_txns_in_block.into(),
create_domain.into(),
create_account.into(),
create_asset.into(),
])?;
Ok((account_id, asset_definition_id))
}
1 change: 1 addition & 0 deletions crates/iroha_test_network/Cargo.toml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -38,3 +38,4 @@ fslock = "0.2.1"
serde = { workspace = true, features = ["derive"] }
derive_more = { workspace = true }
which = "6.0.3"
nix = { version = "0.29.0", features = ["signal"] }
Loading

0 comments on commit 9d61ee4

Please sign in to comment.