Allow acceptable block ID range (#634)

dkg-gadget/src/async_protocols/incoming.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use dkg_primitives::types::{DKGError, DKGMessage, DKGMsgPayload, SessionId, SignedDKGMessage};
-use dkg_runtime_primitives::{crypto::Public, MaxAuthorities};
+use dkg_runtime_primitives::{associated_block_id_acceptable, crypto::Public, MaxAuthorities};
 use futures::Stream;
 use round_based::Msg;
 use sp_runtime::traits::Get;
@@ -127,7 +127,10 @@ impl TransformIncoming for SignedDKGMessage<Public> {
 					if self.msg.session_id == this_session_id {
 						logger
 							.checkpoint_message_raw(self.msg.payload.payload(), "CP-2.3-incoming");
-						if associated_block_id == &self.msg.associated_block_id {
+						if associated_block_id_acceptable(
+							associated_block_id,
+							self.msg.associated_block_id,
+						) {
 							logger.checkpoint_message_raw(
 								self.msg.payload.payload(),
 								"CP-2.4-incoming",

dkg-gadget/src/async_protocols/keygen/handler.rs

@@ -110,7 +110,7 @@ where
 			DKGMsgStatus::QUEUED => KeygenRound::QUEUED,
 		};
 		let i = params.party_i;
-		let associated_round_id = params.associated_block_id.clone();
+		let associated_round_id = params.associated_block_id;
 		let channel_type: ProtocolType<<BI as BlockchainInterface>::MaxProposalLength> =
 			ProtocolType::Keygen { ty, i, t, n, associated_block_id: associated_round_id };
 		new_inner(

dkg-gadget/src/async_protocols/mod.rs

@@ -75,7 +75,7 @@ pub struct AsyncProtocolParameters<
 	pub best_authorities: Arc<Vec<(KeygenPartyId, Public)>>,
 	pub authority_public_key: Arc<Public>,
 	pub party_i: KeygenPartyId,
-	pub associated_block_id: Vec<u8>,
+	pub associated_block_id: u64,
 	pub batch_id_gen: Arc<AtomicU64>,
 	pub handle: AsyncProtocolRemote<BI::Clock>,
 	pub session_id: SessionId,
@@ -136,7 +136,7 @@ impl<
 			engine: self.engine.clone(),
 			keystore: self.keystore.clone(),
 			current_validator_set: self.current_validator_set.clone(),
-			associated_block_id: self.associated_block_id.clone(),
+			associated_block_id: self.associated_block_id,
 			best_authorities: self.best_authorities.clone(),
 			authority_public_key: self.authority_public_key.clone(),
 			party_i: self.party_i,
@@ -290,31 +290,31 @@ pub enum ProtocolType<MaxProposalLength: Get<u32> + Clone + Send + Sync + std::f
 		i: KeygenPartyId,
 		t: u16,
 		n: u16,
-		associated_block_id: Vec<u8>,
+		associated_block_id: u64,
 	},
 	Offline {
 		unsigned_proposal: Arc<UnsignedProposal<MaxProposalLength>>,
 		i: OfflinePartyId,
 		s_l: Vec<KeygenPartyId>,
 		local_key: Arc<LocalKey<Secp256k1>>,
-		associated_block_id: Vec<u8>,
+		associated_block_id: u64,
 	},
 	Voting {
 		offline_stage: Arc<CompletedOfflineStage>,
 		unsigned_proposal: Arc<UnsignedProposal<MaxProposalLength>>,
 		i: OfflinePartyId,
-		associated_block_id: Vec<u8>,
+		associated_block_id: u64,
 	},
 }
 
 impl<MaxProposalLength: Get<u32> + Clone + Send + Sync + std::fmt::Debug + 'static>
 	ProtocolType<MaxProposalLength>
 {
-	pub const fn get_associated_block_id(&self) -> &Vec<u8> {
+	pub const fn get_associated_block_id(&self) -> u64 {
 		match self {
-			Self::Keygen { associated_block_id: associated_round_id, .. } => associated_round_id,
-			Self::Offline { associated_block_id: associated_round_id, .. } => associated_round_id,
-			Self::Voting { associated_block_id: associated_round_id, .. } => associated_round_id,
+			Self::Keygen { associated_block_id, .. } |
+			Self::Offline { associated_block_id, .. } |
+			Self::Voting { associated_block_id, .. } => *associated_block_id,
 		}
 	}
 
@@ -617,7 +617,7 @@ where
 
 			let id = params.authority_public_key.as_ref().clone();
 			let unsigned_dkg_message = DKGMessage {
-				associated_block_id: params.associated_block_id.clone(),
+				associated_block_id: params.associated_block_id,
 				sender_id: id,
 				recipient_id: maybe_recipient_id,
 				status,

dkg-gadget/src/async_protocols/remote.rs

@@ -36,7 +36,7 @@ pub struct AsyncProtocolRemote<C> {
 	current_round_blame: tokio::sync::watch::Receiver<CurrentRoundBlame>,
 	pub(crate) current_round_blame_tx: Arc<tokio::sync::watch::Sender<CurrentRoundBlame>>,
 	pub(crate) session_id: SessionId,
-	pub(crate) associated_block_id: Vec<u8>,
+	pub(crate) associated_block_id: u64,
 	pub(crate) logger: DebugLogger,
 	status_history: Arc<Mutex<Vec<MetaHandlerStatus>>>,
 }
@@ -63,7 +63,7 @@ impl<C: Clone> Clone for AsyncProtocolRemote<C> {
 			session_id: self.session_id,
 			logger: self.logger.clone(),
 			status_history: self.status_history.clone(),
-			associated_block_id: self.associated_block_id.clone(),
+			associated_block_id: self.associated_block_id,
 		}
 	}
 }
@@ -84,7 +84,7 @@ impl<C: AtLeast32BitUnsigned + Copy + Send> AsyncProtocolRemote<C> {
 		at: C,
 		session_id: SessionId,
 		logger: DebugLogger,
-		associated_block_id: Vec<u8>,
+		associated_block_id: u64,
 	) -> Self {
 		let (stop_tx, stop_rx) = tokio::sync::mpsc::unbounded_channel();
 		let (tx_keygen_signing, rx_keygen_signing) = tokio::sync::mpsc::unbounded_channel();

dkg-gadget/src/async_protocols/sign/handler.rs

@@ -149,7 +149,7 @@ where
 			i: offline_i,
 			s_l: s_l.clone(),
 			local_key: Arc::new(local_key.clone()),
-			associated_block_id: params.associated_block_id.clone(),
+			associated_block_id: params.associated_block_id,
 		};
 		let s_l_raw = s_l.into_iter().map(|party_i| *party_i.as_ref()).collect();
 		new_inner(
@@ -178,7 +178,7 @@ where
 				offline_stage: Arc::new(completed_offline_stage.clone()),
 				unsigned_proposal: Arc::new(unsigned_proposal.clone()),
 				i: offline_i,
-				associated_block_id: params.associated_block_id.clone(),
+				associated_block_id: params.associated_block_id,
 			};
 
 			params.logger.round_event(
@@ -224,7 +224,7 @@ where
 			let id = params.authority_public_key.as_ref().clone();
 			// now, broadcast the data
 			let unsigned_dkg_message = DKGMessage {
-				associated_block_id: params.associated_block_id.clone(),
+				associated_block_id: params.associated_block_id,
 				sender_id: id,
 				// No recipient for this message, it is broadcasted
 				recipient_id: None,

dkg-gadget/src/gossip_messages/misbehaviour_report.rs

@@ -148,7 +148,7 @@ where
 		let status =
 			if report.session_id == 0 { DKGMsgStatus::ACTIVE } else { DKGMsgStatus::QUEUED };
 		let message = DKGMessage::<AuthorityId> {
-			associated_block_id: vec![],
+			associated_block_id: 0,
 			sender_id: public.clone(),
 			// We need to gossip this misbehaviour, so no specific recipient.
 			recipient_id: None,

dkg-gadget/src/gossip_messages/public_key_gossip.rs

@@ -152,7 +152,7 @@ pub(crate) fn gossip_public_key<B, C, BE, GE>(
 		let status =
 			if msg.session_id == 0u64 { DKGMsgStatus::ACTIVE } else { DKGMsgStatus::QUEUED };
 		let message = DKGMessage::<AuthorityId> {
-			associated_block_id: vec![], // we don't need to associate this message with a block
+			associated_block_id: 0, // we don't need to associate this message with a block
 			sender_id: public.clone(),
 			// we need to gossip the final public key to all parties, so no specific recipient in
 			// this case.

dkg-gadget/src/signing_manager/work_manager.rs

@@ -6,6 +6,7 @@ use dkg_primitives::{
 	crypto::Public,
 	types::{DKGError, SignedDKGMessage},
 };
+use dkg_runtime_primitives::associated_block_id_acceptable;
 use parking_lot::RwLock;
 use sp_api::BlockT;
 use std::{
@@ -220,8 +221,10 @@ impl<B: BlockT> WorkManager<B> {
 		for task in lock.enqueued_tasks.iter() {
 			if task.handle.session_id == msg.msg.session_id &&
 				&task.task_hash == msg_unsigned_proposal_hash &&
-				msg.msg.associated_block_id == task.handle.associated_block_id
-			{
+				associated_block_id_acceptable(
+					task.handle.associated_block_id,
+					msg.msg.associated_block_id,
+				) {
 				self.logger.debug(format!(
 					"Message is for this ENQUEUED signing execution in session: {}",
 					task.handle.session_id

dkg-gadget/src/worker.rs

@@ -30,6 +30,7 @@ use multi_party_ecdsa::protocols::multi_party_ecdsa::gg_2020::state_machine::key
 use parking_lot::RwLock;
 use sc_client_api::{Backend, FinalityNotification};
 use sc_keystore::LocalKeystore;
+use sp_arithmetic::traits::SaturatedConversion;
 use sp_core::ecdsa;
 use sp_runtime::traits::{Block, Get, Header, NumberFor};
 use std::{
@@ -312,13 +313,9 @@ where
 		let authority_public_key = Arc::new(authority_public_key);
 
 		let now = self.get_latest_block_number();
-		let associated_block_id = associated_block.encode();
-		let mut status_handle = AsyncProtocolRemote::new(
-			now,
-			session_id,
-			self.logger.clone(),
-			associated_block_id.clone(),
-		);
+		let associated_block_id: u64 = associated_block.saturated_into();
+		let mut status_handle =
+			AsyncProtocolRemote::new(now, session_id, self.logger.clone(), associated_block_id);
 		// Fetch the active key. This requires rotating the key to have happened with
 		// full certainty in order to ensure the right key is being used to make signatures.
 		let active_local_key = match stage {

dkg-primitives/src/types.rs

@@ -57,8 +57,8 @@ pub struct DKGMessage<AuthorityId> {
 	pub session_id: SessionId,
 	/// enum for active or queued
 	pub status: DKGMsgStatus,
-	/// The bytes of the round ID
-	pub associated_block_id: Vec<u8>,
+	/// The round ID
+	pub associated_block_id: u64,
 }
 
 #[derive(Debug, Clone, Decode, Encode)]

dkg-runtime-primitives/src/lib.rs

@@ -83,6 +83,21 @@ pub const KEYGEN_TIMEOUT: u32 = 10;
 /// The sign timeout limit in blocks before we consider proposal as stalled
 pub const SIGN_TIMEOUT: u32 = 10;
 
+/// So long as the associated block id is within this tolerance, we consider the message as
+/// deliverable. This should be less than the SIGN_TIMEOUT
+pub const ASSOCIATED_BLOCK_ID_MESSAGE_DELIVERY_TOLERANCE: u64 = (SIGN_TIMEOUT - 2) as u64;
+
+pub const fn associated_block_id_acceptable(expected: u64, received: u64) -> bool {
+	// favor explicit logic for readability
+	let is_acceptable_above = received >= expected &&
+		received <= expected.saturating_add(ASSOCIATED_BLOCK_ID_MESSAGE_DELIVERY_TOLERANCE);
+	let is_acceptable_below = received < expected &&
+		received >= expected.saturating_sub(ASSOCIATED_BLOCK_ID_MESSAGE_DELIVERY_TOLERANCE);
+	let is_equal = expected == received;
+
+	is_acceptable_above || is_acceptable_below || is_equal
+}
+
 // Engine ID for DKG
 pub const DKG_ENGINE_ID: sp_runtime::ConsensusEngineId = *b"WDKG";
 
@@ -314,3 +329,47 @@ sp_api::decl_runtime_apis! {
 		fn should_execute_new_keygen() -> bool;
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use crate::{
+		associated_block_id_acceptable, ASSOCIATED_BLOCK_ID_MESSAGE_DELIVERY_TOLERANCE,
+		SIGN_TIMEOUT,
+	};
+
+	#[test]
+	fn assert_value() {
+		assert!(ASSOCIATED_BLOCK_ID_MESSAGE_DELIVERY_TOLERANCE > 0);
+		assert!(ASSOCIATED_BLOCK_ID_MESSAGE_DELIVERY_TOLERANCE < SIGN_TIMEOUT as _);
+	}
+
+	#[test]
+	fn test_range_above() {
+		let current_block: u64 = 10;
+		assert!(associated_block_id_acceptable(current_block, current_block));
+		assert!(associated_block_id_acceptable(current_block, current_block + 1));
+		assert!(associated_block_id_acceptable(
+			current_block,
+			current_block + ASSOCIATED_BLOCK_ID_MESSAGE_DELIVERY_TOLERANCE
+		));
+		assert!(!associated_block_id_acceptable(
+			current_block,
+			current_block + ASSOCIATED_BLOCK_ID_MESSAGE_DELIVERY_TOLERANCE + 1
+		));
+	}
+
+	#[test]
+	fn test_range_below() {
+		let current_block: u64 = 10;
+		assert!(associated_block_id_acceptable(current_block, current_block));
+		assert!(associated_block_id_acceptable(current_block, current_block - 1));
+		assert!(associated_block_id_acceptable(
+			current_block,
+			current_block - ASSOCIATED_BLOCK_ID_MESSAGE_DELIVERY_TOLERANCE
+		));
+		assert!(!associated_block_id_acceptable(
+			current_block,
+			current_block - ASSOCIATED_BLOCK_ID_MESSAGE_DELIVERY_TOLERANCE - 1
+		));
+	}
+}