CSUB-202 OCW Lock

pallets/creditcoin/src/lib.rs

@@ -73,8 +73,11 @@ pub mod pallet {
 		pallet_prelude::*,
 	};
 	use ocw::errors::VerificationFailureCause;
+	use sp_runtime::offchain::storage_lock::{BlockAndTime, StorageLock};
+	use sp_runtime::offchain::Duration;
 	use sp_runtime::traits::{
-		IdentifyAccount, Saturating, UniqueSaturatedFrom, UniqueSaturatedInto, Verify,
+		IdentifyAccount, SaturatedConversion, Saturating, UniqueSaturatedFrom, UniqueSaturatedInto,
+		Verify,
 	};
 
 	/// Configure the pallet by specifying the parameters and types on which it depends.
@@ -608,10 +611,27 @@ pub mod pallet {
 			};
 
 			for (deadline, id, task) in PendingTasks::<T>::iter() {
-				let storage_key = (deadline, &id).encode();
-				let status = ocw::LocalVerificationStatus::new(&storage_key);
-				if status.is_complete() {
+				let storage_key = crate::ocw::tasks::storage_key(&id);
+				let offset =
+					T::UnverifiedTaskTimeout::get().saturated_into::<u32>().saturating_sub(2u32);
+
+				let mut lock = StorageLock::<BlockAndTime<frame_system::Pallet<T>>>::with_block_and_time_deadline(
+					&storage_key,
+					offset,
+					Duration::from_millis(0),
+				);
+
+				let guard = match lock.try_lock() {
+					Ok(g) => g,
+					Err(_) => continue,
+				};
+
+				if match &id {
+					TaskId::VerifyTransfer(id) => Transfers::<T>::contains_key(id),
+					TaskId::CollectCoins(id) => CollectedCoins::<T>::contains_key(id),
+				} {
 					log::debug!("Already handled Task ({:?}, {:?})", deadline, id);
+					guard.forget();
 					continue;
 				}
 
@@ -623,13 +643,13 @@ pub mod pallet {
 						match Self::offchain_signed_tx(auth_id.clone(), |_| {
 							Call::persist_task_output { deadline, task_output: output.clone() }
 						}) {
+							Ok(_) => guard.forget(),
 							Err(e) => {
 								log::error!(
 									"Failed to send persist dispatchable transaction: {:?}",
 									e
 								)
 							},
-							Ok(_) => status.mark_complete(),
 						}
 					},
 					Err((task, ocw::OffchainError::InvalidTask(cause))) => {
@@ -644,7 +664,7 @@ pub mod pallet {
 									"Failed to send fail dispatchable transaction: {:?}",
 									e
 								),
-								Ok(_) => status.mark_complete(),
+								Ok(_) => guard.forget(),
 							}
 						}
 					},

pallets/creditcoin/src/mock.rs

@@ -10,23 +10,23 @@ use frame_support::{
 	traits::{ConstU32, ConstU64, GenesisBuild, Get, Hooks},
 };
 use frame_system as system;
-use parking_lot::RwLock;
+pub(crate) use parking_lot::RwLock;
 use serde_json::Value;
 use sp_core::H256;
 use sp_keystore::{testing::KeyStore, KeystoreExt, SyncCryptoStore};
+pub(crate) use sp_runtime::offchain::testing::OffchainState;
 use sp_runtime::{
 	offchain::{
 		storage::StorageValueRef,
-		testing::{
-			OffchainState, PendingRequest, PoolState, TestOffchainExt, TestTransactionPoolExt,
-		},
+		testing::{PendingRequest, PoolState, TestOffchainExt, TestTransactionPoolExt},
 		OffchainDbExt, OffchainWorkerExt, TransactionPoolExt,
 	},
 	testing::{Header, TestXt},
 	traits::{BlakeTwo256, Extrinsic as ExtrinsicT, IdentifyAccount, IdentityLookup, Verify},
 	MultiSignature, RuntimeAppPublic,
 };
