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: wallet validation #6765

Merged
merged 2 commits into from
Jan 28, 2025
Merged

fix: wallet validation #6765

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f05ce4a
fix validation
SWvheerden Jan 27, 2025
7d70edc
clippy
SWvheerden Jan 27, 2025
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
107 changes: 59 additions & 48 deletions base_layer/wallet/src/output_manager_service/service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -567,6 +567,7 @@ where
}
}

#[allow(clippy::too_many_lines)]
fn validate_outputs(&mut self) -> Result<u64, OutputManagerError> {
let current_base_node = self
.resources
Expand All @@ -586,6 +587,7 @@ where
let mut base_node_watch = self.resources.connectivity.get_current_base_node_watcher();
let event_publisher = self.resources.event_publisher.clone();
let validation_in_progress = self.validation_in_progress.clone();
let mut base_node_service_event_stream = self.base_node_service.get_event_stream();
tokio::spawn(async move {
// Note: We do not want the validation task to be queued
let mut _lock = match validation_in_progress.try_lock() {
Expand All @@ -604,60 +606,69 @@ where
return;
},
};

let exec_fut = txo_validation.execute();
tokio::pin!(exec_fut);
loop {
tokio::select! {
result = &mut exec_fut => {
match result {
Ok(id) => {
info!(
target: LOG_TARGET,
"UTXO Validation Protocol (Id: {}) completed successfully", id
);
return;
},
Err(OutputManagerProtocolError { id, error }) => {
warn!(
target: LOG_TARGET,
"Error completing UTXO Validation Protocol (Id: {}): {}", id, error
);
let event_payload = match error {
OutputManagerError::InconsistentBaseNodeDataError(_) |
OutputManagerError::BaseNodeChanged |
OutputManagerError::Shutdown |
OutputManagerError::RpcError(_) =>
OutputManagerEvent::TxoValidationCommunicationFailure(id),
_ => OutputManagerEvent::TxoValidationInternalFailure(id),
};
if let Err(e) = event_publisher.send(Arc::new(event_payload)) {
'outer: loop {
let local_run = txo_validation.clone();
let exec_fut = local_run.execute();
tokio::pin!(exec_fut);
loop {
tokio::select! {
result = &mut exec_fut => {
match result {
Ok(id) => {
info!(
target: LOG_TARGET,
"UTXO Validation Protocol (Id: {}) completed successfully", id
);
return;
},
Err(OutputManagerProtocolError { id, error }) => {
warn!(
target: LOG_TARGET,
"Error completing UTXO Validation Protocol (Id: {}): {}", id, error
);
let event_payload = match error {
OutputManagerError::InconsistentBaseNodeDataError(_) |
OutputManagerError::BaseNodeChanged |
OutputManagerError::Shutdown |
OutputManagerError::RpcError(_) =>
OutputManagerEvent::TxoValidationCommunicationFailure(id),
_ => OutputManagerEvent::TxoValidationInternalFailure(id),
};
if let Err(e) = event_publisher.send(Arc::new(event_payload)) {
debug!(
target: LOG_TARGET,
"Error sending event because there are no subscribers: {:?}", e
);
}

return;
},
}
},
_ = shutdown.wait() => {
debug!(target: LOG_TARGET, "TXO Validation Protocol (Id: {}) shutting down because the system \
is shutting down", id);
return;
},
event = base_node_service_event_stream.recv() => {
if let Ok(bn_event) = event {
if let BaseNodeEvent::NewBlockDetected(_hash, _height) = *bn_event {
debug!(target: LOG_TARGET, "TXO Validation Protocol (Id: {}) resetting because base node height changed", id);
continue 'outer;
}
}
}
_ = base_node_watch.changed() => {
if let Some(peer) = base_node_watch.borrow().as_ref() {
if peer.get_current_peer().node_id != current_base_node {
debug!(
target: LOG_TARGET,
"Error sending event because there are no subscribers: {:?}", e
"TXO Validation Protocol (Id: {}) resetting to run again because base node changed", id
);
continue 'outer;
}

return;
},
}
},
_ = shutdown.wait() => {
debug!(target: LOG_TARGET, "TXO Validation Protocol (Id: {}) shutting down because the system \
is shutting down", id);
return;
},
_ = base_node_watch.changed() => {
if let Some(peer) = base_node_watch.borrow().as_ref() {
if peer.get_current_peer().node_id != current_base_node {
debug!(
target: LOG_TARGET,
"TXO Validation Protocol (Id: {}) cancelled because base node changed", id
);
return;
}
}

}
}
}
Expand Down
1 change: 1 addition & 0 deletions base_layer/wallet/src/output_manager_service/tasks/txo_validation_task.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ use crate::{

const LOG_TARGET: &str = "wallet::output_service::txo_validation_task";

#[derive(Clone)]
pub struct TxoValidationTask<TBackend, TWalletConnectivity> {
operation_id: u64,
db: OutputManagerDatabase<TBackend>,
Expand Down
Loading