Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix inconsistency state before commit #1709

Merged
merged 3 commits into from
Jul 22, 2023
Merged

Fix inconsistency state before commit #1709

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
59a7fcc
fix changes in finalize_block
yito88 Jul 12, 2023
f1669e7
add changelog
yito88 Jul 12, 2023
d87df2a
test/shell/finalize_block: add some txs to DB commit test
tzemanovic Jul 13, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .changelog/unreleased/bug-fixes/1709-fix_changes_before_commit.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
- Fix inconsistency state before commit
([\#1709](https://github.com/anoma/namada/issues/1709))
250 changes: 138 additions & 112 deletions apps/src/lib/node/ledger/shell/finalize_block.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,6 @@ where
let tx_hash_key =
replay_protection::get_tx_hash_key(&tx_hash);
self.wl_storage
.storage
.delete(&tx_hash_key)
.expect("Error while deleting tx hash from storage");
}
Expand All @@ -221,16 +220,14 @@ where
processed_tx.header_hash().0,
));
self.wl_storage
.storage
.write(&wrapper_tx_hash_key, vec![])
.write_bytes(&wrapper_tx_hash_key, vec![])
.expect("Error while writing tx hash to storage");

let inner_tx_hash_key = replay_protection::get_tx_hash_key(
&tx.clone().update_header(TxType::Raw).header_hash(),
);
self.wl_storage
.storage
.write(&inner_tx_hash_key, vec![])
.write_bytes(&inner_tx_hash_key, vec![])
.expect("Error while writing tx hash to storage");

#[cfg(not(feature = "mainnet"))]
Expand All @@ -256,11 +253,7 @@ where
match balance.checked_sub(wrapper_fees) {
Some(amount) => {
self.wl_storage
.storage
.write(
&balance_key,
amount.try_to_vec().unwrap(),
)
.write(&balance_key, amount)
.unwrap();
}
None => {
Expand All @@ -271,12 +264,9 @@ where
if reject {
// Burn remaining funds
self.wl_storage
.storage
.write(
&balance_key,
Amount::native_whole(0)
.try_to_vec()
.unwrap(),
Amount::native_whole(0),
)
.unwrap();
tx_event["info"] =
Expand Down Expand Up @@ -481,6 +471,7 @@ where
);
stats.increment_errored_txs();

self.wl_storage.drop_tx();
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Does this only drop the current tx or does it clear the entire wl_storage? From reading the wl_storage code I'm not entirely sure.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

only the current tx

// If transaction type is Decrypted and failed because of
// out of gas, remove its hash from storage to allow
// rewrapping it
Expand All @@ -491,15 +482,16 @@ where
let tx_hash_key =
replay_protection::get_tx_hash_key(&hash);
self.wl_storage
.storage
.delete(&tx_hash_key)
.expect(
"Error while deleting tx hash key from storage",
);
// Apply only to remove its hash,
// since all other changes have already been dropped
self.wl_storage.commit_tx();
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we need to do this?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it is to commit the removal of the tx_hash_key above

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, it's a bit tricky. But we have to commit_tx() to apply changes on the write log when the following commit phase.

}
}

self.wl_storage.drop_tx();
tx_event["gas_used"] = self
.gas_meter
.get_current_transaction_gas()
Expand Down Expand Up @@ -1026,6 +1018,85 @@ mod test_finalize_block {
FinalizeBlock, ProcessedTx,
};

/// Make a wrapper tx and a processed tx from the wrapped tx that can be
/// added to `FinalizeBlock` request.
fn mk_wrapper_tx(
shell: &TestShell,
keypair: &common::SecretKey,
) -> (Tx, ProcessedTx) {
let mut wrapper_tx =
Tx::new(TxType::Wrapper(Box::new(WrapperTx::new(
Fee {
amount: MIN_FEE_AMOUNT,
token: shell.wl_storage.storage.native_token.clone(),
},
keypair.ref_to(),
Epoch(0),
Default::default(),
#[cfg(not(feature = "mainnet"))]
None,
))));
wrapper_tx.header.chain_id = shell.chain_id.clone();
wrapper_tx.set_code(Code::new("wasm_code".as_bytes().to_owned()));
wrapper_tx.set_data(Data::new(
"Encrypted transaction data".as_bytes().to_owned(),
));
wrapper_tx.add_section(Section::Signature(Signature::new(
wrapper_tx.sechashes(),
keypair,
)));
let tx = wrapper_tx.to_bytes();
(
wrapper_tx,
ProcessedTx {
tx,
result: TxResult {
code: ErrorCodes::Ok.into(),
info: "".into(),
},
},
)
}