-use std::{cell::Cell, collections::HashMap, sync::Arc};
+pub(crate) use std::sync::Arc;
+use std::{cell::Cell, collections::HashMap};
 
 type UncheckedExtrinsic = frame_system::mocking::MockUncheckedExtrinsic<Test>;
 type Block = frame_system::mocking::MockBlock<Test>;

pallets/creditcoin/src/ocw.rs

@@ -77,35 +77,5 @@ impl<T: Config> Pallet<T> {
 	}
 }
 
-pub(crate) struct LocalVerificationStatus<'a> {
-	storage_ref: StorageValueRef<'a>,
-	key: &'a [u8],
-}
-
-impl<'a> LocalVerificationStatus<'a> {
-	pub(crate) fn new(storage_key: &'a [u8]) -> Self {
-		Self { storage_ref: StorageValueRef::persistent(storage_key), key: storage_key }
-	}
-
-	pub(crate) fn is_complete(&self) -> bool {
-		match self.storage_ref.get::<()>() {
-			Ok(Some(())) => true,
-			Ok(None) => false,
-			Err(e) => {
-				log::warn!(
-					"Failed to decode offchain storage for {}: {:?}",
-					hex::encode(self.key),
-					e
-				);
-				true
-			},
-		}
-	}
-
-	pub(crate) fn mark_complete(&self) {
-		self.storage_ref.set(&());
-	}
-}
-
 #[cfg(test)]
 mod tests;

pallets/creditcoin/src/ocw/tasks.rs

@@ -6,7 +6,16 @@ use crate::types::{
 	CollectedCoins, Task, TaskOutput, Transfer, UnverifiedCollectedCoins, UnverifiedTransfer,
 };
 use crate::{CollectedCoinsId, Config, TaskData, TransferId};
+use codec::Encode;
+pub use sp_runtime::offchain::storage_lock::{BlockAndTime, Lockable, StorageLock};
 use sp_runtime::traits::{UniqueSaturatedFrom, UniqueSaturatedInto};
+use sp_std::vec::Vec;
+
+#[inline]
+pub(crate) fn storage_key<Id: Encode>(id: &Id) -> Vec<u8> {
+	const TASK_GUARD: &[u8] = b"creditcoin/task/guard/";
+	id.using_encoded(|encoded_id| TASK_GUARD.iter().chain(encoded_id).copied().collect())
+}
 
 impl<AccountId, BlockNum, Hash, Moment> UnverifiedTransfer<AccountId, BlockNum, Hash, Moment>
 where

pallets/creditcoin/src/ocw/tasks/collect_coins.rs

@@ -108,7 +108,7 @@ impl<T: Config> Pallet<T> {
 }
 
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
 
 	use super::*;
 	use crate::TaskId;
@@ -156,13 +156,13 @@ mod tests {
 		input_bytes.into()
 	});
 
