Skip to content

chore: propagate errors from assets and storage #4154

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

Open
metalurgical wants to merge 1 commit into
base: main
Choose a base branch
from
Open

chore: propagate errors from assets and storage #4154

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
541 changes: 301 additions & 240 deletions crates/node/src/assets.rs
View file
Open in desktop

Large diffs are not rendered by default.

16 changes: 11 additions & 5 deletions crates/node/src/assets/cleanup.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -255,8 +255,10 @@ mod tests {
reconstruction_threshold,
)
.unwrap();
let id = triple_store.generate_and_reserve_id();
triple_store.add_owned(id, make_triple(&all_participants));
let id = triple_store.generate_and_reserve_id().unwrap();
triple_store
.add_owned(id, make_triple(&all_participants))
.unwrap();
let v2_key = triple_v2_key(reconstruction_threshold, id);
// Sanity: the row really is in the TripleV2 column.
assert!(db.get(DBCol::TripleV2, &v2_key).unwrap().is_some());
Expand Down Expand Up @@ -324,8 +326,10 @@ mod tests {
reconstruction_threshold,
)
.unwrap();
let id = triple_store.generate_and_reserve_id();
triple_store.add_owned(id, make_triple(&active_subset));
let id = triple_store.generate_and_reserve_id().unwrap();
triple_store
.add_owned(id, make_triple(&active_subset))
.unwrap();
let v2_key = triple_v2_key(reconstruction_threshold, id);