/// Make a wrapper tx and a processed tx from the wrapped tx that can be
/// added to `FinalizeBlock` request.
fn mk_decrypted_tx(
shell: &mut TestShell,
keypair: &common::SecretKey,
) -> ProcessedTx {
let tx_code = TestWasms::TxNoOp.read_bytes();
let mut outer_tx = Tx::new(TxType::Wrapper(Box::new(WrapperTx::new(
Fee {
amount: MIN_FEE_AMOUNT,
token: shell.wl_storage.storage.native_token.clone(),
},
keypair.ref_to(),
Epoch(0),
Default::default(),
#[cfg(not(feature = "mainnet"))]
None,
))));
outer_tx.header.chain_id = shell.chain_id.clone();
outer_tx.set_code(Code::new(tx_code));
outer_tx.set_data(Data::new(
"Decrypted transaction data".as_bytes().to_owned(),
));
shell.enqueue_tx(outer_tx.clone());
outer_tx.update_header(TxType::Decrypted(DecryptedTx::Decrypted {
#[cfg(not(feature = "mainnet"))]
has_valid_pow: false,
}));
outer_tx.decrypt(<EllipticCurve as PairingEngine>::G2Affine::prime_subgroup_generator())
.expect("Test failed");
ProcessedTx {
tx: outer_tx.to_bytes(),
result: TxResult {
code: ErrorCodes::Ok.into(),
info: "".into(),
},
}
}

