diff --git a/cumulus/parachains/runtimes/collectives/collectives-westend/src/lib.rs b/cumulus/parachains/runtimes/collectives/collectives-westend/src/lib.rs
index 5e087832f0e8..5890e73b45b2 100644
--- a/cumulus/parachains/runtimes/collectives/collectives-westend/src/lib.rs
+++ b/cumulus/parachains/runtimes/collectives/collectives-westend/src/lib.rs
@@ -610,11 +610,15 @@ parameter_types! {
 #[cfg(not(feature = "runtime-benchmarks"))]
 parameter_types! {
 	pub const MaxScheduledPerBlock: u32 = 50;
+	pub const MaxScheduledBlocks: u32 = 50;
+	pub const MaxStaleTaskAge: u32 = 10;
 }
 
 #[cfg(feature = "runtime-benchmarks")]
 parameter_types! {
 	pub const MaxScheduledPerBlock: u32 = 200;
+	pub const MaxScheduledBlocks: u32 = 200;
+	pub const MaxStaleTaskAge: u32 = 40;
 }
 
 impl pallet_scheduler::Config for Runtime {
@@ -628,6 +632,9 @@ impl pallet_scheduler::Config for Runtime {
 	type WeightInfo = weights::pallet_scheduler::WeightInfo<Runtime>;
 	type OriginPrivilegeCmp = EqualOrGreatestRootCmp;
 	type Preimages = Preimage;
+	type BlockNumberProvider = frame_system::Pallet<Runtime>;
+	type MaxScheduledBlocks = MaxScheduledBlocks;
+	type MaxStaleTaskAge = MaxStaleTaskAge;
 }
 
 parameter_types! {
diff --git a/polkadot/runtime/rococo/src/lib.rs b/polkadot/runtime/rococo/src/lib.rs
index f165091beda4..0b64a9221ad0 100644
--- a/polkadot/runtime/rococo/src/lib.rs
+++ b/polkadot/runtime/rococo/src/lib.rs
@@ -244,6 +244,8 @@ parameter_types! {
 	pub MaximumSchedulerWeight: Weight = Perbill::from_percent(80) *
 		BlockWeights::get().max_block;
 	pub const MaxScheduledPerBlock: u32 = 50;
+	pub const MaxScheduledBlocks: u32 = 50;
+	pub const MaxStaleTaskAge: u32 = 10;
 	pub const NoPreimagePostponement: Option<u32> = Some(10);
 }
 
@@ -344,6 +346,9 @@ impl pallet_scheduler::Config for Runtime {
 	type WeightInfo = weights::pallet_scheduler::WeightInfo<Runtime>;
 	type OriginPrivilegeCmp = OriginPrivilegeCmp;
 	type Preimages = Preimage;
+	type BlockNumberProvider = frame_system::Pallet<Runtime>;
+	type MaxScheduledBlocks = MaxScheduledBlocks;
+	type MaxStaleTaskAge = MaxStaleTaskAge;
 }
 
 parameter_types! {
diff --git a/polkadot/runtime/westend/src/lib.rs b/polkadot/runtime/westend/src/lib.rs
index 935b62c23388..203d0c958dbd 100644
--- a/polkadot/runtime/westend/src/lib.rs
+++ b/polkadot/runtime/westend/src/lib.rs
@@ -234,6 +234,8 @@ parameter_types! {
 	pub MaximumSchedulerWeight: frame_support::weights::Weight = Perbill::from_percent(80) *
 		BlockWeights::get().max_block;
 	pub const MaxScheduledPerBlock: u32 = 50;
+	pub const MaxScheduledBlocks: u32 = 50;
+	pub const MaxStaleTaskAge: u32 = 10;
 	pub const NoPreimagePostponement: Option<u32> = Some(10);
 }
 
@@ -250,6 +252,9 @@ impl pallet_scheduler::Config for Runtime {
 	type WeightInfo = weights::pallet_scheduler::WeightInfo<Runtime>;
 	type OriginPrivilegeCmp = frame_support::traits::EqualPrivilegeOnly;
 	type Preimages = Preimage;
+	type BlockNumberProvider = frame_system::Pallet<Runtime>;
+	type MaxScheduledBlocks = MaxScheduledBlocks;
+	type MaxStaleTaskAge = MaxStaleTaskAge;
 }
 
 parameter_types! {
diff --git a/prdoc/pr_6362.prdoc b/prdoc/pr_6362.prdoc
new file mode 100644
index 000000000000..57586d1423d4
--- /dev/null
+++ b/prdoc/pr_6362.prdoc
@@ -0,0 +1,23 @@
+title: Update Scheduler to Support Relay Chain Block Number Provider
+
+doc:
+  - audience: Runtime Dev
+    description: |
+      This PR adds the ability for the Scheduler pallet to specify its source of the block number.
+      This is needed for the scheduler pallet to work on a parachain which does not produce blocks
+      on a regular schedule, thus can use the relay chain as a block provider. Because blocks are
+      not produced regularly, we cannot make the assumption that the block number increases 
+      monotonically, and thus have a new logic via a `Queue` to handle multiple blocks with valid 
+      agenda passed between them.
+
+      This change only needs a migration for the `Queue`:
+      1. If the `BlockNumberProvider` continues to use the system pallet's block number
+      2. When a pallet deployed on the relay chain is moved to a parachain, but still uses the 
+      relay chain's block number
+
+      However, we would need migrations if the deployed pallets are upgraded on an existing parachain,
+      and the `BlockNumberProvider` uses the relay chain block number.  
+
+crates:
+  - name: pallet-scheduler
+    bump: major
diff --git a/substrate/bin/node/runtime/src/lib.rs b/substrate/bin/node/runtime/src/lib.rs
index 220929fdfd83..aa7d29870605 100644
--- a/substrate/bin/node/runtime/src/lib.rs
+++ b/substrate/bin/node/runtime/src/lib.rs
@@ -504,6 +504,12 @@ impl pallet_scheduler::Config for Runtime {
 	type WeightInfo = pallet_scheduler::weights::SubstrateWeight<Runtime>;
 	type OriginPrivilegeCmp = EqualPrivilegeOnly;
 	type Preimages = Preimage;
+	type BlockNumberProvider = frame_system::Pallet<Runtime>;
+	#[cfg(feature = "runtime-benchmarks")]
+	type MaxScheduledBlocks = ConstU32<512>;
+	#[cfg(not(feature = "runtime-benchmarks"))]
+	type MaxScheduledBlocks = ConstU32<50>;
+	type MaxStaleTaskAge = ConstU32<10>;
 }
 
 impl pallet_glutton::Config for Runtime {
diff --git a/substrate/frame/democracy/src/tests.rs b/substrate/frame/democracy/src/tests.rs
index 777744800684..7aa8795c4981 100644
--- a/substrate/frame/democracy/src/tests.rs
+++ b/substrate/frame/democracy/src/tests.rs
@@ -106,6 +106,9 @@ impl pallet_scheduler::Config for Test {
 	type WeightInfo = ();
 	type OriginPrivilegeCmp = EqualPrivilegeOnly;
 	type Preimages = ();
+	type BlockNumberProvider = frame_system::Pallet<Test>;
+	type MaxScheduledBlocks = ConstU32<100>;
+	type MaxStaleTaskAge = ConstU64<10>;
 }
 
 #[derive_impl(pallet_balances::config_preludes::TestDefaultConfig)]
diff --git a/substrate/frame/referenda/src/mock.rs b/substrate/frame/referenda/src/mock.rs
index 5d36ce137d46..efc88e057c55 100644
--- a/substrate/frame/referenda/src/mock.rs
+++ b/substrate/frame/referenda/src/mock.rs
@@ -81,6 +81,9 @@ impl pallet_scheduler::Config for Test {
 	type WeightInfo = ();
 	type OriginPrivilegeCmp = EqualPrivilegeOnly;
 	type Preimages = Preimage;
+	type BlockNumberProvider = frame_system::Pallet<Test>;
+	type MaxScheduledBlocks = ConstU32<100>;
+	type MaxStaleTaskAge = ConstU64<10>;
 }
 #[derive_impl(pallet_balances::config_preludes::TestDefaultConfig)]
 impl pallet_balances::Config for Test {
diff --git a/substrate/frame/scheduler/src/benchmarking.rs b/substrate/frame/scheduler/src/benchmarking.rs
index ff40e8ef8abf..7f482f194145 100644
--- a/substrate/frame/scheduler/src/benchmarking.rs
+++ b/substrate/frame/scheduler/src/benchmarking.rs
@@ -21,6 +21,7 @@ use alloc::vec;
 use frame_benchmarking::v2::*;
 use frame_support::{
 	ensure,
+	storage::bounded_btree_set::BoundedBTreeSet,
 	traits::{schedule::Priority, BoundedInline},
 	weights::WeightMeter,
 };
@@ -32,7 +33,10 @@ type SystemCall<T> = frame_system::Call<T>;
 type SystemOrigin<T> = <T as frame_system::Config>::RuntimeOrigin;
 
 const SEED: u32 = 0;
-const BLOCK_NUMBER: u32 = 2;
+
+fn block_number<T: Config>() -> u32 {
+	T::MaxScheduledBlocks::get()
+}
 
 fn assert_last_event<T: Config>(generic_event: <T as Config>::RuntimeEvent) {
 	let events = frame_system::Pallet::<T>::events();
@@ -42,6 +46,14 @@ fn assert_last_event<T: Config>(generic_event: <T as Config>::RuntimeEvent) {
 	assert_eq!(event, &system_event);
 }
 
+fn fill_queue<T: Config>(n: u32) {
+	let mut btree_set = BoundedBTreeSet::<BlockNumberFor<T>, T::MaxScheduledBlocks>::default();
+	for i in 0..n {
+		btree_set.try_insert(i.into()).unwrap();
+	}
+	Queue::<T>::put(btree_set);
+}
+
 /// Add `n` items to the schedule.
 ///
 /// For `resolved`:
@@ -49,10 +61,7 @@ fn assert_last_event<T: Config>(generic_event: <T as Config>::RuntimeEvent) {
 /// - `None`: aborted (hash without preimage)
 /// - `Some(true)`: hash resolves into call if possible, plain call otherwise
 /// - `Some(false)`: plain call
-fn fill_schedule<T: Config>(
-	when: frame_system::pallet_prelude::BlockNumberFor<T>,
-	n: u32,
-) -> Result<(), &'static str> {
+fn fill_schedule<T: Config>(when: BlockNumberFor<T>, n: u32) -> Result<(), &'static str> {
 	let t = DispatchTime::At(when);
 	let origin: <T as Config>::PalletsOrigin = frame_system::RawOrigin::Root.into();
 	for i in 0..n {
@@ -134,20 +143,25 @@ fn make_origin<T: Config>(signed: bool) -> <T as Config>::PalletsOrigin {
 
 #[benchmarks]
 mod benchmarks {
+	use frame_benchmarking::BenchmarkParameter::s;
 	use super::*;
 
-	// `service_agendas` when no work is done.
+	// `service_agenda` when no work is done.
 	#[benchmark]
 	fn service_agendas_base() {
-		let now = BLOCK_NUMBER.into();
-		IncompleteSince::<T>::put(now - One::one());
+		let now = BlockNumberFor::<T>::from(block_number::<T>());
+		let mut set = BoundedBTreeSet::<_, _>::default();
+		set.try_insert(now + One::one()).unwrap();  // Insert the element
+		Queue::<T>::put(set);
 
 		#[block]
 		{
 			Pallet::<T>::service_agendas(&mut WeightMeter::new(), now, 0);
 		}
 
-		assert_eq!(IncompleteSince::<T>::get(), Some(now - One::one()));
+		let mut expected_set = BoundedBTreeSet::<BlockNumberFor<T>, T::MaxScheduledBlocks>::default();
+		expected_set.try_insert(now + One::one()).unwrap();
+		assert_eq!(Queue::<T>::get(), expected_set);
 	}
 
 	// `service_agenda` when no work is done.
@@ -155,7 +169,7 @@ mod benchmarks {
 	fn service_agenda_base(
 		s: Linear<0, { T::MaxScheduledPerBlock::get() }>,
 	) -> Result<(), BenchmarkError> {
-		let now = BLOCK_NUMBER.into();
+		let now = block_number::<T>().into();
 		fill_schedule::<T>(now, s)?;
 		let mut executed = 0;
 
@@ -173,7 +187,7 @@ mod benchmarks {
 	// dispatched (e.g. due to being overweight).
 	#[benchmark]
 	fn service_task_base() {
-		let now = BLOCK_NUMBER.into();
+		let now = block_number::<T>().into();
 		let task = make_task::<T>(false, false, false, None, 0);
 		// prevent any tasks from actually being executed as we only want the surrounding weight.
 		let mut counter = WeightMeter::with_limit(Weight::zero());
@@ -196,7 +210,7 @@ mod benchmarks {
 	fn service_task_fetched(
 		s: Linear<{ BoundedInline::bound() as u32 }, { T::Preimages::MAX_LENGTH as u32 }>,
 	) {
-		let now = BLOCK_NUMBER.into();
+		let now = block_number::<T>().into();
 		let task = make_task::<T>(false, false, false, Some(s), 0);
 		// prevent any tasks from actually being executed as we only want the surrounding weight.
 		let mut counter = WeightMeter::with_limit(Weight::zero());
@@ -214,7 +228,7 @@ mod benchmarks {
 	// dispatched (e.g. due to being overweight).
 	#[benchmark]
 	fn service_task_named() {
-		let now = BLOCK_NUMBER.into();
+		let now = block_number::<T>().into();
 		let task = make_task::<T>(false, true, false, None, 0);
 		// prevent any tasks from actually being executed as we only want the surrounding weight.
 		let mut counter = WeightMeter::with_limit(Weight::zero());
@@ -232,7 +246,7 @@ mod benchmarks {
 	// dispatched (e.g. due to being overweight).
 	#[benchmark]
 	fn service_task_periodic() {
-		let now = BLOCK_NUMBER.into();
+		let now = block_number::<T>().into();
 		let task = make_task::<T>(true, false, false, None, 0);
 		// prevent any tasks from actually being executed as we only want the surrounding weight.
 		let mut counter = WeightMeter::with_limit(Weight::zero());
@@ -286,27 +300,33 @@ mod benchmarks {
 	fn schedule(
 		s: Linear<0, { T::MaxScheduledPerBlock::get() - 1 }>,
 	) -> Result<(), BenchmarkError> {
-		let when = BLOCK_NUMBER.into();
+		let when = block_number::<T>().into();
 		let periodic = Some((BlockNumberFor::<T>::one(), 100));
 		let priority = 0;
 		// Essentially a no-op call.
 		let call = Box::new(SystemCall::set_storage { items: vec![] }.into());
 
 		fill_schedule::<T>(when, s)?;
+		fill_queue::<T>(T::MaxScheduledBlocks::get() - 1);
 
 		#[extrinsic_call]
 		_(RawOrigin::Root, when, periodic, priority, call);
 
-		ensure!(Agenda::<T>::get(when).len() == s as usize + 1, "didn't add to schedule");
+		ensure!(Agenda::<T>::get(when).len() == (s + 1) as usize, "didn't add to schedule");
+		ensure!(
+			Queue::<T>::get().len() == T::MaxScheduledBlocks::get() as usize,
+			"didn't add to queue"
+		);
 
 		Ok(())
 	}
 
 	#[benchmark]
 	fn cancel(s: Linear<1, { T::MaxScheduledPerBlock::get() }>) -> Result<(), BenchmarkError> {
-		let when = BLOCK_NUMBER.into();
+		let when = (block_number::<T>() - 1).into();
 
 		fill_schedule::<T>(when, s)?;
+		fill_queue::<T>(T::MaxScheduledBlocks::get());
 		assert_eq!(Agenda::<T>::get(when).len(), s as usize);
 		let schedule_origin =
 			T::ScheduleOrigin::try_successful_origin().map_err(|_| BenchmarkError::Weightless)?;
@@ -327,6 +347,10 @@ mod benchmarks {
 			s > 1 || Agenda::<T>::get(when).len() == 0,
 			"remove from schedule if only 1 task scheduled for `when`"
 		);
+		ensure!(
+			s > 1 || Queue::<T>::get().len() == T::MaxScheduledBlocks::get() as usize - 1,
+			"didn't remove from queue"
+		);
 
 		Ok(())
 	}
@@ -336,18 +360,23 @@ mod benchmarks {
 		s: Linear<0, { T::MaxScheduledPerBlock::get() - 1 }>,
 	) -> Result<(), BenchmarkError> {
 		let id = u32_to_name(s);
-		let when = BLOCK_NUMBER.into();
+		let when = block_number::<T>().into();
 		let periodic = Some((BlockNumberFor::<T>::one(), 100));
 		let priority = 0;
 		// Essentially a no-op call.
 		let call = Box::new(SystemCall::set_storage { items: vec![] }.into());
 
 		fill_schedule::<T>(when, s)?;
+		fill_queue::<T>(T::MaxScheduledBlocks::get() - 1);
 
 		#[extrinsic_call]
 		_(RawOrigin::Root, id, when, periodic, priority, call);
 
-		ensure!(Agenda::<T>::get(when).len() == s as usize + 1, "didn't add to schedule");
+		ensure!(Agenda::<T>::get(when).len() == (s + 1) as usize, "didn't add to schedule");
+		ensure!(
+			Queue::<T>::get().len() == T::MaxScheduledBlocks::get() as usize,
+			"didn't add to queue"
+		);
 
 		Ok(())
 	}
@@ -356,9 +385,10 @@ mod benchmarks {
 	fn cancel_named(
 		s: Linear<1, { T::MaxScheduledPerBlock::get() }>,
 	) -> Result<(), BenchmarkError> {
-		let when = BLOCK_NUMBER.into();
+		let when = (block_number::<T>() - 1).into();
 
 		fill_schedule::<T>(when, s)?;
+		fill_queue::<T>(T::MaxScheduledBlocks::get());
 
 		#[extrinsic_call]
 		_(RawOrigin::Root, u32_to_name(0));
@@ -376,6 +406,10 @@ mod benchmarks {
 			s > 1 || Agenda::<T>::get(when).len() == 0,
 			"remove from schedule if only 1 task scheduled for `when`"
 		);
+		ensure!(
+			s > 1 || Queue::<T>::get().len() == T::MaxScheduledBlocks::get() as usize - 1,
+			"didn't remove from queue"
+		);
 
 		Ok(())
 	}
@@ -384,7 +418,7 @@ mod benchmarks {
 	fn schedule_retry(
 		s: Linear<1, { T::MaxScheduledPerBlock::get() }>,
 	) -> Result<(), BenchmarkError> {
-		let when = BLOCK_NUMBER.into();
+		let when = block_number::<T>().into();
 
 		fill_schedule::<T>(when, s)?;
 		let name = u32_to_name(s - 1);
@@ -420,7 +454,7 @@ mod benchmarks {
 	#[benchmark]
 	fn set_retry() -> Result<(), BenchmarkError> {
 		let s = T::MaxScheduledPerBlock::get();
-		let when = BLOCK_NUMBER.into();
+		let when = block_number::<T>().into();
 
 		fill_schedule::<T>(when, s)?;
 		let name = u32_to_name(s - 1);
@@ -445,7 +479,7 @@ mod benchmarks {
 	#[benchmark]
 	fn set_retry_named() -> Result<(), BenchmarkError> {
 		let s = T::MaxScheduledPerBlock::get();
-		let when = BLOCK_NUMBER.into();
+		let when = block_number::<T>().into();
 
 		fill_schedule::<T>(when, s)?;
 		let name = u32_to_name(s - 1);
@@ -470,7 +504,7 @@ mod benchmarks {
 	#[benchmark]
 	fn cancel_retry() -> Result<(), BenchmarkError> {
 		let s = T::MaxScheduledPerBlock::get();
-		let when = BLOCK_NUMBER.into();
+		let when = block_number::<T>().into();
 
 		fill_schedule::<T>(when, s)?;
 		let name = u32_to_name(s - 1);
@@ -491,7 +525,7 @@ mod benchmarks {
 	#[benchmark]
 	fn cancel_retry_named() -> Result<(), BenchmarkError> {
 		let s = T::MaxScheduledPerBlock::get();
-		let when = BLOCK_NUMBER.into();
+		let when = block_number::<T>().into();
 
 		fill_schedule::<T>(when, s)?;
 		let name = u32_to_name(s - 1);
diff --git a/substrate/frame/scheduler/src/lib.rs b/substrate/frame/scheduler/src/lib.rs
index 468099010bf9..6fed247eaa4a 100644
--- a/substrate/frame/scheduler/src/lib.rs
+++ b/substrate/frame/scheduler/src/lib.rs
@@ -100,14 +100,11 @@ use frame_support::{
 	},
 	weights::{Weight, WeightMeter},
 };
-use frame_system::{
-	pallet_prelude::BlockNumberFor,
-	{self as system},
-};
+use frame_system::{self as system};
 use scale_info::TypeInfo;
 use sp_io::hashing::blake2_256;
 use sp_runtime::{
-	traits::{BadOrigin, Dispatchable, One, Saturating, Zero},
+	traits::{BadOrigin, BlockNumberProvider, Dispatchable, One, Saturating, Zero},
 	BoundedVec, DispatchError, RuntimeDebug,
 };
 
@@ -125,6 +122,9 @@ pub type CallOrHashOf<T> =
 pub type BoundedCallOf<T> =
 	Bounded<<T as Config>::RuntimeCall, <T as frame_system::Config>::Hashing>;
 
+pub type BlockNumberFor<T> =
+	<<T as Config>::BlockNumberProvider as BlockNumberProvider>::BlockNumber;
+
 /// The configuration of the retry mechanism for a given task along with its current state.
 #[derive(Clone, Copy, RuntimeDebug, PartialEq, Eq, Encode, Decode, MaxEncodedLen, TypeInfo)]
 pub struct RetryConfig<Period> {
@@ -230,7 +230,7 @@ impl<T: WeightInfo> MarginalWeightInfo for T {}
 pub mod pallet {
 	use super::*;
 	use frame_support::{dispatch::PostDispatchInfo, pallet_prelude::*};
-	use frame_system::pallet_prelude::*;
+	use frame_system::pallet_prelude::{BlockNumberFor as SystemBlockNumberFor, OriginFor};
 
 	/// The in-code storage version.
 	const STORAGE_VERSION: StorageVersion = StorageVersion::new(4);
@@ -238,7 +238,6 @@ pub mod pallet {
 	#[pallet::pallet]
 	#[pallet::storage_version(STORAGE_VERSION)]
 	pub struct Pallet<T>(_);
-
 	/// `system::Config` should always be included in our implied traits.
 	#[pallet::config]
 	pub trait Config: frame_system::Config {
@@ -292,10 +291,23 @@ pub mod pallet {
 
 		/// The preimage provider with which we look up call hashes to get the call.
 		type Preimages: QueryPreimage<H = Self::Hashing> + StorePreimage;
-	}
 
-	#[pallet::storage]
-	pub type IncompleteSince<T: Config> = StorageValue<_, BlockNumberFor<T>>;
+		/// Provider for the block number. Normally, for a parachain this is the `frame_system`
+		/// pallet.
+		type BlockNumberProvider: BlockNumberProvider;
+
+		/// The maximum number of blocks that can be scheduled. Should be equal to 50 if
+		/// runtime-benchmarks feature is not enabled while, it should be 200 is runtime-benchmarks
+		/// feature is enabled
+		#[pallet::constant]
+		type MaxScheduledBlocks: Get<u32>;
+
+		/// The maximum number of blocks that a task can be stale for. Should be equal to 10 if
+		/// runtime-benchmarks feature is not enabled while, it should be 40 is runtime-benchmarks
+		/// feature is enabled
+		#[pallet::constant]
+		type MaxStaleTaskAge: Get<BlockNumberFor<Self>>;
+	}
 
 	/// Items to be executed, indexed by the block number that they should be executed on.
 	#[pallet::storage]
@@ -325,6 +337,11 @@ pub mod pallet {
 	pub(crate) type Lookup<T: Config> =
 		StorageMap<_, Twox64Concat, TaskName, TaskAddress<BlockNumberFor<T>>>;
 
+	/// The queue of block numbers that have scheduled agendas.
+	#[pallet::storage]
+	pub(crate) type Queue<T: Config> =
+		StorageValue<_, BoundedBTreeSet<BlockNumberFor<T>, T::MaxScheduledBlocks>, ValueQuery>;
+
 	/// Events type.
 	#[pallet::event]
 	#[pallet::generate_deposit(pub(super) fn deposit_event)]
@@ -374,9 +391,10 @@ pub mod pallet {
 	}
 
 	#[pallet::hooks]
-	impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
+	impl<T: Config> Hooks<SystemBlockNumberFor<T>> for Pallet<T> {
 		/// Execute the scheduled calls
-		fn on_initialize(now: BlockNumberFor<T>) -> Weight {
+		fn on_initialize(_do_not_use_local_block_number: SystemBlockNumberFor<T>) -> Weight {
+			let now = T::BlockNumberProvider::current_block_number();
 			let mut weight_counter = WeightMeter::with_limit(T::MaximumWeight::get());
 			Self::service_agendas(&mut weight_counter, now, u32::max_value());
 			weight_counter.consumed()
@@ -889,7 +907,7 @@ impl<T: Config> Pallet<T> {
 	fn resolve_time(
 		when: DispatchTime<BlockNumberFor<T>>,
 	) -> Result<BlockNumberFor<T>, DispatchError> {
-		let now = frame_system::Pallet::<T>::block_number();
+		let now = T::BlockNumberProvider::current_block_number();
 
 		let when = match when {
 			DispatchTime::At(x) => x,
@@ -926,17 +944,29 @@ impl<T: Config> Pallet<T> {
 		let mut agenda = Agenda::<T>::get(when);
 		let index = if (agenda.len() as u32) < T::MaxScheduledPerBlock::get() {
 			// will always succeed due to the above check.
-			let _ = agenda.try_push(Some(what));
+			let _ = agenda.try_push(Some(what.clone()));
 			agenda.len() as u32 - 1
 		} else {
 			if let Some(hole_index) = agenda.iter().position(|i| i.is_none()) {
-				agenda[hole_index] = Some(what);
+				agenda[hole_index] = Some(what.clone());
 				hole_index as u32
 			} else {
 				return Err((DispatchError::Exhausted, what))
 			}
 		};
 		Agenda::<T>::insert(when, agenda);
+		Queue::<T>::mutate(|q| {
+			// Ensure q is mutable
+			if !q.contains(&when) {
+				// Check if the set is full before inserting
+				if q.len() >= T::MaxScheduledBlocks::get() as usize {
+					return Err((DispatchError::Exhausted, what)); // Return an error if full
+				}
+				// Insert the block number into the set
+				q.try_insert(when).map_err(|_| (DispatchError::Exhausted, what))?;
+			}
+			Ok(())
+		})?;
 		Ok(index)
 	}
 
@@ -952,6 +982,9 @@ impl<T: Config> Pallet<T> {
 			Some(_) => {},
 			None => {
 				Agenda::<T>::remove(when);
+				Queue::<T>::mutate(|q| {
+					q.remove(&when);
+				});
 			},
 		}
 	}
@@ -1157,24 +1190,51 @@ impl<T: Config> Pallet<T> {
 			return
 		}
 
-		let mut incomplete_since = now + One::one();
-		let mut when = IncompleteSince::<T>::take().unwrap_or(now);
-		let mut executed = 0;
+		let queue = Queue::<T>::get();
+		let end_index = match queue.iter().position(|&x| x >= now) {
+			Some(end_index) => end_index.saturating_add(1),
+			None => {
+				return;
+			},
+		};
+		if end_index == 0 {
+			return;
+		}
 
-		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);
+		let mut index = 0;
+		let queue_len = queue.len();
+		for when in queue.iter().skip(index) {
+			if *when < now.saturating_sub(T::MaxStaleTaskAge::get()) {
+				Agenda::<T>::remove(*when);
+			} else {
+				let mut executed = 0;
+				if !Self::service_agenda(weight, &mut executed, now, *when, max) {
+					break;
+				}
+			}
+			index = index.saturating_add(1);
+			if index >= queue_len {
+				break;
 			}
-			when.saturating_inc();
-			count_down.saturating_dec();
-		}
-		incomplete_since = incomplete_since.min(when);
-		if incomplete_since <= now {
-			IncompleteSince::<T>::put(incomplete_since);
 		}
+
+		Queue::<T>::mutate(|queue| {
+			let mut iter = queue.iter();
+			let mut to_remove = Vec::new(); // Collect items to remove
+
+			// Iterate and collect items to remove
+			for _ in 0..index {
+				iter.next();
+			}
+			for item in iter {
+				to_remove.push(*item);
+			}
+
+			// Now remove the collected items
+			for item in to_remove {
+				queue.remove(&item);
+			}
+		});
 	}
 
 	/// Returns `true` if the agenda was fully completed, `false` if it should be revisited at a
diff --git a/substrate/frame/scheduler/src/migration.rs b/substrate/frame/scheduler/src/migration.rs
index a304689a120c..311af2edee80 100644
--- a/substrate/frame/scheduler/src/migration.rs
+++ b/substrate/frame/scheduler/src/migration.rs
@@ -19,7 +19,6 @@
 
 use super::*;
 use frame_support::traits::OnRuntimeUpgrade;
-use frame_system::pallet_prelude::BlockNumberFor;
 
 #[cfg(feature = "try-runtime")]
 use sp_runtime::TryRuntimeError;
@@ -561,3 +560,86 @@ mod test {
 		system::RawOrigin::Signed(i).into()
 	}
 }
+
+pub mod v5 {
+	use super::*;
+	use frame_support::{
+		migrations::VersionedMigration, pallet_prelude::*, traits::UncheckedOnRuntimeUpgrade,
+	};
+
+	#[frame_support::storage_alias]
+	pub(crate) type IncompleteSince<T: Config> = StorageValue<Pallet<T>, BlockNumberFor<T>>;
+
+	pub struct VersionUncheckedMigrateV4ToV5<T>(core::marker::PhantomData<T>);
+	impl<T: Config> UncheckedOnRuntimeUpgrade for VersionUncheckedMigrateV4ToV5<T> {
+		fn on_runtime_upgrade() -> Weight {
+			IncompleteSince::<T>::kill();
+			let mut weight = T::DbWeight::get().writes(1);
+
+			let mut queue = BoundedBTreeSet::<BlockNumberFor<T>, T::MaxScheduledBlocks>::default();
+			Agenda::<T>::iter().for_each(|(k, _)| {
+				if queue.len() < T::MaxScheduledBlocks::get() as usize {
+					// Attempt to insert only if there's capacity left
+					if queue.try_insert(k).is_ok() {
+						weight.saturating_accrue(T::DbWeight::get().reads(1));
+					}
+				}
+			});
+			Queue::<T>::put(queue);
+			weight.saturating_accrue(T::DbWeight::get().writes(1));
+
+			weight
+		}
+
+		#[cfg(feature = "try-runtime")]
+		fn post_upgrade(_state: Vec<u8>) -> Result<(), TryRuntimeError> {
+			frame_support::ensure!(
+				IncompleteSince::<T>::get().is_none(),
+				"IncompleteSince is not empty after the migration"
+			);
+			Ok(())
+		}
+	}
+
+	pub type MigrateV4ToV5<T> = VersionedMigration<
+		4,
+		5,
+		VersionUncheckedMigrateV4ToV5<T>,
+		Pallet<T>,
+		<T as frame_system::Config>::DbWeight,
+	>;
+}
+
+#[cfg(test)]
+pub mod test_v5 {
+	use frame_support::BoundedBTreeSet;
+	use super::*;
+	use crate::{migration::v5::IncompleteSince, mock::*};
+	use frame_support::testing_prelude::*;
+
+	#[test]
+	fn migration_v4_to_v5_works() {
+		new_test_ext().execute_with(|| {
+			StorageVersion::new(4).put::<Scheduler>();
+			IncompleteSince::<Test>::put(1);
+			Agenda::<Test>::insert::<
+				u64,
+				BoundedVec<Option<ScheduledOf<Test>>, <Test as Config>::MaxScheduledPerBlock>,
+			>(2, bounded_vec![None]);
+			Agenda::<Test>::insert::<
+				u64,
+				BoundedVec<Option<ScheduledOf<Test>>, <Test as Config>::MaxScheduledPerBlock>,
+			>(3, bounded_vec![None]);
+
+			v5::MigrateV4ToV5::<Test>::on_runtime_upgrade();
+
+			assert_eq!(StorageVersion::get::<Scheduler>(), 5);
+			assert_eq!(IncompleteSince::<Test>::get(), None);
+			let mut queue = BoundedBTreeSet::<BlockNumberFor<Test>, <Test as Config>::MaxScheduledBlocks>::default();
+			for block in vec![2, 3] {
+				queue.try_insert(block).expect("BoundedBTreeSet insertion failed");
+			}
+			assert_eq!(Queue::<Test>::get(), queue);
+		});
+	}
+}
diff --git a/substrate/frame/scheduler/src/mock.rs b/substrate/frame/scheduler/src/mock.rs
index 43a964bcf149..07633e5dc88e 100644
--- a/substrate/frame/scheduler/src/mock.rs
+++ b/substrate/frame/scheduler/src/mock.rs
@@ -21,8 +21,10 @@ use super::*;
 
 use crate as scheduler;
 use frame_support::{
-	derive_impl, ord_parameter_types, parameter_types,
-	traits::{ConstU32, Contains, EitherOfDiverse, EqualPrivilegeOnly},
+	derive_impl, ord_parameter_types,
+	pallet_prelude::Hooks,
+	parameter_types,
+	traits::{ConstU32, ConstU64, Contains, EitherOfDiverse, EqualPrivilegeOnly},
 };
 use frame_system::{EnsureRoot, EnsureSignedBy};
 use sp_runtime::{BuildStorage, Perbill};
@@ -227,6 +229,9 @@ impl Config for Test {
 	type WeightInfo = TestWeightInfo;
 	type OriginPrivilegeCmp = EqualPrivilegeOnly;
 	type Preimages = Preimage;
+	type BlockNumberProvider = frame_system::Pallet<Self>;
+	type MaxScheduledBlocks = ConstU32<20>;
+	type MaxStaleTaskAge = ConstU64<10>;
 }
 
 pub type LoggerCall = logger::Call<Test>;
@@ -236,6 +241,12 @@ pub fn new_test_ext() -> sp_io::TestExternalities {
 	t.into()
 }
 
+pub fn go_to_block(n: u64) {
+	System::set_block_number(n);
+	Scheduler::on_initialize(n);
+	Scheduler::on_finalize(n);
+}
+
 pub fn root() -> OriginCaller {
 	system::RawOrigin::Root.into()
 }
diff --git a/substrate/frame/scheduler/src/tests.rs b/substrate/frame/scheduler/src/tests.rs
index 755223934108..27f51f10f92f 100644
--- a/substrate/frame/scheduler/src/tests.rs
+++ b/substrate/frame/scheduler/src/tests.rs
@@ -1636,6 +1636,7 @@ fn on_initialize_weight_is_correct() {
 		));
 
 		// Will include the named periodic only
+		System::set_block_number(1);
 		assert_eq!(
 			Scheduler::on_initialize(1),
 			TestWeightInfo::service_agendas_base() +
@@ -1644,10 +1645,10 @@ fn on_initialize_weight_is_correct() {
 				TestWeightInfo::execute_dispatch_unsigned() +
 				call_weight + Weight::from_parts(4, 0)
 		);
-		assert_eq!(IncompleteSince::<Test>::get(), None);
 		assert_eq!(logger::log(), vec![(root(), 2600u32)]);
 
 		// Will include anon and anon periodic
+		System::set_block_number(2);
 		assert_eq!(
 			Scheduler::on_initialize(2),
 			TestWeightInfo::service_agendas_base() +
@@ -1659,10 +1660,10 @@ fn on_initialize_weight_is_correct() {
 				TestWeightInfo::execute_dispatch_unsigned() +
 				call_weight + Weight::from_parts(2, 0)
 		);
-		assert_eq!(IncompleteSince::<Test>::get(), None);
 		assert_eq!(logger::log(), vec![(root(), 2600u32), (root(), 69u32), (root(), 42u32)]);
 
 		// Will include named only
+		System::set_block_number(3);
 		assert_eq!(
 			Scheduler::on_initialize(3),
 			TestWeightInfo::service_agendas_base() +
@@ -1671,18 +1672,15 @@ fn on_initialize_weight_is_correct() {
 				TestWeightInfo::execute_dispatch_unsigned() +
 				call_weight + Weight::from_parts(1, 0)
 		);
-		assert_eq!(IncompleteSince::<Test>::get(), None);
 		assert_eq!(
 			logger::log(),
 			vec![(root(), 2600u32), (root(), 69u32), (root(), 42u32), (root(), 3u32)]
 		);
 
 		// Will contain none
+		System::set_block_number(4);
 		let actual_weight = Scheduler::on_initialize(4);
-		assert_eq!(
-			actual_weight,
-			TestWeightInfo::service_agendas_base() + TestWeightInfo::service_agenda_base(0)
-		);
+		assert_eq!(actual_weight, TestWeightInfo::service_agendas_base());
 	});
 }
 
@@ -3035,3 +3033,59 @@ fn unavailable_call_is_detected() {
 		assert!(!Preimage::is_requested(&hash));
 	});
 }
+
+#[test]
+fn basic_scheduling_with_skipped_blocks_works() {
+	new_test_ext().execute_with(|| {
+		// Call to schedule
+		let call =
+			RuntimeCall::Logger(LoggerCall::log { i: 42, weight: Weight::from_parts(10, 0) });
+
+		// BaseCallFilter should be implemented to accept `Logger::log` runtime call which is
+		// implemented for `BaseFilter` in the mock runtime
+		assert!(!<Test as frame_system::Config>::BaseCallFilter::contains(&call));
+
+		// Schedule call to be executed at the 4th block
+		assert_ok!(Scheduler::do_schedule(
+			DispatchTime::At(4),
+			None,
+			127,
+			root(),
+			Preimage::bound(call).unwrap()
+		));
+
+		// `log` runtime call should not have executed yet
+		go_to_block(3);
+		assert!(logger::log().is_empty());
+
+		go_to_block(6);
+		// `log` runtime call should have executed at block 4
+		assert_eq!(logger::log(), vec![(root(), 42u32)]);
+
+		go_to_block(100);
+		assert_eq!(logger::log(), vec![(root(), 42u32)]);
+	});
+}
+
+#[test]
+fn stale_task_is_removed() {
+	new_test_ext().execute_with(|| {
+		let call =
+			RuntimeCall::Logger(LoggerCall::log { i: 42, weight: Weight::from_parts(10, 0) });
+
+		// Schedule call to be executed at the 4th block
+		assert_ok!(Scheduler::do_schedule(
+			DispatchTime::At(4),
+			None,
+			127,
+			root(),
+			Preimage::bound(call).unwrap()
+		));
+		assert!(Agenda::<Test>::get(4).len() == 1);
+		assert!(Queue::<Test>::get().contains(&4));
+
+		go_to_block(20);
+		assert!(Agenda::<Test>::get(4).is_empty());
+		assert!(!Queue::<Test>::get().contains(&4));
+	})
+}