// Seed epoch_data, then run cleanup with one participant's TLS rotated
Expand Down Expand Up @@ -386,7 +390,9 @@ mod tests {
// Even an outright-stale peer-owned triple stays put — the peer cleans
// its own.
let peer_id = UniqueId::new(peer, 100, 0);
triple_store.add_unowned(peer_id, make_triple(&all_participants));
triple_store
.add_unowned(peer_id, make_triple(&all_participants))
.unwrap();
let v2_key = triple_v2_key(reconstruction_threshold, peer_id);

delete_stale_triples_and_presignatures(
Expand Down
14 changes: 7 additions & 7 deletions crates/node/src/assets/test_utils.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ impl TestContext {
alive_participants: Arc<Mutex<Vec<ParticipantId>>>,
) -> Self {
let dir = tempfile::tempdir().unwrap();
let db = crate::db::SecretDB::new(dir.path(), [1; 16]).unwrap();
let db = SecretDB::new(dir.path(), [1; 16]).unwrap();
Self {
db,
clock: FakeClock::default(),
Expand Down Expand Up @@ -165,29 +165,29 @@ impl TestContext {
pub fn populate(&self, participants: &[ParticipantId]) {
// Mirror cleanup's view of triples: per-`t` TripleV2 column.
let store = self.new_store::<PairedTriple>(DBCol::TripleV2, self.triple_prefix());
let id = store.generate_and_reserve_id();
store.add_owned(id, make_triple(participants));
let id = store.generate_and_reserve_id().unwrap();
store.add_owned(id, make_triple(participants)).unwrap();

for &d in &self.presign_domain_ids {
let store = self.new_store::<PresignOutputWithParticipants>(
DBCol::Presignature,
d.0.to_be_bytes().to_vec(),
);
let id = store.generate_and_reserve_id();
store.add_owned(id, make_presign(participants));
let id = store.generate_and_reserve_id().unwrap();
store.add_owned(id, make_presign(participants)).unwrap();
}
}

pub fn assert_owned(&self, expected: usize) {
let store = self.new_store::<PairedTriple>(DBCol::TripleV2, self.triple_prefix());
assert_eq!(store.num_owned(), expected);
assert_eq!(store.num_owned().unwrap(), expected);

for &d in &self.presign_domain_ids {
let store = self.new_store::<PresignOutputWithParticipants>(
DBCol::Presignature,
d.0.to_be_bytes().to_vec(),
);
assert_eq!(store.num_owned(), expected);
assert_eq!(store.num_owned().unwrap(), expected);
}
}
}
31 changes: 14 additions & 17 deletions crates/node/src/mpc_client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,7 +159,7 @@ where
let metrics_emitter = tracking::spawn("periodically emits metrics", async move {
loop {
client.emit_metrics();
tokio::time::sleep(std::time::Duration::from_secs(5)).await;
tokio::time::sleep(Duration::from_secs(5)).await;
}
});

Expand All @@ -169,17 +169,14 @@ where
MpcClient::monitor_passive_channels_inner(channel_receiver, self.clone()),
)
};

let self_clone = self.clone();
let monitor_chain = {
let chain_txn_sender = chain_txn_sender.clone();
tracking::spawn(
"monitor chain",
self.clone().monitor_block_updates(
block_update_receiver,
chain_txn_sender,
debug_receiver,
),
)
tracking::spawn("monitor chain", async {
self_clone
.monitor_block_updates(block_update_receiver, chain_txn_sender, debug_receiver)
.await
})
};

let tee_verification_handle = {
Expand Down Expand Up @@ -237,7 +234,7 @@ where

let _ = monitor_passive_channels.await?;
metrics_emitter.await?;
monitor_chain.await?;
monitor_chain.await??;
let _ = robust_ecdsa_background_tasks.await?;
let _ = ecdsa_background_tasks.await?;
let _ = eddsa_background_tasks.await?;
Expand All @@ -254,7 +251,7 @@ where
>,
chain_txn_sender: impl TransactionSender + 'static,
mut debug_receiver: tokio::sync::broadcast::Receiver<DebugRequest>,
) {
) -> anyhow::Result<bool> {
let mut tasks = AutoAbortTaskCollection::new();
let mut pending_signatures =
PendingRequests::<SignatureRequest, contract_args::SignatureRespondArgs>::new(
Expand Down Expand Up @@ -289,7 +286,7 @@ where
let Some(block_update) = block_update else {
// If this branch hits, it means the channel is closed, meaning the
// indexer is being shutdown. So just quit this task.
break;
return Err(anyhow::anyhow!("Block update channel closed"));
};

self.client.update_indexer_height(block_update.block.height.into());
Expand All @@ -305,7 +302,7 @@ where

// TODO(#3031): add batch request and unify stores
for request in &signature_requests.requests {
self.sign_request_store.add(request);
self.sign_request_store.add(request)?;
}

// TODO(#3032): remove completed & finalized requests from store
Expand All @@ -318,7 +315,7 @@ where
block_update.completed_ckds
);
for request in &ckd_requests.requests {
self.ckd_request_store.add(request);
self.ckd_request_store.add(request)?;
}

pending_ckds.notify_new_block(ckd_requests);
Expand All @@ -331,7 +328,7 @@ where
);

for request in &verify_foreign_tx_requests.requests {
self.verify_foreign_tx_request_store.add(request);
self.verify_foreign_tx_request_store.add(request)?;
}
pending_verify_foreign_txs.notify_new_block(verify_foreign_tx_requests);
}
Expand Down Expand Up @@ -512,7 +509,7 @@ where
.unwrap()
.last_response_submission = Some(Clock::real().now());

anyhow::Ok(())
anyhow::Ok(true)
},
);
}
Expand Down
37 changes: 20 additions & 17 deletions crates/node/src/providers/ecdsa/presign.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,12 +64,12 @@ impl EcdsaSignatureProvider {
let parallelism_limiter = Arc::new(tokio::sync::Semaphore::new(config.concurrency));
let mut tasks = AutoAbortTaskCollection::new();
loop {
progress_tracker.update_progress();
progress_tracker.update_progress().unwrap(); // Since return is !
metrics::MPC_OWNED_NUM_PRESIGNATURES_ONLINE
.set(presignature_store.num_owned_ready() as i64);
.set(presignature_store.num_owned_ready().unwrap() as i64); // Since return is !
metrics::MPC_OWNED_NUM_PRESIGNATURES_WITH_OFFLINE_PARTICIPANT
.set(presignature_store.num_owned_offline() as i64);
let my_presignatures_count: usize = presignature_store.num_owned();
.set(presignature_store.num_owned_offline().unwrap() as i64); // Since return is !
let my_presignatures_count: usize = presignature_store.num_owned().unwrap(); // Since return is !
metrics::MPC_OWNED_NUM_PRESIGNATURES_AVAILABLE.set(my_presignatures_count as i64);
let should_generate = my_presignatures_count + in_flight_generations.num_in_flight()
< config.desired_presignatures_to_buffer;
Expand All @@ -79,10 +79,11 @@ impl EcdsaSignatureProvider {
&& in_flight_generations.num_in_flight()
< config.concurrency * 2
{
let id = presignature_store.generate_and_reserve_id();
progress_tracker.set_waiting_for_triples(true);
let (paired_triple_id, (triple0, triple1)) = triple_store.take_owned().await;
progress_tracker.set_waiting_for_triples(false);
let id = presignature_store.generate_and_reserve_id().unwrap(); // Since return is !
progress_tracker.set_waiting_for_triples(true).unwrap(); // Since return is !
let (paired_triple_id, (triple0, triple1)) =
triple_store.take_owned().await.unwrap(); // Since return is !
progress_tracker.set_waiting_for_triples(false).unwrap(); // Since return is !
let participants = participants_from_triples(&triple0, &triple1);
let task_id = EcdsaTaskId::Presignature {
id,
Expand Down Expand Up @@ -129,7 +130,7 @@ impl EcdsaSignatureProvider {
presignature,
participants,
},
);
)?;

anyhow::Ok(())
}),
Expand All @@ -139,10 +140,10 @@ impl EcdsaSignatureProvider {

// If the store is full, try to discard some presignatures which cannot be used right now
if my_presignatures_count >= config.desired_presignatures_to_buffer {
presignature_store.maybe_discard_owned(1).await;
presignature_store.maybe_discard_owned(1).await.unwrap(); // Since return is !
}

tokio::time::sleep(std::time::Duration::from_millis(100)).await;
tokio::time::sleep(Duration::from_millis(100)).await;
}
}

