diff --git a/client/db/src/offchain.rs b/client/db/src/offchain.rs
index 651510d6e887e..f6a0925a08617 100644
--- a/client/db/src/offchain.rs
+++ b/client/db/src/offchain.rs
@@ -67,6 +67,14 @@ impl sp_core::offchain::OffchainStorage for LocalStorage {
 		self.db.commit(tx);
 	}
 
+	fn remove(&mut self, prefix: &[u8], key: &[u8]) {
+		let key: Vec<u8> = prefix.iter().chain(key).cloned().collect();
+		let mut tx = Transaction::new();
+		tx.remove(columns::OFFCHAIN, &key);
+
+		self.db.commit(tx);
+	}
+
 	fn get(&self, prefix: &[u8], key: &[u8]) -> Option<Vec<u8>> {
 		let key: Vec<u8> = prefix.iter().chain(key).cloned().collect();
 		self.db.get(columns::OFFCHAIN, &key)
diff --git a/client/offchain/src/api.rs b/client/offchain/src/api.rs
index 45a82d230c11a..a7f4ecbc5825e 100644
--- a/client/offchain/src/api.rs
+++ b/client/offchain/src/api.rs
@@ -100,6 +100,13 @@ impl<Storage: OffchainStorage> OffchainExt for Api<Storage> {
 		}
 	}
 
+	fn local_storage_clear(&mut self, kind: StorageKind, key: &[u8]) {
+		match kind {
+			StorageKind::PERSISTENT => self.db.remove(STORAGE_PREFIX, key),
+			StorageKind::LOCAL => unavailable_yet(LOCAL_DB),
+		}
+	}
+
 	fn local_storage_compare_and_set(
 		&mut self,
 		kind: StorageKind,
diff --git a/frame/system/src/lib.rs b/frame/system/src/lib.rs
index 4fa826ce89893..b4f438ce03422 100644
--- a/frame/system/src/lib.rs
+++ b/frame/system/src/lib.rs
@@ -114,6 +114,7 @@ use sp_runtime::{
 		MaybeSerialize, MaybeSerializeDeserialize, MaybeMallocSizeOf, StaticLookup, One, Bounded,
 		Dispatchable, DispatchInfoOf, PostDispatchInfoOf,
 	},
+	offchain::storage_lock::BlockNumberProvider,
 };
 
 use sp_core::{ChangesTrieConfiguration, storage::well_known_keys};
@@ -1268,6 +1269,15 @@ impl<T: Trait> Happened<T::AccountId> for CallKillAccount<T> {
 	}
 }
 
+impl<T: Trait> BlockNumberProvider for Module<T>
+{
+	type BlockNumber = <T as Trait>::BlockNumber;
+
+	fn current_block_number() -> Self::BlockNumber {
+		Module::<T>::block_number()
+	}
+}
+
 // Implement StoredMap for a simple single-item, kill-account-on-remove system. This works fine for
 // storing a single item which is required to not be empty/default for the account to exist.
 // Anything more complex will need more sophisticated logic.