/// Check that if a wrapper tx was rejected by [`process_proposal`],
/// check that the correct event is returned. Check that it does
/// not appear in the queue of txs to be decrypted
Expand All @@ -1052,36 +1123,11 @@ mod test_finalize_block {

// create some wrapper txs
for i in 1u64..5 {
let mut wrapper =
Tx::new(TxType::Wrapper(Box::new(WrapperTx::new(
Fee {
amount: MIN_FEE_AMOUNT,
token: shell.wl_storage.storage.native_token.clone(),
},
keypair.ref_to(),
Epoch(0),
Default::default(),
#[cfg(not(feature = "mainnet"))]
None,
))));
wrapper.header.chain_id = shell.chain_id.clone();
wrapper.set_data(Data::new("wasm_code".as_bytes().to_owned()));
wrapper.set_code(Code::new(
format!("transaction data: {}", i).as_bytes().to_owned(),
));
wrapper.add_section(Section::Signature(Signature::new(
wrapper.sechashes(),
&keypair,
)));
let (wrapper, mut processed_tx) = mk_wrapper_tx(&shell, &keypair);
if i > 1 {
processed_txs.push(ProcessedTx {
tx: wrapper.to_bytes(),
result: TxResult {
code: u32::try_from(i.rem_euclid(2))
.expect("Test failed"),
info: "".into(),
},
});
processed_tx.result.code =
u32::try_from(i.rem_euclid(2)).unwrap();
processed_txs.push(processed_tx);
} else {
shell.enqueue_tx(wrapper.clone());
}
Expand Down Expand Up @@ -1247,75 +1293,14 @@ mod test_finalize_block {
.unwrap();

// create two decrypted txs
let tx_code = TestWasms::TxNoOp.read_bytes();
for i in 0..2 {
let mut outer_tx =
Tx::new(TxType::Wrapper(Box::new(WrapperTx::new(
Fee {
amount: MIN_FEE_AMOUNT,
token: shell.wl_storage.storage.native_token.clone(),
},
keypair.ref_to(),
Epoch(0),
Default::default(),
#[cfg(not(feature = "mainnet"))]
None,
))));
outer_tx.header.chain_id = shell.chain_id.clone();
outer_tx.set_code(Code::new(tx_code.clone()));
outer_tx.set_data(Data::new(
format!("Decrypted transaction data: {}", i)
.as_bytes()
.to_owned(),
));
shell.enqueue_tx(outer_tx.clone());
outer_tx.update_header(TxType::Decrypted(DecryptedTx::Decrypted {
#[cfg(not(feature = "mainnet"))]
has_valid_pow: false,
}));
outer_tx.decrypt(<EllipticCurve as PairingEngine>::G2Affine::prime_subgroup_generator())
.expect("Test failed");
processed_txs.push(ProcessedTx {
tx: outer_tx.to_bytes(),
result: TxResult {
code: ErrorCodes::Ok.into(),
info: "".into(),
},
});
for _ in 0..2 {
processed_txs.push(mk_decrypted_tx(&mut shell, &keypair));
}
// create two wrapper txs
for i in 0..2 {
let mut wrapper_tx =
Tx::new(TxType::Wrapper(Box::new(WrapperTx::new(
Fee {
amount: MIN_FEE_AMOUNT,
token: shell.wl_storage.storage.native_token.clone(),
},
keypair.ref_to(),
Epoch(0),
Default::default(),
#[cfg(not(feature = "mainnet"))]
None,
))));
wrapper_tx.header.chain_id = shell.chain_id.clone();
wrapper_tx.set_code(Code::new("wasm_code".as_bytes().to_owned()));
wrapper_tx.set_data(Data::new(
format!("Encrypted transaction data: {}", i)
.as_bytes()
.to_owned(),
));
wrapper_tx.add_section(Section::Signature(Signature::new(
wrapper_tx.sechashes(),
&keypair,
)));
valid_txs.push(wrapper_tx.clone());
processed_txs.push(ProcessedTx {
tx: wrapper_tx.to_bytes(),
result: TxResult {
code: ErrorCodes::Ok.into(),
info: "".into(),
},
});
for _ in 0..2 {
let (tx, processed_tx) = mk_wrapper_tx(&shell, &keypair);
valid_txs.push(tx.clone());
processed_txs.push(processed_tx);
}
// Put the wrapper txs in front of the decrypted txs
processed_txs.rotate_left(2);
Expand Down Expand Up @@ -1734,6 +1719,21 @@ mod test_finalize_block {
shell.wl_storage.storage.next_epoch_min_start_height = BlockHeight(5);
shell.wl_storage.storage.next_epoch_min_start_time = DateTimeUtc::now();

let txs_key = gen_keypair();
// Add unshielded balance for fee payment
let balance_key = token::balance_key(
&shell.wl_storage.storage.native_token,
&Address::from(&txs_key.ref_to()),
);
shell
.wl_storage
.storage
.write(
&balance_key,
Amount::native_whole(1000).try_to_vec().unwrap(),
)
.unwrap();

// Add a proposal to be executed on next epoch change.
let mut add_proposal = |proposal_id, vote| {
let validator = shell.mode.get_validator_address().unwrap().clone();
Expand Down Expand Up @@ -1829,12 +1829,32 @@ mod test_finalize_block {
// Need to supply a proposer address and votes to flow through the
// inflation code
for _ in 0..20 {
// Add some txs
let mut txs = vec![];
// create two decrypted txs
for _ in 0..2 {
txs.push(mk_decrypted_tx(&mut shell, &txs_key));
}
// create two wrapper txs
for _ in 0..2 {
let (_tx, processed_tx) = mk_wrapper_tx(&shell, &txs_key);
txs.push(processed_tx);
}

let req = FinalizeBlock {
txs,
proposer_address: proposer_address.clone(),
votes: votes.clone(),
..Default::default()
};
// merkle tree root before finalize_block
let root_pre = shell.shell.wl_storage.storage.block.tree.root();

let _events = shell.finalize_block(req).unwrap();

// the merkle tree root should not change after finalize_block
let root_post = shell.shell.wl_storage.storage.block.tree.root();
assert_eq!(root_pre.0, root_post.0);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

when I remove the fixes and run the tests they do not yet catch the issue - I think we need to add some txs in here

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

added in f30e158

let new_state = store_block_state(&shell);
// The new state must be unchanged
itertools::assert_equal(
Expand Down Expand Up @@ -2226,6 +2246,8 @@ mod test_finalize_block {
},
};
shell.enqueue_tx(wrapper_tx);
// merkle tree root before finalize_block
let root_pre = shell.shell.wl_storage.storage.block.tree.root();

let _event = &shell
.finalize_block(FinalizeBlock {
Expand All @@ -2234,6 +2256,10 @@ mod test_finalize_block {
})
.expect("Test failed")[0];

// the merkle tree root should not change after finalize_block
let root_post = shell.shell.wl_storage.storage.block.tree.root();
assert_eq!(root_pre.0, root_post.0);
tzemanovic marked this conversation as resolved.
Show resolved Hide resolved

// FIXME: uncomment when proper gas metering is in place
// // Check inner tx hash has been removed from storage
// assert_eq!(event.event_type.to_string(), String::from("applied"));
Expand Down
7 changes: 6 additions & 1 deletion core/src/ledger/storage/wl_storage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,12 @@ where
/// Commit the current block's write log to the storage and commit the block
/// to DB. Starts a new block write log.
pub fn commit_block(&mut self) -> storage_api::Result<()> {
if self.storage.last_epoch != self.storage.block.epoch {
self.storage
.update_epoch_in_merkle_tree()
.into_storage_result()?;
}

let mut batch = D::batch();
self.write_log
.commit_block(&mut self.storage, &mut batch)
Expand Down Expand Up @@ -205,7 +211,6 @@ where
.new_epoch(height, evidence_max_age_num_blocks);
tracing::info!("Began a new epoch {}", self.storage.block.epoch);
}
self.storage.update_epoch_in_merkle_tree()?;
Ok(new_epoch)
}
}
Expand Down