Expand Down Expand Up @@ -270,7 +271,7 @@ impl MpcLeaderCentricComputation<()> for FollowerPresignComputation {
presignature,
participants: channel.participants().to_vec(),
},
);
)?;
Ok(())
}

Expand All @@ -287,16 +288,17 @@ struct PresignatureGenerationProgressTracker {
}

impl PresignatureGenerationProgressTracker {
pub fn set_waiting_for_triples(&self, waiting: bool) {
pub fn set_waiting_for_triples(&self, waiting: bool) -> anyhow::Result<()> {
self.waiting_for_triples
.store(waiting, std::sync::atomic::Ordering::Relaxed);
self.update_progress();
self.update_progress()?;
Ok(())
}

pub fn update_progress(&self) {
pub fn update_progress(&self) -> anyhow::Result<()> {
tracking::set_progress(&format!(
"Presignatures: available: {}/{}; generating: {}{}",
self.presignature_store.num_owned(),
self.presignature_store.num_owned()?,
self.desired_presignatures_to_buffer,
self.in_flight_generations
.load(std::sync::atomic::Ordering::Relaxed),
Expand All @@ -308,6 +310,7 @@ impl PresignatureGenerationProgressTracker {
} else {
""
}
))
));
Ok(())
}
}
2 changes: 1 addition & 1 deletion crates/node/src/providers/ecdsa/sign.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ impl EcdsaSignatureProvider {
) -> anyhow::Result<(Signature, VerifyingKey)> {
let sign_request = self.sign_request_store.get(id).await?;
let keyshare = self.keyshare(sign_request.domain)?;
let (presignature_id, presignature) = keyshare.presignature_store.take_owned().await;
let (presignature_id, presignature) = keyshare.presignature_store.take_owned().await?;
let participants = presignature.participants.clone();
let channel = self.new_channel_for_task(
EcdsaTaskId::Signature {
Expand Down
Loading