#![doc = docify::embed!("src/tests.rs", basic_scheduling_works)]
#![doc = docify::embed!("src/tests.rs", scheduling_with_preimages_works)]
#![cfg_attr(not(feature = "std"), no_std)]
#[cfg(feature = "runtime-benchmarks")]
mod benchmarking;
pub mod migration;
#[cfg(test)]
mod mock;
#[cfg(test)]
mod tests;
pub mod weights;
use codec::{Decode, Encode, MaxEncodedLen};
use frame_support::{
dispatch::{DispatchResult, GetDispatchInfo, Parameter, RawOrigin},
ensure,
traits::{
schedule::{self, DispatchTime, MaybeHashed},
Bounded, CallerTrait, EnsureOrigin, Get, IsType, OriginTrait, PalletInfoAccess,
PrivilegeCmp, QueryPreimage, StorageVersion, StorePreimage,
},
weights::{Weight, WeightMeter},
};
use frame_system::{
pallet_prelude::BlockNumberFor,
{self as system},
};
use scale_info::TypeInfo;
use sp_io::hashing::blake2_256;
use sp_runtime::{
traits::{BadOrigin, Dispatchable, One, Saturating, Zero},
BoundedVec, DispatchError, RuntimeDebug,
};
use sp_std::{borrow::Borrow, cmp::Ordering, marker::PhantomData, prelude::*};
pub use pallet::*;
pub use weights::WeightInfo;
pub type PeriodicIndex = u32;
pub type TaskAddress<BlockNumber> = (BlockNumber, u32);
pub type CallOrHashOf<T> =
MaybeHashed<<T as Config>::RuntimeCall, <T as frame_system::Config>::Hash>;
pub type BoundedCallOf<T> =
Bounded<<T as Config>::RuntimeCall, <T as frame_system::Config>::Hashing>;
#[derive(Clone, Copy, RuntimeDebug, PartialEq, Eq, Encode, Decode, MaxEncodedLen, TypeInfo)]
pub struct RetryConfig<Period> {
total_retries: u8,
remaining: u8,
period: Period,
}
#[cfg_attr(any(feature = "std", test), derive(PartialEq, Eq))]
#[derive(Clone, RuntimeDebug, Encode, Decode)]
struct ScheduledV1<Call, BlockNumber> {
maybe_id: Option<Vec<u8>>,
priority: schedule::Priority,
call: Call,
maybe_periodic: Option<schedule::Period<BlockNumber>>,
}
#[cfg_attr(any(feature = "std", test), derive(PartialEq, Eq))]
#[derive(Clone, RuntimeDebug, Encode, Decode, MaxEncodedLen, TypeInfo)]
pub struct Scheduled<Name, Call, BlockNumber, PalletsOrigin, AccountId> {
maybe_id: Option<Name>,
priority: schedule::Priority,
call: Call,
maybe_periodic: Option<schedule::Period<BlockNumber>>,
origin: PalletsOrigin,
_phantom: PhantomData<AccountId>,
}
impl<Name, Call, BlockNumber, PalletsOrigin, AccountId>
Scheduled<Name, Call, BlockNumber, PalletsOrigin, AccountId>
where
Call: Clone,
PalletsOrigin: Clone,
{
pub fn as_retry(&self) -> Self {
Self {
maybe_id: None,
priority: self.priority,
call: self.call.clone(),
maybe_periodic: None,
origin: self.origin.clone(),
_phantom: Default::default(),
}
}
}
use crate::{Scheduled as ScheduledV3, Scheduled as ScheduledV2};
pub type ScheduledV2Of<T> = ScheduledV2<
Vec<u8>,
<T as Config>::RuntimeCall,
BlockNumberFor<T>,
<T as Config>::PalletsOrigin,
<T as frame_system::Config>::AccountId,
>;
pub type ScheduledV3Of<T> = ScheduledV3<
Vec<u8>,
CallOrHashOf<T>,
BlockNumberFor<T>,
<T as Config>::PalletsOrigin,
<T as frame_system::Config>::AccountId,
>;
pub type ScheduledOf<T> = Scheduled<
TaskName,
BoundedCallOf<T>,
BlockNumberFor<T>,
<T as Config>::PalletsOrigin,
<T as frame_system::Config>::AccountId,
>;
pub(crate) trait MarginalWeightInfo: WeightInfo {
fn service_task(maybe_lookup_len: Option<usize>, named: bool, periodic: bool) -> Weight {
let base = Self::service_task_base();
let mut total = match maybe_lookup_len {
None => base,
Some(l) => Self::service_task_fetched(l as u32),
};
if named {
total.saturating_accrue(Self::service_task_named().saturating_sub(base));
}
if periodic {
total.saturating_accrue(Self::service_task_periodic().saturating_sub(base));
}
total
}
}
impl<T: WeightInfo> MarginalWeightInfo for T {}
#[frame_support::pallet]
pub mod pallet {
use super::*;
use frame_support::{dispatch::PostDispatchInfo, pallet_prelude::*};
use frame_system::pallet_prelude::*;
const STORAGE_VERSION: StorageVersion = StorageVersion::new(4);
#[pallet::pallet]
#[pallet::storage_version(STORAGE_VERSION)]
pub struct Pallet<T>(_);
#[pallet::config]
pub trait Config: frame_system::Config {
type RuntimeEvent: From<Event<Self>> + IsType<<Self as frame_system::Config>::RuntimeEvent>;
type RuntimeOrigin: OriginTrait<PalletsOrigin = Self::PalletsOrigin>
+ From<Self::PalletsOrigin>
+ IsType<<Self as system::Config>::RuntimeOrigin>;
type PalletsOrigin: From<system::RawOrigin<Self::AccountId>>
+ CallerTrait<Self::AccountId>
+ MaxEncodedLen;
type RuntimeCall: Parameter
+ Dispatchable<
RuntimeOrigin = <Self as Config>::RuntimeOrigin,
PostInfo = PostDispatchInfo,
> + GetDispatchInfo
+ From<system::Call<Self>>;
#[pallet::constant]
type MaximumWeight: Get<Weight>;
type ScheduleOrigin: EnsureOrigin<<Self as system::Config>::RuntimeOrigin>;
type OriginPrivilegeCmp: PrivilegeCmp<Self::PalletsOrigin>;
#[pallet::constant]
type MaxScheduledPerBlock: Get<u32>;
type WeightInfo: WeightInfo;
type Preimages: QueryPreimage<H = Self::Hashing> + StorePreimage;
}
#[pallet::storage]
pub type IncompleteSince<T: Config> = StorageValue<_, BlockNumberFor<T>>;
#[pallet::storage]
pub type Agenda<T: Config> = StorageMap<
_,
Twox64Concat,
BlockNumberFor<T>,
BoundedVec<Option<ScheduledOf<T>>, T::MaxScheduledPerBlock>,
ValueQuery,
>;
#[pallet::storage]
pub type Retries<T: Config> = StorageMap<
_,
Blake2_128Concat,
TaskAddress<BlockNumberFor<T>>,
RetryConfig<BlockNumberFor<T>>,
OptionQuery,
>;
#[pallet::storage]
pub(crate) type Lookup<T: Config> =
StorageMap<_, Twox64Concat, TaskName, TaskAddress<BlockNumberFor<T>>>;
#[pallet::event]
#[pallet::generate_deposit(pub(super) fn deposit_event)]
pub enum Event<T: Config> {
Scheduled { when: BlockNumberFor<T>, index: u32 },
Canceled { when: BlockNumberFor<T>, index: u32 },
Dispatched {
task: TaskAddress<BlockNumberFor<T>>,
id: Option<TaskName>,
result: DispatchResult,
},
RetrySet {
task: TaskAddress<BlockNumberFor<T>>,
id: Option<TaskName>,
period: BlockNumberFor<T>,
retries: u8,
},
RetryCancelled { task: TaskAddress<BlockNumberFor<T>>, id: Option<TaskName> },
CallUnavailable { task: TaskAddress<BlockNumberFor<T>>, id: Option<TaskName> },
PeriodicFailed { task: TaskAddress<BlockNumberFor<T>>, id: Option<TaskName> },
RetryFailed { task: TaskAddress<BlockNumberFor<T>>, id: Option<TaskName> },
PermanentlyOverweight { task: TaskAddress<BlockNumberFor<T>>, id: Option<TaskName> },
}
#[pallet::error]
pub enum Error<T> {
FailedToSchedule,
NotFound,
TargetBlockNumberInPast,
RescheduleNoChange,
Named,
}
#[pallet::hooks]
impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
fn on_initialize(now: BlockNumberFor<T>) -> Weight {
let mut weight_counter = WeightMeter::with_limit(T::MaximumWeight::get());
Self::service_agendas(&mut weight_counter, now, u32::max_value());
weight_counter.consumed()
}
}
#[pallet::call]
impl<T: Config> Pallet<T> {
#[pallet::call_index(0)]
#[pallet::weight(<T as Config>::WeightInfo::schedule(T::MaxScheduledPerBlock::get()))]
pub fn schedule(
origin: OriginFor<T>,
when: BlockNumberFor<T>,
maybe_periodic: Option<schedule::Period<BlockNumberFor<T>>>,
priority: schedule::Priority,
call: Box<<T as Config>::RuntimeCall>,
) -> DispatchResult {
T::ScheduleOrigin::ensure_origin(origin.clone())?;
let origin = <T as Config>::RuntimeOrigin::from(origin);
Self::do_schedule(
DispatchTime::At(when),
maybe_periodic,
priority,
origin.caller().clone(),
T::Preimages::bound(*call)?,
)?;
Ok(())
}
#[pallet::call_index(1)]
#[pallet::weight(<T as Config>::WeightInfo::cancel(T::MaxScheduledPerBlock::get()))]
pub fn cancel(origin: OriginFor<T>, when: BlockNumberFor<T>, index: u32) -> DispatchResult {
T::ScheduleOrigin::ensure_origin(origin.clone())?;
let origin = <T as Config>::RuntimeOrigin::from(origin);
Self::do_cancel(Some(origin.caller().clone()), (when, index))?;
Ok(())
}
#[pallet::call_index(2)]
#[pallet::weight(<T as Config>::WeightInfo::schedule_named(T::MaxScheduledPerBlock::get()))]
pub fn schedule_named(
origin: OriginFor<T>,
id: TaskName,
when: BlockNumberFor<T>,
maybe_periodic: Option<schedule::Period<BlockNumberFor<T>>>,
priority: schedule::Priority,
call: Box<<T as Config>::RuntimeCall>,
) -> DispatchResult {
T::ScheduleOrigin::ensure_origin(origin.clone())?;
let origin = <T as Config>::RuntimeOrigin::from(origin);
Self::do_schedule_named(
id,
DispatchTime::At(when),
maybe_periodic,
priority,
origin.caller().clone(),
T::Preimages::bound(*call)?,
)?;
Ok(())
}
#[pallet::call_index(3)]
#[pallet::weight(<T as Config>::WeightInfo::cancel_named(T::MaxScheduledPerBlock::get()))]
pub fn cancel_named(origin: OriginFor<T>, id: TaskName) -> DispatchResult {
T::ScheduleOrigin::ensure_origin(origin.clone())?;
let origin = <T as Config>::RuntimeOrigin::from(origin);
Self::do_cancel_named(Some(origin.caller().clone()), id)?;
Ok(())
}
#[pallet::call_index(4)]
#[pallet::weight(<T as Config>::WeightInfo::schedule(T::MaxScheduledPerBlock::get()))]
pub fn schedule_after(
origin: OriginFor<T>,
after: BlockNumberFor<T>,
maybe_periodic: Option<schedule::Period<BlockNumberFor<T>>>,
priority: schedule::Priority,
call: Box<<T as Config>::RuntimeCall>,
) -> DispatchResult {
T::ScheduleOrigin::ensure_origin(origin.clone())?;
let origin = <T as Config>::RuntimeOrigin::from(origin);
Self::do_schedule(
DispatchTime::After(after),
maybe_periodic,
priority,
origin.caller().clone(),
T::Preimages::bound(*call)?,
)?;
Ok(())
}
#[pallet::call_index(5)]
#[pallet::weight(<T as Config>::WeightInfo::schedule_named(T::MaxScheduledPerBlock::get()))]
pub fn schedule_named_after(
origin: OriginFor<T>,
id: TaskName,
after: BlockNumberFor<T>,
maybe_periodic: Option<schedule::Period<BlockNumberFor<T>>>,
priority: schedule::Priority,
call: Box<<T as Config>::RuntimeCall>,
) -> DispatchResult {
T::ScheduleOrigin::ensure_origin(origin.clone())?;
let origin = <T as Config>::RuntimeOrigin::from(origin);
Self::do_schedule_named(
id,
DispatchTime::After(after),
maybe_periodic,
priority,
origin.caller().clone(),
T::Preimages::bound(*call)?,
)?;
Ok(())
}
#[pallet::call_index(6)]
#[pallet::weight(<T as Config>::WeightInfo::set_retry())]
pub fn set_retry(
origin: OriginFor<T>,
task: TaskAddress<BlockNumberFor<T>>,
retries: u8,
period: BlockNumberFor<T>,
) -> DispatchResult {
T::ScheduleOrigin::ensure_origin(origin.clone())?;
let origin = <T as Config>::RuntimeOrigin::from(origin);
let (when, index) = task;
let agenda = Agenda::<T>::get(when);
let scheduled = agenda
.get(index as usize)
.and_then(Option::as_ref)
.ok_or(Error::<T>::NotFound)?;
Self::ensure_privilege(origin.caller(), &scheduled.origin)?;
Retries::<T>::insert(
(when, index),
RetryConfig { total_retries: retries, remaining: retries, period },
);
Self::deposit_event(Event::RetrySet { task, id: None, period, retries });
Ok(())
}
#[pallet::call_index(7)]
#[pallet::weight(<T as Config>::WeightInfo::set_retry_named())]
pub fn set_retry_named(
origin: OriginFor<T>,
id: TaskName,
retries: u8,
period: BlockNumberFor<T>,
) -> DispatchResult {
T::ScheduleOrigin::ensure_origin(origin.clone())?;
let origin = <T as Config>::RuntimeOrigin::from(origin);
let (when, agenda_index) = Lookup::<T>::get(&id).ok_or(Error::<T>::NotFound)?;
let agenda = Agenda::<T>::get(when);
let scheduled = agenda
.get(agenda_index as usize)
.and_then(Option::as_ref)
.ok_or(Error::<T>::NotFound)?;
Self::ensure_privilege(origin.caller(), &scheduled.origin)?;
Retries::<T>::insert(
(when, agenda_index),
RetryConfig { total_retries: retries, remaining: retries, period },
);
Self::deposit_event(Event::RetrySet {
task: (when, agenda_index),
id: Some(id),
period,
retries,
});
Ok(())
}
#[pallet::call_index(8)]
#[pallet::weight(<T as Config>::WeightInfo::cancel_retry())]
pub fn cancel_retry(
origin: OriginFor<T>,
task: TaskAddress<BlockNumberFor<T>>,
) -> DispatchResult {
T::ScheduleOrigin::ensure_origin(origin.clone())?;
let origin = <T as Config>::RuntimeOrigin::from(origin);
Self::do_cancel_retry(origin.caller(), task)?;
Self::deposit_event(Event::RetryCancelled { task, id: None });
Ok(())
}
#[pallet::call_index(9)]
#[pallet::weight(<T as Config>::WeightInfo::cancel_retry_named())]
pub fn cancel_retry_named(origin: OriginFor<T>, id: TaskName) -> DispatchResult {
T::ScheduleOrigin::ensure_origin(origin.clone())?;
let origin = <T as Config>::RuntimeOrigin::from(origin);
let task = Lookup::<T>::get(&id).ok_or(Error::<T>::NotFound)?;
Self::do_cancel_retry(origin.caller(), task)?;
Self::deposit_event(Event::RetryCancelled { task, id: Some(id) });
Ok(())
}
}
}
impl<T: Config> Pallet<T> {
pub fn migrate_v1_to_v4() -> Weight {
use migration::v1 as old;
let mut weight = T::DbWeight::get().reads_writes(1, 1);
let keys = old::Agenda::<T>::iter_keys().collect::<Vec<_>>();
for key in keys {
weight.saturating_accrue(T::DbWeight::get().reads(1));
if let Err(_) = old::Agenda::<T>::try_get(&key) {
weight.saturating_accrue(T::DbWeight::get().writes(1));
old::Agenda::<T>::remove(&key);
log::warn!("Deleted undecodable agenda");
}
}
Agenda::<T>::translate::<
Vec<Option<ScheduledV1<<T as Config>::RuntimeCall, BlockNumberFor<T>>>>,
_,
>(|_, agenda| {
Some(BoundedVec::truncate_from(
agenda
.into_iter()
.map(|schedule| {
weight.saturating_accrue(T::DbWeight::get().reads_writes(1, 1));
schedule.and_then(|schedule| {
if let Some(id) = schedule.maybe_id.as_ref() {
let name = blake2_256(id);
if let Some(item) = old::Lookup::<T>::take(id) {
Lookup::<T>::insert(name, item);
}
weight.saturating_accrue(T::DbWeight::get().reads_writes(2, 2));
}
let call = T::Preimages::bound(schedule.call).ok()?;
if call.lookup_needed() {
weight.saturating_accrue(T::DbWeight::get().reads_writes(0, 1));
}
Some(Scheduled {
maybe_id: schedule.maybe_id.map(|x| blake2_256(&x[..])),
priority: schedule.priority,
call,
maybe_periodic: schedule.maybe_periodic,
origin: system::RawOrigin::Root.into(),
_phantom: Default::default(),
})
})
})
.collect::<Vec<_>>(),
))
});
#[allow(deprecated)]
frame_support::storage::migration::remove_storage_prefix(
Self::name().as_bytes(),
b"StorageVersion",
&[],
);
StorageVersion::new(4).put::<Self>();
weight + T::DbWeight::get().writes(2)
}
pub fn migrate_v2_to_v4() -> Weight {
use migration::v2 as old;
let mut weight = T::DbWeight::get().reads_writes(1, 1);
let keys = old::Agenda::<T>::iter_keys().collect::<Vec<_>>();
for key in keys {
weight.saturating_accrue(T::DbWeight::get().reads(1));
if let Err(_) = old::Agenda::<T>::try_get(&key) {
weight.saturating_accrue(T::DbWeight::get().writes(1));
old::Agenda::<T>::remove(&key);
log::warn!("Deleted undecodable agenda");
}
}
Agenda::<T>::translate::<Vec<Option<ScheduledV2Of<T>>>, _>(|_, agenda| {
Some(BoundedVec::truncate_from(
agenda
.into_iter()
.map(|schedule| {
weight.saturating_accrue(T::DbWeight::get().reads_writes(1, 1));
schedule.and_then(|schedule| {
if let Some(id) = schedule.maybe_id.as_ref() {
let name = blake2_256(id);
if let Some(item) = old::Lookup::<T>::take(id) {
Lookup::<T>::insert(name, item);
}
weight.saturating_accrue(T::DbWeight::get().reads_writes(2, 2));
}
let call = T::Preimages::bound(schedule.call).ok()?;
if call.lookup_needed() {
weight.saturating_accrue(T::DbWeight::get().reads_writes(0, 1));
}
Some(Scheduled {
maybe_id: schedule.maybe_id.map(|x| blake2_256(&x[..])),
priority: schedule.priority,
call,
maybe_periodic: schedule.maybe_periodic,
origin: schedule.origin,
_phantom: Default::default(),
})
})
})
.collect::<Vec<_>>(),
))
});
#[allow(deprecated)]
frame_support::storage::migration::remove_storage_prefix(
Self::name().as_bytes(),
b"StorageVersion",
&[],
);
StorageVersion::new(4).put::<Self>();
weight + T::DbWeight::get().writes(2)
}
#[allow(deprecated)]
pub fn migrate_v3_to_v4() -> Weight {
use migration::v3 as old;
let mut weight = T::DbWeight::get().reads_writes(2, 1);
let blocks = old::Agenda::<T>::iter_keys().collect::<Vec<_>>();
for block in blocks {
weight.saturating_accrue(T::DbWeight::get().reads(1));
if let Err(_) = old::Agenda::<T>::try_get(&block) {
weight.saturating_accrue(T::DbWeight::get().writes(1));
old::Agenda::<T>::remove(&block);
log::warn!("Deleted undecodable agenda of block: {:?}", block);
}
}
Agenda::<T>::translate::<Vec<Option<ScheduledV3Of<T>>>, _>(|block, agenda| {
log::info!("Migrating agenda of block: {:?}", &block);
Some(BoundedVec::truncate_from(
agenda
.into_iter()
.map(|schedule| {
weight.saturating_accrue(T::DbWeight::get().reads_writes(1, 1));
schedule
.and_then(|schedule| {
if let Some(id) = schedule.maybe_id.as_ref() {
let name = blake2_256(id);
if let Some(item) = old::Lookup::<T>::take(id) {
Lookup::<T>::insert(name, item);
log::info!("Migrated name for id: {:?}", id);
} else {
log::error!("No name in Lookup for id: {:?}", &id);
}
weight.saturating_accrue(T::DbWeight::get().reads_writes(2, 2));
} else {
log::info!("Schedule is unnamed");
}
let call = match schedule.call {
MaybeHashed::Hash(h) => {
let bounded = Bounded::from_legacy_hash(h);
if let Err(err) = T::Preimages::peek::<
<T as Config>::RuntimeCall,
>(&bounded)
{
log::error!(
"Dropping undecodable call {:?}: {:?}",
&h,
&err
);
return None
}
weight.saturating_accrue(T::DbWeight::get().reads(1));
log::info!("Migrated call by hash, hash: {:?}", h);
bounded
},
MaybeHashed::Value(v) => {
let call = T::Preimages::bound(v)
.map_err(|e| {
log::error!("Could not bound Call: {:?}", e)
})
.ok()?;
if call.lookup_needed() {
weight.saturating_accrue(
T::DbWeight::get().reads_writes(0, 1),
);
}
log::info!(
"Migrated call by value, hash: {:?}",
call.hash()
);
call
},
};
Some(Scheduled {
maybe_id: schedule.maybe_id.map(|x| blake2_256(&x[..])),
priority: schedule.priority,
call,
maybe_periodic: schedule.maybe_periodic,
origin: schedule.origin,
_phantom: Default::default(),
})
})
.or_else(|| {
log::info!("Schedule in agenda for block {:?} is empty - nothing to do here.", &block);
None
})
})
.collect::<Vec<_>>(),
))
});
#[allow(deprecated)]
frame_support::storage::migration::remove_storage_prefix(
Self::name().as_bytes(),
b"StorageVersion",
&[],
);
StorageVersion::new(4).put::<Self>();
weight + T::DbWeight::get().writes(2)
}
}
impl<T: Config> Pallet<T> {
pub fn migrate_origin<OldOrigin: Into<T::PalletsOrigin> + codec::Decode>() {
Agenda::<T>::translate::<
Vec<
Option<
Scheduled<
TaskName,
BoundedCallOf<T>,
BlockNumberFor<T>,
OldOrigin,
T::AccountId,
>,
>,
>,
_,
>(|_, agenda| {
Some(BoundedVec::truncate_from(
agenda
.into_iter()
.map(|schedule| {
schedule.map(|schedule| Scheduled {
maybe_id: schedule.maybe_id,
priority: schedule.priority,
call: schedule.call,
maybe_periodic: schedule.maybe_periodic,
origin: schedule.origin.into(),
_phantom: Default::default(),
})
})
.collect::<Vec<_>>(),
))
});
}
fn resolve_time(
when: DispatchTime<BlockNumberFor<T>>,
) -> Result<BlockNumberFor<T>, DispatchError> {
let now = frame_system::Pallet::<T>::block_number();
let when = match when {
DispatchTime::At(x) => x,
DispatchTime::After(x) => now.saturating_add(x).saturating_add(One::one()),
};
if when <= now {
return Err(Error::<T>::TargetBlockNumberInPast.into())
}
Ok(when)
}
fn place_task(
when: BlockNumberFor<T>,
what: ScheduledOf<T>,
) -> Result<TaskAddress<BlockNumberFor<T>>, (DispatchError, ScheduledOf<T>)> {
let maybe_name = what.maybe_id;
let index = Self::push_to_agenda(when, what)?;
let address = (when, index);
if let Some(name) = maybe_name {
Lookup::<T>::insert(name, address)
}
Self::deposit_event(Event::Scheduled { when: address.0, index: address.1 });
Ok(address)
}
fn push_to_agenda(
when: BlockNumberFor<T>,
what: ScheduledOf<T>,
) -> Result<u32, (DispatchError, ScheduledOf<T>)> {
let mut agenda = Agenda::<T>::get(when);
let index = if (agenda.len() as u32) < T::MaxScheduledPerBlock::get() {
let _ = agenda.try_push(Some(what));
agenda.len() as u32 - 1
} else {
if let Some(hole_index) = agenda.iter().position(|i| i.is_none()) {
agenda[hole_index] = Some(what);
hole_index as u32
} else {
return Err((DispatchError::Exhausted, what))
}
};
Agenda::<T>::insert(when, agenda);
Ok(index)
}
fn cleanup_agenda(when: BlockNumberFor<T>) {
let mut agenda = Agenda::<T>::get(when);
match agenda.iter().rposition(|i| i.is_some()) {
Some(i) if agenda.len() > i + 1 => {
agenda.truncate(i + 1);
Agenda::<T>::insert(when, agenda);
},
Some(_) => {},
None => {
Agenda::<T>::remove(when);
},
}
}
fn do_schedule(
when: DispatchTime<BlockNumberFor<T>>,
maybe_periodic: Option<schedule::Period<BlockNumberFor<T>>>,
priority: schedule::Priority,
origin: T::PalletsOrigin,
call: BoundedCallOf<T>,
) -> Result<TaskAddress<BlockNumberFor<T>>, DispatchError> {
let when = Self::resolve_time(when)?;
let lookup_hash = call.lookup_hash();
let maybe_periodic = maybe_periodic
.filter(|p| p.1 > 1 && !p.0.is_zero())
.map(|(p, c)| (p, c - 1));
let task = Scheduled {
maybe_id: None,
priority,
call,
maybe_periodic,
origin,
_phantom: PhantomData,
};
let res = Self::place_task(when, task).map_err(|x| x.0)?;
if let Some(hash) = lookup_hash {
T::Preimages::request(&hash);
}
Ok(res)
}
fn do_cancel(
origin: Option<T::PalletsOrigin>,
(when, index): TaskAddress<BlockNumberFor<T>>,
) -> Result<(), DispatchError> {
let scheduled = Agenda::<T>::try_mutate(when, |agenda| {
agenda.get_mut(index as usize).map_or(
Ok(None),
|s| -> Result<Option<Scheduled<_, _, _, _, _>>, DispatchError> {
if let (Some(ref o), Some(ref s)) = (origin, s.borrow()) {
Self::ensure_privilege(o, &s.origin)?;
};
Ok(s.take())
},
)
})?;
if let Some(s) = scheduled {
T::Preimages::drop(&s.call);
if let Some(id) = s.maybe_id {
Lookup::<T>::remove(id);
}
Retries::<T>::remove((when, index));
Self::cleanup_agenda(when);
Self::deposit_event(Event::Canceled { when, index });
Ok(())
} else {
return Err(Error::<T>::NotFound.into())
}
}
fn do_reschedule(
(when, index): TaskAddress<BlockNumberFor<T>>,
new_time: DispatchTime<BlockNumberFor<T>>,
) -> Result<TaskAddress<BlockNumberFor<T>>, DispatchError> {
let new_time = Self::resolve_time(new_time)?;
if new_time == when {
return Err(Error::<T>::RescheduleNoChange.into())
}
let task = Agenda::<T>::try_mutate(when, |agenda| {
let task = agenda.get_mut(index as usize).ok_or(Error::<T>::NotFound)?;
ensure!(!matches!(task, Some(Scheduled { maybe_id: Some(_), .. })), Error::<T>::Named);
task.take().ok_or(Error::<T>::NotFound)
})?;
Self::cleanup_agenda(when);
Self::deposit_event(Event::Canceled { when, index });
Self::place_task(new_time, task).map_err(|x| x.0)
}
fn do_schedule_named(
id: TaskName,
when: DispatchTime<BlockNumberFor<T>>,
maybe_periodic: Option<schedule::Period<BlockNumberFor<T>>>,
priority: schedule::Priority,
origin: T::PalletsOrigin,
call: BoundedCallOf<T>,
) -> Result<TaskAddress<BlockNumberFor<T>>, DispatchError> {
if Lookup::<T>::contains_key(&id) {
return Err(Error::<T>::FailedToSchedule.into())
}
let when = Self::resolve_time(when)?;
let lookup_hash = call.lookup_hash();
let maybe_periodic = maybe_periodic
.filter(|p| p.1 > 1 && !p.0.is_zero())
.map(|(p, c)| (p, c - 1));
let task = Scheduled {
maybe_id: Some(id),
priority,
call,
maybe_periodic,
origin,
_phantom: Default::default(),
};
let res = Self::place_task(when, task).map_err(|x| x.0)?;
if let Some(hash) = lookup_hash {
T::Preimages::request(&hash);
}
Ok(res)
}
fn do_cancel_named(origin: Option<T::PalletsOrigin>, id: TaskName) -> DispatchResult {
Lookup::<T>::try_mutate_exists(id, |lookup| -> DispatchResult {
if let Some((when, index)) = lookup.take() {
let i = index as usize;
Agenda::<T>::try_mutate(when, |agenda| -> DispatchResult {
if let Some(s) = agenda.get_mut(i) {
if let (Some(ref o), Some(ref s)) = (origin, s.borrow()) {
Self::ensure_privilege(o, &s.origin)?;
Retries::<T>::remove((when, index));
T::Preimages::drop(&s.call);
}
*s = None;
}
Ok(())
})?;
Self::cleanup_agenda(when);
Self::deposit_event(Event::Canceled { when, index });
Ok(())
} else {
return Err(Error::<T>::NotFound.into())
}
})
}
fn do_reschedule_named(
id: TaskName,
new_time: DispatchTime<BlockNumberFor<T>>,
) -> Result<TaskAddress<BlockNumberFor<T>>, DispatchError> {
let new_time = Self::resolve_time(new_time)?;
let lookup = Lookup::<T>::get(id);
let (when, index) = lookup.ok_or(Error::<T>::NotFound)?;
if new_time == when {
return Err(Error::<T>::RescheduleNoChange.into())
}
let task = Agenda::<T>::try_mutate(when, |agenda| {
let task = agenda.get_mut(index as usize).ok_or(Error::<T>::NotFound)?;
task.take().ok_or(Error::<T>::NotFound)
})?;
Self::cleanup_agenda(when);
Self::deposit_event(Event::Canceled { when, index });
Self::place_task(new_time, task).map_err(|x| x.0)
}
fn do_cancel_retry(
origin: &T::PalletsOrigin,
(when, index): TaskAddress<BlockNumberFor<T>>,
) -> Result<(), DispatchError> {
let agenda = Agenda::<T>::get(when);
let scheduled = agenda
.get(index as usize)
.and_then(Option::as_ref)
.ok_or(Error::<T>::NotFound)?;
Self::ensure_privilege(origin, &scheduled.origin)?;
Retries::<T>::remove((when, index));
Ok(())
}
}
enum ServiceTaskError {
Unavailable,
Overweight,
}
use ServiceTaskError::*;
impl<T: Config> Pallet<T> {
fn service_agendas(weight: &mut WeightMeter, now: BlockNumberFor<T>, max: u32) {
if weight.try_consume(T::WeightInfo::service_agendas_base()).is_err() {
return
}
let mut incomplete_since = now + One::one();
let mut when = IncompleteSince::<T>::take().unwrap_or(now);
let mut executed = 0;
let max_items = T::MaxScheduledPerBlock::get();
let mut count_down = max;
let service_agenda_base_weight = T::WeightInfo::service_agenda_base(max_items);
while count_down > 0 && when <= now && weight.can_consume(service_agenda_base_weight) {
if !Self::service_agenda(weight, &mut executed, now, when, u32::max_value()) {
incomplete_since = incomplete_since.min(when);
}
when.saturating_inc();
count_down.saturating_dec();
}
incomplete_since = incomplete_since.min(when);
if incomplete_since <= now {
IncompleteSince::<T>::put(incomplete_since);
}
}
fn service_agenda(
weight: &mut WeightMeter,
executed: &mut u32,
now: BlockNumberFor<T>,
when: BlockNumberFor<T>,
max: u32,
) -> bool {
let mut agenda = Agenda::<T>::get(when);
let mut ordered = agenda
.iter()
.enumerate()
.filter_map(|(index, maybe_item)| {
maybe_item.as_ref().map(|item| (index as u32, item.priority))
})
.collect::<Vec<_>>();
ordered.sort_by_key(|k| k.1);
let within_limit = weight
.try_consume(T::WeightInfo::service_agenda_base(ordered.len() as u32))
.is_ok();
debug_assert!(within_limit, "weight limit should have been checked in advance");
let mut postponed = (ordered.len() as u32).saturating_sub(max);
let mut dropped = 0;
for (agenda_index, _) in ordered.into_iter().take(max as usize) {
let task = match agenda[agenda_index as usize].take() {
None => continue,
Some(t) => t,
};
let base_weight = T::WeightInfo::service_task(
task.call.lookup_len().map(|x| x as usize),
task.maybe_id.is_some(),
task.maybe_periodic.is_some(),
);
if !weight.can_consume(base_weight) {
postponed += 1;
break
}
let result = Self::service_task(weight, now, when, agenda_index, *executed == 0, task);
agenda[agenda_index as usize] = match result {
Err((Unavailable, slot)) => {
dropped += 1;
slot
},
Err((Overweight, slot)) => {
postponed += 1;
slot
},
Ok(()) => {
*executed += 1;
None
},
};
}
if postponed > 0 || dropped > 0 {
Agenda::<T>::insert(when, agenda);
} else {
Agenda::<T>::remove(when);
}
postponed == 0
}
fn service_task(
weight: &mut WeightMeter,
now: BlockNumberFor<T>,
when: BlockNumberFor<T>,
agenda_index: u32,
is_first: bool,
mut task: ScheduledOf<T>,
) -> Result<(), (ServiceTaskError, Option<ScheduledOf<T>>)> {
if let Some(ref id) = task.maybe_id {
Lookup::<T>::remove(id);
}
let (call, lookup_len) = match T::Preimages::peek(&task.call) {
Ok(c) => c,
Err(_) => {
Self::deposit_event(Event::CallUnavailable {
task: (when, agenda_index),
id: task.maybe_id,
});
return Err((Unavailable, Some(task)))
},
};
let _ = weight.try_consume(T::WeightInfo::service_task(
lookup_len.map(|x| x as usize),
task.maybe_id.is_some(),
task.maybe_periodic.is_some(),
));
match Self::execute_dispatch(weight, task.origin.clone(), call) {
Err(()) if is_first => {
T::Preimages::drop(&task.call);
Self::deposit_event(Event::PermanentlyOverweight {
task: (when, agenda_index),
id: task.maybe_id,
});
Err((Unavailable, Some(task)))
},
Err(()) => Err((Overweight, Some(task))),
Ok(result) => {
let failed = result.is_err();
let maybe_retry_config = Retries::<T>::take((when, agenda_index));
Self::deposit_event(Event::Dispatched {
task: (when, agenda_index),
id: task.maybe_id,
result,
});
match maybe_retry_config {
Some(retry_config) if failed => {
Self::schedule_retry(weight, now, when, agenda_index, &task, retry_config);
},
_ => {},
}
if let &Some((period, count)) = &task.maybe_periodic {
if count > 1 {
task.maybe_periodic = Some((period, count - 1));
} else {
task.maybe_periodic = None;
}
let wake = now.saturating_add(period);
match Self::place_task(wake, task) {
Ok(new_address) =>
if let Some(retry_config) = maybe_retry_config {
Retries::<T>::insert(new_address, retry_config);
},
Err((_, task)) => {
T::Preimages::drop(&task.call);
Self::deposit_event(Event::PeriodicFailed {
task: (when, agenda_index),
id: task.maybe_id,
});
},
}
} else {
T::Preimages::drop(&task.call);
}
Ok(())
},
}
}
fn execute_dispatch(
weight: &mut WeightMeter,
origin: T::PalletsOrigin,
call: <T as Config>::RuntimeCall,
) -> Result<DispatchResult, ()> {
let base_weight = match origin.as_system_ref() {
Some(&RawOrigin::Signed(_)) => T::WeightInfo::execute_dispatch_signed(),
_ => T::WeightInfo::execute_dispatch_unsigned(),
};
let call_weight = call.get_dispatch_info().weight;
let max_weight = base_weight.saturating_add(call_weight);
if !weight.can_consume(max_weight) {
return Err(())
}
let dispatch_origin = origin.into();
let (maybe_actual_call_weight, result) = match call.dispatch(dispatch_origin) {
Ok(post_info) => (post_info.actual_weight, Ok(())),
Err(error_and_info) =>
(error_and_info.post_info.actual_weight, Err(error_and_info.error)),
};
let call_weight = maybe_actual_call_weight.unwrap_or(call_weight);
let _ = weight.try_consume(base_weight);
let _ = weight.try_consume(call_weight);
Ok(result)
}
fn schedule_retry(
weight: &mut WeightMeter,
now: BlockNumberFor<T>,
when: BlockNumberFor<T>,
agenda_index: u32,
task: &ScheduledOf<T>,
retry_config: RetryConfig<BlockNumberFor<T>>,
) {
if weight
.try_consume(T::WeightInfo::schedule_retry(T::MaxScheduledPerBlock::get()))
.is_err()
{
Self::deposit_event(Event::RetryFailed {
task: (when, agenda_index),
id: task.maybe_id,
});
return;
}
let RetryConfig { total_retries, mut remaining, period } = retry_config;
remaining = match remaining.checked_sub(1) {
Some(n) => n,
None => return,
};
let wake = now.saturating_add(period);
match Self::place_task(wake, task.as_retry()) {
Ok(address) => {
Retries::<T>::insert(address, RetryConfig { total_retries, remaining, period });
},
Err((_, task)) => {
T::Preimages::drop(&task.call);
Self::deposit_event(Event::RetryFailed {
task: (when, agenda_index),
id: task.maybe_id,
});
},
}
}
fn ensure_privilege(
left: &<T as Config>::PalletsOrigin,
right: &<T as Config>::PalletsOrigin,
) -> Result<(), DispatchError> {
if matches!(T::OriginPrivilegeCmp::cmp_privilege(left, right), Some(Ordering::Less) | None)
{
return Err(BadOrigin.into());
}
Ok(())
}
}
impl<T: Config> schedule::v2::Anon<BlockNumberFor<T>, <T as Config>::RuntimeCall, T::PalletsOrigin>
for Pallet<T>
{
type Address = TaskAddress<BlockNumberFor<T>>;
type Hash = T::Hash;
fn schedule(
when: DispatchTime<BlockNumberFor<T>>,
maybe_periodic: Option<schedule::Period<BlockNumberFor<T>>>,
priority: schedule::Priority,
origin: T::PalletsOrigin,
call: CallOrHashOf<T>,
) -> Result<Self::Address, DispatchError> {
let call = call.as_value().ok_or(DispatchError::CannotLookup)?;
let call = T::Preimages::bound(call)?.transmute();
Self::do_schedule(when, maybe_periodic, priority, origin, call)
}
fn cancel((when, index): Self::Address) -> Result<(), ()> {
Self::do_cancel(None, (when, index)).map_err(|_| ())
}
fn reschedule(
address: Self::Address,
when: DispatchTime<BlockNumberFor<T>>,
) -> Result<Self::Address, DispatchError> {
Self::do_reschedule(address, when)
}
fn next_dispatch_time((when, index): Self::Address) -> Result<BlockNumberFor<T>, ()> {
Agenda::<T>::get(when).get(index as usize).ok_or(()).map(|_| when)
}
}
impl<T: Config> schedule::v2::Named<BlockNumberFor<T>, <T as Config>::RuntimeCall, T::PalletsOrigin>
for Pallet<T>
{
type Address = TaskAddress<BlockNumberFor<T>>;
type Hash = T::Hash;
fn schedule_named(
id: Vec<u8>,
when: DispatchTime<BlockNumberFor<T>>,
maybe_periodic: Option<schedule::Period<BlockNumberFor<T>>>,
priority: schedule::Priority,
origin: T::PalletsOrigin,
call: CallOrHashOf<T>,
) -> Result<Self::Address, ()> {
let call = call.as_value().ok_or(())?;
let call = T::Preimages::bound(call).map_err(|_| ())?.transmute();
let name = blake2_256(&id[..]);
Self::do_schedule_named(name, when, maybe_periodic, priority, origin, call).map_err(|_| ())
}
fn cancel_named(id: Vec<u8>) -> Result<(), ()> {
let name = blake2_256(&id[..]);
Self::do_cancel_named(None, name).map_err(|_| ())
}
fn reschedule_named(
id: Vec<u8>,
when: DispatchTime<BlockNumberFor<T>>,
) -> Result<Self::Address, DispatchError> {
let name = blake2_256(&id[..]);
Self::do_reschedule_named(name, when)
}
fn next_dispatch_time(id: Vec<u8>) -> Result<BlockNumberFor<T>, ()> {
let name = blake2_256(&id[..]);
Lookup::<T>::get(name)
.and_then(|(when, index)| Agenda::<T>::get(when).get(index as usize).map(|_| when))
.ok_or(())
}
}
impl<T: Config> schedule::v3::Anon<BlockNumberFor<T>, <T as Config>::RuntimeCall, T::PalletsOrigin>
for Pallet<T>
{
type Address = TaskAddress<BlockNumberFor<T>>;
type Hasher = T::Hashing;
fn schedule(
when: DispatchTime<BlockNumberFor<T>>,
maybe_periodic: Option<schedule::Period<BlockNumberFor<T>>>,
priority: schedule::Priority,
origin: T::PalletsOrigin,
call: BoundedCallOf<T>,
) -> Result<Self::Address, DispatchError> {
Self::do_schedule(when, maybe_periodic, priority, origin, call)
}
fn cancel((when, index): Self::Address) -> Result<(), DispatchError> {
Self::do_cancel(None, (when, index)).map_err(map_err_to_v3_err::<T>)
}
fn reschedule(
address: Self::Address,
when: DispatchTime<BlockNumberFor<T>>,
) -> Result<Self::Address, DispatchError> {
Self::do_reschedule(address, when).map_err(map_err_to_v3_err::<T>)
}
fn next_dispatch_time(
(when, index): Self::Address,
) -> Result<BlockNumberFor<T>, DispatchError> {
Agenda::<T>::get(when)
.get(index as usize)
.ok_or(DispatchError::Unavailable)
.map(|_| when)
}
}
use schedule::v3::TaskName;
impl<T: Config> schedule::v3::Named<BlockNumberFor<T>, <T as Config>::RuntimeCall, T::PalletsOrigin>
for Pallet<T>
{
type Address = TaskAddress<BlockNumberFor<T>>;
type Hasher = T::Hashing;
fn schedule_named(
id: TaskName,
when: DispatchTime<BlockNumberFor<T>>,
maybe_periodic: Option<schedule::Period<BlockNumberFor<T>>>,
priority: schedule::Priority,
origin: T::PalletsOrigin,
call: BoundedCallOf<T>,
) -> Result<Self::Address, DispatchError> {
Self::do_schedule_named(id, when, maybe_periodic, priority, origin, call)
}
fn cancel_named(id: TaskName) -> Result<(), DispatchError> {
Self::do_cancel_named(None, id).map_err(map_err_to_v3_err::<T>)
}
fn reschedule_named(
id: TaskName,
when: DispatchTime<BlockNumberFor<T>>,
) -> Result<Self::Address, DispatchError> {
Self::do_reschedule_named(id, when).map_err(map_err_to_v3_err::<T>)
}
fn next_dispatch_time(id: TaskName) -> Result<BlockNumberFor<T>, DispatchError> {
Lookup::<T>::get(id)
.and_then(|(when, index)| Agenda::<T>::get(when).get(index as usize).map(|_| when))
.ok_or(DispatchError::Unavailable)
}
}
fn map_err_to_v3_err<T: Config>(err: DispatchError) -> DispatchError {
if err == DispatchError::from(Error::<T>::NotFound) {
DispatchError::Unavailable
} else {
err
}
}