-	static TX_HASH: Lazy<String> = Lazy::new(|| {
+	pub(crate) static TX_HASH: Lazy<String> = Lazy::new(|| {
 		let responses = &*RESPONSES;
 		let val = responses["eth_getTransactionByHash"].result.clone().unwrap()["hash"].clone();
 		serde_json::from_value(val).unwrap()
 	});
 
-	static RPC_RESPONSE_AMOUNT: Lazy<sp_core::U256> = Lazy::new(|| {
+	pub(crate) static RPC_RESPONSE_AMOUNT: Lazy<sp_core::U256> = Lazy::new(|| {
 		let transfer_fn = burn_vested_cc_abi();
 
 		let inputs = transfer_fn.decode_input(&(INPUT.0)[4..]).unwrap();
@@ -177,6 +177,7 @@ mod tests {
 
 	use std::convert::TryFrom;
 
+	use alloc::sync::Arc;
 	use assert_matches::assert_matches;
 	use codec::Decode;
 	use ethereum_types::{H160, U64};
@@ -186,7 +187,8 @@ mod tests {
 
 	use crate::helpers::non_paying_error;
 	use crate::mock::{
-		roll_by_with_ocw, set_rpc_uri, AccountId, ExtBuilder, MockedRpcRequests, Origin, Test,
+		roll_by_with_ocw, set_rpc_uri, AccountId, ExtBuilder, MockedRpcRequests, OffchainState,
+		Origin, RwLock, Test,
 	};
 	use crate::ocw::{
 		errors::{OffchainError, VerificationFailureCause as Cause},
@@ -198,6 +200,16 @@ mod tests {
 	use crate::Pallet as Creditcoin;
 	use crate::{ocw::rpc::JsonRpcResponse, ExternalAddress};
 
+	/// call from externalities context
+	pub(crate) fn mock_rpc_for_collect_coins(state: &Arc<RwLock<OffchainState>>) {
+		let dummy_url = "dummy";
+		set_rpc_uri(&CONTRACT_CHAIN, &dummy_url);
+
+		let mut rpcs =
+			MockedRpcRequests::new(dummy_url, &*TX_HASH, &*BLOCK_NUMBER_STR, &*RESPONSES);
+		rpcs.mock_get_block_number(&mut *state.write());
+	}
+
 	struct PassingCollectCoins {
 		to: ExternalAddress,
 		receipt: EthTransactionReceipt,
@@ -460,7 +472,6 @@ mod tests {
 	}
 
 	#[test]
-	// #[tracing_test::traced_test]
 	fn persist_collect_coins() {
 		let mut ext = ExtBuilder::default();
 		let acct_pubkey = ext.generate_authority();
@@ -511,7 +522,6 @@ mod tests {
 	}
 
 	#[test]
-	#[tracing_test::traced_test]
 	fn persist_unregistered_address() {
 		let mut ext = ExtBuilder::default();
 		let acct_pubkey = ext.generate_authority();
@@ -540,7 +550,6 @@ mod tests {
 	}
 
 	#[test]
-	#[tracing_test::traced_test]
 	fn persist_more_than_max_balance_should_error() {
 		let mut ext = ExtBuilder::default();
 		let acct_pubkey = ext.generate_authority();
@@ -580,7 +589,6 @@ mod tests {
 	}
 
 	#[test]
-	#[tracing_test::traced_test]
 	fn request_persisted_not_reentrant() {
 		let mut ext = ExtBuilder::default();
 		let acct_pubkey = ext.generate_authority();
@@ -622,7 +630,6 @@ mod tests {
 	}
 
 	#[test]
-	#[tracing_test::traced_test]
 	fn request_pending_not_reentrant() {
 		let mut ext = ExtBuilder::default();
 		ext.generate_authority();
@@ -749,12 +756,7 @@ mod tests {
 		let mut ext = ExtBuilder::default();
 		ext.generate_authority();
 		ext.build_offchain_and_execute_with_state(|state, pool| {
-			let dummy_url = "dummy";
-			set_rpc_uri(&CONTRACT_CHAIN, &dummy_url);
-
-			let mut rpcs =
-				MockedRpcRequests::new(dummy_url, &*TX_HASH, &*BLOCK_NUMBER_STR, &*RESPONSES);
-			rpcs.mock_get_block_number(&mut state.write());
+			mock_rpc_for_collect_coins(&state);
 
 			let (acc, addr, sign, _) = generate_address_with_proof("collector");
 
@@ -836,17 +838,11 @@ mod tests {
 	}
 
 	#[test]
-	#[tracing_test::traced_test]
 	fn unverified_collect_coins_are_removed() {
 		let mut ext = ExtBuilder::default();
 		ext.generate_authority();
 		ext.build_offchain_and_execute_with_state(|state, _| {
-			let dummy_url = "dummy";
-			set_rpc_uri(&CONTRACT_CHAIN, &dummy_url);
-
-			let mut rpcs =
-				MockedRpcRequests::new(dummy_url, &*TX_HASH, &*BLOCK_NUMBER_STR, &*RESPONSES);
-			rpcs.mock_get_block_number(&mut state.write());
+			mock_rpc_for_collect_coins(&state);
 
 			let (acc, addr, sign, _) = generate_address_with_proof("collector");
 

pallets/creditcoin/src/ocw/tasks/tests.rs

@@ -0,0 +1,109 @@
+#![cfg(test)]
+
+use super::{StorageLock, Task};
+use crate::{
+	mock::{roll_by_with_ocw, set_rpc_uri, AccountId, ExtBuilder, MockedRpcRequests, Origin, Test},
+	ocw::{
+		errors::{OffchainError, VerificationFailureCause as Cause},
+		rpc::{EthTransaction, EthTransactionReceipt, JsonRpcResponse},
+		EncodeLike, ETH_CONFIRMATIONS,
+	},
+	pallet::{Config, Store},
+	tests::{generate_address_with_proof, RefstrExt},
+	types::CollectedCoinsId,
+	ExternalAddress, Pallet as Creditcoin,
+};
+use assert_matches::assert_matches;
+use codec::Decode;
+use ethereum_types::{H160, U64};
+use frame_support::{assert_noop, assert_ok, once_cell::sync::Lazy};
+use frame_system::Pallet as System;
+use pallet_timestamp::Pallet as Timestamp;
+use sp_core::offchain::Duration;
+use sp_io::offchain;
+use sp_runtime::{
+	offchain::{
+		storage::StorageValueRef,
+		storage_lock::{BlockAndTime, Lockable, Time},
+	},
+	traits::{BlockNumberProvider, IdentifyAccount},
+};
+use std::convert::{TryFrom, TryInto};
+use std::sync::{
+	atomic::{AtomicU8, Ordering},
+	Arc,
+};
+use std::thread;
+
+#[test]
+fn lock_released_when_guard_is_dropped() {
+	let ext = ExtBuilder::default();
+	ext.build_offchain_and_execute_with_state(|state, _| {
+		let key = b"id_1";
+		let mut l1 = StorageLock::<'_, Time>::new(key);
+		let g = l1.try_lock();
+		assert!(g.is_ok());
+		drop(g);
+		assert!(state.read().persistent_storage.get(key).is_none());
+	});
+}
+
+#[test]
+fn lock_guard_is_kept_alive() {
+	let ext = ExtBuilder::default();
+	ext.build_offchain_and_execute_with_state(|_, _| {
+		let mut l1 = StorageLock::<'_, Time>::new(b"id_1");
+		let g = l1.try_lock();
+		g.expect("ok").forget();
+		let g = l1.try_lock();
+		assert!(g.is_err());
+	});
+}
+
+#[test]
+fn lock_expires() {
+	let ext = ExtBuilder::default();
+	ext.build_offchain_and_execute_with_state(|_, _| {
+		System::<Test>::set_block_number(1);
+		let mut l1 = StorageLock::<'_, BlockAndTime<System<Test>>>::with_block_and_time_deadline(
+			b"id_1",
+			1,
+			Duration::from_millis(0),
+		);
+		let g = l1.try_lock().expect("ok");
+		g.forget();
+		System::<Test>::set_block_number(3);
+		let sleep_until = offchain::timestamp().add(Duration::from_millis(1));
+		offchain::sleep_until(sleep_until);
+		let g = l1.try_lock();
+		assert!(g.is_ok());
+	});
+}
+
+#[test]
+fn lock_mutual_exclusion() {
+	let ext = ExtBuilder::default();
+	ext.build_offchain_and_execute_with_state(|state, _| {
+		let mut l1 = StorageLock::<'_, Time>::new(b"id_1");
+		let mut l2 = StorageLock::<'_, Time>::new(b"id_2");
+
+		let g1 = l1.try_lock().expect("ok");
+		g1.forget();
+		let g1 = l1.try_lock();
+		assert!(g1.is_err());
+		//won't release because it was not a guard.
+		drop(g1);
+		let g2 = l2.try_lock().expect("ok");
+		drop(g2);
+		let g2 = l2.try_lock();
+		assert!(g2.is_ok());
+
+		let g1 = l1.try_lock();
+		assert!(g1.is_err());
+		drop(g2);
+		let deadline = state.read().persistent_storage.get(b"id_2");
+		assert!(deadline.is_none());
+		let deadline = state.read().persistent_storage.get(b"id_1");
+		assert!(deadline.is_some());
+	});
+}