Merge pull request #500 from Cerebellum-Network/fix/clear-inspection-…

…ocw-guard

Clear inspection OCW guard on early return

Cargo.toml

@@ -53,6 +53,7 @@ static_assertions = { version = "1.1.0" }
 url = { version = "2.4.1" }
 array-bytes = { version = "6.1" }
 itertools = { version = "0.13.0", default-features = false, features = ["use_alloc"] }
+scopeguard = { version = "1.2.0", default-features = false }
 
 # Substrate Dependencies
 # Please keey format such that:

pallets/ddc-verification/Cargo.toml

@@ -27,6 +27,7 @@ polkadot-ckb-merkle-mountain-range = { workspace = true }
 prost = { version = "0.13", default-features = false, features = ["prost-derive"] }
 rand = { workspace = true, features = ["small_rng", "alloc"], default-features = false }
 scale-info = { workspace = true }
+scopeguard = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 serde_with = { version = "3", default-features = false, features = ["base64", "macros"] }
@@ -60,6 +61,7 @@ std = [
   "frame-support/std",
   "frame-system/std",
   "scale-info/std",
+  "scopeguard/use_std",
   "sp-std/std",
   "scale-info/std",
   "sp-runtime/std",

pallets/ddc-verification/src/lib.rs

@@ -15,6 +15,7 @@ use base64ct::{Base64, Encoding};
 #[cfg(feature = "runtime-benchmarks")]
 use ddc_primitives::traits::{BucketManager, ClusterCreator, CustomerDepositor};
 use ddc_primitives::{
+	ocw_mutex::OcwMutex,
 	traits::{
 		ClusterManager, ClusterValidator, CustomerVisitor, NodeManager, PayoutProcessor,
 		StorageUsageProvider, ValidatorVisitor,
@@ -26,7 +27,7 @@ use ddc_primitives::{
 };
 use frame_support::{
 	pallet_prelude::*,
-	traits::{Currency, Get, OneSessionHandler},
+	traits::{Currency, Get, OneSessionHandler, StorageVersion},
 };
 use frame_system::{
 	offchain::{Account, AppCrypto, CreateSignedTransaction, SendSignedTransaction, Signer},
@@ -101,16 +102,14 @@ pub mod pallet {
 	use super::*;
 
 	/// The current storage version.
-	const STORAGE_VERSION: frame_support::traits::StorageVersion =
-		frame_support::traits::StorageVersion::new(1);
+	const STORAGE_VERSION: StorageVersion = StorageVersion::new(1);
 
 	const _SUCCESS_CODE: u16 = 200;
 	const _BUF_SIZE: usize = 128;
 	const RESPONSE_TIMEOUT: u64 = 20000;
 	pub const BUCKETS_AGGREGATES_FETCH_BATCH_SIZE: usize = 100;
 	pub const NODES_AGGREGATES_FETCH_BATCH_SIZE: usize = 10;
-	pub const IS_RUNNING_KEY: &[u8] = b"offchain::validator::is_running";
-	pub const IS_RUNNING_VALUE: &[u8] = &[1];
+	pub const OCW_MUTEX_ID: &[u8] = b"inspection";
 
 	/// Delta usage of a bucket includes only the delta usage for the processing era reported by
 	/// collectors. This usage can be verified of unverified by inspectors.
@@ -837,15 +836,13 @@ pub mod pallet {
 				return;
 			}
 
-			// Allow only one instance of the offchain worker to run at a time.
-			if !local_storage_compare_and_set(
-				StorageKind::PERSISTENT,
-				IS_RUNNING_KEY,
-				None,
-				IS_RUNNING_VALUE,
-			) {
+			// Allow only one instance of the off-chain worker to run at the same time.
+			let mut ocw_mutex = OcwMutex::new(OCW_MUTEX_ID.to_vec());
+			if !ocw_mutex.try_lock() {
+				log::debug!("Another inspection OCW is already running, terminating...",);
 				return;
 			}
+			log::debug!("OCW mutex 0x{} locked.", hex::encode(ocw_mutex.local_storage_key()));
 
 			let verification_account = unwrap_or_log_error!(
 				Self::collect_verification_pub_key(),
@@ -887,9 +884,6 @@ pub mod pallet {
 
 				Self::submit_errors(&errors, &verification_account, &signer);
 			}
-
-			// Allow the next invocation of the offchain worker hook to run.
-			local_storage_clear(StorageKind::PERSISTENT, IS_RUNNING_KEY);
 		}
 	}
 

pallets/ddc-verification/src/migrations.rs

@@ -1,13 +1,11 @@
-use frame_support::{migration, traits::OnRuntimeUpgrade};
+use frame_support::{migration, pallet_prelude::*, traits::OnRuntimeUpgrade};
 use log;
 
 use super::*;
 
 const LOG_TARGET: &str = "ddc-verification";
 
 pub mod v1 {
-	use frame_support::pallet_prelude::*;
-
 	use super::*;
 
 	pub fn migrate_to_v1<T: Config>() -> Weight {

primitives/Cargo.toml

@@ -14,11 +14,13 @@ codec = { workspace = true }
 frame-support = { workspace = true }
 frame-system = { workspace = true }
 hex = { workspace = true }
+log = { workspace = true }
 polkadot-ckb-merkle-mountain-range = { workspace = true }
 scale-info = { workspace = true }
 serde = { workspace = true }
 sp-application-crypto = { workspace = true }
 sp-core = { workspace = true }
+sp-io = { workspace = true }
 sp-runtime = { workspace = true }
 sp-std = { workspace = true }
 

primitives/src/lib.rs

@@ -16,6 +16,8 @@ use sp_std::collections::btree_set::BTreeSet;
 
 pub mod traits;
 
+pub mod ocw_mutex;
+
 parameter_types! {
 	pub MaxHostLen: u8 = 255;
 	pub MaxDomainLen: u8 = 255;

primitives/src/ocw_mutex.rs

@@ -0,0 +1,62 @@
+use codec::Encode;
+pub use sp_io::offchain::{
+	local_storage_clear, local_storage_compare_and_set, local_storage_get, local_storage_set,
+};
+use sp_runtime::offchain::StorageKind;
+use sp_std::prelude::Vec;
+
+pub const LOCKED_VALUE: &[u8] = &[1];
+pub const PREFIX: &[u8] = b"ocwmu";
+pub const RESET_VALUE: &[u8] = &[0];
+
+/// Intended for allowing only one instance of an off-chain worker with the same ID to run at the
+/// same time.
+///
+/// It takes advantage of the off-chain storage and supports state reset by an
+/// [offchain_localStorageSet](https://polkadot.js.org/docs/kusama/rpc#localstoragesetkind-storagekind-key-bytes-value-bytes-null)
+/// RPC zeroing the `ocwmu{id}` key to fix a [poisoned](https://doc.rust-lang.org/std/sync/struct.Mutex.html#poisoning)
+/// persistent key. Make sure there is no running instance of the OCW when resetting the state.
+pub struct OcwMutex {
+	key: Vec<u8>,
+	locked: bool,
+}
+
+impl OcwMutex {
+	pub fn new(id: Vec<u8>) -> Self {
+		Self { key: (PREFIX, id).encode(), locked: false }
+	}
+
+	pub fn try_lock(&mut self) -> bool {
+		if self.locked {
+			return false;
+		}
+
+		if !local_storage_compare_and_set(StorageKind::PERSISTENT, &self.key, None, LOCKED_VALUE) {
+			match local_storage_get(StorageKind::PERSISTENT, &self.key) {
+				Some(value) if value == RESET_VALUE => {
+					log::warn!("OCW mutex key '{:?}' cleared.", &self.key);
+					local_storage_clear(StorageKind::PERSISTENT, &self.key);
+
+					return self.try_lock();
+				},
+				_ => return false,
+			}
+		}
+
+		self.locked = true;
+
+		true
+	}
+
+	pub fn local_storage_key(&self) -> &[u8] {
+		&self.key
+	}
+}
+
+impl Drop for OcwMutex {
+	fn drop(&mut self) {
+		if self.locked {
+			local_storage_clear(StorageKind::PERSISTENT, &self.key);
+		}
+	}
+}

runtime/cere-dev/src/lib.rs

@@ -153,7 +153,7 @@ pub const VERSION: RuntimeVersion = RuntimeVersion {
 	// and set impl_version to 0. If only runtime
 	// implementation changes and behavior does not, then leave spec_version as
 	// is and increment impl_version.
-	spec_version: 64000,
+	spec_version: 64001,
 	impl_version: 0,
 	apis: RUNTIME_API_VERSIONS,
 	transaction_version: 24,
@@ -1431,7 +1431,7 @@ pub type SignedPayload = generic::SignedPayload<RuntimeCall, SignedExtra>;
 pub type CheckedExtrinsic = generic::CheckedExtrinsic<AccountId, RuntimeCall, SignedExtra>;
 
 /// Runtime migrations
-type Migrations = (pallet_ddc_payouts::migrations::v2::MigrateToV2<Runtime>,);
+type Migrations = ();
 
 /// Executive: handles dispatch to the various modules.
 pub type Executive = frame_executive::Executive<

runtime/cere/src/lib.rs

@@ -146,7 +146,7 @@ pub const VERSION: RuntimeVersion = RuntimeVersion {
 	// and set impl_version to 0. If only runtime
 	// implementation changes and behavior does not, then leave spec_version as
 	// is and increment impl_version.
-	spec_version: 64000,
+	spec_version: 64001,
 	impl_version: 0,
 	apis: RUNTIME_API_VERSIONS,
 	transaction_version: 24,
@@ -1452,22 +1452,22 @@ pub type CheckedExtrinsic = generic::CheckedExtrinsic<AccountId, RuntimeCall, Si
 // const IDENTITY_MIGRATION_KEY_LIMIT: u64 = u64::MAX; // for `pallet_identity` migration below
 
 /// Runtime migrations
-type Migrations = (
-	// Migrations related to substrate version upgrades
-	// pallet_nomination_pools::migration::versioned::V5toV6<Runtime>,
-	// pallet_nomination_pools::migration::versioned::V6ToV7<Runtime>,
-	// pallet_nomination_pools::migration::versioned::V7ToV8<Runtime>,
-	// pallet_staking::migrations::v14::MigrateToV14<Runtime>,
-	// pallet_grandpa::migrations::MigrateV4ToV5<Runtime>,
-	// pallet_identity::migration::versioned::V0ToV1<Runtime, IDENTITY_MIGRATION_KEY_LIMIT>,
-
-	// The 'Unreleased' migration enables DAC Verification, that atm. is enabled at QANET only.
-	// Uncomment this line when DAC is ready for TESTNET and MAINNET migrations::Unreleased,
-	// migrations::Unreleased,
-
-	// Migrations for DAC and Payouts on QANET
-	pallet_ddc_payouts::migrations::v2::MigrateToV2<Runtime>,
-);
+// type Migrations = (
+// 	// Migrations related to substrate version upgrades
+// 	// pallet_nomination_pools::migration::versioned::V5toV6<Runtime>,
+// 	// pallet_nomination_pools::migration::versioned::V6ToV7<Runtime>,
+// 	// pallet_nomination_pools::migration::versioned::V7ToV8<Runtime>,
+// 	// pallet_staking::migrations::v14::MigrateToV14<Runtime>,
+// 	// pallet_grandpa::migrations::MigrateV4ToV5<Runtime>,
+// 	// pallet_identity::migration::versioned::V0ToV1<Runtime, IDENTITY_MIGRATION_KEY_LIMIT>,
+
+// 	// The 'Unreleased' migration enables DAC Verification, that atm. is enabled at QANET only.
+// 	// Uncomment this line when DAC is ready for TESTNET and MAINNET migrations::Unreleased,
+// 	// migrations::Unreleased,
+
+// 	// Migrations for DAC and Payouts on QANET
+// );
+type Migrations = ();
 
 pub mod migrations {
 	use super::*;