diff --git a/primitives/core/src/offchain/mod.rs b/primitives/core/src/offchain/mod.rs
index 1d77e10f59ca7..b2ff3552135ce 100644
--- a/primitives/core/src/offchain/mod.rs
+++ b/primitives/core/src/offchain/mod.rs
@@ -37,6 +37,9 @@ pub trait OffchainStorage: Clone + Send + Sync {
 	/// Persist a value in storage under given key and prefix.
 	fn set(&mut self, prefix: &[u8], key: &[u8], value: &[u8]);
 
+	/// Clear a storage entry under given key and prefix.
+	fn remove(&mut self, prefix: &[u8], key: &[u8]);
+
 	/// Retrieve a value from storage under given key and prefix.
 	fn get(&self, prefix: &[u8], key: &[u8]) -> Option<Vec<u8>>;
 
@@ -219,7 +222,7 @@ pub struct Duration(u64);
 
 impl Duration {
 	/// Create new duration representing given number of milliseconds.
-	pub fn from_millis(millis: u64) -> Self {
+	pub const fn from_millis(millis: u64) -> Self {
 		Duration(millis)
 	}
 
@@ -346,9 +349,15 @@ pub trait Externalities: Send {
 	/// Sets a value in the local storage.
 	///
 	/// Note this storage is not part of the consensus, it's only accessible by
-	/// offchain worker tasks running on the same machine. It IS persisted between runs.
+	/// offchain worker tasks running on the same machine. It _is_ persisted between runs.
 	fn local_storage_set(&mut self, kind: StorageKind, key: &[u8], value: &[u8]);
 
+	/// Removes a value in the local storage.
+	///
+	/// Note this storage is not part of the consensus, it's only accessible by
+	/// offchain worker tasks running on the same machine. It _is_ persisted between runs.
+	fn local_storage_clear(&mut self, kind: StorageKind, key: &[u8]);
+
 	/// Sets a value in the local storage if it matches current value.
 	///
 	/// Since multiple offchain workers may be running concurrently, to prevent
@@ -357,7 +366,7 @@ pub trait Externalities: Send {
 	/// Returns `true` if the value has been set, `false` otherwise.
 	///
 	/// Note this storage is not part of the consensus, it's only accessible by
-	/// offchain worker tasks running on the same machine. It IS persisted between runs.
+	/// offchain worker tasks running on the same machine. It _is_ persisted between runs.
 	fn local_storage_compare_and_set(
 		&mut self,
 		kind: StorageKind,
@@ -370,7 +379,7 @@ pub trait Externalities: Send {
 	///
 	/// If the value does not exist in the storage `None` will be returned.
 	/// Note this storage is not part of the consensus, it's only accessible by
-	/// offchain worker tasks running on the same machine. It IS persisted between runs.
+	/// offchain worker tasks running on the same machine. It _is_ persisted between runs.
 	fn local_storage_get(&mut self, kind: StorageKind, key: &[u8]) -> Option<Vec<u8>>;
 
 	/// Initiates a http request given HTTP verb and the URL.
@@ -513,6 +522,10 @@ impl<T: Externalities + ?Sized> Externalities for Box<T> {
 		(&mut **self).local_storage_set(kind, key, value)
 	}
 
+	fn local_storage_clear(&mut self, kind: StorageKind, key: &[u8]) {
+		(&mut **self).local_storage_clear(kind, key)
+	}
+
 	fn local_storage_compare_and_set(
 		&mut self,
 		kind: StorageKind,
@@ -618,6 +631,11 @@ impl<T: Externalities> Externalities for LimitedExternalities<T> {
 		self.externalities.local_storage_set(kind, key, value)
 	}
 
+	fn local_storage_clear(&mut self, kind: StorageKind, key: &[u8]) {
+		self.check(Capability::OffchainWorkerDbWrite, "local_storage_clear");
+		self.externalities.local_storage_clear(kind, key)
+	}
+
 	fn local_storage_compare_and_set(
 		&mut self,
 		kind: StorageKind,
diff --git a/primitives/core/src/offchain/storage.rs b/primitives/core/src/offchain/storage.rs
index 1826015b0d0cc..52a7bbe857d9d 100644
--- a/primitives/core/src/offchain/storage.rs
+++ b/primitives/core/src/offchain/storage.rs
@@ -51,6 +51,11 @@ impl OffchainStorage for InMemOffchainStorage {
 		self.storage.insert(key, value.to_vec());
 	}
 
+	fn remove(&mut self, prefix: &[u8], key: &[u8]) {
+		let key: Vec<u8> = prefix.iter().chain(key).cloned().collect();
+		self.storage.remove(&key);
+	}
+
 	fn get(&self, prefix: &[u8], key: &[u8]) -> Option<Vec<u8>> {
 		let key: Vec<u8> = prefix.iter().chain(key).cloned().collect();
 		self.storage.get(&key).cloned()
diff --git a/primitives/core/src/offchain/testing.rs b/primitives/core/src/offchain/testing.rs
index 5e25e433a3cd5..76cf8915f2054 100644
--- a/primitives/core/src/offchain/testing.rs
+++ b/primitives/core/src/offchain/testing.rs
@@ -73,10 +73,10 @@ pub struct OffchainState {
 	pub persistent_storage: InMemOffchainStorage,
 	/// Local storage
 	pub local_storage: InMemOffchainStorage,
-	/// Current timestamp (unix millis)
-	pub timestamp: u64,
 	/// A supposedly random seed.
 	pub seed: [u8; 32],
+	/// A timestamp simulating the current time.
+	pub timestamp: Timestamp,
 }
 
 impl OffchainState {
@@ -160,11 +160,11 @@ impl offchain::Externalities for TestOffchainExt {
 	}
 
 	fn timestamp(&mut self) -> Timestamp {
-		Timestamp::from_unix_millis(self.0.read().timestamp)
+		self.0.read().timestamp
 	}
 
-	fn sleep_until(&mut self, _deadline: Timestamp) {
-		unimplemented!("not needed in tests so far")
+	fn sleep_until(&mut self, deadline: Timestamp) {
+		self.0.write().timestamp = deadline;
 	}
 
 	fn random_seed(&mut self) -> [u8; 32] {
@@ -179,6 +179,14 @@ impl offchain::Externalities for TestOffchainExt {
 		}.set(b"", key, value);
 	}
 
+	fn local_storage_clear(&mut self, kind: StorageKind, key: &[u8]) {
+		let mut state = self.0.write();
+		match kind {
+			StorageKind::LOCAL => &mut state.local_storage,
+			StorageKind::PERSISTENT => &mut state.persistent_storage,
+		}.remove(b"", key);
+	}
+
 	fn local_storage_compare_and_set(
 		&mut self,
 		kind: StorageKind,
diff --git a/primitives/io/src/lib.rs b/primitives/io/src/lib.rs
index f28f3e2c9552e..8d81a84c4c88a 100644
--- a/primitives/io/src/lib.rs
+++ b/primitives/io/src/lib.rs
@@ -796,6 +796,16 @@ pub trait Offchain {
 			.local_storage_set(kind, key, value)
 	}
 
+	/// Remove a value from the local storage.
+	///
+	/// Note this storage is not part of the consensus, it's only accessible by
+	/// offchain worker tasks running on the same machine. It IS persisted between runs.
+	fn local_storage_clear(&mut self, kind: StorageKind, key: &[u8]) {
+		self.extension::<OffchainExt>()
+			.expect("local_storage_clear can be called only in the offchain worker context")
+			.local_storage_clear(kind, key)
+	}
+
 	/// Sets a value in the local storage if it matches current value.
 	///
 	/// Since multiple offchain workers may be running concurrently, to prevent
diff --git a/primitives/runtime/src/offchain/mod.rs b/primitives/runtime/src/offchain/mod.rs
index 427b54468f48d..fe5844ce3004b 100644
--- a/primitives/runtime/src/offchain/mod.rs
+++ b/primitives/runtime/src/offchain/mod.rs
@@ -19,5 +19,6 @@
 
 pub mod http;
 pub mod storage;
+pub mod storage_lock;
 
 pub use sp_core::offchain::*;
diff --git a/primitives/runtime/src/offchain/storage.rs b/primitives/runtime/src/offchain/storage.rs
index f8dcd73fa2bc6..2f62d400c0b95 100644
--- a/primitives/runtime/src/offchain/storage.rs
+++ b/primitives/runtime/src/offchain/storage.rs
@@ -50,6 +50,11 @@ impl<'a> StorageValueRef<'a> {
 		})
 	}
 
+	/// Remove the associated value from the storage.
+	pub fn clear(&mut self) {
+		sp_io::offchain::local_storage_clear(self.kind, self.key)
+	}
+
 	/// Retrieve & decode the value from storage.
 	///
 	/// Note that if you want to do some checks based on the value
@@ -67,7 +72,8 @@ impl<'a> StorageValueRef<'a> {
 	/// Function `f` should return a new value that we should attempt to write to storage.
 	/// This function returns:
 	/// 1. `Ok(Ok(T))` in case the value has been successfully set.
-	/// 2. `Ok(Err(T))` in case the value was returned, but it couldn't have been set.
+	/// 2. `Ok(Err(T))` in case the value was calculated by the passed closure `f`,
+	///    but it could not be stored.
 	/// 3. `Err(_)` in case `f` returns an error.
 	pub fn mutate<T, E, F>(&self, f: F) -> Result<Result<T, T>, E> where
 		T: codec::Codec,
diff --git a/primitives/runtime/src/offchain/storage_lock.rs b/primitives/runtime/src/offchain/storage_lock.rs
new file mode 100644
index 0000000000000..60bf9f04772dd
--- /dev/null
+++ b/primitives/runtime/src/offchain/storage_lock.rs
@@ -0,0 +1,516 @@
+// Copyright 2019-2020 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+//! # Off-chain Storage Lock
+//!
+//! A storage-based lock with a defined expiry time.
+//!
+//! The lock is using Local Storage and allows synchronizing access to critical
+//! section of your code for concurrently running Off-chain Workers. Usage of
+//! `PERSISTENT` variant of the storage persists the lock value across a full node
+//! restart or re-orgs.
+//!
+//! A use case for the lock is to make sure that a particular section of the
+//! code is only run by one Off-chain Worker at a time. This may include
+//! performing a side-effect (i.e. an HTTP call) or alteration of single or
+//! multiple Local Storage entries.
+//!
+//! One use case would be collective updates of multiple data items or append /
+//! remove of i.e. sets, vectors which are stored in the off-chain storage DB.
+//!
+//! ## Example:
+//!
+//! ```rust
+//! # use codec::{Decode, Encode, Codec};
+//! // in your off-chain worker code
+//! use sp_runtime::offchain::{
+//!		storage::StorageValueRef,
+//!		storage_lock::{StorageLock, Time},
+//! };
+//!
+//! fn append_to_in_storage_vec<'a, T>(key: &'a [u8], _: T) where T: Codec {
+//!    // `access::lock` defines the storage entry which is used for
+//!    // persisting the lock in the underlying database.
+//!    // The entry name _must_ be unique and can be interpreted as a
+//!    // unique mutex instance reference tag.
+//!    let mut lock = StorageLock::<Time>::new(b"access::lock");
+//!    {
+//!         let _guard = lock.lock();
+//!         let acc = StorageValueRef::persistent(key);
+//!         let v: Vec<T> = acc.get::<Vec<T>>().unwrap().unwrap();
+//!         // modify `v` as desired
+//!         // i.e. perform some heavy computation with
+//!         // side effects that should only be done once.
+//!         acc.set(&v);
+//!         // drop `_guard` implicitly at end of scope
+//!    }
+//! }
+//! ```
+
+use crate::offchain::storage::StorageValueRef;
+use crate::traits::AtLeast32Bit;
+use codec::{Codec, Decode, Encode};
+use sp_core::offchain::{Duration, Timestamp};
+use sp_io::offchain;
+
+/// Default expiry duration for time based locks in milliseconds.
+const STORAGE_LOCK_DEFAULT_EXPIRY_DURATION: Duration = Duration::from_millis(20_000);
+
+/// Default expiry duration for block based locks in blocks.
+const STORAGE_LOCK_DEFAULT_EXPIRY_BLOCKS: u32 = 4;
+
+/// Time between checks if the lock is still being held in milliseconds.
+const STORAGE_LOCK_PER_CHECK_ITERATION_SNOOZE_MIN: Duration = Duration::from_millis(100);
+const STORAGE_LOCK_PER_CHECK_ITERATION_SNOOZE_MAX: Duration = Duration::from_millis(10);
+
+/// Lockable item for use with a persisted storage lock.
+///
+/// Bound for an item that has a stateful ordered meaning
+/// without explicitly requiring `Ord` trait in general.
+pub trait Lockable: Sized {
+	/// An instant type specifying i.e. a point in time.
+	type Deadline: Sized + Codec + Clone;
+
+	/// Calculate the deadline based on a current state
+	/// such as time or block number and derives the deadline.
+	fn deadline(&self) -> Self::Deadline;
+
+	/// Verify the deadline has not expired compared to the
+	/// current state, i.e. time or block number.
+	fn has_expired(deadline: &Self::Deadline) -> bool;
+
+	/// Snooze at least until `deadline` is reached.
+	///
+	/// Note that `deadline` is only passed to allow optimizations
+	/// for `Lockables` which have a time based component.
+	fn snooze(_deadline: &Self::Deadline) {
+		sp_io::offchain::sleep_until(
+			offchain::timestamp().add(STORAGE_LOCK_PER_CHECK_ITERATION_SNOOZE_MAX),
+		);
+	}
+}
+
+/// Lockable based on the current timestamp with a configurable expiration time.
+#[derive(Encode, Decode)]
+pub struct Time {
+	/// How long the lock will stay valid once `fn lock(..)` or
+	/// `fn try_lock(..)` successfully acquire a lock.
+	expiration_duration: Duration,
+}
+
+impl Default for Time {
+	fn default() -> Self {
+		Self {
+			expiration_duration: STORAGE_LOCK_DEFAULT_EXPIRY_DURATION,
+		}
+	}
+}
+
+impl Lockable for Time {
+	type Deadline = Timestamp;
+
+	fn deadline(&self) -> Self::Deadline {
+		offchain::timestamp().add(self.expiration_duration)
+	}
+
+	fn has_expired(deadline: &Self::Deadline) -> bool {
+		offchain::timestamp() > *deadline
+	}
+
+	fn snooze(deadline: &Self::Deadline) {
+		let now = offchain::timestamp();
+		let remainder: Duration = now.diff(&deadline);
+		// do not snooze the full duration, but instead snooze max 100ms
+		// it might get unlocked in another thread
+		use core::cmp::{max, min};
+		let snooze = max(
+			min(remainder, STORAGE_LOCK_PER_CHECK_ITERATION_SNOOZE_MAX),
+			STORAGE_LOCK_PER_CHECK_ITERATION_SNOOZE_MIN,
+		);
+		sp_io::offchain::sleep_until(now.add(snooze));
+	}
+}
+
+/// A deadline based on block number and time.
+#[derive(Encode, Decode, Eq, PartialEq)]
+pub struct BlockAndTimeDeadline<B: BlockNumberProvider> {
+	/// The block number until which the lock is still valid _at least_.
+	pub block_number: <B as BlockNumberProvider>::BlockNumber,
+	/// The timestamp until which the lock is still valid _at least_.
+	pub timestamp: Timestamp,
+}
+
+impl<B: BlockNumberProvider> Clone for BlockAndTimeDeadline<B> {
+	fn clone(&self) -> Self {
+		Self {
+			block_number: self.block_number.clone(),
+			timestamp: self.timestamp.clone(),
+		}
+	}
+}
+
+impl<B: BlockNumberProvider> Default for BlockAndTimeDeadline<B> {
+	/// Provide the current state of block number and time.
+	fn default() -> Self {
+		Self {
+			block_number: B::current_block_number() + STORAGE_LOCK_DEFAULT_EXPIRY_BLOCKS.into(),
+			timestamp: offchain::timestamp().add(STORAGE_LOCK_DEFAULT_EXPIRY_DURATION),
+		}
+	}
+}
+
+/// Lockable based on block number and timestamp.
+///
+/// Expiration is defined if both, block number _and_ timestamp
+/// expire.
+pub struct BlockAndTime<B: BlockNumberProvider> {
+	/// Relative block number offset, which is used to determine
+	/// the block number part of the deadline.
+	expiration_block_number_offset: u32,
+	/// Relative duration, used to derive the time based part of
+	/// the deadline.
+	expiration_duration: Duration,
+
+	#[doc(hidden)]
+	_phantom: core::marker::PhantomData<B>,
+}
+
+impl<B: BlockNumberProvider> Default for BlockAndTime<B> {
+	fn default() -> Self {
+		Self {
+			expiration_block_number_offset: STORAGE_LOCK_DEFAULT_EXPIRY_BLOCKS,
+			expiration_duration: STORAGE_LOCK_DEFAULT_EXPIRY_DURATION,
+			_phantom: core::marker::PhantomData::<B>,
+		}
+	}
+}
+
+// derive not possible, since `B` does not necessarily implement `trait Clone`
+impl<B: BlockNumberProvider> Clone for BlockAndTime<B> {
+	fn clone(&self) -> Self {
+		Self {
+			expiration_block_number_offset: self.expiration_block_number_offset.clone(),
+			expiration_duration: self.expiration_duration,
+			_phantom: core::marker::PhantomData::<B>,
+		}
+	}
+}
+
+impl<B: BlockNumberProvider> Lockable for BlockAndTime<B> {
+	type Deadline = BlockAndTimeDeadline<B>;
+
+	fn deadline(&self) -> Self::Deadline {
+		let block_number = <B as BlockNumberProvider>::current_block_number()
+			+ self.expiration_block_number_offset.into();
+		BlockAndTimeDeadline {
+			timestamp: offchain::timestamp().add(self.expiration_duration),
+			block_number,
+		}
+	}
+
+	fn has_expired(deadline: &Self::Deadline) -> bool {
+		offchain::timestamp() > deadline.timestamp
+			&& <B as BlockNumberProvider>::current_block_number() > deadline.block_number
+	}
+
+	fn snooze(deadline: &Self::Deadline) {
+		let now = offchain::timestamp();
+		let remainder: Duration = now.diff(&(deadline.timestamp));
+		use core::cmp::{max, min};
+		let snooze = max(
+			min(remainder, STORAGE_LOCK_PER_CHECK_ITERATION_SNOOZE_MAX),
+			STORAGE_LOCK_PER_CHECK_ITERATION_SNOOZE_MIN,
+		);
+		sp_io::offchain::sleep_until(now.add(snooze));
+	}
+}
+
+/// Storage based lock.
+///
+/// A lock that is persisted in the DB and provides the ability to guard against
+/// concurrent access in an off-chain worker, with a defined expiry deadline
+/// based on the concrete [`Lockable`](Self::Lockable) implementation.
+pub struct StorageLock<'a, L = Time> {
+	// A storage value ref which defines the DB entry representing the lock.
+	value_ref: StorageValueRef<'a>,
+	lockable: L,
+}
+
+impl<'a, L: Lockable + Default> StorageLock<'a, L> {
+	/// Create a new storage lock with a `default()` instance of type `L`.
+	pub fn new(key: &'a [u8]) -> Self {
+		Self::with_lockable(key, Default::default())
+	}
+}
+
+impl<'a, L: Lockable> StorageLock<'a, L> {
+	/// Create a new storage lock with an explicit instance of a lockable `L`.
+	pub fn with_lockable(key: &'a [u8], lockable: L) -> Self {
+		Self {
+			value_ref: StorageValueRef::<'a>::persistent(key),
+			lockable,
+		}
+	}
+
+	/// Internal lock helper to avoid lifetime conflicts.
+	fn try_lock_inner(
+		&mut self,
+		new_deadline: L::Deadline,
+	) -> Result<(), <L as Lockable>::Deadline> {
+		let res = self.value_ref.mutate(
+			|s: Option<Option<L::Deadline>>|
+			-> Result<<L as Lockable>::Deadline, <L as Lockable>::Deadline> {
+				match s {
+					// no lock set, we can safely acquire it
+					None => Ok(new_deadline),
+					// write was good, but read failed
+					Some(None) => Ok(new_deadline),
+					// lock is set, but it is expired. We can re-acquire it.
+					Some(Some(deadline)) if <L as Lockable>::has_expired(&deadline) =>
+						Ok(new_deadline),
+					// lock is present and is still active
+					Some(Some(deadline)) => Err(deadline),
+				}
+			},
+		);
+		match res {
+			Ok(Ok(_)) => Ok(()),
+			Ok(Err(deadline)) => Err(deadline),
+			Err(e) => Err(e),
+		}
+	}
+
+	/// A single attempt to lock using the storage entry.
+	///
+	/// Returns a lock guard on success, otherwise an error containing the
+	/// `<Self::Lockable>::Deadline` in for the currently active lock
+	/// by another task `Err(<L as Lockable>::Deadline)`.
+	pub fn try_lock(&mut self) -> Result<StorageLockGuard<'a, '_, L>, <L as Lockable>::Deadline> {
+		self.try_lock_inner(self.lockable.deadline())?;
+		Ok(StorageLockGuard::<'a, '_> { lock: Some(self) })
+	}
+
+	/// Repeated lock attempts until the lock is successfully acquired.
+	///
+	/// If one uses `fn forget(..)`, it is highly likely `fn try_lock(..)`
+	/// is the correct API to use instead of `fn lock(..)`, since that might
+	/// never unlock in the anticipated span i.e. when used with `BlockAndTime`
+	/// during a certain block number span.
+	pub fn lock(&mut self) -> StorageLockGuard<'a, '_, L> {
+		while let Err(deadline) = self.try_lock_inner(self.lockable.deadline()) {
+			L::snooze(&deadline);
+		}
+		StorageLockGuard::<'a, '_, L> { lock: Some(self) }
+	}
+
+	/// Explicitly unlock the lock.
+	fn unlock(&mut self) {
+		self.value_ref.clear();
+	}
+}
+
+/// RAII style guard for a lock.
+pub struct StorageLockGuard<'a, 'b, L: Lockable> {
+	lock: Option<&'b mut StorageLock<'a, L>>,
+}
+
+impl<'a, 'b, L: Lockable> StorageLockGuard<'a, 'b, L> {
+	/// Consume the guard but **do not** unlock the underlying lock.
+	///
+	/// Can be used to implement a grace period after doing some
+	/// heavy computations and sending a transaction to be included
+	/// on-chain. By forgetting the lock, it will stay locked until
+	/// its expiration deadline is reached while the off-chain worker
+	/// can already exit.
+	pub fn forget(mut self) {
+		let _ = self.lock.take();
+	}
+}
+
+impl<'a, 'b, L: Lockable> Drop for StorageLockGuard<'a, 'b, L> {
+	fn drop(&mut self) {
+		if let Some(lock) = self.lock.take() {
+			lock.unlock();
+		}
+	}
+}
+
+impl<'a> StorageLock<'a, Time> {
+	/// Explicitly create a time based storage lock with a non-default
+	/// expiration timeout.
+	pub fn with_deadline(key: &'a [u8], expiration_duration: Duration) -> Self {
+		Self {
+			value_ref: StorageValueRef::<'a>::persistent(key),
+			lockable: Time {
+				expiration_duration: expiration_duration,
+			},
+		}
+	}
+}
+
+impl<'a, B> StorageLock<'a, BlockAndTime<B>>
+where
+	B: BlockNumberProvider,
+{
+	/// Explicitly create a time and block number based storage lock with
+	/// a non-default expiration duration and block number offset.
+	pub fn with_block_and_time_deadline(
+		key: &'a [u8],
+		expiration_block_number_offset: u32,
+		expiration_duration: Duration,
+	) -> Self {
+		Self {
+			value_ref: StorageValueRef::<'a>::persistent(key),
+			lockable: BlockAndTime::<B> {
+				expiration_block_number_offset,
+				expiration_duration,
+				_phantom: core::marker::PhantomData,
+			},
+		}
+	}
+
+	/// Explicitly create a time and block number based storage lock with
+	/// the default expiration duration and a non-default block number offset.
+	pub fn with_block_deadline(key: &'a [u8], expiration_block_number_offset: u32) -> Self {
+		Self {
+			value_ref: StorageValueRef::<'a>::persistent(key),
+			lockable: BlockAndTime::<B> {
+				expiration_block_number_offset,
+				expiration_duration: STORAGE_LOCK_DEFAULT_EXPIRY_DURATION,
+				_phantom: core::marker::PhantomData,
+			},
+		}
+	}
+}
+
+/// Bound for a block number source
+/// used with [`BlockAndTime<BlockNumberProvider>`](Self::BlockAndTime).
+pub trait BlockNumberProvider {
+	/// Type of `BlockNumber` to provide.
+	type BlockNumber: Codec + Clone + Ord + Eq + AtLeast32Bit;
+	/// Returns the current block number.
+	///
+	/// Provides an abstraction over an arbitrary way of providing the
+	/// current block number.
+	///
+	/// In case of using crate `sp_runtime` without the crate `frame`
+	/// system, it is already implemented for
+	/// `frame_system::Module<T: Trait>` as:
+	///
+	/// ```ignore
+	/// fn current_block_number() -> Self {
+	///     frame_system::Module<Trait>::block_number()
+	/// }
+	/// ```
+	/// .
+	fn current_block_number() -> Self::BlockNumber;
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use sp_core::offchain::{testing, OffchainExt, OffchainStorage};
+	use sp_io::TestExternalities;
+
+	const VAL_1: u32 = 0u32;
+	const VAL_2: u32 = 0xFFFF_FFFFu32;
+
+	#[test]
+	fn storage_lock_write_unlock_lock_read_unlock() {
+		let (offchain, state) = testing::TestOffchainExt::new();
+		let mut t = TestExternalities::default();
+		t.register_extension(OffchainExt::new(offchain));
+
+		t.execute_with(|| {
+			let mut lock = StorageLock::<'_, Time>::new(b"lock_1");
+
+			let val = StorageValueRef::persistent(b"protected_value");
+
+			{
+				let _guard = lock.lock();
+
+				val.set(&VAL_1);
+
+				assert_eq!(val.get::<u32>(), Some(Some(VAL_1)));
+			}
+
+			{
+				let _guard = lock.lock();
+				val.set(&VAL_2);
+
+				assert_eq!(val.get::<u32>(), Some(Some(VAL_2)));
+			}
+		});
+		// lock must have been cleared at this point
+		assert_eq!(state.read().persistent_storage.get(b"", b"lock_1"), None);
+	}
+
+	#[test]
+	fn storage_lock_and_forget() {
+		let (offchain, state) = testing::TestOffchainExt::new();
+		let mut t = TestExternalities::default();
+		t.register_extension(OffchainExt::new(offchain));
+
+		t.execute_with(|| {
+			let mut lock = StorageLock::<'_, Time>::new(b"lock_2");
+
+			let val = StorageValueRef::persistent(b"protected_value");
+
+			let guard = lock.lock();
+
+			val.set(&VAL_1);
+
+			assert_eq!(val.get::<u32>(), Some(Some(VAL_1)));
+
+			guard.forget();
+		});
+		// lock must have been cleared at this point
+		let opt = state.read().persistent_storage.get(b"", b"lock_2");
+		assert!(opt.is_some());
+	}
+
+	#[test]
+	fn storage_lock_and_let_expire_and_lock_again() {
+		let (offchain, state) = testing::TestOffchainExt::new();
+		let mut t = TestExternalities::default();
+		t.register_extension(OffchainExt::new(offchain));
+
+		t.execute_with(|| {
+			let sleep_until = offchain::timestamp().add(Duration::from_millis(500));
+			let lock_expiration = Duration::from_millis(200);
+
+			let mut lock = StorageLock::<'_, Time>::with_deadline(b"lock_3", lock_expiration);
+
+			{
+				let guard = lock.lock();
+				guard.forget();
+			}
+
+			// assure the lock expires
+			offchain::sleep_until(sleep_until);
+
+			let mut lock = StorageLock::<'_, Time>::new(b"lock_3");
+			let res = lock.try_lock();
+			assert!(res.is_ok());
+			let guard = res.unwrap();
+			guard.forget();
+		});
+
+		// lock must have been cleared at this point
+		let opt = state.read().persistent_storage.get(b"", b"lock_3");
+		assert!(opt.is_some());
+	}
+}